API reference¶

The public API is re-exported from the top-level aiointercept package.

The mock¶

class aiointercept.aiointercept(mock_external_urls=False, passthrough=None, passthrough_unmatched=False, param=None, **kwargs)[source]¶

Bases: object

Mock aiohttp requests by routing them through a real test server.

Starts a real aiohttp.web.Application on localhost and directs the client’s requests to it. In the default mode you point the client at server_url; with mock_external_urls=True the DNS resolver is patched at the class level so any aiohttp request to a registered host is transparently intercepted. Register responses with add() or the per-method shortcuts (get(), post(), …), then inspect what was sent via requests and the assert_* helpers.

Use it as an async context manager, as a decorator, or drive the lifecycle explicitly with start() / stop():

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

Variables:

server_url – Base URL of the running test server, e.g. "http://127.0.0.1:54321". Set once start() has run.
server_host – Host the test server is bound to.
server_port – Port the test server is listening on.
requests – Mapping of (METHOD, normalized URL) to the list of AiointerceptRequest objects received, in arrival order.

Parameters:

mock_external_urls (bool)
passthrough (Sequence[str] | None)
passthrough_unmatched (bool)
param (str | None)
kwargs (Any)

Create a mock.

Parameters:

mock_external_urls (bool) – When True, patch the DNS resolver at the process level so requests to any registered host are intercepted, even those made by third-party libraries. When False (default), only requests sent to server_url are intercepted and no global state is modified.
passthrough (Sequence[str] | None) – Hosts whose requests bypass the mock and reach the real network. Requires mock_external_urls=True.
passthrough_unmatched (bool) – When True, proxy every unmatched request to the real network instead of failing it. Requires mock_external_urls=True.
param (str | None) – Keyword-argument name under which the mock is injected when the instance is used as a decorator. Defaults to appending it as the last positional argument.
kwargs (Any)

async start()[source]¶

Start the test server and install patches (if any).

Called automatically by __aenter__. Call it directly from a framework’s setup hook (e.g. asyncSetUp) when not using the context manager. After this returns, server_url is available. Pair every start() with a stop().

Return type:: None

async stop()[source]¶

Stop the test server and remove any DNS/SSL patches.

Called automatically by __aexit__. When DNS patching is active, the patches are only fully removed once the last concurrently-running instance stops.

Return type:: None

add(url, method='GET', status=200, body=b'', json=None, payload=None, headers=None, repeat=False, content_type=None, callback=None, reason=None, exception=None)[source]¶

Register a mock handler for url and method.

Parameters:

url (URL | str | Pattern[str]) – Target URL as str, URL, or compiled re.Pattern.
method (str) – HTTP method (case-insensitive, default GET).
status (int) – Response status code.
body (str | bytes) – Raw response body (str is UTF-8 encoded; default empty bytes).
json (Any) – Response body as a JSON-serialisable object (overrides body).
payload (Any) – Alias for json.
headers (Mapping[str, str] | None) – Additional response headers.
repeat (bool | int) – True to respond indefinitely; integer N to respond N times; False or 0 to respond once (default).
content_type (str | None) – Override the Content-Type response header.
callback (Callable[[...], CallbackResult | Awaitable[CallbackResult]] | None) – Callable (url, *, headers, query, json) → CallbackResult (sync or async). Takes precedence over body / json / status.
reason (str | None) – HTTP reason phrase.
exception (Exception | bool | None) – Any truthy value registers a handler that closes the connection, causing ClientConnectionError on the client. Passing a specific exception instance logs a warning; pass exception=True to suppress it.

Return type:

None

get(url, **kwargs)[source]¶

Register a mock GET handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

post(url, **kwargs)[source]¶

Register a mock POST handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

put(url, **kwargs)[source]¶

Register a mock PUT handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

patch(url, **kwargs)[source]¶

Register a mock PATCH handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

delete(url, **kwargs)[source]¶

Register a mock DELETE handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

head(url, **kwargs)[source]¶

Register a mock HEAD handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

options(url, **kwargs)[source]¶

Register a mock OPTIONS handler. See add() for all keyword arguments.

Parameters:

url (URL | str | Pattern[str])
kwargs (Any)

Return type:

None

clear()[source]¶

Clear all recorded requests and registered handlers.

Return type:: None

assert_called()[source]¶

Assert that at least one request was made.

