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(theregistriesandresponses.matchersmodules referenced here are stable from 0.21+;OrderedRegistrylives inresponses.registries).requests >= 2.31andpytest >= 8.0.responsesintercepts onlyrequests; forhttpxuserespxinstead, as covered in Mocking Network and HTTP Calls.
Solution
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:
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:
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
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.
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.
Edge cases and failure modes
- Trailing slash and query mismatches.
https://api.example.com/itemsand.../items/are different URLs, and a query string the code adds but the stub omits will not match. Usematchers.query_param_matcher(orquery_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
OrderedRegistryso an extra call raises instead of silently replaying. - Passthrough leakage.
responses.add_passthru(prefix)lets specific hosts reach the real network — combine it withpytest-socketallow-hosts so the rest of the suite stays blocked, as described in Mocking Network and HTTP Calls. - Wrong patch surface for non-requests clients.
responsesdoes nothing forhttpx,aiohttp, or rawurllib. 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.
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.
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.
Related
- Mocking Network and HTTP Calls — the transport-interception strategy behind
responses, plusrespxforhttpxand socket-level blocking withpytest-socket. - Where to Patch: Understanding mock.patch Targets — the same lookup-vs-definition discipline that decides whether an adapter-level patch actually intercepts your client.
- Mock vs MagicMock vs AsyncMock — when to use each — when a hand-rolled
Sessionstub beats a canned-response registry, and when it does not. - Injecting fakes vs mocks in constructors — passing a pre-built
Sessionin so the client stays swappable under test.
← Back to Mocking Network and HTTP Calls