"""Step definitions for features/mcp_adapter_coverage_r3.feature. Covers previously-uncovered lines in src/cleveragents/mcp/adapter.py: - L131: MCPTransport.connect() raises NotImplementedError - L135-136: MCPTransport.call() raises NotImplementedError - L348-352: dispatch_notification catches listener exceptions - L531-535: register_tools builds slot_dicts from inferred resource slots """ from __future__ import annotations from typing import Any from behave import given, then, when from behave.runner import Context from cleveragents.mcp.adapter import ( MCPServerConfig, MCPToolAdapter, MCPTransport, ) from cleveragents.tool.registry import ToolRegistry from features.mocks.mock_mcp_transport import MockMCPTransport # --------------------------------------------------------------- # Helpers # --------------------------------------------------------------- def _mock_tool(name: str, desc: str = "", schema: dict | None = None) -> dict: return { "name": name, "description": desc or f"Mock tool {name}", "inputSchema": schema or {}, } def _stdio_config(name: str = "test-server") -> MCPServerConfig: return MCPServerConfig(name=name, transport="stdio", command="echo") # --------------------------------------------------------------- # Given - base MCPTransport # --------------------------------------------------------------- @given("mcpcov3 a bare MCPTransport instance") def step_mcpcov3_bare_transport(context: Context) -> None: context.mcpcov3_transport = MCPTransport() context.mcpcov3_error = None # --------------------------------------------------------------- # Given - notification listeners # --------------------------------------------------------------- @given("mcpcov3 a connected adapter with a notification listener that raises") def step_mcpcov3_adapter_with_failing_listener(context: Context) -> None: config = _stdio_config() transport = MockMCPTransport() adapter = MCPToolAdapter(config=config, transport=transport) adapter.connect() # Track calls to the healthy listener context.mcpcov3_healthy_calls: list[tuple[str, dict]] = [] def _failing_listener(method: str, params: dict[str, Any]) -> None: raise ValueError("listener exploded") def _healthy_listener(method: str, params: dict[str, Any]) -> None: context.mcpcov3_healthy_calls.append((method, params)) # Register the failing listener first, then the healthy one adapter.add_notification_listener(_failing_listener) adapter.add_notification_listener(_healthy_listener) context.mcpcov3_adapter = adapter context.mcpcov3_error = None @given("mcpcov3 a connected adapter with only failing notification listeners") def step_mcpcov3_adapter_all_failing_listeners(context: Context) -> None: config = _stdio_config() transport = MockMCPTransport() adapter = MCPToolAdapter(config=config, transport=transport) adapter.connect() def _failing_listener_1(method: str, params: dict[str, Any]) -> None: raise RuntimeError("boom one") def _failing_listener_2(method: str, params: dict[str, Any]) -> None: raise TypeError("boom two") adapter.add_notification_listener(_failing_listener_1) adapter.add_notification_listener(_failing_listener_2) context.mcpcov3_adapter = adapter context.mcpcov3_error = None # --------------------------------------------------------------- # Given - tools with resource-slot-triggering schemas # --------------------------------------------------------------- @given("mcpcov3 a connected adapter with a tool having file_path in its schema") def step_mcpcov3_tool_with_file_path(context: Context) -> None: schema: dict[str, Any] = { "type": "object", "properties": { "file_path": {"type": "string"}, "content": {"type": "string"}, }, } tools = [_mock_tool("write_file", schema=schema)] results = {"write_file": {"status": "ok"}} transport = MockMCPTransport(tools=tools, invoke_results=results) config = _stdio_config() adapter = MCPToolAdapter(config=config, transport=transport) adapter.connect() context.mcpcov3_adapter = adapter context.mcpcov3_error = None @given("mcpcov3 a connected adapter with a tool having directory in its schema") def step_mcpcov3_tool_with_directory(context: Context) -> None: schema: dict[str, Any] = { "type": "object", "properties": { "directory": {"type": "string"}, }, } tools = [_mock_tool("list_dir", schema=schema)] results = {"list_dir": {"entries": []}} transport = MockMCPTransport(tools=tools, invoke_results=results) config = _stdio_config() adapter = MCPToolAdapter(config=config, transport=transport) adapter.connect() context.mcpcov3_adapter = adapter context.mcpcov3_error = None @given("mcpcov3 a connected adapter with a tool having repo_path in its schema") def step_mcpcov3_tool_with_repo_path(context: Context) -> None: schema: dict[str, Any] = { "type": "object", "properties": { "repo_path": {"type": "string"}, }, } tools = [_mock_tool("git_status", schema=schema)] results = {"git_status": {"clean": True}} transport = MockMCPTransport(tools=tools, invoke_results=results) config = _stdio_config() adapter = MCPToolAdapter(config=config, transport=transport) adapter.connect() context.mcpcov3_adapter = adapter context.mcpcov3_error = None @given( "mcpcov3 a connected adapter with a tool having file_path and directory in its schema" ) def step_mcpcov3_tool_with_file_and_dir(context: Context) -> None: schema: dict[str, Any] = { "type": "object", "properties": { "file_path": {"type": "string"}, "directory": {"type": "string"}, }, } tools = [_mock_tool("copy_file", schema=schema)] results = {"copy_file": {"copied": True}} transport = MockMCPTransport(tools=tools, invoke_results=results) config = _stdio_config() adapter = MCPToolAdapter(config=config, transport=transport) adapter.connect() context.mcpcov3_adapter = adapter context.mcpcov3_error = None @given("mcpcov3 an empty tool registry") def step_mcpcov3_empty_registry(context: Context) -> None: context.mcpcov3_registry = ToolRegistry() # --------------------------------------------------------------- # When - base transport methods # --------------------------------------------------------------- @when("mcpcov3 I call connect on the base transport") def step_mcpcov3_call_connect(context: Context) -> None: context.mcpcov3_error = None try: dummy_config = _stdio_config("dummy") context.mcpcov3_transport.connect(dummy_config) except Exception as exc: context.mcpcov3_error = exc @when('mcpcov3 I call call on the base transport with method "tools/list" and params') def step_mcpcov3_call_call(context: Context) -> None: context.mcpcov3_error = None try: context.mcpcov3_transport.call("tools/list", {}) except Exception as exc: context.mcpcov3_error = exc # --------------------------------------------------------------- # When - notification dispatch # --------------------------------------------------------------- @when('mcpcov3 I dispatch notification "{method}" with no params') def step_mcpcov3_dispatch_notification(context: Context, method: str) -> None: context.mcpcov3_error = None try: context.mcpcov3_adapter.dispatch_notification(method) except Exception as exc: context.mcpcov3_error = exc # --------------------------------------------------------------- # When - register tools # --------------------------------------------------------------- @when('mcpcov3 I register tools with namespace "{namespace}"') def step_mcpcov3_register_tools(context: Context, namespace: str) -> None: context.mcpcov3_error = None try: context.mcpcov3_registered = context.mcpcov3_adapter.register_tools( registry=context.mcpcov3_registry, namespace=namespace, ) except Exception as exc: context.mcpcov3_error = exc # --------------------------------------------------------------- # Then - error assertions # --------------------------------------------------------------- @then("mcpcov3 the error should be a NotImplementedError") def step_mcpcov3_error_is_not_implemented(context: Context) -> None: assert context.mcpcov3_error is not None, ( "Expected NotImplementedError but no error occurred" ) assert isinstance(context.mcpcov3_error, NotImplementedError), ( f"Expected NotImplementedError, got {type(context.mcpcov3_error).__name__}: " f"{context.mcpcov3_error}" ) @then("mcpcov3 no error should propagate") def step_mcpcov3_no_error(context: Context) -> None: assert context.mcpcov3_error is None, ( f"Expected no error, but got: {context.mcpcov3_error}" ) @then("mcpcov3 the healthy listener should have received the notification") def step_mcpcov3_healthy_listener_received(context: Context) -> None: assert len(context.mcpcov3_healthy_calls) == 1, ( f"Expected 1 healthy call, got {len(context.mcpcov3_healthy_calls)}" ) method, params = context.mcpcov3_healthy_calls[0] assert method == "tools/list_changed" assert isinstance(params, dict) # --------------------------------------------------------------- # Then - registry / slot assertions # --------------------------------------------------------------- @then('mcpcov3 the registry should contain tool "{tool_name}"') def step_mcpcov3_registry_has_tool(context: Context, tool_name: str) -> None: assert context.mcpcov3_error is None, ( f"Registration failed: {context.mcpcov3_error}" ) spec = context.mcpcov3_registry.get(tool_name) assert spec is not None, f"Tool '{tool_name}' not found in registry" @then('mcpcov3 tool "{tool_name}" source_metadata should have resource_slots') def step_mcpcov3_tool_has_resource_slots(context: Context, tool_name: str) -> None: spec = context.mcpcov3_registry.get(tool_name) assert spec is not None, f"Tool '{tool_name}' not found in registry" metadata = spec.source_metadata assert metadata is not None, "source_metadata is None" slots = metadata.get("resource_slots") assert slots is not None, "resource_slots key missing from source_metadata" assert len(slots) > 0, "resource_slots is empty" # Store for later assertions context.mcpcov3_slots = slots @then('mcpcov3 resource slot {idx:d} should have name "{name}" and type "{rtype}"') def step_mcpcov3_slot_name_type( context: Context, idx: int, name: str, rtype: str ) -> None: slots = context.mcpcov3_slots assert idx < len(slots), f"Slot index {idx} out of range (have {len(slots)} slots)" slot = slots[idx] assert slot["name"] == name, f"Expected slot name '{name}', got '{slot['name']}'" assert slot["resource_type"] == rtype, ( f"Expected resource_type '{rtype}', got '{slot['resource_type']}'" ) # Also verify access and binding are present as strings (lines 534-535) assert "access" in slot, "Slot missing 'access' key" assert "binding" in slot, "Slot missing 'binding' key" assert isinstance(slot["access"], str), ( f"access should be str, got {type(slot['access'])}" ) assert isinstance(slot["binding"], str), ( f"binding should be str, got {type(slot['binding'])}" ) @then( 'mcpcov3 tool "{tool_name}" source_metadata should have resource_slots with count {count:d}' ) def step_mcpcov3_tool_slots_count(context: Context, tool_name: str, count: int) -> None: spec = context.mcpcov3_registry.get(tool_name) assert spec is not None, f"Tool '{tool_name}' not found in registry" metadata = spec.source_metadata assert metadata is not None, "source_metadata is None" slots = metadata.get("resource_slots") assert slots is not None, "resource_slots key missing from source_metadata" assert len(slots) == count, f"Expected {count} slots, got {len(slots)}"