"""Step definitions for A2A facade and stubs Behave scenarios.""" from __future__ import annotations import json from typing import Any from behave import given, then, use_step_matcher, when from behave.runner import Context from pydantic import ValidationError try: from cleveragents.a2a.errors import ( A2aError, A2aNotAvailableError, A2aOperationNotFoundError, A2aVersionMismatchError, ) from cleveragents.a2a.events import A2aEventQueue from cleveragents.a2a.facade import A2aLocalFacade from cleveragents.a2a.models import ( A2aErrorDetail, A2aEvent, A2aRequest, A2aResponse, ) from cleveragents.a2a.transport import A2aHttpTransport from cleveragents.a2a.versioning import A2aVersionNegotiator except ImportError: pass # a2a module not available from cleveragents.core.exceptions import CleverAgentsError use_step_matcher("re") # A2aLocalFacade — creation and service registration @given(r"a new A2aLocalFacade with no services") def step_facade_no_services(context: Context) -> None: context.facade = A2aLocalFacade() @given(r"a new A2aLocalFacade with services (?P.+)") def step_facade_with_services(context: Context, services_json: str) -> None: services = json.loads(services_json) context.facade = A2aLocalFacade(services=services) @then(r"the facade should be created successfully") def step_facade_created(context: Context) -> None: assert context.facade is not None @when(r'I register a service named "(?P[^"]+)" on the facade') def step_register_service(context: Context, name: str) -> None: context.facade.register_service(name, object()) @then(r"the service should be registered successfully") def step_service_registered(context: Context) -> None: pass @when(r"I try to register a service with an empty name") def step_register_empty_name(context: Context) -> None: context.caught_error = None context.error = None try: context.facade.register_service("", object()) except ValueError as exc: context.caught_error = exc context.error = exc @then(r"an A2A ValueError should be raised") def step_value_error_raised(context: Context) -> None: assert isinstance(context.caught_error, ValueError), ( f"Expected ValueError, got {type(context.caught_error)}" ) # A2aLocalFacade — dispatch operations @when(r'I dispatch operation "(?P[^"]+)" with params (?P.+)') def step_dispatch_operation(context: Context, operation: str, params_json: str) -> None: params: dict[str, Any] = json.loads(params_json) request = A2aRequest(method=operation, params=params) context.response = context.facade.dispatch(request) @then(r'the response status should be "(?P[^"]+)"') def step_response_status(context: Context, status: str) -> None: is_ok = context.response.error is None expected_ok = status == "ok" assert is_ok == expected_ok, ( f"Expected status '{status}', got error={context.response.error}" ) @then(r'response data includes key "(?P[^"]+)"') def step_response_data_has_key(context: Context, key: str) -> None: assert key in context.response.result, ( f"Key '{key}' not found in response data: {context.response.result}" ) @then(r'response data key "(?P[^"]+)" equals "(?P[^"]+)"') def step_response_data_key_value(context: Context, key: str, value: str) -> None: assert key in context.response.result, f"Key '{key}' not found in response data" actual = context.response.result[key] assert str(actual) == value, f"Expected '{value}', got '{actual}'" # A2aLocalFacade — unknown operation @when(r'I dispatch an unknown operation "(?P[^"]+)"') def step_dispatch_unknown(context: Context, operation: str) -> None: context.caught_error = None context.error = None try: request = A2aRequest(method=operation) context.facade.dispatch(request) except A2aOperationNotFoundError as exc: context.caught_error = exc context.error = exc @then(r"an A2aOperationNotFoundError should be raised") def step_op_not_found_raised(context: Context) -> None: assert isinstance(context.caught_error, A2aOperationNotFoundError) @then(r'the error operation attribute should be "(?P[^"]+)"') def step_error_operation_attr(context: Context, operation: str) -> None: assert context.caught_error.operation == operation # A2aLocalFacade — list_operations @when(r"I call list_operations on the facade") def step_list_operations(context: Context) -> None: context.operations = context.facade.list_operations() @then(r'the operations list should contain "(?P[^"]+)"') def step_operations_contains(context: Context, operation: str) -> None: assert operation in context.operations, f"'{operation}' not in {context.operations}" @then(r"the operations list should have (?P\d+) items") def step_operations_count(context: Context, count: str) -> None: expected = int(count) assert len(context.operations) == expected, ( f"Expected {expected}, got {len(context.operations)}" ) # A2aHttpTransport @given(r"a new A2aHttpTransport") def step_transport(context: Context) -> None: context.transport = A2aHttpTransport() @when(r"I try to send a request via the transport") def step_transport_send(context: Context) -> None: context.caught_error = None context.error = None try: request = A2aRequest(method="test.op") context.transport.send(request) except A2aNotAvailableError as exc: context.caught_error = exc context.error = exc @when(r'I try to connect via the transport to "(?P[^"]+)"') def step_transport_connect(context: Context, url: str) -> None: context.caught_error = None context.error = None try: context.transport.connect(url) except A2aNotAvailableError as exc: context.caught_error = exc context.error = exc @when(r"I try to disconnect the transport") def step_transport_disconnect(context: Context) -> None: context.caught_error = None context.error = None try: context.transport.disconnect() except A2aNotAvailableError as exc: context.caught_error = exc context.error = exc @then(r"an A2aNotAvailableError should be raised") def step_not_available_raised(context: Context) -> None: assert isinstance(context.caught_error, A2aNotAvailableError), ( f"Expected A2aNotAvailableError, got {type(context.caught_error)}" ) @then(r"the transport should not be connected") def step_transport_not_connected(context: Context) -> None: assert context.transport.is_connected() is False # A2aEventQueue — local mode @given(r"a new A2aEventQueue") def step_event_queue(context: Context) -> None: context.queue = A2aEventQueue() context.callback_events: list[A2aEvent] = [] @given(r"I subscribe locally with a callback") def step_subscribe_local(context: Context) -> None: def _cb(event: A2aEvent) -> None: context.callback_events.append(event) context.subscription_id = context.queue.subscribe_local(_cb) @when(r'I publish an event with type "(?P[^"]+)"') def step_publish_event(context: Context, event_type: str) -> None: event = A2aEvent(event_type=event_type) context.queue.publish(event) @when(r"I publish (?P\d+) events") def step_publish_n_events(context: Context, count: str) -> None: for i in range(int(count)): event = A2aEvent(event_type=f"test.event.{i}") context.queue.publish(event) @when(r"I get events with limit (?P\d+)") def step_get_events(context: Context, limit: str) -> None: context.fetched_events = context.queue.get_events(limit=int(limit)) @then(r"the event queue should have (?P\d+) event") def step_queue_count(context: Context, count: str) -> None: expected = int(count) events = context.queue.get_events() assert len(events) == expected, f"Expected {expected}, got {len(events)}" @then(r'the callback should have been called with event type "(?P[^"]+)"') def step_callback_called(context: Context, event_type: str) -> None: assert len(context.callback_events) > 0, "Callback was not called" assert context.callback_events[-1].event_type == event_type @when(r"I unsubscribe using the subscription id") def step_unsubscribe(context: Context) -> None: context.unsubscribe_result = context.queue.unsubscribe(context.subscription_id) @when(r"I unsubscribe using a non-existent subscription id") def step_unsubscribe_nonexistent(context: Context) -> None: context.unsubscribe_result = context.queue.unsubscribe("NONEXISTENT_ID") @then(r"the unsubscribe should return True") def step_unsubscribe_true(context: Context) -> None: assert context.unsubscribe_result is True @then(r"the unsubscribe should return False") def step_unsubscribe_false(context: Context) -> None: assert context.unsubscribe_result is False @then(r"I should receive (?P\d+) events") def step_received_count(context: Context, count: str) -> None: expected = int(count) assert len(context.fetched_events) == expected, ( f"Expected {expected}, got {len(context.fetched_events)}" ) # A2aEventQueue — remote stub @when(r'I try to subscribe remotely to "(?P[^"]+)"') def step_remote_subscribe(context: Context, endpoint: str) -> None: context.caught_error = None context.error = None try: context.queue.subscribe_remote(endpoint) except A2aNotAvailableError as exc: context.caught_error = exc context.error = exc # A2aVersionNegotiator @given(r"a new A2aVersionNegotiator") def step_version_negotiator(context: Context) -> None: context.negotiator = A2aVersionNegotiator() @when(r'I negotiate version "(?P[^"]+)"') def step_negotiate(context: Context, version: str) -> None: context.negotiated_version = context.negotiator.negotiate(version) @then(r'the negotiated version should be "(?P[^"]+)"') def step_negotiated(context: Context, version: str) -> None: assert context.negotiated_version == version @when(r'I try to negotiate version "(?P[^"]+)"') def step_negotiate_fail(context: Context, version: str) -> None: context.caught_error = None context.error = None try: context.negotiator.negotiate(version) except A2aVersionMismatchError as exc: context.caught_error = exc context.error = exc @then(r"an A2aVersionMismatchError should be raised") def step_version_mismatch_raised(context: Context) -> None: assert isinstance(context.caught_error, A2aVersionMismatchError) @then(r'the error requested_version should be "(?P[^"]+)"') def step_error_requested_version(context: Context, version: str) -> None: assert context.caught_error.requested_version == version @then(r'version "(?P[^"]+)" should be supported') def step_version_supported(context: Context, version: str) -> None: assert context.negotiator.is_supported(version) is True @then(r'version "(?P[^"]+)" should not be supported') def step_version_not_supported(context: Context, version: str) -> None: assert context.negotiator.is_supported(version) is False @then(r'the current version should be "(?P[^"]+)"') def step_current_version(context: Context, version: str) -> None: assert context.negotiator.get_current() == version # A2aRequest model validation @when(r'I create an A2aRequest with operation "(?P[^"]+)"') def step_create_request(context: Context, operation: str) -> None: context.request = A2aRequest(method=operation) @then(r"the request should have a non-empty request_id") def step_request_has_id(context: Context) -> None: assert context.request.id, "id should not be empty" @then(r'the request a2a_version should be "(?P[^"]+)"') def step_request_version(context: Context, version: str) -> None: assert context.request.jsonrpc == "2.0" @when(r"I try to create an A2aRequest with empty operation") def step_create_request_empty(context: Context) -> None: context.caught_error = None context.error = None try: A2aRequest(method="") except ValidationError as exc: context.caught_error = exc context.error = exc # Note: "a validation error should be raised" is defined in domain_models_steps.py # Note: "the error message should contain" is defined in service_steps.py # We reuse those shared step definitions. # A2aResponse model validation @when( r'I create an A2aResponse with status "(?P[^"]+)" and request_id "(?P[^"]+)"' ) def step_create_response(context: Context, status: str, rid: str) -> None: if status == "ok": context.response = A2aResponse(id=rid, result={}) else: from cleveragents.a2a.models import A2aErrorDetail context.response = A2aResponse( id=rid, error=A2aErrorDetail(code=status, message=status) ) @then(r"the response should be valid") def step_response_valid(context: Context) -> None: assert context.response is not None @when(r'I try to create an A2aResponse with status "(?P[^"]+)"') def step_create_response_invalid(context: Context, status: str) -> None: context.caught_error = None context.error = None try: # A2aResponse requires either result or error (not both, not neither) # An invalid status means we try to create one with neither result nor error A2aResponse(id="REQ") except ValidationError as exc: context.caught_error = exc context.error = exc # A2aErrorDetail model validation @when( r'I create an A2aErrorDetail with code "(?P[^"]+)" and message "(?P[^"]+)"' ) def step_create_error_detail(context: Context, code: str, msg: str) -> None: context.error_detail = A2aErrorDetail(code=code, message=msg) @then(r"the error detail should be valid") def step_error_detail_valid(context: Context) -> None: assert context.error_detail is not None @when(r"I try to create an A2aErrorDetail with empty code") def step_create_error_detail_empty(context: Context) -> None: context.caught_error = None context.error = None try: A2aErrorDetail(code="", message="test") except ValidationError as exc: context.caught_error = exc context.error = exc # A2aEvent model construction @when(r'I create an A2aEvent with type "(?P[^"]+)"') def step_create_event(context: Context, event_type: str) -> None: context.event = A2aEvent(event_type=event_type) @then(r"the event should have a non-empty event_id") def step_event_has_id(context: Context) -> None: assert context.event.event_id, "event_id should not be empty" @then(r"the event should have a non-empty timestamp") def step_event_has_timestamp(context: Context) -> None: assert context.event.timestamp, "timestamp should not be empty" # Error hierarchy @then(r"A2aError should be a subclass of CleverAgentsError") def step_a2a_error_hierarchy(context: Context) -> None: assert issubclass(A2aError, CleverAgentsError) @then(r"A2aNotAvailableError should be a subclass of A2aError") def step_not_available_hierarchy(context: Context) -> None: assert issubclass(A2aNotAvailableError, A2aError) @then(r"A2aVersionMismatchError should be a subclass of A2aError") def step_version_mismatch_hierarchy(context: Context) -> None: assert issubclass(A2aVersionMismatchError, A2aError) @then(r"A2aOperationNotFoundError should be a subclass of A2aError") def step_op_not_found_hierarchy(context: Context) -> None: assert issubclass(A2aOperationNotFoundError, A2aError) @when(r"I create an A2aNotAvailableError with default message") def step_create_not_available_default(context: Context) -> None: context.caught_error = A2aNotAvailableError() context.error = context.caught_error # "the error message should contain" is provided by service_steps.py # Reset step matcher to parse (default) so subsequent step files are not affected use_step_matcher("parse")