"""Step definitions for M6 A2A facade, event queue, transport, and version tests. Split from ``m6_autonomy_acceptance_steps.py`` to stay under the project's 500-line guideline. All step names keep the ``m6 smoke`` prefix to avoid ``AmbiguousStep`` conflicts. """ from __future__ import annotations import json from pathlib import Path from unittest.mock import MagicMock from behave import given, then, when from behave.runner import Context from cleveragents.a2a.errors import ( 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 _FIXTURES_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "m6" # ----------------------------------------------------------------------- # Background # ----------------------------------------------------------------------- @given("a m6 smoke test runner") def step_m6_smoke_runner(context: Context) -> None: """Initialise the m6 smoke test context.""" context.m6_result = None context.m6_response = None context.m6_error = None context.m6_profile = None context.m6_guard_result = None context.m6_operations = None context.m6_event_queue = None context.m6_callback_calls = [] context.m6_subscription_id = None context.m6_version_result = None context.m6_version_supported = None context.m6_transport_connected = None context.m6_profiles_list = [] @given("a m6 smoke A2A local facade") def step_m6_smoke_facade(context: Context) -> None: """Create an A2aLocalFacade for testing.""" context.m6_facade = A2aLocalFacade() # ----------------------------------------------------------------------- # Fixture loading — A2A facade flows # ----------------------------------------------------------------------- @when("I m6 smoke load the A2A facade flows fixture") def step_m6_smoke_load_facade_fixture(context: Context) -> None: with open(_FIXTURES_DIR / "a2a_facade_flows.json") as f: context.m6_fixture_data = json.load(f) @then("the m6 smoke facade fixture should have a session lifecycle entry") def step_m6_smoke_facade_session(context: Context) -> None: names = [f["name"] for f in context.m6_fixture_data["fixtures"]] assert "session_lifecycle" in names @then("the m6 smoke facade fixture should have a plan lifecycle entry") def step_m6_smoke_facade_plan(context: Context) -> None: names = [f["name"] for f in context.m6_fixture_data["fixtures"]] assert "plan_lifecycle" in names # ----------------------------------------------------------------------- # A2A facade dispatch operations # ----------------------------------------------------------------------- @when('I m6 smoke dispatch "{operation}" with params {params_json}') def step_m6_smoke_dispatch( context: Context, operation: str, params_json: str, ) -> None: params = json.loads(params_json) request = A2aRequest(method=operation, params=params) context.m6_response = context.m6_facade.dispatch(request) @then('the m6 smoke response status should be "{status}"') def step_m6_smoke_response_status(context: Context, status: str) -> None: assert context.m6_response is not None if status == "ok": assert context.m6_response.result is not None, ( f"Expected ok response but got error: {context.m6_response.error}" ) else: assert context.m6_response.error is not None, ( f"Expected error response but got result: {context.m6_response.result}" ) @then('the m6 smoke response data should contain key "{key}"') def step_m6_smoke_response_key(context: Context, key: str) -> None: assert context.m6_response is not None data = context.m6_response.result or {} assert key in data, f"Key '{key}' not found in response data: {data}" @then('the m6 smoke response data "{key}" should equal "{value}"') def step_m6_smoke_response_value( context: Context, key: str, value: str, ) -> None: assert context.m6_response is not None data = context.m6_response.result or {} assert data[key] == value, f"Expected {key}={value!r}, got {data.get(key)!r}" # ----------------------------------------------------------------------- # A2A facade error handling # ----------------------------------------------------------------------- @when('I m6 smoke dispatch unknown operation "{operation}"') def step_m6_smoke_dispatch_unknown(context: Context, operation: str) -> None: request = A2aRequest(method=operation, params={}) try: context.m6_facade.dispatch(request) context.m6_error = None except A2aOperationNotFoundError as exc: context.m6_error = exc @then("the m6 smoke facade should raise A2aOperationNotFoundError") def step_m6_smoke_error_op_not_found(context: Context) -> None: assert isinstance(context.m6_error, A2aOperationNotFoundError) @when("I m6 smoke dispatch with a non-A2aRequest object") def step_m6_smoke_dispatch_invalid(context: Context) -> None: try: context.m6_facade.dispatch("not a request") # type: ignore[arg-type] context.m6_error = None except TypeError as exc: context.m6_error = exc @then("the m6 smoke facade should raise TypeError") def step_m6_smoke_error_type(context: Context) -> None: assert isinstance(context.m6_error, TypeError) # ----------------------------------------------------------------------- # A2A facade service registration and list operations # ----------------------------------------------------------------------- @when('I m6 smoke register service "{name}" on the facade') def step_m6_smoke_register_service(context: Context, name: str) -> None: mock_service = MagicMock() context.m6_facade.register_service(name, mock_service) @then('the m6 smoke facade should have service "{name}"') def step_m6_smoke_has_service(context: Context, name: str) -> None: assert name in context.m6_facade._services @when("I m6 smoke list facade operations") def step_m6_smoke_list_ops(context: Context) -> None: context.m6_operations = context.m6_facade.list_operations() @then('the m6 smoke operations should include "{operation}"') def step_m6_smoke_ops_include(context: Context, operation: str) -> None: assert operation in context.m6_operations @then("the m6 smoke operations count should be {count:d}") def step_m6_smoke_ops_count(context: Context, count: int) -> None: assert len(context.m6_operations) == count # ----------------------------------------------------------------------- # A2A event queue # ----------------------------------------------------------------------- @given("a m6 smoke A2A event queue") def step_m6_smoke_event_queue(context: Context) -> None: context.m6_event_queue = A2aEventQueue() context.m6_callback_calls = [] @when('I m6 smoke publish an event with type "{event_type}"') def step_m6_smoke_publish_event(context: Context, event_type: str) -> None: event = A2aEvent(event_type=event_type, data={"test": True}) context.m6_event_queue.publish(event) @then("the m6 smoke event queue should have {count:d} event") def step_m6_smoke_event_count(context: Context, count: int) -> None: events = context.m6_event_queue.get_events() assert len(events) == count @then('the m6 smoke last event type should be "{event_type}"') def step_m6_smoke_last_event_type(context: Context, event_type: str) -> None: events = context.m6_event_queue.get_events() assert events[-1].event_type == event_type @when("I m6 smoke subscribe a local callback") def step_m6_smoke_subscribe_local(context: Context) -> None: def _callback(event: A2aEvent) -> None: context.m6_callback_calls.append(event) context.m6_subscription_id = context.m6_event_queue.subscribe_local(_callback) @then("the m6 smoke callback should have been called once") def step_m6_smoke_callback_called(context: Context) -> None: assert len(context.m6_callback_calls) == 1 @when("I m6 smoke unsubscribe the callback") def step_m6_smoke_unsubscribe(context: Context) -> None: context.m6_event_queue.unsubscribe(context.m6_subscription_id) @then("the m6 smoke callback should not have been called") def step_m6_smoke_callback_not_called(context: Context) -> None: assert len(context.m6_callback_calls) == 0 @when("I m6 smoke close the event queue") def step_m6_smoke_close_queue(context: Context) -> None: context.m6_event_queue.close() @then("the m6 smoke publishing should raise RuntimeError") def step_m6_smoke_publish_after_close(context: Context) -> None: event = A2aEvent(event_type="after.close", data={}) try: context.m6_event_queue.publish(event) raise AssertionError("Expected RuntimeError") except RuntimeError: pass @when('I m6 smoke attempt remote subscribe to "{endpoint}"') def step_m6_smoke_remote_subscribe(context: Context, endpoint: str) -> None: try: context.m6_event_queue.subscribe_remote(endpoint) context.m6_error = None except A2aNotAvailableError as exc: context.m6_error = exc @then("the m6 smoke facade should raise A2aNotAvailableError") def step_m6_smoke_error_not_available(context: Context) -> None: assert isinstance(context.m6_error, A2aNotAvailableError) # ----------------------------------------------------------------------- # A2A HTTP transport stub # ----------------------------------------------------------------------- @when("I m6 smoke attempt transport send") def step_m6_smoke_transport_send(context: Context) -> None: transport = A2aHttpTransport() request = A2aRequest(method="plan.create", params={}) try: transport.send(request) context.m6_error = None except A2aNotAvailableError as exc: context.m6_error = exc @when('I m6 smoke attempt transport connect to "{url}"') def step_m6_smoke_transport_connect(context: Context, url: str) -> None: transport = A2aHttpTransport() try: transport.connect(url) context.m6_error = None except A2aNotAvailableError as exc: context.m6_error = exc @when("I m6 smoke attempt transport disconnect") def step_m6_smoke_transport_disconnect(context: Context) -> None: transport = A2aHttpTransport() try: transport.disconnect() context.m6_error = None except A2aNotAvailableError as exc: context.m6_error = exc @when("I m6 smoke check transport is_connected") def step_m6_smoke_transport_connected(context: Context) -> None: transport = A2aHttpTransport() context.m6_transport_connected = transport.is_connected() @then("the m6 smoke transport should not be connected") def step_m6_smoke_transport_not_connected(context: Context) -> None: assert context.m6_transport_connected is False # ----------------------------------------------------------------------- # A2A version negotiation # ----------------------------------------------------------------------- @when('I m6 smoke negotiate A2A version "{version}"') def step_m6_smoke_negotiate(context: Context, version: str) -> None: negotiator = A2aVersionNegotiator() try: context.m6_version_result = negotiator.negotiate(version) context.m6_error = None except A2aVersionMismatchError as exc: context.m6_error = exc @then('the m6 smoke negotiated version should be "{version}"') def step_m6_smoke_negotiated(context: Context, version: str) -> None: assert context.m6_version_result == version @then("the m6 smoke facade should raise A2aVersionMismatchError") def step_m6_smoke_error_version(context: Context) -> None: assert isinstance(context.m6_error, A2aVersionMismatchError) @when('I m6 smoke check if version "{version}" is supported') def step_m6_smoke_version_supported(context: Context, version: str) -> None: negotiator = A2aVersionNegotiator() context.m6_version_supported = negotiator.is_supported(version) @then("the m6 smoke version support should be true") def step_m6_smoke_version_true(context: Context) -> None: assert context.m6_version_supported is True @then("the m6 smoke version support should be false") def step_m6_smoke_version_false(context: Context) -> None: assert context.m6_version_supported is False # ----------------------------------------------------------------------- # A2A model validation # ----------------------------------------------------------------------- @when("I m6 smoke create A2aRequest with empty operation") def step_m6_smoke_invalid_request(context: Context) -> None: try: A2aRequest(method="") context.m6_error = None except ValueError as exc: context.m6_error = exc @when('I m6 smoke create A2aResponse with invalid status "{status}"') def step_m6_smoke_invalid_response(context: Context, status: str) -> None: try: # A2aResponse requires either result or error (not both, not neither) A2aResponse(id="test") context.m6_error = None except ValueError as exc: context.m6_error = exc @when("I m6 smoke create A2aEvent with empty event_type") def step_m6_smoke_invalid_event(context: Context) -> None: try: A2aEvent(event_type="") context.m6_error = None except ValueError as exc: context.m6_error = exc @when("I m6 smoke create A2aErrorDetail with empty code") def step_m6_smoke_invalid_error_detail(context: Context) -> None: try: A2aErrorDetail(code="", message="test") context.m6_error = None except ValueError as exc: context.m6_error = exc