httpx_mock

Mock for httpx.AsyncClient and httpx.Client in tests. Supports routing by URL, headers, query params, custom matchers, response sequences, network errors, and stateful mock services.

httpx_mock can be used as a small declarative helper for one-off endpoint mocks, but its main strength is the class-based Router: you can write realistic in-memory services with internal state, conditional branching, and multi-step workflows directly in your tests.

Why httpx_mock?

• Stateful Router classes: model real mock services with mutable state, request-dependent behavior, and workflow transitions.
• Expressive matchers: compose routing rules with & and |, for example Path("/users") & Params(active=1).
• Low boilerplate for simple tests: use inline dicts for static responses, sequences, or transport errors.
• Sync and async support: patch both httpx.Client and httpx.AsyncClient.
• Request assertions: inspect request history with assert_requested_once, assert_request_count, and raw t.requests.

For more involved patterns - combined matchers on a multi-route Router, lambda matchers/handlers, pytest.mark.parametrize together with the function decorator, and the isolation trade-offs of Router classes vs. instances - see the runnable examples in tests/examples/test_examples_router.py, tests/examples/test_examples_lambda.py, tests/examples/test_examples_parametrize.py, and tests/examples/test_examples_shared_router_state.py.

Table of contents

• Requirements
• Installation
• Stateful routers
• Router state isolation
• Matchers
• Usage
• Request history and assertions
• Fallback: nested mocks, passthrough, and strict
• License

Requirements

• Python >= 3.12
• httpx >= 0.27

Installation

pip install httpx-mock

Import:

import httpx
from httpx_mock import httpx_mock
from httpx_mock.matcher import Path, Exact, Regex, StartsWith, Host, Headers, Params, AnyMatcher
from httpx_mock.provider import Router

---

Stateful routers

Use a Router class when a test needs a mock service, not just a static response. Router instances can keep state between requests made inside the same mock context, which is useful for CRUD flows, polling APIs, retries, task state machines, and tests that need conditional responses.

import json

import httpx

from httpx_mock import httpx_mock
from httpx_mock.matcher import AnyMatcher, Path
from httpx_mock.provider import Router


class UserTaskRouter(Router):
    """Mock service with state stored inside the router instance."""

    def __init__(self) -> None:
        super().__init__()
        self.tasks: dict[str, str] = {}
        self.history: list[httpx.Request] = []

    @Router.route.post(Path("/tasks"))
    async def create_task(self, request: httpx.Request):
        self.history.append(request)
        payload = json.loads(request.content or b"{}")
        task_id = f"task_{len(self.tasks) + 1}"
        self.tasks[task_id] = "processing"
        return httpx.Response(201, json={
            "id": task_id,
            "status": "processing",
            "name": payload.get("name"),
        })

    @Router.route.get(Path("/tasks/{task_id}"))
    async def get_task_status(self, request: httpx.Request):
        self.history.append(request)
        task_id = request.url.path.rsplit("/", 1)[-1]
        status = self.tasks.get(task_id)

        if status is None:
            return httpx.Response(404)

        # Simulate polling: the first read returns the old state and advances it.
        self.tasks[task_id] = "completed"
        return {"id": task_id, "status": status}


@httpx_mock(UserTaskRouter)
async def test_task_polling_flow():
    async with httpx.AsyncClient(base_url="http://api") as client:
        created = await client.post("/tasks", json={"name": "import-users"})
        first_poll = await client.get(f"/tasks/{created.json()['id']}")
        second_poll = await client.get(f"/tasks/{created.json()['id']}")

    assert created.status_code == 201
    assert first_poll.json()["status"] == "processing"
    assert second_poll.json()["status"] == "completed"

Router handlers may be def or async def. They can either accept the current httpx.Request or take no arguments. Path parameters are matched by Path("/tasks/{task_id}"), but they are not injected as separate handler arguments; read them from request.url.path when you need them.

When more than one route could match the same request (overlapping matchers, or a specific route plus a catch-all), the one declared first - top to bottom in the class body - wins, regardless of the methods' names.

Handler return values are normalized for convenience:

Return value	Response
httpx.Response(...)	returned as-is
None	httpx.Response(200)
dict or list	httpx.Response(200, json=value)
str	httpx.Response(200, content=value.encode())
bytes	httpx.Response(200, content=value)

If a handler raises an exception, the test sees that exception. This keeps broken test doubles visible instead of hiding them behind a fake HTTP response. Return httpx.Response(500, ...) explicitly when the service under test should receive a 500 response.

---

Router state isolation

State isolation depends on whether you pass a Router class or a Router instance.

Isolated state by default:

