diff --git a/test/unit/adapters/openai_agents/test_runner_spawn_patch.py b/test/unit/adapters/openai_agents/test_runner_spawn_patch.py index e166b526..4d45b946 100644 --- a/test/unit/adapters/openai_agents/test_runner_spawn_patch.py +++ b/test/unit/adapters/openai_agents/test_runner_spawn_patch.py @@ -61,7 +61,7 @@ def teardown_method(self) -> None: async def test_patched_run_sets_spawn_ctx(self, monkeypatch: pytest.MonkeyPatch) -> None: captured: list[SpawnContext | None] = [] - async def capturing_run(agent: object, *, input: object, **kwargs: object) -> str: + async def capturing_run(agent: object, /, *, input: object, **kwargs: object) -> str: captured.append(_SPAWN_CTX.get()) return "done" @@ -79,7 +79,7 @@ async def capturing_run(agent: object, *, input: object, **kwargs: object) -> st @pytest.mark.asyncio async def test_spawn_ctx_is_reset_after_run(self, monkeypatch: pytest.MonkeyPatch) -> None: - async def passthrough_run(agent: object, *, input: object, **kwargs: object) -> str: + async def passthrough_run(agent: object, /, *, input: object, **kwargs: object) -> str: return "ok" monkeypatch.setattr(FakeRunner, "run", classmethod(passthrough_run)) @@ -90,7 +90,7 @@ async def passthrough_run(agent: object, *, input: object, **kwargs: object) -> @pytest.mark.asyncio async def test_spawn_ctx_reset_on_exception(self, monkeypatch: pytest.MonkeyPatch) -> None: - async def failing_run(agent: object, *, input: object, **kwargs: object) -> str: + async def failing_run(agent: object, /, *, input: object, **kwargs: object) -> str: raise RuntimeError("runner failed") monkeypatch.setattr(FakeRunner, "run", classmethod(failing_run)) diff --git a/test/unit/negative_control.py b/test/unit/negative_control.py new file mode 100644 index 00000000..d2f72f88 --- /dev/null +++ b/test/unit/negative_control.py @@ -0,0 +1,176 @@ +"""Reusable enforcement-truth negative-control fixture (AAASM-5529). + +A test that only asserts ``pytest.raises(PolicyViolationError)`` proves the SDK +produced a refusal, not that the refusal *prevented* anything: a tool whose body +has no observable effect would satisfy the same assertion. These helpers give a +denied tool a real, externally-observable effect — a file on disk, an HTTP +request delivered to a live loopback listener — so a deny can be asserted as the +*absence* of that effect and the matching allow as its *presence*. + +Every control built on this fixture is used as a pair: + +* **positive control** — policy allows, the effect is observed. Without it, + "no file on disk" is equally well explained by "the tool never ran at all", + and the negative control proves nothing. +* **negative control** — policy denies, the same effect is absent. + +The effects are deliberately real (``pathlib``, ``http.server``) rather than +``Mock`` call counters: a recorded call is evidence of intent, and Epic +AAASM-5526 exists because intent-level evidence is what over-claimed enforcement +looks like. +""" + +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from typing import Any +from urllib import request as urllib_request + + +@dataclass +class FileSideEffect: + """A filesystem-backed side effect rooted at ``path``. + + ``write`` really creates the file and ``occurred`` really stats it, so an + assertion over ``occurred()`` is an assertion about the world rather than + about the SDK's own bookkeeping. + """ + + path: Path + + def write(self, content: str) -> str: + self.path.write_text(content, encoding="utf-8") + return str(self.path) + + def occurred(self) -> bool: + return self.path.exists() + + def content(self) -> str | None: + if not self.path.exists(): + return None + return self.path.read_text(encoding="utf-8") + + +@dataclass +class ReceivedRequest: + method: str + path: str + body: str + + +class _RecordingHandler(BaseHTTPRequestHandler): + """Records every request instead of serving anything.""" + + received: list[ReceivedRequest] + + def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler contract + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length).decode("utf-8") if length else "" + type(self).received.append(ReceivedRequest(method="POST", path=self.path, body=body)) + self.send_response(204) + self.end_headers() + + def log_message(self, _format: str, *_args: Any) -> None: + """Silence the default stderr access log so test output stays readable.""" + return None + + +@dataclass +class NetworkSideEffect: + """A loopback HTTP listener that records every request it receives. + + A denied tool must leave ``requests`` empty. Because the positive control + exercises the same live listener, an empty log is evidence the egress did + not happen rather than evidence it could not have. + """ + + url: str + _server: HTTPServer + _thread: threading.Thread + requests: list[ReceivedRequest] = field(default_factory=list) + + def call(self, body: str) -> int: + req = urllib_request.Request(self.url, data=body.encode("utf-8"), method="POST") + with urllib_request.urlopen(req, timeout=5) as response: # noqa: S310 - fixed loopback URL + return int(response.status) + + def occurred(self) -> bool: + return len(self.requests) > 0 + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5) + + +def start_network_side_effect() -> NetworkSideEffect: + """Start a loopback listener on an ephemeral port and return its fixture.""" + received: list[ReceivedRequest] = [] + handler = type("_BoundRecordingHandler", (_RecordingHandler,), {"received": received}) + server = HTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return NetworkSideEffect( + url=f"http://127.0.0.1:{server.server_port}/exfiltrate", + _server=server, + _thread=thread, + requests=received, + ) + + +@dataclass +class RecordedResult: + """One post-execution audit record the governed path emitted.""" + + tool_name: str + agent_id: str | None + run_id: str | None + result: str + + +class AuditRecordingInterceptor: + """Delegates every governance call to ``inner`` and records the audit hook. + + Only the post-execution ``record_result`` hook is added — the authoritative + verdict still comes from the wrapped interceptor, so the deny under test is + the real one. This exists because the SDK's ``GatewayClient`` implements no + audit sink of its own (the interceptor is the only one), and AAASM-5529 + requires deny/allow evidence to carry agent and tool identity. + """ + + def __init__(self, inner: Any) -> None: + self._inner = inner + self.records: list[RecordedResult] = [] + + def check_tool_start(self, **kwargs: Any) -> Any: + return self._inner.check_tool_start(**kwargs) + + def record_result( + self, + *, + tool_name: str, + result: str, + agent_id: str | None = None, + run_id: str | None = None, + ) -> None: + self.records.append(RecordedResult(tool_name=tool_name, agent_id=agent_id, run_id=run_id, result=result)) + + def __getattr__(self, name: str) -> Any: + return getattr(self._inner, name) + + +def tool_args_of(query_call: tuple[Any, ...]) -> dict[str, Any]: + """Decode the ``tool_args_json`` the SDK presented to the native runtime. + + ``FakeRuntimeClient.query_calls`` entries are + ``(agent_id, action_type, tool_name, tool_args_json)``. + """ + raw = query_call[3] + if not raw: + return {} + decoded = json.loads(raw) + return decoded if isinstance(decoded, dict) else {} diff --git a/test/unit/test_quickstart_negative_control.py b/test/unit/test_quickstart_negative_control.py new file mode 100644 index 00000000..493a212f --- /dev/null +++ b/test/unit/test_quickstart_negative_control.py @@ -0,0 +1,344 @@ +"""Enforcement-truth negative controls for the documented Python quick-start. + +AAASM-5529, Epic AAASM-5526. + +``docs/quick-start.md`` tells a reader that after ``init_assembly(...)`` "every +tool call from now on goes through the policy gate", and that a denied tool +surfaces as a ``ToolExecutionBlockedError``. Existing tests of that claim assert +either the returned verdict dict or an empty ``executed`` list captured by a +closure. Neither shows that the *effect the tool exists to produce* was +prevented. + +Each control here therefore: + +1. runs the real ``init_assembly`` so the interceptor under test is the one the + SDK actually builds (a genuine ``RuntimeQueryInterceptor`` over a fake native + runtime, not a hand-rolled stand-in); +2. drives the SDK's own governed-call chain, ``run_governed_async_tool`` — the + shared pre-execution gate behind the Google ADK and Pydantic AI quick-start + tabs — so the SDK, not the test, decides whether the tool body runs; +3. asserts a real side effect (a file on disk, an HTTP request delivered to a + live loopback listener) is present under allow and absent under deny. + +The side-effect assertion is made *before* the exception assertion on purpose. +Asserting the exception first would short-circuit the test when enforcement is +removed, leaving the absence check unexercised — the falsification run would +then only ever prove "no error was raised", which is precisely the weak evidence +this suite replaces. + +The ``FALSIFICATION`` cases run the identical function with governance removed. +They must observe the side effect; if they stop doing so, every deny assertion +here has become vacuous. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any + +import pytest + +from agent_assembly import init_assembly +from agent_assembly.adapters._shared.tool_governance import run_governed_async_tool +from agent_assembly.core import assembly as core_assembly +from agent_assembly.core.runtime_interceptor import build_governance_interceptor +from agent_assembly.exceptions import ToolExecutionBlockedError + +from .core._fake_core import FakeRuntimeClient, install_fake_core +from .negative_control import ( + AuditRecordingInterceptor, + FileSideEffect, + NetworkSideEffect, + start_network_side_effect, + tool_args_of, +) + +# A non-loopback https gateway: the register-endpoint TLS guard (AAASM-4655) +# fail-closes a plaintext http:// register channel to a non-loopback host. +_GW_URL = "https://gateway.test" +_API_KEY = "test-key" +_AGENT_ID = "quickstart-negative-control-agent" + + +@pytest.fixture(autouse=True) +def _cleanup_active_context() -> None: + """Release the process-singleton context so each control inits cleanly.""" + active = core_assembly._ACTIVE_CONTEXT + if active is not None and not active.is_shutdown: + active.shutdown() + core_assembly._ACTIVE_CONTEXT = None + + +@pytest.fixture +def file_effect(tmp_path: Path) -> FileSideEffect: + return FileSideEffect(path=tmp_path / "denied-write.txt") + + +@pytest.fixture +def network_effect() -> Any: + effect = start_network_side_effect() + try: + yield effect + finally: + effect.close() + + +class _GovernedQuickStart: + """The real ``init_assembly`` context plus the interceptor it produced.""" + + def __init__(self, context: Any, interceptor: AuditRecordingInterceptor, runtime: FakeRuntimeClient) -> None: + self.context = context + self.interceptor = interceptor + self.runtime = runtime + + def call(self, tool_name: str, tool_args: dict[str, Any], body: Any) -> Any: + """Run ``body`` through the SDK's governed-tool chain.""" + return asyncio.run( + run_governed_async_tool( + self.interceptor, + enforce=True, + tool_name=tool_name, + tool_args=tool_args, + agent_id=_AGENT_ID, + run_id="run-1", + invoke_original=body, + ) + ) + + +def _init_quickstart(monkeypatch: pytest.MonkeyPatch, *, decision: str, reason: str = "") -> _GovernedQuickStart: + """Run the documented ``init_assembly`` and capture the interceptor it built. + + ``build_governance_interceptor`` is wrapped rather than replaced, so the real + ``_register_adapters`` runs and the interceptor handed back is the SDK's own. + """ + runtime = FakeRuntimeClient(decision=decision, reason=reason) + install_fake_core(monkeypatch, runtime) + monkeypatch.setattr( + core_assembly, + "_start_network_layer", + lambda **_kwargs: ("sdk-only", core_assembly._noop_shutdown), + ) + + captured: list[AuditRecordingInterceptor] = [] + + def _spy(*args: Any, **kwargs: Any) -> Any: + interceptor = AuditRecordingInterceptor(build_governance_interceptor(*args, **kwargs)) + captured.append(interceptor) + return interceptor + + monkeypatch.setattr(core_assembly, "build_governance_interceptor", _spy) + + context = init_assembly( + gateway_url=_GW_URL, + api_key=_API_KEY, + agent_id=_AGENT_ID, + mode="sdk-only", + enforcement_mode="enforce", + ) + assert captured, "init_assembly did not build a governance interceptor" + return _GovernedQuickStart(context, captured[0], runtime) + + +def _settle(call: Any) -> Any: + """Run ``call`` and return its result or its exception, never raising. + + Keeps the side-effect assertion reachable even when the governed call + unexpectedly succeeds — see the module docstring. + """ + try: + return call() + except Exception as error: # noqa: BLE001 - the exception is the value under test + return error + + +class TestFilesystemSideEffect: + def test_positive_control_allowed_write_creates_the_file( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="allow") + try: + quickstart.call("write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("allowed")) + finally: + quickstart.context.shutdown() + + assert file_effect.occurred() is True + assert file_effect.content() == "allowed" + + def test_negative_control_denied_write_leaves_no_file( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="deny", reason="policy forbids disk writes") + try: + outcome = _settle( + lambda: quickstart.call( + "write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("denied") + ) + ) + finally: + quickstart.context.shutdown() + + # The load-bearing assertion: the effect the tool exists to produce is + # absent from the filesystem, not merely that an error was raised. + assert file_effect.occurred() is False + assert file_effect.content() is None + assert isinstance(outcome, ToolExecutionBlockedError) + assert "policy forbids disk writes" in str(outcome) + + def test_falsification_the_same_write_ungoverned_creates_the_file(self, file_effect: FileSideEffect) -> None: + # No init_assembly, no interceptor — enforcement removed. If this does + # not write, the negative control above is vacuous. + file_effect.write("ungoverned") + + assert file_effect.occurred() is True + assert file_effect.content() == "ungoverned" + + +class TestNetworkSideEffect: + def test_positive_control_allowed_egress_reaches_the_listener( + self, monkeypatch: pytest.MonkeyPatch, network_effect: NetworkSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="allow") + try: + status = quickstart.call( + "send_http_request", + {"url": network_effect.url}, + lambda: network_effect.call("allowed-payload"), + ) + finally: + quickstart.context.shutdown() + + assert status == 204 + assert len(network_effect.requests) == 1 + assert network_effect.requests[0].body == "allowed-payload" + + def test_negative_control_denied_egress_never_reaches_the_listener( + self, monkeypatch: pytest.MonkeyPatch, network_effect: NetworkSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="deny", reason="egress denied") + try: + outcome = _settle( + lambda: quickstart.call( + "send_http_request", + {"url": network_effect.url}, + lambda: network_effect.call("denied-payload"), + ) + ) + finally: + quickstart.context.shutdown() + + # The listener is live and was reachable throughout — the positive + # control proves that on the same fixture — so zero received requests is + # evidence the egress did not happen, not that it could not have. + assert network_effect.occurred() is False + assert network_effect.requests == [] + assert isinstance(outcome, ToolExecutionBlockedError) + + def test_falsification_the_same_egress_ungoverned_reaches_the_listener( + self, network_effect: NetworkSideEffect + ) -> None: + assert network_effect.call("ungoverned-payload") == 204 + + assert network_effect.occurred() is True + assert network_effect.requests[0].body == "ungoverned-payload" + + +class TestDenyIsAttributable: + def test_the_runtime_saw_the_agent_and_tool_the_deny_was_decided_against( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="deny", reason="policy forbids disk writes") + try: + outcome = _settle( + lambda: quickstart.call( + "write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("denied") + ) + ) + finally: + quickstart.context.shutdown() + + # Absence first, as in every other control here: an exception assertion + # placed ahead of it aborts the test before the side effect is checked, + # so the absence would never be exercised by the falsification run. + assert file_effect.occurred() is False + assert isinstance(outcome, ToolExecutionBlockedError) + + # Identity as presented to the authoritative policy query, not as the + # test reconstructed it: an anonymous deny is not usable evidence. + assert len(quickstart.runtime.query_calls) == 1 + agent_id, action_type, tool_name, _args_json = quickstart.runtime.query_calls[0] + assert agent_id == _AGENT_ID + assert action_type == "tool_call" + assert tool_name == "write_to_disk" + assert tool_args_of(quickstart.runtime.query_calls[0]) == {"path": str(file_effect.path)} + + def test_an_allowed_call_is_recorded_with_the_same_identity( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + quickstart = _init_quickstart(monkeypatch, decision="allow") + try: + quickstart.call("write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("allowed")) + finally: + quickstart.context.shutdown() + + assert file_effect.occurred() is True + assert len(quickstart.interceptor.records) == 1 + record = quickstart.interceptor.records[0] + assert record.tool_name == "write_to_disk" + assert record.agent_id == _AGENT_ID + assert record.run_id == "run-1" + + +class TestDegradedRuntimeCannotLookProtected: + def test_an_unavailable_native_runtime_denies_rather_than_silently_allowing( + self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect + ) -> None: + """Under enforce, no native extension must block — not pass through. + + AAASM-5526 forbids a degraded path presenting as protected. The control + proves the posture is real by the same standard as every other one here: + the side effect is absent. + """ + monkeypatch.setattr( + core_assembly, + "_start_network_layer", + lambda **_kwargs: ("sdk-only", core_assembly._noop_shutdown), + ) + captured: list[Any] = [] + + def _spy(*args: Any, **kwargs: Any) -> Any: + interceptor = build_governance_interceptor(*args, **kwargs) + captured.append(interceptor) + return interceptor + + monkeypatch.setattr(core_assembly, "build_governance_interceptor", _spy) + + # No install_fake_core: agent_assembly._core is absent, so the SDK has no + # authoritative verdict source at all. + context = init_assembly( + gateway_url=_GW_URL, + api_key=_API_KEY, + agent_id=_AGENT_ID, + mode="sdk-only", + enforcement_mode="enforce", + ) + try: + outcome = _settle( + lambda: asyncio.run( + run_governed_async_tool( + captured[0], + enforce=True, + tool_name="write_to_disk", + tool_args={"path": str(file_effect.path)}, + agent_id=_AGENT_ID, + run_id="run-1", + invoke_original=lambda: file_effect.write("degraded"), + ) + ) + ) + finally: + context.shutdown() + + assert file_effect.occurred() is False + assert isinstance(outcome, ToolExecutionBlockedError)