How tripwire Works¶
This guide explains tripwire's architecture: how the sandbox intercepts external calls, how interactions are recorded and asserted, and how the plugin system ties it all together.
The Three Guarantees¶
tripwire enforces three rules that most mocking libraries leave silent:
- Every call must be pre-authorized. If your code makes an external call with no registered mock, tripwire raises
UnmockedInteractionErrorimmediately, not at teardown. - Every recorded interaction must be explicitly asserted. If an interaction is recorded but never asserted, tripwire raises
UnassertedInteractionsErrorat teardown. - Every registered mock must actually be triggered. If you register a mock that never fires, tripwire raises
UnusedMocksErrorat teardown.
Together, these guarantees mean that when a test passes, you know exactly what happened -- not just that nothing crashed.
Sandbox Lifecycle¶
The core entry point is with tripwire:, which creates a sandbox -- a controlled environment where all external calls are intercepted. Here is the exact sequence of events:
Entering the sandbox¶
- ContextVar set. The
_active_verifierContextVar is set to the currentStrictVerifier. This is how interceptors know which verifier to route calls to. - Plugin activation. Each registered plugin's
activate()method is called in registration order. This installs interceptors (monkeypatches) on the target libraries. If any plugin fails to activate, all previously activated plugins are deactivated and the error propagates.
def _enter(self) -> StrictVerifier:
self._token = _active_verifier.set(self._verifier)
for plugin in self._verifier._plugins:
plugin.activate()
return self._verifier
- Your code runs. Inside the
withblock, every external call hits an interceptor instead of the real implementation.
Exiting the sandbox¶
- Plugin deactivation. Each plugin's
deactivate()method is called in reverse order, removing the monkeypatches. - ContextVar reset. The
_active_verifierContextVar is reset to its previous value. - Assertions and verification. After the sandbox exits, you make your assertions. At test teardown,
verify_all()checks guarantees 2 and 3 -- any unasserted interactions or unused mocks cause failures.
The SandboxContext supports both with and async with, using the same _enter() and _exit() methods.
Interception Model¶
tripwire intercepts external calls through class-level monkeypatching. The design uses two key patterns: module-level capture of originals and class-level reference counting.
Module-level capture of originals¶
When a plugin module is first imported, it captures references to the original (unpatched) methods at module scope. For example, the HTTP plugin does this:
# Captured at import time, before any patches
_HTTPX_ORIGINAL_HANDLE = httpx.HTTPTransport.handle_request
_REQUESTS_ORIGINAL_SEND = requests.adapters.HTTPAdapter.send
These references serve two purposes: they are used by conflict detection to identify whether another library (like respx or responses) has already patched the same methods, and they are the values that get restored when the sandbox exits.
Class-level reference counting¶
Plugins use class-level _install_count and _install_lock attributes to handle nested sandboxes correctly. The patches are installed on the first activation and removed only when the last sandbox exits:
class HttpPlugin(BasePlugin):
_install_count: int = 0
_install_lock: threading.Lock = threading.Lock()
def activate(self) -> None:
with HttpPlugin._install_lock:
if HttpPlugin._install_count == 0:
self.check_conflicts()
self.install_patches()
HttpPlugin._install_count += 1
def deactivate(self) -> None:
with HttpPlugin._install_lock:
HttpPlugin._install_count = max(0, HttpPlugin._install_count - 1)
if HttpPlugin._install_count == 0:
self.restore_patches()
This means patches are shared across all verifier instances. The reference count is class-level (not instance-level), so two concurrent sandboxes both using HttpPlugin share the same interceptors. The ContextVar routing (described next) ensures each intercepted call reaches the correct verifier.
ContextVar Routing¶
The central question for any interceptor is: "which verifier should I report to?" tripwire answers this with a ContextVar:
_active_verifier: contextvars.ContextVar[StrictVerifier | None] = contextvars.ContextVar(
"tripwire_active_verifier", default=None
)
When an interceptor fires, it calls get_verifier_or_raise(source_id), which reads the ContextVar and returns the active verifier. If no sandbox is active (the ContextVar is None), it raises SandboxNotActiveError.
Why ContextVar?¶
Python's contextvars.ContextVar is both thread-safe and async-safe. Each thread gets its own value, and each asyncio.Task gets an independent copy. This means:
- Multiple threads can run separate sandboxes concurrently without interference.
- Multiple async tasks can run separate sandboxes concurrently without interference.
- No global mutable state, no locks needed for routing.
tripwire uses three ContextVars:
| ContextVar | Purpose |
|---|---|
_active_verifier |
Points interceptors to the current verifier. Set on sandbox enter, reset on exit. |
_current_test_verifier |
Points module-level API functions (tripwire.mock(), tripwire.assert_interaction()) to the per-test verifier. Managed by the pytest fixture. |
_any_order_depth |
Tracks nesting depth of in_any_order() blocks. When > 0, assertions match in any order. |
Timeline and Interactions¶
Every intercepted call is recorded as an Interaction on a shared Timeline owned by the StrictVerifier.
The Interaction dataclass¶
@dataclass
class Interaction:
source_id: str # e.g., "http:request" or "mock:db.query"
sequence: int # assigned atomically by Timeline.append()
details: dict[str, Any] # plugin-specific fields (method, url, args, etc.)
plugin: BasePlugin # the plugin that recorded this interaction
_asserted: bool = False # flipped to True by mark_asserted()
The source_id identifies which plugin and source produced the interaction (e.g., "http:request" for HTTP calls, "mock:db.query" for a mock named db with method query). The details dict holds the plugin-specific data that test authors assert against.
Thread-safe sequence numbering¶
The Timeline uses a threading.Lock to assign monotonically increasing sequence numbers:
class Timeline:
def append(self, interaction: Interaction) -> None:
with self._lock:
interaction.sequence = self._sequence
self._sequence += 1
self._interactions.append(interaction)
Sequence numbers establish a total ordering of all interactions across all plugins. This ordering is what assert_interaction() checks by default -- assertions must match in the order interactions were recorded.
Recording guard¶
The BasePlugin.record() method sets a _recording_in_progress ContextVar before appending to the timeline. If any code calls Timeline.mark_asserted() while recording is in progress, tripwire raises AutoAssertError. This is a runtime guard against the auto-assert anti-pattern -- plugins must never mark interactions as asserted during recording.
Mock Queues¶
Plugins use a FIFO queue pattern for mock configurations. When you register a mock, the configuration is appended to a queue. When an intercepted call matches, the first matching configuration is popped from the front of the queue.
For MockPlugin, each MethodProxy owns its own deque[MockConfig]:
class MethodProxy:
def __init__(self, ...):
self._config_queue: deque[MockConfig] = deque()
def returns(self, value: Any) -> MethodProxy:
self._config_queue.append(MockConfig(..., side_effect=_ReturnValue(value)))
return self
When the mock is called, the first config is consumed:
If the queue is empty and no wraps object is configured, UnmockedInteractionError is raised. This enforces guarantee 1: every call must be pre-authorized.
The FIFO pattern means you can chain multiple configurations to handle sequential calls:
db = tripwire.mock("db")
db.query.returns(["row1"]).returns(["row2"])
# First call returns ["row1"], second returns ["row2"]
Side effects come in three flavors: _ReturnValue (return a value), _RaiseException (raise an exception), and _CallFn (call a function with the intercepted arguments).
Assertion Model¶
Assertions happen in two phases: inline assertions during the test, and teardown verification at the end.
Inline assertions: assert_interaction()¶
When you call assert_interaction() (or a plugin helper like http.assert_request()), tripwire:
-
Finds the next unasserted interaction by peeking at the timeline. In normal mode, this is strictly the next unasserted interaction in sequence order. Inside an
in_any_order()block, it searches all unasserted interactions for a match. -
Checks source_id. The interaction's
source_idmust match the source argument'ssource_id. -
Enforces completeness. The plugin's
assertable_fields()method returns the set of fields that must appear in the assertion. Any missing field raisesMissingAssertionFieldsError. By default, every key ininteraction.detailsis assertable -- you cannot silently skip fields. -
Checks field values. The plugin's
matches()method compares expected values against actual values. If they do not match,InteractionMismatchErroris raised with a detailed hint. -
Marks asserted. If everything matches, the interaction is marked as asserted on the timeline.
Assertions are blocked inside the sandbox. Calling assert_interaction() while the sandbox is active raises AssertionInsideSandboxError. This enforces a clean separation: record inside the sandbox, assert outside.
Teardown verification: verify_all()¶
At test teardown, verify_all() enforces guarantees 2 and 3:
- Unasserted interactions: any interaction still marked
_asserted=FalseraisesUnassertedInteractionsError. - Unused mocks: each plugin's
get_unused_mocks()is called. Any mock configuration that was never consumed raisesUnusedMocksError.
If both violations exist, they are combined into a single VerificationError so you see all problems at once.
Plugin Registry¶
tripwire uses a registry to manage its built-in plugins and supports entry points for third-party plugins.
Built-in registry¶
The PLUGIN_REGISTRY tuple in _registry.py lists every built-in plugin with its metadata:
@dataclass(frozen=True)
class PluginEntry:
name: str # e.g., "http"
import_path: str # e.g., "tripwire.plugins.http"
class_name: str # e.g., "HttpPlugin"
availability_check: str # dependency check strategy
default_enabled: bool # False for opt-in plugins
Auto-discovery and availability¶
When a StrictVerifier is created, it calls resolve_enabled_plugins() to determine which plugins to instantiate. The resolution logic:
- If
enabled_pluginsis set in config, only those plugins are loaded (allowlist). - If
disabled_pluginsis set, all default-enabled plugins are loaded except those (blocklist). - If neither is set, all default-enabled plugins whose dependencies are available are loaded.
Availability is checked via the availability_check field:
| Value | Meaning |
|---|---|
"always" |
No optional dependencies; always available |
"httpx+requests" |
Multiple modules; all must be importable |
"redis" |
Single module; must be importable |
"flag:module:attr" |
Read a boolean flag from a plugin module |
Plugins whose dependencies are not installed are silently skipped -- unless they were explicitly listed in enabled_plugins, which raises TripwireConfigError.
Third-party plugins via entry points¶
After built-in plugins are loaded, tripwire discovers third-party plugins registered under the tripwire.plugins entry point group:
This allows library authors to ship tripwire plugins that activate automatically when installed.
Deduplication¶
The _register_plugin() method on StrictVerifier silently skips duplicate plugin types. If a plugin class is registered both by the built-in registry and by an entry point, only the first instance is kept.
pytest Integration¶
tripwire ships as a pytest plugin, registered via the tripwire entry point. It provides two fixtures:
_tripwire_auto_verifier (autouse)¶
This fixture runs automatically for every test. It:
- Creates a fresh
StrictVerifier(which auto-instantiates all enabled plugins). - Sets the
_current_test_verifierContextVar so module-level functions liketripwire.mock()andtripwire.http.mock_response()can find the verifier. - Yields the verifier to the test.
- On teardown, resets the ContextVar and calls
verify_all().
@pytest.fixture(autouse=True)
def _tripwire_auto_verifier() -> Generator[StrictVerifier, None, None]:
verifier = StrictVerifier()
token = _current_test_verifier.set(verifier)
yield verifier
_current_test_verifier.reset(token)
verifier.verify_all()
Because it is autouse, test authors do not need to request it. Every test gets a verifier, and every test gets verify_all() at teardown. If a test does not use tripwire at all, verify_all() is a no-op (no interactions, no mocks, nothing to verify).
tripwire_verifier (explicit)¶
For tests that need direct access to the verifier instance (e.g., to manually construct plugins), the tripwire_verifier fixture exposes the same verifier created by the autouse fixture.
The sandbox is not automatic¶
The pytest plugin creates the verifier and runs verification, but it does not activate the sandbox automatically. The test author controls sandbox lifetime with with tripwire: or async with tripwire:. This is intentional: mock registration and assertions happen outside the sandbox, and only the code under test runs inside it.