Retry logic is simple to write and easy to get subtly wrong: one attempt too many, a backoff that never caps, jitter that can go negative, a deadline that is checked before sleeping instead of after. Every one of those bugs is invisible in a test that just checks the final result, and a test that exercises real exponential backoff takes seconds or minutes per run. The result is that most suites test retries once, lightly, and then never touch them again.
The cost of leaving them untested is higher than it looks. Retry logic runs precisely when a dependency is struggling, so its bugs appear during incidents: an extra attempt multiplies load on a service already overloaded, a missing cap turns a brief outage into minutes of stalled requests, and synchronised retries without jitter hammer a recovering service at the same instant from every client. Injecting the sleep function changes that. The retry helper asks for a delay, the test records what was asked for and returns instantly, and the assertion can state the exact schedule the specification requires. A test that would have waited fifteen seconds runs in microseconds and checks more than it ever could with real time.
Prerequisites
- Python 3.9+;
unittest.mockfor scripting failures. pytest >= 8.0.- The injected-clock pattern from injecting a clock instead of patching datetime.
Solution
import random
import time
from typing import Callable, TypeVar
T = TypeVar("T")
def retry(fn: Callable[[], T], *, attempts: int = 4, base: float = 0.5,
cap: float = 4.0, jitter: float = 0.0,
sleep: Callable[[float], None] = time.sleep,
rng: random.Random = random.Random()) -> T:
for attempt in range(attempts):
try:
return fn()
except ConnectionError:
if attempt == attempts - 1:
raise # out of attempts
delay = min(cap, base * 2 ** attempt) # exponential, capped
if jitter:
delay += rng.uniform(-jitter, jitter) * delay
sleep(max(0.0, delay))
raise AssertionError("unreachable")
from unittest.mock import Mock
import pytest
def test_backoff_schedule_is_exponential_and_capped():
slept: list[float] = []
fn = Mock(side_effect=[ConnectionError] * 4 + ["ok"])
result = retry(fn, attempts=5, base=0.5, cap=2.0, sleep=slept.append)
assert result == "ok"
assert fn.call_count == 5
assert slept == [0.5, 1.0, 2.0, 2.0] # doubled, then held at the cap
def test_gives_up_after_the_last_attempt():
slept: list[float] = []
fn = Mock(side_effect=ConnectionError("down"))
with pytest.raises(ConnectionError):
retry(fn, attempts=3, sleep=slept.append)
assert fn.call_count == 3 # not a fourth
assert len(slept) == 2 # no sleep after the final failure
[0.5, 1.0, 2.0, 2.0] checks the doubling, the cap and the absence of a trailing sleep in one assertion — 5.5 seconds of real waiting, verified instantly.Why this works
The retry helper's behaviour has two parts: how many times it calls the function, and how long it waits between calls. The side_effect script controls the first — the function fails exactly as many times as the test says — and the injected sleep observes the second, recording each requested delay without actually waiting. Together they reduce a time-dependent process to two lists the test can compare exactly.
The give-up test is the one teams most often skip, and it catches the most common bugs: an extra attempt after the limit, and a pointless sleep after the final failure that delays the error reaching the caller. Both show up immediately as a wrong call_count or an extra entry in slept.
Edge cases and failure modes
- Jitter going negative. Symmetric jitter around a small base can produce a negative delay, which
time.sleeprejects. Clamp to zero, and test it with a seeded generator that draws the extreme. - No cap. Exponential backoff without a cap reaches minutes after ten attempts. Assert the cap explicitly.
- Retrying the wrong exceptions. Retrying a
ValueErrorfrom bad input wastes attempts on something that will never succeed. Test that non-transient errors propagate immediately with no sleep. - Deadlines checked at the wrong time. A total deadline must account for the next sleep, or the helper sleeps past it. Inject a fake clock and assert the helper gives up before the deadline, not after.
- Async retries. Inject
asyncio.sleepas an async callable; the recorder becomesasync def record(d): slept.append(d).
Deadlines as well as attempt limits
An attempt limit bounds how many times the code tries; it does not bound how long the caller waits. Four attempts with a two-second cap can take seven seconds, and a caller with a five-second budget of its own needs the retry helper to respect that. A total deadline — give up once the next sleep would cross it — is the fix, and it is exactly the kind of logic that is untestable with real time and trivial with a fake clock.
The fake clock's sleep advances its own time, so the helper's view of "now" moves forward exactly as it would in production. The test sets a deadline, scripts permanent failure, and asserts two things: the helper raised before the clock passed the deadline, and it did not sleep across it. The second assertion is the subtle one. A helper that checks the deadline only at the top of each loop iteration will happily sleep past it and then notice, which means the caller waits longer than its budget allows — a bug that appears in production as cascading timeouts upstream.
Testing it this way also pins down the boundary precisely. With a deadline of five seconds and delays of 0.5, 1 and 2, the helper should make its fourth attempt at 3.5 seconds and then give up rather than sleeping two more; a fake clock makes that sequence observable and assertable to the millisecond. With real time, the same behaviour could only be checked approximately, and slowly, and would still vary with how busy the machine running the test happened to be.
Testing jitter reproducibly
Jitter exists to stop many clients retrying in lockstep after a shared outage, and it makes delays random by design. Testing it needs a seeded generator and assertions on bounds rather than values.
import random
def test_jitter_stays_within_bounds_and_never_negative():
slept: list[float] = []
fn = Mock(side_effect=[ConnectionError] * 5 + ["ok"])
retry(fn, attempts=6, base=0.1, cap=1.0, jitter=0.5,
sleep=slept.append, rng=random.Random(20260918))
bases = [0.1, 0.2, 0.4, 0.8, 1.0]
for delay, base in zip(slept, bases):
assert 0.0 <= delay <= base * 1.5 # within ±50%, clamped at zero
assert delay >= base * 0.5 or delay == 0.0
The fixed seed makes the draws reproducible, so a failure here is a real defect rather than bad luck, while the bounds-based assertion keeps the test valid if the jitter formula is refined. A second test with a generator whose uniform always returns the lower extreme — a tiny stub — confirms the clamp at zero, which a seeded run might never hit.
Library-based retries
Most production code uses a retry library — tenacity, backoff, stamina — rather than a hand-written loop, and the same technique applies because every serious library exposes its sleep function. tenacity.Retrying(sleep=...) takes it as an argument; decorated functions expose it as fn.retry.sleep, which a test can replace for its duration. stamina offers a testing mode that disables waiting entirely and a context manager that caps attempts.
The assertions do not change: script the collaborator's failures, record the requested delays, and compare both the attempt count and the schedule against the policy the code declares. Doing this for library-based retries matters as much as for hand-written ones, because the policy — which exceptions are retried, the base, the multiplier, the cap, the stop condition — is configuration written by the team, and configuration is where retry bugs live. A test that pins the recorded schedule to the documented policy turns an unnoticed change to a decorator argument into a failing test on the pull request that changed it. That is the cheapest possible moment to discuss whether the change was intended. Afterwards, the same change is discovered during an outage.
Frequently Asked Questions
How do I test exponential backoff without the test taking minutes?
Inject the sleep function. Production passes time.sleep; the test passes a recorder that appends each requested delay to a list and returns immediately. The test then asserts the exact schedule, such as [0.5, 1.0, 2.0], in microseconds.
How do I test jitter if the delays are random? Inject a seeded random generator as well, and assert on bounds rather than exact values: each delay lies within the jitter range around its base. With a fixed seed the delays are also reproducible if a precise check is needed.
Can tenacity-based code be tested the same way?
Yes. tenacity's Retrying accepts a sleep argument, and its wait strategies are pure functions of the attempt number. Pass a recording sleep in tests, or patch the decorated function's retry.sleep attribute, and assert the recorded schedule.
Related
- Injecting a Clock Instead of Patching datetime — a clock whose sleep advances time.
- Driving Mocks with side_effect Sequences — scripting the failures.
- Simulating Timeouts and Connection Errors — the network errors retries respond to.
- Seeding random and NumPy for Reproducible Tests — seeding the jitter generator.
← Back to Controlling Time and Randomness in Tests