Isolation & Contracts

Mocking requests with the responses Library

You need to test code that calls a third-party REST API through the requests library — including its retry logic, custom headers, and error handling — without the test ever touching the network. The responses library patches requests at the transport-adapter layer, so your real Session runs while canned responses are replayed and every call is recorded for assertion. This guide covers the registration API (@responses.activate, responses.add), ordered and unordered registries, matchers for query strings and bodies, and the assert_all_requests_are_fired guard that turns a forgotten stub into a failing test instead of silent dead code.

Prerequisites

  • responses >= 0.25 (the registries and responses.matchers modules referenced here are stable from 0.21+; OrderedRegistry lives in responses.registries).
  • requests >= 2.31 and pytest >= 8.0.
  • responses intercepts only requests; for httpx use respx instead, as covered in Mocking Network and HTTP Calls.

Solution

Python
import requests
import responses
from responses import matchers

# Code under test: a thin client with one retry on 429.
def fetch_page(session: requests.Session, page: int) -> dict:
    for _ in range(2):                                   # one retry
        resp = session.get(
            "https://api.example.com/items",
            params={"page": page},
            headers={"Accept": "application/json"},
        )
        if resp.status_code == 429:
            continue                                     # retry once on rate-limit
        resp.raise_for_status()
        return resp.json()
    resp.raise_for_status()
    return resp.json()


@responses.activate                                      # patches requests for this test only
def test_fetch_page_retries_then_succeeds():
    # First registration -> 429; second registration -> 200. Same method+URL,
    # consumed in order, so call 1 gets the 429 and call 2 gets the 200.
    responses.add(
        responses.GET,
        "https://api.example.com/items",
        json={"error": "rate_limit"},
        status=429,
        match=[matchers.query_param_matcher({"page": "2"})],  # only fire for page=2
    )
    responses.add(
        responses.GET,
        "https://api.example.com/items",
        json={"items": [1, 2, 3]},
        status=200,
        match=[matchers.query_param_matcher({"page": "2"})],
    )

    with requests.Session() as session:                  # the REAL client
        result = fetch_page(session, page=2)

    assert result == {"items": [1, 2, 3]}

    # responses records every intercepted call in order.
    assert len(responses.calls) == 2
    assert responses.calls[0].response.status_code == 429
    assert responses.calls[1].response.status_code == 200
    # The query string and header reached the transport unchanged.
    assert responses.calls[1].request.params == {"page": "2"}
    assert responses.calls[1].request.headers["Accept"] == "application/json"

For finer control over consumption order, use an explicit registry:

Python
import responses
from responses import registries

# OrderedRegistry forces responses to be consumed strictly in registration
# order and errors if a request arrives out of sequence.
@responses.activate(registry=registries.OrderedRegistry)
def test_strict_ordering():
    responses.add(responses.GET, "https://api.example.com/a", json={"step": 1})
    responses.add(responses.GET, "https://api.example.com/b", json={"step": 2})
    import requests
    assert requests.get("https://api.example.com/a").json()["step"] == 1
    assert requests.get("https://api.example.com/b").json()["step"] == 2

The context-manager form makes the firing guard explicit:

Python
import responses

def test_context_manager_guard():
    # assert_all_requests_are_fired defaults to True here: an unused
    # registration fails the test on context exit.
    with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
        rsps.add(responses.GET, "https://api.example.com/used", json={"ok": True})
        import requests
        requests.get("https://api.example.com/used")
        # If you added an unused stub, the block would raise AssertionError on exit.

Why this works

Where responses intercepts a requests call A downward flow: your code calls session.get with params and headers; the real requests Session runs and encodes params, injects headers, and builds the Request; that reaches the responses-installed HTTPAdapter, which swaps only the response bytes. The path down to the urllib3 connection pool, socket, and network is blocked, so no bytes leave the process. Instead the adapter pops the next matching entry off an ordered registration queue (first a 429, then a 200, and finally the last one repeats), and the canned response is replayed back up so the real Session decodes it through Response.json. Where responses cuts into a requests call replayed up · Session runs .json() Your code session.get(url, params=…, headers=…) requests.Session (real, runs) encode params · inject headers · build Request responses HTTPAdapter (interception) swaps only the response bytes off the wire no bytes leave the process urllib3 pool · socket · network never reached pop next ordered registration queue GET /items → 429 consumed GET /items → 200 next … then last entry repeats Everything above the adapter is the production code path; only the bytes below it are canned. One entry is consumed per matching request — which is what makes retry and pagination sequences testable.
Where responses cuts into the request path: the custom adapter answers below your client code and above urllib3, so no socket is ever opened.

responses registers a custom HTTPAdapter on the requests Session machinery, so requests never reach urllib3's connection pool or a socket. Because interception happens below your client code, the real Session, parameter encoding, header injection, and Response.json() decoding all execute exactly as in production — the only thing replaced is the bytes coming back off the wire. Registrations for the same method and URL form an ordered queue that is consumed one per matching request, which is what makes retry and pagination sequences testable, and assert_all_requests_are_fired inverts the check so a stub you forgot to exercise becomes a loud failure rather than silent coverage rot.

Matching more than the URL

A stub that matches on URL alone passes even when the client sends the wrong method, body or headers — which is how a broken request payload ships with a green suite. responses exposes matchers that turn each of those into an assertion, and the registration reads as a specification of the outgoing call.

Python
import json
import requests
import responses
from responses import matchers

