time.sleep. In tenacity the sleeper is a constructor argument, Retrying(sleep=...), so pass a callable that records the delay instead of performing it, and the list it collects is the backoff schedule you assert. Deadlines need a fake clock as well, because stop_after_delay measures elapsed time rather than counting sleeps, and a recorder advances nothing.This test takes thirty-one seconds and proves almost nothing:
\n\ndef test_retries_on_503(client):\n client.transport.responses = [503, 503, 200]\n fetch_order(client, \"ord_1\")\n assert client.transport.get.call_count == 3It passes if the backoff schedule is [0.5, 1.0] and it passes if the schedule is [30, 0.001]. It passes if the deadline is never enforced. It would pass if the code retried a 400 Bad Request, because that is also three calls. And it costs half a minute of every CI run to establish a number the code makes trivially true.
Retry logic has four separable behaviours: the schedule, the stop rule, the retry predicate, and the final error. A call count observes none of them. All four become fast, exact assertions once the things that make retrying slow are arguments rather than globals.
\n\nThe instinct is to patch the clock. It is unnecessary, because tenacity's controller takes the sleeper as its first parameter. The signature is explicit about the type and the default:
\n\nclass tenacity.Retrying(\n sleep: Callable[[Union[int, float]], None] = <function sleep>,\n stop: StopBaseT = <stop_never>,\n wait: WaitBaseT = <wait_none>,\n retry: RetryBaseT = <retry_if_exception_type>,\n before=..., after=..., before_sleep=None,\n reraise: bool = False,\n retry_error_cls: Type[RetryError] = RetryError,\n retry_error_callback=None,\n)Two things worth noticing in that default line before going further. stop defaults to stop_never and wait defaults to wait_none. A bare controller retries forever, immediately. Neither is a sane production setting, which means both must be supplied, which means both are worth a test.
The async controller mirrors it with an awaitable: AsyncRetrying(sleep: Callable[[float], Awaitable[Any]] = <function sleep>). Same technique, one async def.
So the injection is a list append:
\n\n# Wrong. monkeypatch.setattr replaces the attribute and undoes it after the\n# test, which is fine — but it is global for the duration. It silences every\n# sleep in the process, including ones in libraries you are not testing, and\n# it does not work at all if the module under test did `from time import\n# sleep` at import time, because that binding is a different object.\ndef test_backoff(monkeypatch):\n monkeypatch.setattr(time, \"sleep\", lambda _: None)\n ...\n\n\n# Right. The policy is constructed with a sleeper that records instead of\n# sleeping. Nothing global changes, and the recording is the evidence.\ndef test_backoff():\n sleeps: list[float] = []\n\n retrying = Retrying(\n sleep=sleeps.append,\n wait=wait_exponential(multiplier=0.5, exp_base=2, max=8),\n stop=stop_after_attempt(4),\n retry=retry_if_exception_type(ServiceUnavailable),\n reraise=True,\n )This is the whole trick, and it generalises past tenacity: any retry helper worth using takes its sleeper as a parameter, and one that does not is a retry helper you cannot test.
\n\nWith a recorder in place, the schedule is data. Script the transport with a list. unittest.mock documents that \"if side_effect is an iterable then each call to the mock will return the next value from the iterable\", and an iterable may mix exceptions and return values:
def test_two_failures_then_success_backs_off_exponentially():\n sleeps: list[float] = []\n transport = Mock(side_effect=[\n ServiceUnavailable(\"503\"),\n ServiceUnavailable(\"503\"),\n {\"id\": \"ord_1\", \"status\": \"paid\"},\n ])\n\n result = Retrying(\n sleep=sleeps.append,\n wait=wait_exponential(multiplier=0.5, exp_base=2, max=8),\n stop=stop_after_attempt(4),\n retry=retry_if_exception_type(ServiceUnavailable),\n reraise=True,\n )(transport, \"ord_1\")\n\n assert result == {\"id\": \"ord_1\", \"status\": \"paid\"}\n # Three attempts means exactly two waits, and both values are checked.\n assert sleeps == [0.5, 1.0]That runs in microseconds and it fails if the multiplier changes, if the base changes, if a wait is skipped, or if an extra wait is inserted. The call count is now implied: two sleeps between three attempts, so there is nothing left to assert separately.
\n\nOne detail about wait_exponential(multiplier=1, max=4.611686018427388e+18, exp_base=2, min=0): that default max is roughly 2⁶², which is to say there is effectively no cap unless you set one. A test that asserts the cap is honoured is one line and catches a real production hazard:
def test_backoff_is_capped():\n sleeps: list[float] = []\n transport = Mock(side_effect=[ServiceUnavailable(\"503\")] * 8)\n\n with pytest.raises(ServiceUnavailable):\n Retrying(\n sleep=sleeps.append,\n wait=wait_exponential(multiplier=0.5, exp_base=2, max=4),\n stop=stop_after_attempt(8),\n reraise=True,\n )(transport)\n\n assert sleeps == [0.5, 1.0, 2.0, 4.0, 4.0, 4.0, 4.0]\n assert max(sleeps) <= 4Jitter is where people give up and go back to counting calls, because the schedule is no longer a fixed list. It does not have to be, because the formula is documented. wait_exponential_jitter(initial=1, max=..., exp_base=2, jitter=1) states it outright: \"the wait time is min(initial * 2**n + random.uniform(0, jitter), maximum) where n is the retry count.\"
From that you get an exact interval per attempt without knowing the random value at all:
\n\ndef test_jittered_backoff_stays_within_its_documented_interval():\n sleeps: list[float] = []\n transport = Mock(side_effect=[ServiceUnavailable(\"503\")] * 5)\n\n with pytest.raises(ServiceUnavailable):\n Retrying(\n sleep=sleeps.append,\n wait=wait_exponential_jitter(initial=0.5, exp_base=2,\n jitter=0.25, max=4),\n stop=stop_after_attempt(5),\n reraise=True,\n )(transport)\n\n for n, actual in enumerate(sleeps):\n base = min(0.5 * 2 ** n, 4)\n # The floor is the un-jittered term; the ceiling adds the jitter,\n # and the cap applies to the sum.\n assert base <= actual <= min(base + 0.25, 4)This is a better test than pinning random.uniform to a fixed value would be, because it holds for every draw rather than for one. Patching the random source is still the right move when you need a specific value (a retry that must land exactly on a deadline boundary, say), but bounds are the default.
Which jitter strategy to use is a design question the docs answer directly, and the answer is more specific than \"add jitter.\" wait_exponential's intervals \"are fixed (i.e. there is no jitter), so this strategy is suitable for balancing retries against latency when a required resource is unavailable for an unknown duration, but not suitable for resolving contention between multiple processes for a shared resource. Use wait_random_exponential for the latter case.\"
Read that as a decision rule. One client waiting for a service to come back does not need jitter. Forty workers hammering one row lock do, because without it they retry in lockstep forever. If your retry exists to survive contention and its strategy has no jitter, that is a bug the bounds test above will never catch, so it belongs in a review checklist, not a test.
\n\nHere is where the recorder alone stops being enough. stop_after_delay(max_delay) is documented as \"stop when the time from the first attempt >= limit\". It reads a clock. A sleeper that only appends to a list advances no clock, so under a naive recorder the deadline never fires and a deadline test silently becomes an attempt-count test.
You could patch whatever clock stop_after_delay consults, but that is an internal you would be pinning to a version. tenacity offers a supported alternative: a stop rule may be any callable. The docs specify the contract: my_stop(retry_state) takes \"info about current retry invocation\" and returns \"whether or not retrying should stop\" as a bool. So the deadline can read a clock you own:
class FakeClock:\n \"\"\"A clock that only moves when the code under test sleeps.\"\"\"\n\n def __init__(self) -> None:\n self.now = 0.0\n self.sleeps: list[float] = []\n\n def sleep(self, seconds: float) -> None:\n self.sleeps.append(seconds)\n self.now += seconds\n\n def deadline(self, limit: float):\n \"\"\"A tenacity stop callable enforcing `limit` against this clock.\"\"\"\n return lambda retry_state: self.now >= limitdef test_gives_up_at_the_deadline_rather_than_the_attempt_limit():\n clock = FakeClock()\n transport = Mock(side_effect=[ServiceUnavailable(\"503\")] * 20)\n\n with pytest.raises(ServiceUnavailable):\n Retrying(\n sleep=clock.sleep,\n wait=wait_exponential(multiplier=0.5, exp_base=2, max=8),\n # Both limits, so whichever binds first is explicit.\n stop=stop_any(clock.deadline(3.0), stop_after_attempt(20)),\n reraise=True,\n )(transport)\n\n assert clock.sleeps == [0.5, 1.0, 2.0] # 3.5s elapsed at the third wake\n assert transport.call_count == 4 # the deadline stopped it, not the capThe last two assertions together are the point: the test proves the deadline ended the retry loop and not the attempt cap. Assert one without the other and the test cannot tell you which rule fired. stop_any and stop_all exist for exactly this combination: \"stop if any of the stop condition is valid\" and \"stop if all the stop conditions are valid\", and stating both limits explicitly is what makes the question answerable.
In production the same policy uses stop_after_delay(3.0), which measures against a real clock. Keeping the limit a parameter of the policy rather than a literal inside it is what lets the test substitute its own rule without reaching into the library.
This one bites during a real incident, which is the worst time to discover it. tenacity's default is not to re-raise: \"normally when your function fails its final time (and will not be retried again based on your settings), a RetryError is raised. The exception your code encountered will be shown somewhere in the middle of the stack trace.\"
So a caller writing except ServiceUnavailable catches nothing, and an alert grouping on exception type groups every retry exhaustion together regardless of cause. The switch is documented alongside it: \"if you would rather see the exception your code encountered at the end of the stack trace (where it is most visible), you can set reraise=True.\"
Test whichever one you have chosen, because both are defensible and the silent one is wrong:
\n\n# With reraise=True — the caller sees the real failure and never has to\n# import tenacity. This is usually right at a library boundary.\ndef test_final_failure_surfaces_the_transport_error():\n transport = Mock(side_effect=[ServiceUnavailable(\"503\")] * 3)\n\n with pytest.raises(ServiceUnavailable) as excinfo:\n Retrying(sleep=lambda _: None, stop=stop_after_attempt(3),\n reraise=True)(transport)\n\n assert \"503\" in str(excinfo.value)\n\n\n# Without it — assert the wrapper, then reach through it, so the test still\n# pins the cause rather than accepting any exhaustion.\ndef test_final_failure_retains_the_cause_inside_retryerror():\n transport = Mock(side_effect=[ServiceUnavailable(\"503\")] * 3)\n\n with pytest.raises(RetryError) as excinfo:\n Retrying(sleep=lambda _: None, stop=stop_after_attempt(3))(transport)\n\n cause = excinfo.value.last_attempt.exception()\n assert isinstance(cause, ServiceUnavailable)The retry predicate is the behaviour most suites never test, and the one most likely to cause damage. Retrying a 400 wastes time; retrying a non-idempotent POST creates duplicate charges.
The negative test is three lines and it is the cheapest high-value test in this article:
\n\n@pytest.mark.parametrize(\n \"error\",\n [\n pytest.param(BadRequest(\"400\"), id=\"client-error-is-not-transient\"),\n pytest.param(Unauthorized(\"401\"), id=\"auth-error-will-not-self-heal\"),\n pytest.param(Conflict(\"409\"), id=\"conflict-needs-a-new-request\"),\n ],\n)\ndef test_does_not_retry_permanent_failures(error):\n sleeps: list[float] = []\n transport = Mock(side_effect=error)\n\n with pytest.raises(type(error)):\n Retrying(\n sleep=sleeps.append,\n wait=wait_exponential(multiplier=0.5),\n stop=stop_after_attempt(4),\n retry=retry_if_exception_type(ServiceUnavailable),\n reraise=True,\n )(transport)\n\n assert sleeps == [] # nothing waited\n assert transport.call_count == 1 # nothing repeatedassert sleeps == [] is the assertion that matters. A call count of one can also mean the retry never ran for an unrelated reason; an empty sleep list says the policy evaluated the predicate and declined.
Tests establish the schedule. Production needs to show it, and tenacity exposes both halves.
\n\nThe decorated function carries its own counters: \"you can access the statistics about the retry made over a function by using the retry attribute attached to the function and its statistics attribute\": fetch_order.retry.statistics after a call. Useful in a test as a cross-check, and useful in an incident as a cheap read.
For logs, the callback you want is the one that fires only when another attempt is coming. The docs name it precisely: \"it's also possible to only log failures that are going to be retried. Normally retries happen after a wait interval, so the keyword argument is called before_sleep.\"
@retry(\n stop=stop_any(stop_after_delay(3.0), stop_after_attempt(5)),\n wait=wait_exponential_jitter(initial=0.5, jitter=0.25, max=4),\n retry=retry_if_exception_type(ServiceUnavailable),\n reraise=True,\n # Fires per retried failure, not per attempt: one line per wait.\n before_sleep=before_sleep_log(logger, logging.DEBUG),\n)\ndef fetch_order(order_id: str) -> dict: ...There are matching before and after callbacks with before_log and after_log helpers if you want every attempt rather than every retry. before_sleep is the one that gives you a log line per wait, which is what makes a production schedule reconstructable after the fact.
Search your retry tests for call_count and for anything that patches time.sleep. Both are signals that the schedule is untested. Replace the patch with an injected sleeper and the count with an equality assertion on the recorded list. The second change usually deletes the first assertion rather than joining it.
Then check the two defaults. Anywhere a retry policy is constructed, confirm stop and wait are both passed explicitly, because the library's defaults are retry-forever and wait-never. And confirm retry= names a specific exception type rather than being left to catch everything.
Done looks like a retry test file that runs in under a second and contains, for each policy, four assertions: the exact sleep list, the stop rule that fired, an empty sleep list for a permanent error, and the exception type a caller will actually catch.
\n\nA clock-freezing library will make the deadline test work, and it is heavier than it needs to be here. The FakeClock above is five lines, moves only when the code under test sleeps, and makes the coupling between sleeping and elapsed time explicit, which is the thing the test is about. Reach for a general time-freezing tool when the code under test reads wall-clock time for its own reasons, such as formatting timestamps into a payload, not merely because it retries.
Separate the policy object from the transport, and let the caller supply it. That is what makes the tests in this article possible without a fake HTTP server, and it also means a batch job and a request handler can use the same client with different deadlines, which they should, since a user-facing request cannot afford a three-second retry budget that a nightly job welcomes.
\n\nThe same way, with an awaitable recorder. AsyncRetrying's sleeper is typed Callable[[float], Awaitable[Any]], so an async def that appends and returns satisfies it. Everything else (the schedule assertion, the bounds check, the empty-list negative test) is unchanged, which is the payoff of injecting rather than patching: the technique does not care whether the sleep was blocking.
If the API returns a retry hint, the policy must read it, and that is a different wait strategy rather than a tweak to an exponential one: the schedule now depends on the response, so it belongs in a custom wait callback that receives the retry state. Test it the same way: script the transport to return specific hints and assert the recorded sleeps match them. The bug this catches is a policy that computes exponential backoff and ignores the header entirely, which no call-count test can see.
\n\nAll of it, and the first change is the same: make the sleeper a parameter with a default rather than a direct call. A hand-rolled loop that calls time.sleep(delay) inline cannot be tested without patching a global; one that calls self._sleep(delay) can. The rest of the article is then about which four things to assert, and those are independent of whose retry loop it is.
sleep is a constructor parameter typed Callable[[Union[int, float]], None] defaulting to the real sleep, which is the injection point this article is built on; that AsyncRetrying takes Callable[[float], Awaitable[Any]]; and the remaining defaults, including stop=stop_never, wait=wait_none, reraise=False and retry_error_cls=RetryError.wait_exponential signature including its effectively unbounded default max, and its guidance that fixed intervals are \"suitable for balancing retries against latency when a required resource is unavailable for an unknown duration, but not suitable for resolving contention between multiple processes for a shared resource\"; the wait_exponential_jitter formula min(initial * 2**n + random.uniform(0, jitter), maximum); and wait_none, wait_fixed, wait_random, wait_chain and wait_combine.stop_after_attempt stops \"when the previous attempt >= max_attempt\" while stop_after_delay stops \"when the time from the first attempt >= limit\", a clock reading rather than a sleep count, and that stop_any and stop_all combine conditions with any and all semantics.RetryError is raised\" and \"the exception your code encountered will be shown somewhere in the middle of the stack trace\", and that reraise=True instead surfaces it \"at the end of the stack trace (where it is most visible)\".retry attribute attached to the function and its statistics attribute\", which makes the attempt record readable both in a test and during an incident.before, after and before_sleep callbacks with their before_log, after_log and before_sleep_log helpers, and the stated reason for the third's name: it logs \"only … failures that are going to be retried\", because \"normally retries happen after a wait interval\".side_effect is an iterable then each call to the mock will return the next value from the iterable\", and that an exception class or instance in that position is raised when the mock is called. This is what allows one list to script a sequence of failures followed by a success.my_stop(retry_state) receives \"info about current retry invocation\" and returns a bool for \"whether or not retrying should stop\". This is how a deadline can be enforced against a clock the test owns without patching library internals. The same section documents the equivalent hooks for wait, retry, before, after and before_sleep.monkeypatch modifications \"will be undone after the requesting test or fixture has finished\", the property that makes patching safe but still global for the duration, which is why an injected sleeper is preferable when the library already accepts one.