Initial version

tests/test_safety.py

@@ -0,0 +1,97 @@
+"""Safety checks as tests (plan §13).
+
+* a read-only API key must fail cleanly on every mutating tool
+* failed workflows leave no partial state (no comment, no finished step)
+* the API key never appears in any tool result
+"""
+
+import json
+
+import pytest
+from mcp.client import Client
+
+from fakes import make_template
+from mcp_server.tools import register_tools
+
+TOKEN_SAFE_MESSAGE = "labvoice-secret-key-do-not-leak"
+
+MUTATING_TOOLS = {
+    "complete_next_protocol_step": {"experiment_id": 123},
+    "complete_protocol_step": {"experiment_id": 123, "step_id": 9},
+    "record_protocol_observation": {"experiment_id": 123, "comment": "note"},
+    "create_experiment_from_template": {"template_id": 7},
+    "adjust_inventory": {"amount": 5.0, "unit": "mL", "container_id": 31},
+}
+
+
+@pytest.fixture
+async def mcp_client(patch_build_services, client):
+    from mcp.server.fastmcp import FastMCP
+
+    patch_build_services()
+    client.templates[7] = make_template()
+    mcp = FastMCP("labvoice-safety")
+    register_tools(mcp)
+    async with Client(mcp, raise_exceptions=False) as c:
+        yield c
+
+
+def _state_fingerprint(client):
+    experiment = client.experiments[123]
+    return {
+        "steps": [(s.id, s.finished) for s in experiment.steps],
+        "comments": [c.body for c in experiment.comments],
+        "containers": [
+            (i.id, c.sub_id, c.qty_stored)
+            for i in client.items.values()
+            for c in i.containers
+        ],
+    }
+
+
+@pytest.mark.parametrize("tool", sorted(MUTATING_TOOLS))
+async def test_read_only_key_fails_cleanly_on_every_mutating_tool(
+    mcp_client, client, tool
+):
+    """Plan §13: read-only key ⇒ every mutation fails with a clear message."""
+    client.read_only = True
+    before = _state_fingerprint(client)
+
+    result = await mcp_client.call_tool(tool, MUTATING_TOOLS[tool])
+    payload = None
+    structured = getattr(result, "structured_content", None)
+    texts = " ".join(getattr(b, "text", "") for b in getattr(result, "content", []))
+    if structured is not None:
+        payload = structured
+    assert getattr(result, "is_error", False) or (
+        payload is not None and payload.get("ok") is False
+    ) or "permission" in texts.lower() or "read-only" in texts.lower(), (
+        f"{tool} must not silently succeed with a read-only key"
+    )
+
+    assert _state_fingerprint(client) == before, f"{tool} must not mutate anything"
+
+
+async def test_aborted_workflow_leaves_no_partial_state(mcp_client, client, resolver_stub):
+    """Unresolvable resource ⇒ no finished step, no comment, no stock change."""
+    resolver_stub.mappings.clear()
+    before = _state_fingerprint(client)
+
+    result = await mcp_client.call_tool(
+        "complete_next_protocol_step", {"experiment_id": 123}
+    )
+    texts = " ".join(getattr(b, "text", "") for b in getattr(result, "content", []))
+    assert result.is_error or "clarification" in texts.lower() or "not know" in texts.lower()
+
+    assert _state_fingerprint(client) == before
+
+
+async def test_error_results_never_contain_the_api_key(mcp_client, client, resolver_stub):
+    client.read_only = True
+    resolver_stub.mappings.clear()
+    for tool, args in MUTATING_TOOLS.items():
+        result = await mcp_client.call_tool(tool, dict(args))
+        texts = " ".join(getattr(b, "text", "") for b in getattr(result, "content", []))
+        structured = getattr(result, "structured_content", None)
+        blob = texts + json.dumps(structured, default=str)
+        assert TOKEN_SAFE_MESSAGE not in blob