Return type:: None

assert_not_called()[source]¶

Assert that no requests were made.

Return type:: None

assert_called_once()[source]¶

Assert that exactly one request was made across all URLs.

Return type:: None

assert_any_call(url, method='GET', params=None)[source]¶

Assert that url was called at least once with the given method.

Parameters:

url (URL | str)
method (str)
params (Mapping[str, str] | None)

Return type:

None

assert_called_with(url, method='GET', params=None, data=None, json=None, headers=None, strict_headers=False, **kwargs)[source]¶

Assert that the most recent call to url matched the given arguments.

Parameters:

url (URL | str) – Expected URL (str or URL).
method (str) – Expected HTTP method (default GET).
params (Mapping[str, str] | None) – Query string params merged into url before lookup.
data (str | bytes | Mapping[str, Any] | None) – Expected request body — bytes, str, or a dict (form-encoded via application/x-www-form-urlencoded).
json (Any) – Expected request body as a JSON-serialisable object.
headers (Mapping[str, str] | None) – Expected request headers. By default only the headers listed here are checked; auto-added aiohttp headers are ignored. Set strict_headers to compare the full header map.
strict_headers (bool) – When True, the complete set of actual request headers must match headers exactly. Use unittest.mock.ANY as a value to accept any value for a specific key (e.g. Content-Length).
kwargs (Any) – Ignored (present for aioresponses API compatibility).

Return type:

None

assert_called_once_with(url, method='GET', params=None, data=None, json=None, headers=None, strict_headers=False, **kwargs)[source]¶

Assert that exactly one request was made and it matched the given arguments.

Parameters:

url (URL | str)
method (str)
params (Mapping[str, str] | None)
data (str | bytes | Mapping[str, Any] | None)
json (Any)
headers (Mapping[str, str] | None)
strict_headers (bool)
kwargs (Any)

Return type:

None

Callbacks¶

class aiointercept.CallbackResult(method='GET', status=200, body='', content_type='application/json', payload=None, headers=None, response_class=None, reason=None)[source]¶

Bases: object

Result object returned by a callback.

Parameters:

method (str) – HTTP method (default GET; not used by the server handler).
status (int) – HTTP response status code.
body (str | bytes) – Raw response body as str or bytes.
content_type (str) – Content-Type header value. Set to None when headers already carries a Content-Type entry, to avoid colliding with it.
payload (Any) – Response body as a dict; serialized to JSON automatically.
headers (Mapping[str, str] | None) – Additional response headers.
response_class (type[ClientResponse] | None) – Ignored (present for aioresponses API compatibility).
reason (str | None) – HTTP reason phrase.

Requests¶

class aiointercept.AiointerceptRequest(message, payload, protocol, payload_writer, task, loop, *, client_max_size=1048576, state=None, scheme=None, host=None, remote=None)[source]¶

Bases: Request

A subclass of aiohttp.web.Request captured by the mock server.

Instances are the live request objects the test server received, augmented with the fields below so callbacks and assertions can inspect the body that was already consumed off the wire. They are stored in aiointercept.requests and passed to callbacks registered via aiointercept.add().

Variables:

captured_body (bytes) – The raw request body as bytes, read before dispatch.
kwargs (aiointercept.core.AiointerceptRequestKwargs) – A mapping with the parsed headers, query (dict of key → list of values), data (raw body or None if no body), and ``json` (decoded body, or None) — the keyword arguments a callback receives.
canonical_url (yarl.URL) – The normalized request URL with the original scheme restored (e.g. https:// even though the test server received the request over plain HTTP). This is the URL passed to callbacks and used as the aiointercept.requests key.

Parameters:

message (RawRequestMessage)
payload (StreamReader)
protocol (RequestHandler)
payload_writer (AbstractStreamWriter)
task (asyncio.Task[None])
loop (AbstractEventLoop)
client_max_size (int)
state (dict[RequestKey[Any] | str, Any] | None)
scheme (str | None)
host (str | None)
remote (str | None)

captured_body: bytes¶

kwargs: AiointerceptRequestKwargs¶

canonical_url: URL¶

classmethod upgrade(request, captured_body, kwargs, canonical_url)[source]¶

Parameters:

request (Request)
captured_body (bytes | None)
kwargs (AiointerceptRequestKwargs)
canonical_url (URL)

Return type:

AiointerceptRequest