@responses.activate
def test_charge_sends_the_right_payload():
    responses.post(                                   # method is part of the match
        "https://api.example.com/charges",
        json={"id": "ch_1", "status": "succeeded"},
        status=201,
        match=[
            matchers.json_params_matcher({"amount": 500, "currency": "eur"}),
            matchers.header_matcher({"Idempotency-Key": "order-7"}),
        ],
    )

    resp = requests.post(
        "https://api.example.com/charges",
        json={"amount": 500, "currency": "eur"},
        headers={"Idempotency-Key": "order-7"},
        timeout=5,
    )

    assert resp.status_code == 201
    assert resp.json()["status"] == "succeeded"
    # The request that was actually sent is available for further assertions.
    sent = responses.calls[0].request
    assert json.loads(sent.body)["amount"] == 500

When a matcher fails, responses raises ConnectionError with a diff describing which registered stub was closest and why it did not match — read that message rather than adding a looser stub, because a loosened matcher is a deleted assertion.

Three matchers cover most real APIs. json_params_matcher pins the decoded body and is the right choice for JSON APIs; urlencoded_params_matcher does the same for form posts. query_param_matcher pins the query string independently of order, which matters because requests preserves the order you pass parameters in and a dict literal's order is an implementation detail. header_matcher checks only the headers you name, so authentication and idempotency headers can be asserted without freezing the whole header set.

Two operational notes. assert_all_requests_are_fired defaults to True inside the @responses.activate decorator, so a stub the code never calls fails the test at teardown — keep it on; it catches dead code paths. And responses.calls is ordered, so a test that expects a retry can assert both attempts and their bodies rather than asserting a call count and hoping.

How a stub is selected for an outgoing request A vertical decision flow showing how responses picks a stub: the URL and method are compared first, then every registered matcher runs against the request, then the first fully matching stub replies while an unmatched request raises ConnectionError with a diff of the closest stub. How a stub is selected for an outgoing request URL and method compared registration order preserved A wrong verb never matches matchers run in order body, query and headers A loosened matcher is a deleted assertion first full match replies the stub returns its payload responses.calls records what was sent no match: ConnectionError error names the closest stub Read the diff before adding a stub
Matching is ordered and strict: the first stub that satisfies every matcher wins, and an unmatched request is an error rather than a passthrough.

Edge cases and failure modes

  • Trailing slash and query mismatches. https://api.example.com/items and .../items/ are different URLs, and a query string the code adds but the stub omits will not match. Use matchers.query_param_matcher (or query_string_matcher) rather than baking params into the URL.
  • Unconsumed registrations. With assert_all_requests_are_fired=True, a leftover stub fails the test. Keep it on to catch dead branches; turn it off only for deliberately optional fallbacks.
  • Queue exhaustion. Once all registrations for a URL are consumed, the last one repeats for further calls. If you expect an error after N calls, register exactly N responses with an OrderedRegistry so an extra call raises instead of silently replaying.
  • Passthrough leakage. responses.add_passthru(prefix) lets specific hosts reach the real network — combine it with pytest-socket allow-hosts so the rest of the suite stays blocked, as described in Mocking Network and HTTP Calls.
  • Wrong patch surface for non-requests clients. responses does nothing for httpx, aiohttp, or raw urllib. Getting the interception layer right is the same discipline as choosing where to patch: you patch where the object is looked up, not where it is defined.

Registering failure paths deliberately

Retry and timeout logic is the part of a client most likely to be wrong and least likely to be tested, because reproducing a flaky upstream by hand is tedious. responses makes it a two-line registration: register the same URL twice and the stubs are consumed in order, so the first call fails and the second succeeds.

Python
import requests, responses
from requests.adapters import HTTPAdapter
from requests.exceptions import ConnectionError as ReqConnectionError

@responses.activate
def test_retries_once_then_succeeds():
    responses.get("https://api.example.com/orders", body=ReqConnectionError("reset"))
    responses.get("https://api.example.com/orders", json={"orders": []}, status=200)

    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=1))
    resp = session.get("https://api.example.com/orders", timeout=2)

    assert resp.status_code == 200
    assert len(responses.calls) == 2        # the retry actually happened

Passing an exception instance as body makes the stub raise instead of reply, which is how you test timeout handling without waiting for one. A status=500 stub covers the retry-on-server-error path, and registering it twice proves the client gives up rather than looping forever.

A retrying client against two registered stubs A timeline of one client call: the first registered stub raises a connection error, the adapter retries, the second stub returns a 200 response, and the test then asserts that exactly two calls were recorded. A retrying client against two registered stubs call 1 stub raises ConnectionError body retry adapter backs off max_retries=1 call 2 stub replies 200 JSON payload returned assert len(calls) == 2 retry proven, not assumed
Stubs registered against the same URL are consumed in order, which turns retry behaviour into an ordinary assertion.

Frequently Asked Questions

What does assert_all_requests_are_fired do in responses? When True it fails the test if any registered response was never matched by a request, catching dead stubs and skipped code paths. It defaults to True for the RequestsMock context manager and can be toggled per block.

How do I match a request by query string or JSON body with responses? Pass matchers from responses.matchers to responses.add, for example matchers.query_param_matcher({'page': '2'}) or matchers.json_params_matcher({'id': 7}). A registration only fires when every supplied matcher passes against the incoming request.

Can responses return a different response on each call to the same URL? Yes. Register the same method and URL multiple times; responses consumes registrations in order, so the first call gets the first registration and so on. The last registration repeats once the queue is exhausted unless you use an OrderedRegistry.

Does responses work with requests sessions created inside the code under test? Yes. responses patches the HTTPAdapter send path used by every Session, including sessions the code constructs internally, so you do not need a seam to inject the session. The one exception is a client that bypasses requests entirely — an httpx or raw urllib3 call is unaffected and needs its own interception layer.

← Back to Mocking Network and HTTP Calls