forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
336 lines
12 KiB
Python
336 lines
12 KiB
Python
"""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)}"
|