API Testing · lesson 8 of 11
Timeouts and Retries Against a Flaky Endpoint
Extend the local test API with a slow route and one that fails a fixed number of times, then measure the difference between a single request, a retrying session, and one with too few retries configured.
By SkillByExample · Published
A test suite that never sees a timeout or a retry is testing the happy path only — and the happy path is rarely where production incidents come from.
A server with a slow route and a flaky one
# local_api.py
import json
import time
from http.server import BaseHTTPRequestHandler
class FlakyAPIHandler(BaseHTTPRequestHandler):
def _send_json(self, status, payload):
body = json.dumps(payload).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
if self.path == "/flaky/books":
self.server.attempt_count += 1
if self.server.attempt_count <= self.server.fail_times:
self._send_json(503, {"error": "temporarily unavailable"})
else:
self._send_json(200, [{"id": 1, "title": "Dune"}])
return
if self.path == "/slow/books":
time.sleep(0.5)
self._send_json(200, [{"id": 1, "title": "Dune"}])
return
self._send_json(404, {"error": "not found"})
def log_message(self, format, *args):
pass
# conftest.py
import threading
from http.server import ThreadingHTTPServer
import pytest
from local_api import FlakyAPIHandler
class QuietServer(ThreadingHTTPServer):
def handle_error(self, request, client_address):
pass # a client that gave up early (timeout) shouldn't print a server-side traceback
@pytest.fixture
def local_api_url():
server = QuietServer(("127.0.0.1", 0), FlakyAPIHandler)
server.attempt_count = 0
server.fail_times = 2
port = server.server_address[1]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
yield f"http://127.0.0.1:{port}"
server.shutdown()
thread.join()
/flaky/books fails with 503 on the first two calls, then succeeds — fail_times=2 is fixed
per test since the fixture is function-scoped. /slow/books always succeeds, just slowly.
Timeouts
import pytest
import requests
def test_timeout_raises_when_server_is_slow(local_api_url):
with pytest.raises(requests.exceptions.Timeout):
requests.get(f"{local_api_url}/slow/books", timeout=0.1)
test_timeout.py::test_timeout_raises_when_server_is_slow PASSED [100%]
============================== 1 passed in 0.58s ==============================
The server sleeps 0.5s; the client gives up at 0.1s and raises Timeout rather than
waiting. Without timeout=, requests has no default and would simply wait the full 0.5s —
or forever, against a server that never responds at all.
No retry: one call, one failure
def test_single_request_fails_against_flaky_endpoint(local_api_url):
response = requests.get(f"{local_api_url}/flaky/books")
assert response.status_code == 503
test_flaky_no_retry.py::test_single_request_fails_against_flaky_endpoint PASSED [100%]
requests.get() makes exactly one attempt. This “passes” only because the test expected the
503 — against a real client, this is the failure a retry policy exists to prevent.
Retrying with urllib3’s Retry
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def test_session_with_retry_recovers_from_flaky_endpoint(local_api_url):
retry = Retry(total=3, backoff_factor=0, status_forcelist=[503])
session = requests.Session()
session.mount("http://", HTTPAdapter(max_retries=retry))
response = session.get(f"{local_api_url}/flaky/books")
assert response.status_code == 200
assert response.json()[0]["title"] == "Dune"
test_flaky_with_retry.py::test_session_with_retry_recovers_from_flaky_endpoint PASSED [100%]
============================== 1 passed in 0.58s ==============================
status_forcelist=[503] tells Retry which status codes count as retryable; total=3 allows
up to three retries. session.mount("http://", ...) attaches the retry-aware adapter to every
request the session makes on that scheme — no per-call retry code needed.
The off-by-one that still fails
def test_too_few_retries_still_fails(local_api_url):
retry = Retry(total=1, backoff_factor=0, status_forcelist=[503])
session = requests.Session()
session.mount("http://", HTTPAdapter(max_retries=retry))
response = session.get(f"{local_api_url}/flaky/books")
assert response.status_code == 200
_______________________ test_too_few_retries_still_fails _______________________
...
except MaxRetryError as e:
...
if isinstance(e.reason, ResponseError):
> raise RetryError(e, request=request)
E requests.exceptions.RetryError: HTTPConnectionPool(host='127.0.0.1', port=58859):
Max retries exceeded with url: /flaky/books
(Caused by ResponseError('too many 503 error responses'))
..\venv\Lib\site-packages\requests\adapters.py:720: RetryError
total counts retries, not attempts — total=1 means one retry after the first request,
two attempts total. The endpoint fails twice before succeeding on the third, so two attempts
isn’t enough. total=2 (matching the earlier passing example’s margin) is the minimum that
actually covers this endpoint’s failure count.
Practice
1. Confirm a generous timeout succeeds against the slow endpoint instead of raising.
def test_fast_enough_timeout_succeeds(local_api_url):
response = requests.get(f"{local_api_url}/slow/books", timeout=1)
assert response.status_code == 200
test_practice.py::test_fast_enough_timeout_succeeds PASSED [ 33%]
1 second comfortably covers the 0.5s the handler sleeps — timeouts should be set to what a
slow-but-healthy response actually needs, not shaved as close to the edge as possible.
2. Find the minimum total that succeeds against a two-failure endpoint, and confirm it.
def test_retry_total_equal_to_fail_times_plus_one_succeeds(local_api_url):
retry = Retry(total=2, backoff_factor=0, status_forcelist=[503])
session = requests.Session()
session.mount("http://", HTTPAdapter(max_retries=retry))
response = session.get(f"{local_api_url}/flaky/books")
assert response.status_code == 200
test_practice.py::test_retry_total_equal_to_fail_times_plus_one_succeeds PASSED [ 66%]
total must be at least equal to fail_times — one retry per failure, plus the initial
attempt that doesn’t count against total.
3. Measure that a nonzero backoff_factor actually adds delay, rather than assuming it.
import time
def test_backoff_factor_measurably_delays_retries(local_api_url):
retry = Retry(total=3, backoff_factor=0.2, status_forcelist=[503])
session = requests.Session()
session.mount("http://", HTTPAdapter(max_retries=retry))
started = time.monotonic()
session.get(f"{local_api_url}/flaky/books")
elapsed = time.monotonic() - started
assert elapsed > 0.2
test_practice.py::test_backoff_factor_measurably_delays_retries PASSED [100%]
backoff_factor=0.2 with exponential growth means the second retry alone waits about 0.2s —
measuring the elapsed wall time is the only way to confirm backoff is actually happening, as
opposed to just trusting the config value is doing something.
Next: testing pagination and rate-limit headers.