Mock server¶

See the Mock server guide and Proxy testing guide for usage.

Services¶

asyncly.srvmocker.service.start_service `async` ¶

start_service(
    routes: Iterable[MockRoute],
    *,
    ssl_context: SSLContext | None = None,
) -> AsyncGenerator[MockService, None]

Start a real aiohttp.TestServer for the given routes.

An async context manager. On enter it binds the server to a free port and yields a MockService whose url points at it; on exit it shuts the server down.

Parameters:

Name	Type	Description	Default
`routes`	`Iterable[MockRoute]`	The routes to serve. Several routes may share a `(method, path)` and be disambiguated by their `Match`.	required
`ssl_context`	`SSLContext \| None`	If given, serve over HTTPS; `service.url` then reports `scheme="https"`.	`None`

Yields:

Name	Type	Description
`MockService`	`AsyncGenerator[MockService, None]`	Handle to register responses and assert on requests.

Example

async with start_service([MockRoute("GET", "/x", "ok")]) as service:
    service.register("ok", JsonResponse({"ok": True}))

Source code in asyncly/srvmocker/service.py

@asynccontextmanager
async def start_service(
    routes: Iterable[MockRoute],
    *,
    ssl_context: SSLContext | None = None,
) -> AsyncGenerator[MockService, None]:
    """Start a real `aiohttp.TestServer` for the given routes.

    An async context manager. On enter it binds the server to a free port and
    yields a [`MockService`][asyncly.srvmocker.MockService] whose ``url`` points
    at it; on exit it shuts the server down.

    Args:
        routes: The routes to serve. Several routes may share a
            ``(method, path)`` and be disambiguated by their
            [`Match`][asyncly.srvmocker.Match].
        ssl_context: If given, serve over HTTPS; ``service.url`` then reports
            ``scheme="https"``.

    Yields:
        MockService: Handle to register responses and assert on requests.

    Example:
        ```python
        async with start_service([MockRoute("GET", "/x", "ok")]) as service:
            service.register("ok", JsonResponse({"ok": True}))
        ```
    """
    app = Application()
    routes_list = list(routes)
    handler_names = frozenset(r.handler_name for r in routes_list)
    mock_service = MockService(
        history=list(),
        history_map=defaultdict(list),
        url=URL(),
        handlers=dict(),
        _handler_names=handler_names,
    )
    app[SERVICE_KEY] = mock_service

    groups: dict[tuple[str, str], list[MockRoute]] = defaultdict(list)
    for route in routes_list:
        groups[(route.method, route.path)].append(route)
    for (method, path), group in groups.items():
        app.router.add_route(method=method, path=path, handler=build_dispatcher(group))

    server = TestServer(app)
    await server.start_server(ssl=ssl_context)
    mock_service.set_url(server.make_url(""))
    try:
        yield mock_service
    finally:
        await server.close()

asyncly.srvmocker.proxy.start_proxy `async` ¶

start_proxy(
    *, auth: BasicAuth | None = None
) -> AsyncGenerator[MockProxyService, None]

Start an in-process forwarding HTTP proxy for tests.

An async context manager. It records every request that passes through and forwards it to the absolute target the client requested (typically another start_service), relaying the response verbatim. Only plain HTTP targets are supported (no CONNECT tunnelling).

Parameters:

Name	Type	Description	Default
`auth`	`BasicAuth \| None`	If given, require a matching `Proxy-Authorization` header. Requests without it (or with wrong credentials) get a `407 Proxy Authentication Required` and are not forwarded.	`None`

Yields:

Name	Type	Description
`MockProxyService`	`AsyncGenerator[MockProxyService, None]`	Handle exposing the proxy `url` and request history.

Source code in asyncly/srvmocker/proxy.py

