"""Step definitions for a2a_facade_coverage.feature. Targets uncovered lines in ``src/cleveragents/a2a/facade.py``: | Lines | Code path | |-----------|------------------------------------------------------------| | 359-363 | _cleanup_session_devcontainers exception handler | | 501-502 | _handle_plan_cancel with service (non-stub path) | | 505-512 | _handle_plan_tree full method | | 516-524 | _handle_plan_explain stub | | 526-528 | _handle_plan_correct stub | | 530-540 | _handle_plan_artifacts both paths | | 542-544 | _handle_plan_prompt stub | | 546-548 | _handle_plan_rollback stub | | 550-560 | _handle_plan_list with service | | 566-567 | _handle_registry_list_stub | | 573-574 | _handle_context_stub | | 580-584 | _handle_health_check / _handle_diagnostics_run | | 590-591 | _handle_sync_stub | | 597-598 | _handle_namespace_stub | | 191-210 | dispatch error mapping path | | 169-170 | dispatch TypeError for non-request | | 212-220 | register_service + cache invalidation | | 315-317 | session create with service | | 329-335 | session close with service (empty + valid session_id) | | 486-490 | event subscribe with wired queue | | 441-445 | registry list tools with service | | 451-462 | registry list resources with service | | 374-382 | plan create with service | """ from __future__ import annotations import json from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, use_step_matcher, when from behave.runner import Context try: from cleveragents.a2a.facade import A2aLocalFacade from cleveragents.a2a.models import A2aRequest except ImportError: pass # a2a module not available from cleveragents.core.exceptions import PlanError use_step_matcher("re") # --------------------------------------------------------------------------- # Mock helpers # --------------------------------------------------------------------------- class _MockSession: """Minimal session stub.""" def __init__(self, session_id: str = "MOCK-SESSION-001") -> None: self.session_id = session_id class _MockPlanIdentity: """Minimal PlanIdentity stub.""" def __init__(self, plan_id: str = "MOCK-PLAN-001") -> None: self.plan_id = plan_id class _MockPlan: """Minimal plan stub with all attributes used by handlers.""" def __init__(self, plan_id: str = "MOCK-PLAN-001") -> None: self.identity = _MockPlanIdentity(plan_id) self.phase = MagicMock() self.phase.value = "strategize" self.state = MagicMock() self.state.value = "queued" self.decision_root_id = "DECISION-ROOT-001" self.changeset_id = "CHANGESET-001" self.sandbox_refs = ["sandbox-ref-1", "sandbox-ref-2"] class _MockToolSpec: """Minimal ToolSpec stub.""" def __init__(self, name: str, description: str) -> None: self.name = name self.description = description class _MockResource: """Minimal resource stub.""" def __init__(self, resource_id: str, name: str, resource_type_name: str) -> None: self.resource_id = resource_id self.name = name self.resource_type_name = resource_type_name def _build_mock_session_service() -> MagicMock: svc = MagicMock() svc.create.return_value = _MockSession() svc.delete.return_value = None return svc def _build_full_mock_plan_lifecycle_service() -> MagicMock: """Build a mock PlanLifecycleService with all methods.""" svc = MagicMock() svc.use_action.return_value = _MockPlan() svc.execute_plan.return_value = _MockPlan() svc.get_plan.return_value = _MockPlan() svc.apply_plan.return_value = _MockPlan() svc.cancel_plan.return_value = None svc.list_plans.return_value = [_MockPlan("PLAN-A"), _MockPlan("PLAN-B")] return svc def _build_raising_plan_lifecycle_service() -> MagicMock: """Build a mock that raises PlanError on execute_plan.""" svc = MagicMock() svc.execute_plan.side_effect = PlanError("plan execution failed") return svc def _build_mock_tool_registry() -> MagicMock: registry = MagicMock() registry.list_tools.return_value = [ _MockToolSpec("local/tool-a", "Tool A description"), _MockToolSpec("local/tool-b", "Tool B description"), ] return registry def _build_mock_resource_registry_service() -> MagicMock: svc = MagicMock() svc.list_resources.return_value = [ _MockResource("RES-001", "my-repo", "git-checkout"), ] return svc def _build_mock_event_queue() -> MagicMock: queue = MagicMock() queue.subscribe_local.return_value = "SUB-MOCK-001" return queue # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given(r"a facade-cov facade with no services") def step_fc_facade_no_services(context: Context) -> None: context.fc_facade = A2aLocalFacade() @given(r"a facade-cov facade with a mock SessionService") def step_fc_facade_session(context: Context) -> None: context.fc_facade = A2aLocalFacade( services={"session_service": _build_mock_session_service()} ) @given(r"a facade-cov facade with a full mock PlanLifecycleService") def step_fc_facade_full_plan(context: Context) -> None: context.fc_facade = A2aLocalFacade( services={"plan_lifecycle_service": _build_full_mock_plan_lifecycle_service()} ) @given(r"a facade-cov facade with a raising PlanLifecycleService") def step_fc_facade_raising_plan(context: Context) -> None: context.fc_facade = A2aLocalFacade( services={"plan_lifecycle_service": _build_raising_plan_lifecycle_service()} ) @given(r"a facade-cov facade with a mock ToolRegistry") def step_fc_facade_tool_registry(context: Context) -> None: context.fc_facade = A2aLocalFacade( services={"tool_registry": _build_mock_tool_registry()} ) @given(r"a facade-cov facade with a mock ResourceRegistryService") def step_fc_facade_resource_registry(context: Context) -> None: context.fc_facade = A2aLocalFacade( services={"resource_registry_service": _build_mock_resource_registry_service()} ) @given(r"a facade-cov facade with a mock EventQueue") def step_fc_facade_event_queue(context: Context) -> None: context.fc_facade = A2aLocalFacade( services={"event_queue": _build_mock_event_queue()} ) @given(r"a facade-cov facade with no services and a failing cleanup service") def step_fc_facade_failing_cleanup(context: Context) -> None: context.fc_facade = A2aLocalFacade() # Patch the CleanupService so stop_active_devcontainers raises, # exercising the except Exception block at lines 359-363. context._fc_cleanup_patcher = patch( "cleveragents.application.services.cleanup_service.CleanupService" ) mock_cls = context._fc_cleanup_patcher.start() mock_cls.stop_active_devcontainers.side_effect = RuntimeError( "cleanup service unavailable" ) def _stop_patcher(): context._fc_cleanup_patcher.stop() context.add_cleanup(_stop_patcher) # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when( r'I dispatch facade-cov operation "(?P[^"]+)" ' r"with params (?P.+)" ) def step_fc_dispatch(context: Context, operation: str, params_json: str) -> None: params: dict[str, Any] = json.loads(params_json) request = A2aRequest(method=operation, params=params) context.fc_response = context.fc_facade.dispatch(request) @when(r"I try to dispatch a non-A2aRequest object via the facade-cov facade") def step_fc_dispatch_non_request(context: Context) -> None: context.fc_caught_error = None try: context.fc_facade.dispatch("not-a-request") # type: ignore[arg-type] except TypeError as exc: context.fc_caught_error = exc @when(r'I register a facade-cov service "(?P[^"]+)" with a mock') def step_fc_register_service(context: Context, name: str) -> None: context.fc_facade.register_service(name, _build_mock_session_service()) @when(r"I try to register a facade-cov service with empty name") def step_fc_register_empty_name(context: Context) -> None: context.fc_caught_error = None try: context.fc_facade.register_service("", object()) except ValueError as exc: context.fc_caught_error = exc # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then(r'the facade-cov response status should be "(?P[^"]+)"') def step_fc_response_status(context: Context, status: str) -> None: assert (context.fc_response.error is None) == (status == "ok"), ( f"Expected status '{status}', got error={context.fc_response.error}" ) @then( r'the facade-cov response data key "(?P[^"]+)" should equal "(?P[^"]+)"' ) def step_fc_data_key_equals(context: Context, key: str, value: str) -> None: actual = context.fc_response.result.get(key) assert str(actual) == value, f"Expected data['{key}'] = '{value}', got '{actual}'" @then(r'the facade-cov response data key "(?P[^"]+)" should be true') def step_fc_data_key_true(context: Context, key: str) -> None: actual = context.fc_response.result.get(key) assert actual is True, f"Expected data['{key}'] to be True, got {actual!r}" @then(r'the facade-cov response data should contain key "(?P[^"]+)"') def step_fc_data_has_key(context: Context, key: str) -> None: assert key in context.fc_response.result, ( f"Expected key '{key}' in response data, " f"got: {list(context.fc_response.result.keys())}" ) @then(r'the facade-cov response data should not contain key "(?P[^"]+)"') def step_fc_data_no_key(context: Context, key: str) -> None: assert key not in context.fc_response.result, ( f"Expected key '{key}' not in response data, but found it" ) @then(r"the facade-cov response error should contain 'session_id is required'") def step_fc_error_contains_session_id(context: Context) -> None: assert context.fc_response.error is not None, "Expected an error in the response" assert "session_id is required" in context.fc_response.error.message, ( f"Expected error containing 'session_id is required', got: {context.fc_response.error.message}" ) @then(r"the facade-cov response error should not be None") def step_fc_error_not_none(context: Context) -> None: assert context.fc_response.error is not None, "Expected an error in the response" @then(r"the facade-cov response should have timing_ms set") def step_fc_timing_set(context: Context) -> None: # timing_ms removed from wire format per JSON-RPC 2.0 compliance assert context.fc_response.result is not None, "Expected result to be set" @then(r'a facade-cov TypeError should be raised with message "(?P[^"]+)"') def step_fc_type_error_raised(context: Context, msg: str) -> None: assert isinstance(context.fc_caught_error, TypeError), ( f"Expected TypeError, got {type(context.fc_caught_error)}" ) assert msg in str(context.fc_caught_error), ( f"Expected message containing '{msg}', got '{context.fc_caught_error}'" ) @then(r"a facade-cov ValueError should be raised") def step_fc_value_error_raised(context: Context) -> None: assert isinstance(context.fc_caught_error, ValueError), ( f"Expected ValueError, got {type(context.fc_caught_error)}" ) @then(r"the second facade-cov session create should use the new service") def step_fc_session_create_used_service(context: Context) -> None: # After registering a new service and dispatching again, # the response should reflect the mock service return value assert context.fc_response.error is None assert context.fc_response.result["session_id"] == "MOCK-SESSION-001" @then(r"the facade-cov response tools list should have (?P\d+) items") def step_fc_tools_count(context: Context, count: str) -> None: tools = context.fc_response.result.get("tools", []) assert len(tools) == int(count), f"Expected {count} tools, got {len(tools)}" @then(r"the facade-cov response resources list should have (?P\d+) items") def step_fc_resources_count(context: Context, count: str) -> None: resources = context.fc_response.result.get("resources", []) assert len(resources) == int(count), ( f"Expected {count} resources, got {len(resources)}" ) # Reset step matcher to parse (default) so subsequent step files are not affected use_step_matcher("parse")