"""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 from features.steps._a2a_code_map import A2A_CODE_MAP 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 @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)}" ) # 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 import errors as _a2a_errors from cleveragents.a2a.models import A2aErrorDetail int_code: int = A2A_CODE_MAP.get(status, _a2a_errors.INTERNAL_ERROR) context.response = A2aResponse( id=rid, error=A2aErrorDetail(code=int_code, 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: from cleveragents.a2a import errors as _a2a_errors int_code: int = A2A_CODE_MAP.get( code, int(code) if code.lstrip("-").isdigit() else _a2a_errors.INTERNAL_ERROR ) context.error_detail = A2aErrorDetail(code=int_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 # A2aHttpTransport — server-mode connect / disconnect @then(r"the transport should be connected when using url") def step_transport_is_connected(context: Context) -> None: assert context.transport.is_connected() is True @then(r"the transport should be in unconnected state") def step_transport_unconnected_state(context: Context) -> None: assert context.transport.is_connected() is False # ----- connect steps ----- @when(r'I try to connect via the transport to "(?P.*)"') def step_transport_try_connect(context: Context, url: str) -> None: context.caught_error = None context.error = None try: context.transport.connect(url) except ValueError as exc: context.caught_error = exc context.error = exc @when(r"I connect the transport to (?P.+)") def step_transport_connect(context: Context, url: str) -> None: context.transport.connect(url.strip("'\"")) @when(r"the transport is connected to (?P.+)") def step_transport_connected(context: Context, url: str) -> None: context.transport.connect(url.strip("'\"")) @when(r"I disconnect the transport") @then(r"I disconnect the transport") def step_transport_disconnect(context: Context) -> None: context.transport.disconnect() # ----- connect value error handling ----- @then(r"an A2aValueError should be raised with message containing (?P.+)") def step_connect_value_error(context: Context, reason: str) -> None: assert isinstance(context.caught_error, ValueError), ( f"Expected ValueError, got {type(context.caught_error)}" ) @then(r"the ValueError message should contain (?P.+)") def step_connect_value_error_message(context: Context, text: str) -> None: assert context.caught_error is not None assert isinstance(context.caught_error, ValueError) assert text in str(context.caught_error), ( f"Expected '{text}' in '{context.caught_error}'" ) # ----- disconnect / unconnected steps ----- @when(r"the transport\s+is in unconnected state") def step_force_unconnected(context: Context) -> None: context.transport._is_connected = False @then(r"the transport should not be connected") def step_transport_should_not_be_connected(context: Context) -> None: assert context.transport.is_connected() is False # ----- send non-connected operation (RuntimeError) ----- @when(r"I try to send a request via the unconnected transport") def step_send_unconnected(context: Context) -> None: context.caught_error = None context.error = None try: request = A2aRequest(method="test.op") context.transport.send(request) except RuntimeError as exc: context.caught_error = exc context.error = exc @when(r"I dispatch a non-connected operation via the transport") def step_nonconnected_op(context: Context) -> None: context.caught_error = None context.error = None try: request = A2aRequest(method="test.op") context.transport.send(request) except RuntimeError as exc: context.caught_error = exc context.error = exc @then(r"an A2aRuntimeError should be raised for the transport") def step_runtime_error(context: Context) -> None: assert isinstance(context.caught_error, RuntimeError), ( f"Expected RuntimeError, got {type(context.caught_error)}" ) # ----- send non-A2aRequest (TypeError) ----- @when(r"I try to send a non-A2aRequest via the transport") def step_send_non_a2arequest(context: Context) -> None: context.caught_error = None context.error = None try: context.transport.send("not an A2aRequest") except TypeError as exc: context.caught_error = exc context.error = exc @then(r"an A2aTypeError should be raised for the transport") def step_type_error(context: Context) -> None: assert isinstance(context.caught_error, TypeError), ( f"Expected TypeError, got {type(context.caught_error)}" ) # ----- HTTP response mocking ----- @when(r"HTTP responses are mocked with JSON-RPC success response") def step_mock_jsonrpc_success(context: Context) -> None: from unittest.mock import MagicMock mock_resp = MagicMock() mock_resp.status = 200 mock_resp.read.return_value = json.dumps( { "jsonrpc": "2.0", "id": "mock-req-id", "result": {"status": "ok"}, } ).encode("utf-8") context.mock_response = mock_resp @when(r"HTTP responses are mocked with HTTP 500 status") def step_mock_http_500(context: Context) -> None: from unittest.mock import MagicMock mock_resp = MagicMock() mock_resp.status = 500 mock_resp.read.return_value = json.dumps( { "jsonrpc": "2.0", "error": {"code": -32603, "message": "Internal Server Error"}, } ).encode("utf-8") context.mock_response = mock_resp @when(r"I send a request (?P.+) via the transport") def step_send_with_mock(context: Context, method_str: str) -> None: from unittest.mock import patch if hasattr(context, "mock_response"): with patch( "cleveragents.a2a.transport.urlopen", return_value=context.mock_response, ): request = A2aRequest(method="test.op") context.response = context.transport.send(request) else: request = A2aRequest(method="test.op") context.response = context.transport.send(request) # ----- response assertions ----- @then(r"the response has no error") def step_response_no_error(context: Context) -> None: assert context.response.error is None, ( f"Expected no error, got {context.response.error}" ) @then(r"the response has an error") def step_response_has_error(context: Context) -> None: assert context.response.error is not None, "Expected an error in response" # Reset step matcher to parse (default) so subsequent step files are not affected use_step_matcher("parse")