Interbolt
Guides

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

  • @guard does 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.
  • Reporter and ApprovalResolver are the two injectable seams, both with inert defaults, so they are mocked with stock unittest.mock.Mock/AsyncMock (or pytest-mock's mocker fixture), with no monkeypatching of internals required.
  • Policy testing is check() (or runtime.check()) called with synthetic args and taint, asserted against the returned Decision; there is no separate simulate function.
  • InMemoryReporter is the assertion surface for what was emitted, across four lists: events, decisions, findings, and endorsements.

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_id

The 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 allowed

For 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_APPROVAL

Inside 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.

# 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.

On this page