@asynccontextmanager
async def start_proxy(
    *,
    auth: BasicAuth | None = None,
) -> AsyncGenerator[MockProxyService, None]:
    """Start an in-process forwarding HTTP proxy for tests.

    An async context manager. It records every request that passes through and
    forwards it to the absolute target the client requested (typically another
    [`start_service`][asyncly.srvmocker.start_service]), relaying the response
    verbatim. Only plain HTTP targets are supported (no ``CONNECT`` tunnelling).

    Args:
        auth: If given, require a matching ``Proxy-Authorization`` header.
            Requests without it (or with wrong credentials) get a
            ``407 Proxy Authentication Required`` and are not forwarded.

    Yields:
        MockProxyService: Handle exposing the proxy ``url`` and request history.
    """
    proxy = MockProxyService(url=URL())
    expected_auth = auth.encode() if auth is not None else None

    # Relay target responses verbatim (HTTP, uncompressed): keep the body and
    # its Content-Encoding untouched instead of auto-decompressing.
    forward_session = ClientSession(auto_decompress=False)

    async def _handler(request: BaseRequest) -> Response:
        body = await request.read()
        proxy.history.append(RequestHistory(request=request, body=body))

        if expected_auth is not None:
            provided = request.headers.get("Proxy-Authorization")
            if provided != expected_auth:
                return Response(
                    status=HTTPStatus.PROXY_AUTHENTICATION_REQUIRED,
                    reason="Proxy Authentication Required",
                    headers={"Proxy-Authenticate": "Basic"},
                )

        async with forward_session.request(
            method=request.method,
            url=request.url,
            headers=_relay_headers(request.headers, drop=_HOP_BY_HOP),
            data=body or None,
            allow_redirects=False,
        ) as upstream:
            payload = await upstream.read()
            return Response(
                status=upstream.status,
                reason=upstream.reason,
                headers=_relay_headers(upstream.headers, drop=_HOP_BY_HOP_RESPONSE),
                body=payload,
            )

    server = RawTestServer(_handler)
    try:
        await server.start_server()
        proxy.set_url(server.make_url(""))
        yield proxy
    finally:
        await server.close()
        await forward_session.close()

asyncly.srvmocker.models.MockService `dataclass` ¶

MockService(
    history: MutableSequence[RequestHistory],
    history_map: MutableMapping[
        str, MutableSequence[RequestHistory]
    ],
    url: URL,
    handlers: MutableMapping[str, BaseMockResponse],
    _handler_names: frozenset[str] = frozenset(),
)

Handle to a running mock server.

Returned by start_service. Use it to register responses, read the recorded request history, and assert on what a client sent. The url attribute is the base URL to point a client at.

register ¶

register(name: str, resp: BaseMockResponse) -> None

Register (or replace) the response returned for a handler name.

Parameters:

Name	Type	Description	Default
`name`	`str`	A `handler_name` declared on one of the routes.	required
`resp`	`BaseMockResponse`	The response to return. May be re-registered mid-test to change behavior.	required

Source code in asyncly/srvmocker/models.py

def register(self, name: str, resp: BaseMockResponse) -> None:
    """Register (or replace) the response returned for a handler name.

    Args:
        name: A ``handler_name`` declared on one of the routes.
        resp: The response to return. May be re-registered mid-test to
            change behavior.
    """
    if self._handler_names and name not in self._handler_names:
        import warnings

        warnings.warn(
            f"register() called with unknown handler_name {name!r}; "
            f"declared handlers: {sorted(self._handler_names)}. "
            "This will raise UnknownHandlerError in asyncly 0.7.",
            DeprecationWarning,
            stacklevel=2,
        )
    self.handlers[name] = resp

get_calls ¶

get_calls(name: str) -> list[RequestHistory]

Return all recorded calls for a handler, oldest first.

Source code in asyncly/srvmocker/models.py

def get_calls(self, name: str) -> list[RequestHistory]:
    """Return all recorded calls for a handler, oldest first."""
    return list(self.history_map.get(name, []))

last_call ¶

last_call(name: str) -> RequestHistory

Return the most recent call for a handler.

Raises:

Type	Description
`AssertionError`	If the handler recorded no calls.

Source code in asyncly/srvmocker/models.py

def last_call(self, name: str) -> RequestHistory:
    """Return the most recent call for a handler.

    Raises:
        AssertionError: If the handler recorded no calls.
    """
    calls = self.history_map.get(name) or []
    if not calls:
        raise AssertionError(f"no calls recorded for handler {name!r}")
    return calls[-1]

assert_called ¶

assert_called(
    name: str,
    *,
    times: int | None = None,
    json: object = None,
    body: bytes | None = None,
    headers: dict[str, str] | None = None,
    query: dict[str, str] | None = None,
) -> None

Assert that a matching request was received.

With only times, asserts the exact call count. With predicates, asserts at least one recorded call matched all of them: json and body match exactly, while headers and query match as a subset.

Parameters:

