"""The Tool contract and registry. The behaviour under test is mostly refusal: a Tool that is disabled must not run, one that needs approval must not act, and a failing Tool must not take the workflow down with it. """ from __future__ import annotations import pytest from app.api.client import AgentApiError, NotFound from app.tools.base import ExecutionPolicy, Tool, ToolResult from app.tools.client_tools import GetClientTool, SearchClientTool from app.tools.registry import ToolRegistry class FakeCms: """Stands in for the CMS. The Tools must never notice.""" def __init__(self, *, clients=None, raises=None): self._clients = clients if clients is not None else [] self._raises = raises self.calls: list[tuple[str, dict]] = [] def search_clients(self, **criteria): self.calls.append(("search", criteria)) if self._raises: raise self._raises return {"data": self._clients, "meta": {"total": len(self._clients)}} def get_client(self, client_id: int): self.calls.append(("get", {"client_id": client_id})) if self._raises: raise self._raises return {"id": client_id, "name": "Test Client"} class TestExecutionPolicy: def test_a_disabled_tool_refuses_to_run(self): class DeleteEverything(Tool): name = "delete_everything" description = "Never allowed." policy = ExecutionPolicy.DISABLED def run(self, **kwargs): # pragma: no cover - must never be reached raise AssertionError("A disabled tool ran.") result = DeleteEverything().execute() assert not result.ok assert "disabled" in result.error def test_an_approval_tool_proposes_instead_of_acting(self): class UpdateClient(Tool): name = "update_client" description = "Writes to a real client." policy = ExecutionPolicy.REQUIRES_APPROVAL def run(self, **kwargs): # pragma: no cover - must never be reached raise AssertionError("An approval-gated tool acted.") result = UpdateClient().execute(client_id=7, city="Karachi") # Not an error — the correct outcome. The proposal goes to a human. assert result.error == "approval_required" assert result.data["proposed"] == {"client_id": 7, "city": "Karachi"} def test_the_default_policy_is_disabled(self): class Forgetful(Tool): name = "forgetful" description = "Author forgot to declare a policy." def run(self, **kwargs): # pragma: no cover raise AssertionError("Ran without a declared policy.") # The unsafe default must be unreachable, not merely discouraged. assert Forgetful.policy is ExecutionPolicy.DISABLED assert not Forgetful().execute().ok class TestToolContract: def test_a_tool_must_declare_a_name_and_description(self): with pytest.raises(TypeError, match="name"): class Nameless(Tool): description = "Has no name." policy = ExecutionPolicy.AUTOMATIC def run(self, **kwargs): return ToolResult.success() def test_a_raising_tool_returns_a_failure_rather_than_propagating(self): class Exploding(Tool): name = "exploding" description = "Throws." policy = ExecutionPolicy.AUTOMATIC def run(self, **kwargs): raise ValueError("boom") result = Exploding().execute() # One broken Tool must not end a workflow that could still recover. assert not result.ok assert "boom" in result.error def test_duration_is_recorded(self): class Quick(Tool): name = "quick" description = "Returns immediately." policy = ExecutionPolicy.AUTOMATIC def run(self, **kwargs): return ToolResult.success({"x": 1}) assert Quick().execute().duration_ms is not None class TestSearchClientTool: def test_it_returns_matches(self): tool = SearchClientTool(FakeCms(clients=[{"id": 1, "name": "Ali"}])) result = tool.execute(query="Ali") assert result.ok assert result.data["total"] == 1 def test_a_single_match_is_flagged_unambiguous(self): tool = SearchClientTool(FakeCms(clients=[{"id": 1}])) assert tool.execute(query="Ali").data["unambiguous"] is True def test_multiple_matches_are_flagged_ambiguous(self): tool = SearchClientTool(FakeCms(clients=[{"id": 1}, {"id": 2}])) # A workflow must be able to branch on this rather than assume the first # hit — picking silently is how a document is filed against the wrong client. assert tool.execute(query="Ali").data["unambiguous"] is False def test_an_empty_query_is_refused(self): cms = FakeCms(clients=[{"id": 1}]) tool = SearchClientTool(cms) result = tool.execute(query=" ") assert not result.ok assert cms.calls == [], "An empty search must not reach the CMS at all." def test_an_api_failure_becomes_a_tool_failure(self): tool = SearchClientTool(FakeCms(raises=AgentApiError("CMS unreachable"))) result = tool.execute(query="Ali") assert not result.ok assert "unreachable" in result.error class TestGetClientTool: def test_it_fetches_a_client(self): result = GetClientTool(FakeCms()).execute(client_id=42) assert result.ok assert result.data["client"]["id"] == 42 @pytest.mark.parametrize("bad", [0, -1, "seven", None]) def test_an_invalid_id_is_refused_before_calling_the_cms(self, bad): cms = FakeCms() assert not GetClientTool(cms).execute(client_id=bad).ok assert cms.calls == [] def test_an_invisible_client_reads_as_not_found(self): tool = GetClientTool(FakeCms(raises=NotFound("nope", status=404))) result = tool.execute(client_id=99) # The CMS deliberately conflates "absent" and "not yours"; repeating the # distinction here would invent information the API refused to give. assert not result.ok assert "visible" in result.error class TestRegistry: def _tool(self, name: str, policy: ExecutionPolicy) -> Tool: # Built with type() so name and description exist in the class body at # definition time. Assigning them afterwards trips Tool's own guard — # correctly, which is why the helper accommodates it rather than the # other way round. cls = type( f"Tool_{name}", (Tool,), { "name": name, "description": f"{name} tool", "policy": policy, "run": lambda self, **kwargs: ToolResult.success(), }, ) return cls() def test_it_registers_and_retrieves(self): registry = ToolRegistry() registry.register(self._tool("alpha", ExecutionPolicy.AUTOMATIC)) assert registry.has("alpha") assert registry.get("alpha").name == "alpha" def test_registering_the_same_name_twice_is_refused(self): registry = ToolRegistry() registry.register(self._tool("alpha", ExecutionPolicy.AUTOMATIC)) # Silent replacement would let a half-finished Tool shadow a working one. with pytest.raises(ValueError, match="already registered"): registry.register(self._tool("alpha", ExecutionPolicy.AUTOMATIC)) def test_an_unknown_tool_names_what_is_available(self): registry = ToolRegistry() registry.register(self._tool("alpha", ExecutionPolicy.AUTOMATIC)) with pytest.raises(KeyError, match="alpha"): registry.get("missing") def test_executable_excludes_approval_gated_and_disabled_tools(self): registry = ToolRegistry() registry.register(self._tool("readable", ExecutionPolicy.AUTOMATIC)) registry.register(self._tool("writes", ExecutionPolicy.REQUIRES_APPROVAL)) registry.register(self._tool("banned", ExecutionPolicy.DISABLED)) # Offering a model capabilities it cannot exercise invites plans that # cannot be carried out. assert [t.name for t in registry.executable()] == ["readable"] def test_describe_exposes_only_level_one_data(self): registry = ToolRegistry() registry.register(self._tool("alpha", ExecutionPolicy.AUTOMATIC)) entry = registry.describe()[0] # Safe to send to a hosted model (ADR-0002): no client data of any kind. assert set(entry) == {"name", "description", "policy"}