Passing the class creates a fresh router instance every time httpx_mock starts. This is the safest default for independent tests and parametrized cases.

@httpx_mock(UserTaskRouter)
async def test_a():
    ...

@httpx_mock(UserTaskRouter)
async def test_b():
    ...

Each test gets a clean UserTaskRouter.

Idiomatic pytest fixtures:

Use a function-scoped fixture when each test needs a clean service:

@pytest.fixture
async def service_router():
    router = UserTaskRouter()
    async with httpx_mock(router):
        yield router


async def test_user_creation(service_router: UserTaskRouter):
    async with httpx.AsyncClient(base_url="http://api") as client:
        await client.post("/tasks", json={"name": "Alice"})

    assert service_router.tasks

Use an explicit broader scope only when the shared state is part of the scenario:

@pytest.fixture(scope="class")
def stateful_service_router():
    return UserTaskRouter()


class TestRegistrationFlow:
    async def test_1_create_task(self, stateful_service_router):
        async with httpx_mock(stateful_service_router):
            async with httpx.AsyncClient(base_url="http://api") as client:
                await client.post("/tasks", json={"name": "signup"})

    async def test_2_poll_task(self, stateful_service_router):
        async with httpx_mock(stateful_service_router):
            async with httpx.AsyncClient(base_url="http://api") as client:
                response = await client.get("/tasks/task_1")

        assert response.status_code == 200

This shares the UserTaskRouter instance, but each test still starts and stops its own httpx_mock(...) context. If you want the mock itself to cover the whole fixture scope, wrap httpx_mock(...) inside the fixture and yield the transport:

@pytest.fixture(scope="class")
async def stateful_router_transport():
    async with httpx_mock(UserTaskRouter()) as transport:
        yield transport


class TestRegistrationFlow:
    async def test_1_create_task(self, stateful_router_transport: MockTransport):
        async with httpx.AsyncClient(base_url="http://api") as client:
            await client.post("/tasks", json={"name": "signup"})

    async def test_2_poll_task(self, stateful_router_transport: MockTransport):
        async with httpx.AsyncClient(base_url="http://api") as client:
            response = await client.get("/tasks/task_1")

        assert response.status_code == 200

The first version makes the shared router lifetime explicit while keeping the mock context local to each test. The second version makes the fixture own the mock context for its whole scope, so the test body is already inside the mock; it also shares the returned MockTransport and its request history.

For synchronous tests, use the normal context manager:

@pytest.fixture
def service_router_sync():
    router = UserTaskRouter()
    with httpx_mock(router):
        yield router

About shared Router instances:

Passing an instance reuses that exact object, so its mutable state is preserved across every mock context that receives the same instance. Prefer pytest fixtures for this, because the fixture scope makes the lifetime of shared state explicit.

Avoid keeping a module-level router and decorating unrelated tests with it:

# Avoid: hidden shared state between tests.
shared_router = UserTaskRouter()


@httpx_mock(shared_router)
async def test_a():
    ...


@httpx_mock(shared_router)
async def test_b():
    ...

Use a fixture with an explicit scope instead.

---

Matchers

Matcher is the library's central concept. It describes a condition a request must satisfy. Matchers are used everywhere: in route keys, in Router.route, and in assertion methods.

Matcher	Condition
Path("/users/{id}")	path matches the template (FastAPI-style)
Exact("http://api/users")	full URL matches exactly
StartsWith("http://api")	URL starts with the prefix
Regex(r"/users/\d+")	URL matches the regular expression
Host("api.example.com")	matches by host
Headers(x_role="admin")	request contains the given headers
Params(active=1, page=2)	request contains the given query params
AnyMatcher()	any request

Besides Matcher instances, anywhere a matcher is accepted you can pass a string (becomes Exact), a re.Pattern (becomes Regex), or a callable (request) -> bool.

Combining with & and |:

# AND - all conditions must match
Path("/users/{id}") & Params(active=1) & Headers(x_role="admin")

# OR - at least one must match
Path("/users") | Path("/accounts")

# Combination
(Path("/users") | Path("/accounts")) & Headers(x_role="admin")

---

Usage

# Function decorator
@httpx_mock(get={Path("/users/{id}"): httpx.Response(200, json={"id": 1})})
async def test_get_user():
    async with httpx.AsyncClient(base_url="http://api") as c:
        r = await c.get("/users/1")
    assert r.status_code == 200

# Async context manager
async def test_something():
    async with httpx_mock(post={Path("/login"): httpx.Response(200, json={"token": "abc"})}):
        async with httpx.AsyncClient(base_url="http://api") as c:
            r = await c.post("/login")