Name	Type	Description	Default
`name`	`str`	The handler name to check.	required
`times`	`int \| None`	If given, the exact number of calls expected.	`None`
`json`	`object`	Expected parsed JSON body (exact match).	`None`
`body`	`bytes \| None`	Expected raw body bytes (exact match).	`None`
`headers`	`dict[str, str] \| None`	Header key/values that must all be present.	`None`
`query`	`dict[str, str] \| None`	Query key/values that must all be present.	`None`

Raises:

Type	Description
`AssertionError`	If no recorded call satisfies the criteria.

Source code in asyncly/srvmocker/models.py

def assert_called(
    self,
    name: str,
    *,
    times: int | None = None,
    json: object = None,
    body: bytes | None = None,
    headers: dict[str, str] | None = None,
    query: dict[str, str] | None = None,
) -> None:
    """Assert that a matching request was received.

    With only ``times``, asserts the exact call count. With predicates,
    asserts at least one recorded call matched all of them: ``json`` and
    ``body`` match exactly, while ``headers`` and ``query`` match as a
    subset.

    Args:
        name: The handler name to check.
        times: If given, the exact number of calls expected.
        json: Expected parsed JSON body (exact match).
        body: Expected raw body bytes (exact match).
        headers: Header key/values that must all be present.
        query: Query key/values that must all be present.

    Raises:
        AssertionError: If no recorded call satisfies the criteria.
    """
    from asyncly.srvmocker.assertions import call_matches

    calls = self.get_calls(name)
    if times is not None:
        if len(calls) != times:
            raise AssertionError(
                f"handler {name!r}: expected {times} call(s), got {len(calls)}"
            )
        if all(v is None for v in (json, body, headers, query)):
            return
    if not calls:
        raise AssertionError(f"handler {name!r}: expected at least one call, got 0")
    for call in calls:
        if call_matches(call, json=json, body=body, headers=headers, query=query):
            return
    raise AssertionError(
        f"handler {name!r}: none of {len(calls)} call(s) matched the given criteria"
    )

assert_not_called ¶

assert_not_called(name: str) -> None

Assert the handler received no requests.

Raises:

Type	Description
`AssertionError`	If any call was recorded for `name`.

Source code in asyncly/srvmocker/models.py

def assert_not_called(self, name: str) -> None:
    """Assert the handler received no requests.

    Raises:
        AssertionError: If any call was recorded for ``name``.
    """
    calls = self.get_calls(name)
    if calls:
        raise AssertionError(
            f"handler {name!r}: expected no calls, got {len(calls)}"
        )

asyncly.srvmocker.proxy.MockProxyService `dataclass` ¶

MockProxyService(
    url: URL, history: list[RequestHistory] = list()
)

Handle to a running mock forwarding proxy.

Returned by start_proxy. Point a client at it via proxy=proxy.url and assert on the requests that passed through. The history holds every request the proxy received.

assert_called ¶

assert_called(
    *,
    times: int | None = None,
    target: URL | str | None = None,
    method: str | None = None,
    json: object = None,
    body: bytes | None = None,
    headers: dict[str, str] | None = None,
    query: dict[str, str] | None = None,
) -> None

Assert that a matching request passed through the proxy.

Like MockService.assert_called, plus two proxy-specific predicates:

Parameters:

Name	Type	Description	Default
`times`	`int \| None`	If given, the exact number of forwarded requests expected.	`None`
`target`	`URL \| str \| None`	Expected absolute destination URL the client asked for.	`None`
`method`	`str \| None`	Expected HTTP method.	`None`
`json`	`object`	Expected parsed JSON body (exact match).	`None`
`body`	`bytes \| None`	Expected raw body bytes (exact match).	`None`
`headers`	`dict[str, str] \| None`	Header key/values that must all be present (e.g. `Proxy-Authorization`).	`None`
`query`	`dict[str, str] \| None`	Query key/values that must all be present.	`None`

Raises:

Type	Description
`AssertionError`	If no recorded request satisfies the criteria.

Source code in asyncly/srvmocker/proxy.py

