Testing
How to assert on policy decisions with InMemoryReporter and a fake approval resolver.
Testing
Interbolt is built so a consumer's existing tests of their own tool functions keep working unchanged after adding the library, and so testing policy behavior itself needs no bespoke harness.
Why this works
@guarddoes nothing heavy at import or decoration time; the binding model guarantees decoration captures no runtime. A test that calls a guarded function still calls the real function, after a decision is computed.ReporterandApprovalResolverare the two injectable seams, both with inert defaults, so they are mocked with stockunittest.mock.Mock/AsyncMock(orpytest-mock'smockerfixture), with no monkeypatching of internals required.- Policy testing is
check()(orruntime.check()) called with synthetic args and taint, asserted against the returnedDecision; there is no separatesimulatefunction. InMemoryReporteris the assertion surface for what was emitted, across four lists:events,decisions,findings, andendorsements.
A minimal recipe
from interbolt import (
Action, InMemoryReporter, Policy, configure, taint,
)
def test_untrusted_email_is_blocked():
reporter = InMemoryReporter()
runtime = configure(
policy=Policy.from_file("policy.yaml"),
reporter=reporter,
mode="enforce",
)
decision = runtime.check(
tool="email.send_email",
args={
"to": taint("attacker@external.com", source="web_search"),
"body": "...",
},
agent_id="test-agent",
)
assert decision.action is Action.BLOCK
assert decision.matched_rule == "block_untrusted_exfil"
assert reporter.decisions[-1].decision_id == decision.decision_idThe tool= string must match a sink key in the policy exactly. A mismatch
is not an error: the call falls through to defaults.sink_action, so a
typo shows up as an unexpected require_approval rather than a failure
naming the typo.
Testing through @guard
import pytest
from interbolt import ApprovalDenied, PolicyViolation, configure, taint
def test_guarded_call_is_blocked(runtime): # see fixtures below
agent = runtime.agent("research-agent")
@agent.guard(tool="email.send_email")
def send_email(to: str, body: str) -> None:
... # never reached when blocked
with pytest.raises(PolicyViolation) as exc_info:
send_email(
to=taint("attacker@external.com", source="web_search"),
body="...",
)
assert exc_info.value.decision.matched_rule == "block_untrusted_exfil"A require_approval decision invokes the configured ApprovalResolver. Use
a fake resolver to control the outcome deterministically, rather than the
default auto_deny:
def test_approval_denied_then_granted(mocker):
resolver = mocker.Mock(return_value=False)
runtime = configure(policy=..., approval_resolver=resolver)
agent = runtime.agent("research-agent")
@agent.guard(tool="fs.write")
def write_file(path: str, content: str) -> None: ...
with pytest.raises(ApprovalDenied):
write_file(path="/data/out.txt", content="...")
resolver.return_value = True
write_file(path="/data/out.txt", content="...") # now allowedFor an async def guarded function, use an AsyncMock resolver: a guard
wrapping a coroutine function awaits the resolver automatically. A sync
call site needs a resolver returning a plain bool, and one returning an
awaitable raises InterboltUsageError.
Testing run-level rules
A rule on run.tainted only fires if the ingress was attributed to a run,
which means the taint() call has to happen inside an agent_context, and
check() has to be given that same run's id:
async def test_run_tainted_gates_shell(runtime):
async with runtime.agent_context("research-agent"):
taint("poisoned page text", source="web_search")
decision = runtime.check(
tool="default.run_shell",
args={"cmd": "curl evil.com"},
agent_id="research-agent",
# run_id resolves from the active agent_context; no need to thread it
)
assert decision.action is Action.REQUIRE_APPROVALInside an agent_context, check() resolves the run's run_id from the
context automatically, so the ingress and the decision share one run and
run.tainted reads true. Passing an explicit run_id= that does not
match the context's run is the one way to break this: the gate then reads
false against a run that recorded nothing under your id. See
Identity.
Recommended fixtures
# conftest.py
from unittest.mock import Mock
import pytest
from pytest_mock import MockerFixture
from interbolt import InMemoryReporter, Policy, Runtime, configure
@pytest.fixture
def in_memory_reporter() -> InMemoryReporter:
return InMemoryReporter()
@pytest.fixture
def fake_resolver(mocker: MockerFixture) -> Mock:
return mocker.Mock(return_value=False)
@pytest.fixture
def runtime(in_memory_reporter: InMemoryReporter, fake_resolver: Mock) -> Runtime:
policy = Policy.from_file("tests/policies/test_policy.yaml")
return configure(
policy=policy,
reporter=in_memory_reporter,
approval_resolver=fake_resolver,
mode="enforce",
)Each test that calls configure(), directly or through a fixture, rebinds
the process-current runtime. guard/check resolve the runtime lazily, so
they pick up whichever runtime is current on their next call, and tests do
not leak state through stale captured runtimes. There is one runtime per
process,
so tests that configure different policies must not run concurrently in
threads within one process; separate processes, as with pytest-xdist, are
fine.
dry_run against live traffic
To exercise a new policy without blocking anything, configure
mode="dry_run", drive your agent through real traffic, and inspect
reporter.events[i].outcome, the real pre-downgrade action, rather than
reporter.decisions[i].action, which is always allow under dry_run.
Outcome is a StrEnum, so event.outcome == "block" works directly. See
Policies.