interposition

Protocol-agnostic interaction interposition with lifecycle hooks for record, replay, and control.

Overview

Interposition is a Python library for replaying recorded interactions. Unlike VCRpy or other HTTP-specific tools, Interposition does not automatically hook into network libraries.

Instead, it provides a pure logic engine for storage, matching, and replay. You write the adapter for your specific target (HTTP client, database driver, IoT message handler), and Interposition handles the rest.

Key Features:

• Protocol-agnostic: Works with any protocol (HTTP, gRPC, SQL, Pub/Sub, etc.)
• Type-safe: Full mypy strict mode support with Pydantic v2
• Immutable: All data structures are frozen Pydantic models
• Serializable: Built-in JSON/YAML serialization for cassette persistence
• Memory-efficient: O(1) lookup with fingerprint indexing
• Streaming: Generator-based response delivery
• Multi-mode: Supports replay, record, and auto modes

Architecture

Interposition sits behind your application's data access layer. You provide the "Adapter" that captures live traffic or requests replay from the Broker.

+-------------+      +------------------+      +---------------+
| Application | <--> | Your Adapter     | <--> | Interposition |
+-------------+      +------------------+      +---------------+
                            |                          |
                       (Traps calls)              (Manages)
                                                       |
                                                  [Cassette]

Installation

pip install interposition

Practical Integration (Pytest Recipe)

The most common use case is using Interposition as a test fixture. Here is a production-ready recipe for pytest:

import pytest
from interposition import Broker, Cassette, InteractionRequest

@pytest.fixture
def cassette_broker():
    # Load cassette from a JSON file (or create one programmatically)
    with open("tests/fixtures/my_cassette.json", "rb") as f:
        cassette = Cassette.model_validate_json(f.read())
    return Broker(cassette)

def test_user_service(cassette_broker, monkeypatch):
    # 1. Create your adapter (mocking your actual client)
    def mock_fetch(url):
        request = InteractionRequest(
            protocol="http",
            action="GET",
            target=url,
            headers=(),
            body=b"",
        )
        # Delegate to Interposition
        chunks = list(cassette_broker.replay(request))
        return chunks[0].data

    # 2. Inject the adapter
    monkeypatch.setattr("my_app.client.fetch", mock_fetch)

    # 3. Run your test
    from my_app import get_user_name
    assert get_user_name(42) == "Alice"

Protocol-Agnostic Examples

Interposition shines where HTTP-only tools fail.

SQL Database Query

request = InteractionRequest(
    protocol="postgres",
    action="SELECT",
    target="users_table",
    headers=(),
    body=b"SELECT id, name FROM users WHERE id = 42",
)
# Replay returns: b'[(42, "Alice")]'

MQTT / PubSub Message

request = InteractionRequest(
    protocol="mqtt",
    action="subscribe",
    target="sensors/temp/room1",
    headers=(("qos", "1"),),
    body=b"",
)
# Replay returns stream of messages: b'24.5', b'24.6', ...

Usage Guide

Manual Construction (Quick Start)

If you need to build interactions programmatically (e.g., for seeding tests):

from interposition import (
    Broker,
    Cassette,
    Interaction,
    InteractionRequest,
    ResponseChunk,
)

# 1. Define the Request
request = InteractionRequest(
    protocol="api",
    action="query",
    target="users/42",
    headers=(),
    body=b"",
)

# 2. Define the Response
chunks = (
    ResponseChunk(data=b'{"id": 42, "name": "Alice"}', sequence=0),
)

# 3. Create Interaction & Cassette
interaction = Interaction(
    request=request,
    fingerprint=request.fingerprint(),
    response_chunks=chunks,
)
cassette = Cassette(interactions=(interaction,))

# 4. Replay
broker = Broker(cassette=cassette)
response = list(broker.replay(request))

Persistence & Serialization

Interposition models are Pydantic v2 models, making serialization trivial.

# Save to JSON
with open("cassette.json", "w") as f:
    f.write(cassette.model_dump_json(indent=2))

# Load from JSON
with open("cassette.json") as f:
    cassette = Cassette.model_validate_json(f.read())

# Generate JSON Schema
schema = Cassette.model_json_schema()

Streaming Responses

For large files or streaming protocols, responses are yielded lazily:

# The broker returns a generator
for chunk in broker.replay(request):
    print(f"Received chunk: {len(chunk.data)} bytes")

Broker Modes

The Broker supports three modes via the mode parameter:

The BrokerMode type alias is available for type hints:

from interposition import BrokerMode

mode: BrokerMode = "auto"

Live Responder

For record and auto modes, you must provide a live_responder callable that forwards requests to your actual backend:

from interposition import (
    Broker,
    Cassette,
    InteractionRequest,
    ResponseChunk,
)
from collections.abc import Iterable

def my_live_responder(request: InteractionRequest) -> Iterable[ResponseChunk]:
    """Forward request to actual backend and yield response chunks."""
    # Your actual implementation here
    response = your_http_client.request(
        method=request.action,
        url=request.target,
        headers=dict(request.headers),
        data=request.body,
    )
    yield ResponseChunk(data=response.content, sequence=0)

The LiveResponder type alias is available:

from interposition.services import LiveResponder

Record Mode

Use record mode to capture new interactions:

# Start with empty cassette
cassette = Cassette(interactions=())

broker = Broker(
    cassette=cassette,
    mode="record",
    live_responder=my_live_responder,
)

# All requests are forwarded and recorded
response = list(broker.replay(request))

# Save the updated cassette
with open("cassette.json", "w") as f:
    f.write(broker.cassette.model_dump_json(indent=2))

Auto Mode

Use auto mode for hybrid workflows (replay if available, record if not):

# Load existing cassette (may be empty or partial)
with open("cassette.json") as f:
    cassette = Cassette.model_validate_json(f.read())

broker = Broker(
    cassette=cassette,
    mode="auto",
    live_responder=my_live_responder,
)

# Returns recorded response if exists, otherwise forwards and records
response = list(broker.replay(request))

Cassette Store

For automatic cassette persistence during recording, use a CassetteStore. The CassetteStore protocol defines a simple interface for loading and saving cassettes:

from interposition import CassetteStore

class MyCassetteStore:
    """Custom store implementation."""

    def load(self) -> Cassette:
        """Load cassette from storage."""
        ...

    def save(self, cassette: Cassette) -> None:
        """Save cassette to storage."""
        ...

When a cassette_store is provided to the Broker, it automatically saves the cassette after each recorded interaction.

JsonFileCassetteStore

A built-in file-based cassette store using JSON format:

from pathlib import Path
from interposition import Broker, Cassette, JsonFileCassetteStore

# Create store pointing to a JSON file
store = JsonFileCassetteStore(Path("cassettes/my_test.json"))

# Load existing cassette (raises CassetteLoadError if not exists)
cassette = store.load()

# Or start with empty cassette
cassette = Cassette(interactions=())

# Create broker with automatic persistence
broker = Broker(
    cassette=cassette,
    mode="record",
    live_responder=my_live_responder,
    cassette_store=store,  # Auto-saves after each recording
)

# After replay, cassette is automatically saved to file
response = list(broker.replay(request))

By default, load() raises CassetteLoadError if the file doesn't exist. Use create_if_missing=True to return an empty cassette instead — useful for record/auto workflows where the file is created on first save:

store = JsonFileCassetteStore(
    Path("cassettes/my_test.json"),
    create_if_missing=True,
)
cassette = store.load()  # Returns empty Cassette if file doesn't exist

You can also use Broker.from_store() to load the cassette and create a broker in one step:

store = JsonFileCassetteStore(
    Path("cassettes/my_test.json"),
    create_if_missing=True,
)
broker = Broker.from_store(store, mode="auto", live_responder=my_live_responder)

The JsonFileCassetteStore creates parent directories automatically when saving. If saving fails, the error is propagated and response streaming stops (fail-fast).

Error Handling

All interposition exceptions inherit from InterpositionError, allowing you to catch all domain errors with a single handler:

from interposition import InterpositionError

try:
    broker.replay(request)
except InterpositionError as e:
    print(f"Interposition error: {e}")

InteractionNotFoundError: Raised when no matching interaction exists in replay mode:

from interposition import InteractionNotFoundError

try:
    broker.replay(unknown_request)
except InteractionNotFoundError as e:
    print(f"Not recorded: {e.request.target}")

LiveResponderRequiredError: Raised at broker construction when record or auto mode is used without a live_responder:

from interposition import LiveResponderRequiredError

try:
    Broker(cassette=cassette, mode="auto")  # No live_responder!
except LiveResponderRequiredError as e:
    print(f"live_responder required for {e.mode} mode")

InteractionValidationError: Raised when an Interaction fails validation (e.g., fingerprint mismatch or invalid response chunk sequence):

from interposition import Interaction, InteractionValidationError

try:
    # This will fail: fingerprint doesn't match request
    interaction = Interaction(
        request=request,
        fingerprint=wrong_fingerprint,  # Mismatch!
        response_chunks=chunks,
    )
except InteractionValidationError as e:
    print(f"Validation failed: {e}")

CassetteLoadError: Raised when JsonFileCassetteStore.load() fails (file not found, permission denied, corrupted JSON, etc.):

from pathlib import Path
from interposition import CassetteLoadError, JsonFileCassetteStore

store = JsonFileCassetteStore(Path("cassettes/missing.json"))

try:
    cassette = store.load()
except CassetteLoadError as e:
    print(f"Failed to load from {e.path}: {e.__cause__}")

CassetteSaveError: Raised when JsonFileCassetteStore.save() fails due to I/O errors (permission denied, disk full, etc.):

from pathlib import Path
from interposition import CassetteSaveError, JsonFileCassetteStore

store = JsonFileCassetteStore(Path("/readonly/cassette.json"))

try:
    store.save(cassette)
except CassetteSaveError as e:
    print(f"Failed to save to {e.path}: {e.__cause__}")

Version

Access the package version programmatically:

from interposition import __version__

if __version__ < "0.2.0":
    print("Auto mode is not supported")
else:
    print("Auto mode is supported")

Development

Prerequisites

• Python 3.10+
• uv (recommended)

Setup & Testing

# Clone and install
git clone https://github.com/osoekawaitlab/interposition.git
cd interposition
uv pip install -e . --group=dev

# Run tests
nox -s tests

License

MIT