Usage¶

Interception modes¶

`mock_external_urls=False` (default)¶

The server starts on localhost but DNS is not patched. Point your client directly at m.server_url. This is the safest mode — no global state is modified.

async with aiointercept() as m:
    m.get(f"{m.server_url}/api/users", payload=[{"id": 1}])

    async with aiohttp.ClientSession() as session:
        resp = await session.get(f"{m.server_url}/api/users")
        assert resp.status == 200

Use m.server_url as a base_url to keep your code clean:

async with aiointercept() as m:
    m.get(f"{m.server_url}/api/users", payload=[{"id": 1}])

    async with aiohttp.ClientSession(base_url=m.server_url) as session:
        resp = await session.get("/api/users")

`mock_external_urls=True`¶

Patches the DNS resolver at the process level so every aiohttp request is redirected to the mock server — even those made by third-party libraries you cannot modify.

async with aiointercept(mock_external_urls=True) as m:
    m.get("https://api.stripe.com/v1/charges", payload={"data": []})

    # Code under test calls the real Stripe URL internally
    result = await billing_service.list_charges()
    assert result == []

Warning

DNS patching affects the whole process for the duration of the block. It does not intercept requests to bare IP addresses.

Registering mock responses¶

`add(url, method, ...)`¶

m.add(
    url,                    # str | yarl.URL | re.Pattern
    method="GET",           # HTTP method (case-insensitive)
    status=200,
    body=b"",               # raw response body (str is UTF-8 encoded)
    json=None,              # serialized to JSON, overrides body
    payload=None,           # alias for json
    headers=None,           # extra response headers
    content_type=None,      # overrides Content-Type
    repeat=False,           # True = infinite; int N = exactly N times
    callback=None,          # callable or coroutine -> CallbackResult
    reason=None,            # HTTP reason phrase
    exception=None,         # truthy -> close connection (ClientConnectionError)
)

HTTP method shortcuts¶

m.get(url, **kwargs)
m.post(url, **kwargs)
m.put(url, **kwargs)
m.patch(url, **kwargs)
m.delete(url, **kwargs)
m.head(url, **kwargs)
m.options(url, **kwargs)

All shortcuts forward their keyword arguments to add().

Regex patterns¶

Pass a compiled re.Pattern to match a family of URLs:

import re

pattern = re.compile(r"^https://api\.example\.com/users/\d+$")
m.get(pattern, payload={"id": 1, "name": "Alice"})

# Matches https://api.example.com/users/1, /users/42, etc.

Repeat and response queuing¶

# Respond to every request (indefinite):
m.get(url, repeat=True, payload={"ok": True})

# Respond exactly 3 times, then raise ClientConnectionError:
m.get(url, repeat=3, status=200)

# Queue different responses by calling add() multiple times:
m.post(url, status=201, payload={"created": True})
m.post(url, status=409, payload={"error": "conflict"})
# First POST -> 201, second POST -> 409, third POST -> ClientConnectionError

Callbacks¶

Use a callback when the response depends on the request:

from aiointercept import aiointercept, CallbackResult

def echo_callback(url, *, headers, query, json, **kwargs):
    return CallbackResult(status=200, payload={"echo": json})

async def test_echo():
    async with aiointercept() as m:
        m.post(f"{m.server_url}/echo", callback=echo_callback)
        ...

Async callbacks are also supported:

async def async_callback(url, **kwargs):
    await asyncio.sleep(10)
    return CallbackResult(body=b"async response")

async def test_slow():
    async with aiointercept() as m:
        m.get(f"{m.server_url}/slow", callback=async_callback)
        ...

A callback returns a CallbackResult:

Field	Type	Default	Description
`status`	`int`	`200`	Response status code
`body`	`str \| bytes`	`""`	Raw response body
`payload`	`Any`	`None`	Response body serialized to JSON (overrides `body`)
`headers`	`dict[str, str] \| None`	`None`	Extra response headers
`content_type`	`str`	`"application/json"`	`Content-Type` header
`reason`	`str \| None`	`None`	HTTP reason phrase

Passthrough¶

Let specific hosts or all unmatched requests reach the real network. Only available with mock_external_urls=True.

# Specific hosts bypass the mock:
async with aiointercept(
    mock_external_urls=True,
    passthrough=["https://real-api.example.com"],
) as m:
    m.get("https://mocked.example.com/data", payload={"mocked": True})
    # Requests to real-api.example.com go to the real server.

# All unmatched requests go to the real server:
async with aiointercept(
    mock_external_urls=True,
    passthrough_unmatched=True,
) as m:
    m.get("https://mocked.example.com/data", payload={"mocked": True})
    # Any other URL is proxied to the real network.

Inspecting requests¶

All intercepted requests are stored in m.requests, keyed by (METHOD, normalized_url):

from yarl import URL

key = ("POST", URL("https://api.example.com/orders"))
req = m.requests[key][-1]   # most recent request

req.captured_body            # raw bytes body
req.kwargs["json"]           # parsed JSON body (or None)
req.kwargs["query"]          # dict[str, list[str]] - preserves duplicate keys
req.kwargs["headers"]        # raw request headers (multidict)

URLs are normalized: fragments are stripped and query parameters are sorted.

Constructor parameters¶

Parameter	Type	Default	Description
`mock_external_urls`	`bool`	`False`	When `True`, patches the DNS resolver so external URLs are intercepted. When `False`, only requests to `m.server_url` are intercepted.
`passthrough`	`list[str] \| None`	`None`	Hosts whose requests bypass the mock and reach the real network. Requires `mock_external_urls=True`.
`passthrough_unmatched`	`bool`	`False`	Proxy all unmatched requests to the real network. Requires `mock_external_urls=True`.
`param`	`str \| None`	`None`	Kwarg name under which the mock is injected when used as a decorator.