httpx_mock

Mock for httpx.AsyncClient and httpx.Client in tests. Supports routing by URL, headers, and query params.

Routes are defined as plain dicts, matchers compose with &/| instead of stacking special-cased kwargs, and a class-based Router is available when a route needs real logic (state, conditional branching) rather than a static response. Beyond static responses, the same route can return a sequence of responses (for pagination/retry tests) or raise an httpx.TransportError to emulate a network failure. Every mocked client records its request history, so tests can assert on what was actually sent (assert_requested_once, assert_request_count, etc.) instead of only on the response.

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/test_examples_router.py, tests/test_examples_lambda.py, tests/test_examples_parametrize.py, and tests/test_examples_shared_router_state.py.

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

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, in assertion methods.

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")

# Class decorator (each test_* gets an isolated client)
@httpx_mock(get={Path("/ping"): httpx.Response(200)})
class TestMyService:
    async def test_a(self): ...
    async def test_b(self): ...

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 MyRouter(Router):
    @Router.route.get(Path("/flaky"))
    async def flaky(self, request: httpx.Request):
        raise httpx.ReadTimeout("read timed out", request=request)

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

Router — a class with routes:

class MyRouter(Router):
    @Router.route.get(Path("/users/{id}"))
    async def get_user(self, request: httpx.Request):
        return httpx.Response(200, json={"id": 42})

@httpx_mock(routers={AnyMatcher(): MyRouter})
async def test_with_router(): ...

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"))

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+"))

License

MIT — see LICENSE.