def assert_called(
    self,
    *,
    times: int | None = None,
    target: URL | str | None = None,
    method: str | None = None,
    json: object = None,
    body: bytes | None = None,
    headers: dict[str, str] | None = None,
    query: dict[str, str] | None = None,
) -> None:
    """Assert that a matching request passed through the proxy.

    Like `MockService.assert_called`, plus two proxy-specific predicates:

    Args:
        times: If given, the exact number of forwarded requests expected.
        target: Expected absolute destination URL the client asked for.
        method: Expected HTTP method.
        json: Expected parsed JSON body (exact match).
        body: Expected raw body bytes (exact match).
        headers: Header key/values that must all be present (e.g.
            ``Proxy-Authorization``).
        query: Query key/values that must all be present.

    Raises:
        AssertionError: If no recorded request satisfies the criteria.
    """
    calls = self.get_calls()
    if times is not None and len(calls) != times:
        raise AssertionError(f"proxy: expected {times} call(s), got {len(calls)}")
    criteria = (target, method, json, body, headers, query)
    if times is not None and all(v is None for v in criteria):
        return
    if not calls:
        raise AssertionError("proxy: expected at least one call, got 0")
    for call in calls:
        if target is not None and str(call.request.url) != str(URL(target)):
            continue
        if method is not None and call.request.method != method:
            continue
        if call_matches(call, json=json, body=body, headers=headers, query=query):
            return
    raise AssertionError(
        f"proxy: none of {len(calls)} call(s) matched the given criteria"
    )

Routes & history¶

asyncly.srvmocker.models.MockRoute `dataclass` ¶

MockRoute(
    method: str,
    path: str,
    handler_name: str,
    match: Match | None = None,
)

Binds an HTTP method and path to a named handler.

Attributes:

Name	Type	Description
`method`	`str`	HTTP method, e.g. `"GET"`.
`path`	`str`	URL path, e.g. `"/items"`.
`handler_name`	`str`	Label a response is registered under via `MockService.register`.
`match`	`Match \| None`	Optional `Match`; when several routes share a `(method, path)`, the first whose match succeeds wins. A route with no `match` is an always-matching fallback.

asyncly.srvmocker.models.RequestHistory `dataclass` ¶

RequestHistory(request: BaseRequest, body: bytes)

A single recorded request.

Attributes:

Name	Type	Description
`request`	`BaseRequest`	The aiohttp request object (headers, query, url, method).
`body`	`bytes`	The raw request body bytes.

Matching¶

asyncly.srvmocker.matching.Match `dataclass` ¶

Match(
    json: Any = None,
    body: bytes | None = None,
    headers: dict[str, str] | None = None,
    query: dict[str, str] | None = None,
)

Request matcher attached to a MockRoute.

json: parsed JSON body must equal this value exactly. body: raw body bytes must equal this value exactly. headers: every (key, value) here must be present in request headers (subset). query: every (key, value) here must be present in request query (subset).

Match is value-immutable: headers and query dicts are defensively copied at construction so caller-side mutation cannot affect matcher behavior. The instances themselves remain unhashable due to dict fields -- wrap in a tuple of items if you need hashability.

Exceptions¶

asyncly.srvmocker.exceptions.SrvMockerError ¶

Bases: Exception

Base exception for srvmocker.

asyncly.srvmocker.exceptions.SequenceExhausted ¶

Bases: SrvMockerError

Raised by SequenceResponse when responses run out and on_exhausted='raise'.

asyncly.srvmocker.exceptions.UnknownHandlerError ¶

Bases: SrvMockerError

Raised when register() is called with a name not declared in any MockRoute.

Mock server¶

Services¶

asyncly.srvmocker.service.start_service async ¶

asyncly.srvmocker.proxy.start_proxy async ¶

asyncly.srvmocker.models.MockService dataclass ¶

register ¶

get_calls ¶

last_call ¶

assert_called ¶

assert_not_called ¶

asyncly.srvmocker.proxy.MockProxyService dataclass ¶

assert_called ¶

Routes & history¶

asyncly.srvmocker.models.MockRoute dataclass ¶

asyncly.srvmocker.models.RequestHistory dataclass ¶

Matching¶

asyncly.srvmocker.matching.Match dataclass ¶

Exceptions¶

asyncly.srvmocker.exceptions.SrvMockerError ¶

asyncly.srvmocker.exceptions.SequenceExhausted ¶

asyncly.srvmocker.exceptions.UnknownHandlerError ¶

asyncly.srvmocker.service.start_service `async` ¶

asyncly.srvmocker.proxy.start_proxy `async` ¶

asyncly.srvmocker.models.MockService `dataclass` ¶

asyncly.srvmocker.proxy.MockProxyService `dataclass` ¶

asyncly.srvmocker.models.MockRoute `dataclass` ¶

asyncly.srvmocker.models.RequestHistory `dataclass` ¶

asyncly.srvmocker.matching.Match `dataclass` ¶