# Sync context manager
def test_something_sync():
    with httpx_mock(get={Path("/ping"): httpx.Response(200)}):
        with httpx.Client(base_url="http://api") as c:
            r = c.get("/ping")

# Function decorator on a sync test
@httpx_mock(get={Path("/ping"): httpx.Response(200)})
def test_get_ping_sync():
    with httpx.Client(base_url="http://api") as c:
        r = c.get("/ping")
    assert r.status_code == 200

# Class decorator (each test_* gets an isolated client; sync and async test
# methods can be mixed freely in the same class)
@httpx_mock(get={Path("/ping"): httpx.Response(200)})
class TestMyService:
    async def test_a(self): ...
    def test_b(self): ...

Supported method buckets are get, post, put, patch, delete, routers, and any. The any bucket matches any HTTP method:

@httpx_mock(any={Path("/ping"): httpx.Response(200, text="pong")})
async def test_ping_any_method():
    async with httpx.AsyncClient(base_url="http://api") as client:
        assert (await client.get("/ping")).text == "pong"
        assert (await client.delete("/ping")).text == "pong"

Within a request, routes are resolved in this order: routers, then the concrete request method (get, post, ...), then any. If no route matches (and there's no fallback - see Fallback: nested mocks, passthrough, and strict), the mock raises AssertionError by default; pass unmatched_404=True to get httpx.Response(404, json={"message": "Not found"}) instead.

When httpx_mock is active, it patches httpx.Client and httpx.AsyncClient to use the mock transport. If the client is created with its own transport=..., that transport is replaced by the mock transport for the duration of the context or decorated function.

RouteMap vs. HandlerLike:

Every httpx_mock(...) argument accepts either an explicit RouteMap or a single HandlerLike shorthand.

A RouteMap is a dict where each key is a matcher and each value is the handler for that route:

@httpx_mock(get={
    Path("/users/{id}"): httpx.Response(200, json={"id": 1}),
    Path("/health"): httpx.Response(200, json={"ok": True}),
})
async def test_with_route_map(): ...

A HandlerLike is one route value passed directly. httpx_mock attaches it to AnyMatcher() for that bucket:

# Same response for any GET request
@httpx_mock(get=httpx.Response(200, json={"ok": True}))
async def test_with_method_shorthand(): ...

# Same router for all requests
@httpx_mock(UserTaskRouter)
async def test_with_router_shorthand(): ...

In other words, these two forms are equivalent:

@httpx_mock(UserTaskRouter)
async def test_short(): ...

@httpx_mock(routers={AnyMatcher(): UserTaskRouter})
async def test_explicit(): ...

Callable and provider results are wrapped:

Passing a plain function/lambda, a Router (class or instance), or any other ResponseProvider as a handler does not send its return value to the client as-is - it's normalized the same way as a Router method's return value (see the table in Stateful routers): None becomes httpx.Response(200), a dict/list becomes a JSON 200, str/bytes becomes the response body, and an httpx.Response is returned unchanged.

@httpx_mock(get={Path("/ping"): lambda request: "pong"})
async def test_function_provider_is_wrapped_in_a_response():
    async with httpx.AsyncClient(base_url="http://api") as c:
        r = await c.get("/ping")
    assert r.status_code == 200
    assert r.text == "pong"

If you need to control the status code or headers, return an httpx.Response(...) explicitly instead of a bare value.

Response sequences - a list is returned in order, the last item repeats:

@httpx_mock(get={Path("/data"): [
    httpx.Response(200, json={"page": 1}),
    httpx.Response(200, json={"page": 2}),
    httpx.Response(404),
]})
async def test_pagination(): ...

Emulating network errors - pass an httpx.TransportError instance instead of a response:

# A single error for the whole route
@httpx_mock(get={Path("/api"): httpx.TimeoutException("timeout")})
async def test_timeout():
    with pytest.raises(httpx.TimeoutException):
        async with httpx.AsyncClient(base_url="http://api") as c:
            await c.get("/api")

# First request errors, second succeeds (for testing retries)
@httpx_mock(get={Path("/api"): [
    httpx.ConnectError("connection refused"),
    httpx.Response(200, json={"ok": True}),
]})
async def test_retry(): ...

# In a Router - via raise
class SupportRouter(Router):
    @Router.route.get(Path("/flaky"))
    async def flaky(self, request: httpx.Request):
        raise httpx.ReadTimeout("read timed out", request=request)


@httpx_mock(SupportRouter)
async def test_router_handler_can_raise_a_transport_error():
    with pytest.raises(httpx.ReadTimeout):
        async with httpx.AsyncClient(base_url="http://api") as client:
            await client.get("/flaky")

Supported error classes from httpx: TimeoutException, ConnectTimeout, ReadTimeout, WriteTimeout, PoolTimeout, ConnectError, ReadError, WriteError, RemoteProtocolError.

Sync vs. async handlers:

A Router method can be def or async def - pick whichever fits the handler's body. A bare callable/lambda handler (get={Path("/x"): lambda request: httpx.Response(200)}) is always called synchronously, never awaited.

• httpx.AsyncClient works with both sync and async handlers.
• httpx.Client (sync) only works with sync handlers. An async Router handler invoked through the sync client raises RuntimeError immediately. It does not try to bridge it by spinning up a second event loop in a thread, since that can deadlock if the handler awaits an asyncio primitive (Lock/Event/Queue/...) shared with an outer event loop. If a route needs to work with both client types, give it a sync handler.

---

Request history and assertions

The context manager returns a MockTransport; the function decorator accepts a transport parameter:

# Context manager
async def test_something():
    async with httpx_mock(get={Path("/users"): httpx.Response(200, json=[])}) as t:
        async with httpx.AsyncClient(base_url="http://api") as c:
            await c.get("/users")
        t.assert_requested_once("GET", Path("/users"))

# Decorator - declare a transport parameter
@httpx_mock(get={Path("/users"): httpx.Response(200, json=[])})
async def test_something(transport: MockTransport):
    async with httpx.AsyncClient(base_url="http://api") as c:
        await c.get("/users")
    transport.assert_requested_once("GET", Path("/users"))

Method	Checks
assert_requested(method?, matcher?)	at least 1 matching request
assert_requested_once(method?, matcher?)	exactly 1 matching request
assert_not_requested(method?, matcher?)	0 matching requests
assert_request_count(n, method?, matcher?)	exactly N matching requests
t.requests	raw list of all httpx.Request objects

matcher accepts any Matcher, including combined ones:

t.assert_requested("POST", Path("/orders") & Headers(authorization="Bearer token"))
t.assert_requested_once("GET", Path("/users/{id}") & Params(active=1))
t.assert_requested(matcher=Path("/users") | Path("/accounts"))
t.assert_not_requested("DELETE")
t.assert_request_count(3, "GET", re.compile(r"/items/\d+"))

---

Fallback: nested mocks, passthrough, and strict

By default, an unmatched request raises AssertionError - an accidentally-unmocked call fails loudly instead of quietly producing a response your code under test wasn't expecting. Nesting one httpx_mock(...) inside another, passthrough=True, strict=True, and unmatched_404=True all change what happens with an unmatched request, and they interact. The priority, highest first:

1. strict=True - this mock never consults its fallback for an unmatched request; the search for a response ends right here, regardless of nesting or passthrough.
2. Fallback - if not strict, the enclosing mock if this one is nested, otherwise the real network if passthrough=True.
3. unmatched_404=True - if the search above ends with nothing (this mock is strict with no match, or has no fallback, or the whole fallback chain came up empty), this mock returns httpx.Response(404, json={"message": "Not found"}) instead of raising. This is checked on whichever mock the httpx.Client/AsyncClient in your test is actually talking to - not on whichever mock deep in the fallback chain happened to run out of routes.
4. AssertionError(f"Unmocked request: {method} {url}") - if none of the above apply.

strict and unmatched_404 are independent: strict decides whether this mock is even allowed to look further (at its fallback) for an unmatched request; unmatched_404 decides how this mock reports final failure to the client, once nothing anywhere in the chain matched.

Nested mocks compose:

If a request doesn't match anything in the innermost active mock, it falls through to the mock that encloses it, not straight to 404:

async def test_nested_mocks():
    async with httpx_mock(get={Path("/users"): httpx.Response(200, json=[])}) as outer:
        async with httpx_mock(get={Path("/orders"): httpx.Response(200, json=[])}) as inner:
            async with httpx.AsyncClient(base_url="http://api") as c:
                r_orders = await c.get("/orders")   # matched by inner
                r_users = await c.get("/users")     # not in inner -> falls back to outer
        assert r_orders.status_code == 200
        assert r_users.status_code == 200

This is useful for a fixture-level mock covering common endpoints (auth, health checks) plus a per-test mock for the endpoint actually under test - see tests/examples/test_examples_nesting.py for a worked example.

Concurrency: the "currently active mock" is tracked per-contextvars context, not as a single shared attribute on httpx.AsyncClient/httpx.Client. Since asyncio.Task copies its context at creation, sibling httpx_mock(...) scopes running as concurrent tasks (e.g. via asyncio.gather) never see or interfere with each other's active transport, no matter how their lifetimes overlap or in what order they enter/exit - see tests/test_async_concurrency.py for the scenarios this covers.

Gotcha: this only covers httpx_mock(...) used normally, as a decorator or through matching async with/with blocks (which Python itself guarantees exit in reverse order of entry). If you call __aenter__()/__aexit__() (or __enter__()/__exit__()) directly and get the order wrong - e.g. exiting an outer mock while a nested one is still open, or exiting the same mock twice - httpx_mock raises RuntimeError/IndexError immediately instead of silently leaving the wrong mock active. See tests/test_misuse_guards.py.

Sibling isolation: nested mocks entered one after another (not one inside the other) don't see each other's routes - each sibling only sees its own routes plus whatever the still-active enclosing mock provides:

async def test_sibling_isolation():
    async with httpx_mock(get={Path("/outer"): httpx.Response(200)}):
        async with httpx_mock(get={Path("/inner-a"): httpx.Response(200)}):
            ...  # sees /inner-a and /outer
        async with httpx_mock(get={Path("/inner-b"): httpx.Response(200)}):
            async with httpx.AsyncClient(base_url="http://api") as c:
                with pytest.raises(AssertionError):
                    await c.get("/inner-a")  # not registered here, and not leaked from the previous sibling

Multi-level request history: a request is recorded in .requests only on the mock whose own routes actually matched it - not on any mock it merely cascaded through on the way there. In the example above, inner.requests has just /orders and outer.requests has just /users, even though /users was first offered to inner and only matched at outer. Since each mock's lifetime and scope differ, its .requests reflects only what it itself is responsible for mocking.

Passthrough to the real network:

Pass passthrough=True to send unmatched requests to the real network instead of failing - useful when testing code that calls many different APIs and you only care about mocking one of them:

@httpx_mock(get={Path("/the-one-endpoint-i-care-about"): httpx.Response(200)}, passthrough=True)
async def test_something(): ...

See tests/examples/test_examples_passthrough.py for a runnable, hermetic version (the real transport is monkeypatched, so no network access actually happens).

Every passthrough request emits a httpx_mock.client.PassthroughWarning naming the method and URL, so an unintentionally-unmocked route (a typo in a Path(...) template, for example) stays visible in pytest's warning summary instead of silently hitting a real API.

can_passthrough inheritance: a nested mock doesn't need to repeat passthrough=True if an enclosing mock already has it - the network destination is decided by whichever mock in the chain is outermost / has no fallback of its own.

Gotcha: setting passthrough=True on a mock nested inside one where passthrough isn't reachable anywhere in the fallback chain raises immediately:

ValueError: passthrough=True has no effect here: this mock is nested inside
another mock that doesn't have passthrough enabled anywhere in its
fallback chain. Enable passthrough on the enclosing httpx_mock(...)
instead, or remove it here.

Shared connection pool: two independent, non-nested passthrough=True mocks (used one after another, not nested in each other) share the same process-wide real transport and connection pool - httpx_mock builds it lazily on first use and caches it, rather than opening a fresh one per mock. This only matters if you're relying on connection-level behavior (e.g. inspecting or monkeypatching the transport instance itself); the requests each mock sees are still independent.

strict=True:

Every mock already raises AssertionError on an unmatched request by default (see above), so plain httpx_mock(...) already catches accidentally-unmocked calls. Use strict=True specifically when this mock is nested inside another (or has passthrough=True upstream) and you want to disable that fallback for it - i.e. force this mock's own routes to be the entire story, with nothing beyond them consulted:

async def test_something():
    async with httpx_mock(get={Path("/users"): httpx.Response(200, json=[])}) as outer:
        async with httpx_mock(get={Path("/orders"): httpx.Response(200, json=[])}, strict=True):
            async with httpx.AsyncClient(base_url="http://api") as c:
                await c.get("/users")  # raises, even though outer has this route - strict skips the fallback

strict=True always takes priority over both nesting and passthrough: it never consults the fallback, regardless of what's upstream.

unmatched_404=True:

Pass unmatched_404=True to get the old, lenient behavior back for a given mock - an unmatched request (once nothing in its own routes or its fallback chain matches) becomes httpx.Response(404, json={"message": "Not found"}) instead of raising:

@httpx_mock(get={Path("/users"): httpx.Response(200, json=[])}, unmatched_404=True)
async def test_something():
    async with httpx.AsyncClient(base_url="http://api") as c:
        r = await c.get("/orders")
    assert r.status_code == 404

---

License

MIT - see LICENSE.