Messy retry loops fail extracts in a predictable way. Unpinned helpers change attempt counts and delay math. Freeze sleep arguments and retryable error classes first. Then move only the loop into a helper.
Core rule
Do not extract a retry helper from mixed networking code. Pin sleep arguments and retryable exception types first. After those pins pass, move only the loop.
Leave jitter, logging, and HTTP details untouched. Those behaviors need separate goldens later. One extract should change structure, not timing policy.
The usual breakage
A god function loads data and retries timeouts inline. It also maps business errors in the same block. A model asked to clean retries will invent new policy. It often retries Exception instead of two classes.
Wall-clock elapsed time is the wrong golden. Process scheduling noise will fail that suite. Spy time.sleep instead of measuring seconds on a clock. The argument list is the backoff contract.
Observables to freeze
Capture four fields for every fixture row.
- Attempt count against the loader mock.
- Exact
time.sleepargument sequence. - Exception class that escapes after exhaustion.
- Sleep count after a mid-loop success.
Use this decision table before writing tests.
| Injected loader behavior | Attempts | Sleep args | Escapes as |
|---|---|---|---|
| TimeoutError then success | 2 | [0.1] |
no exception |
| TimeoutError three times | 3 | [0.1, 0.2] |
TimeoutError |
| ConnectionError then success | 2 | [0.1] |
no exception |
| ValueError on first call | 1 | [] |
ValueError |
| success on first call | 1 | [] |
no exception |
The table is the product behavior. Comments in the messy function are not the contract. If a later extract changes one cell, reject the patch.
Artifact: stdlib characterization harness
The following Python is a proposed local example. It is not a live benchmark and claims no production timings. It uses only the standard library. Drop both files into an empty directory.
File invoice_fetch.py
# Proposed local example. Not production networking code.
from __future__ import annotations
import time
from typing import Callable, TypeVar
T = TypeVar("T")
RETRYABLE = (TimeoutError, ConnectionError)
DELAYS = (0.1, 0.2, 0.4)
def fetch_invoice(load: Callable[[], T]) -> T:
last_exc: BaseException | None = None
for index, delay in enumerate(DELAYS):
try:
return load()
except RETRYABLE as exc:
last_exc = exc
if index < len(DELAYS) - 1:
time.sleep(delay)
continue
assert last_exc is not None
raise last_exc
File test_invoice_fetch_chars.py
# Proposed characterization suite. Label unexecuted until you run it.
import unittest
from unittest.mock import patch
import invoice_fetch
class SleepRecorder:
def __init__(self) -> None:
self.calls: list[float] = []
def __call__(self, seconds: float) -> None:
self.calls.append(seconds)
class Loader:
def __init__(self, outcomes: list[object]) -> None:
self.outcomes = list(outcomes)
self.calls = 0
def __call__(self):
self.calls += 1
item = self.outcomes.pop(0)
if isinstance(item, BaseException):
raise item
return item
class CharacterizeFetchInvoice(unittest.TestCase):
def _run(self, outcomes):
recorder = SleepRecorder()
loader = Loader(outcomes)
with patch.object(invoice_fetch.time, "sleep", recorder):
try:
value = invoice_fetch.fetch_invoice(loader)
error = None
except BaseException as exc:
value = None
error = exc
return loader.calls, recorder.calls, value, error
def test_success_first_try_does_not_sleep(self):
attempts, sleeps, value, error = self._run(["ok"])
self.assertEqual(attempts, 1)
self.assertEqual(sleeps, [])
self.assertEqual(value, "ok")
self.assertIsNone(error)
def test_timeout_then_success_sleeps_once(self):
attempts, sleeps, value, error = self._run(
[TimeoutError("t"), "ok"]
)
self.assertEqual(attempts, 2)
self.assertEqual(sleeps, [0.1])
self.assertEqual(value, "ok")
self.assertIsNone(error)
def test_three_timeouts_do_not_sleep_after_last(self):
attempts, sleeps, value, error = self._run(
[TimeoutError("a"), TimeoutError("b"), TimeoutError("c")]
)
self.assertEqual(attempts, 3)
self.assertEqual(sleeps, [0.1, 0.2])
self.assertIsNone(value)
self.assertIsInstance(error, TimeoutError)
def test_value_error_is_not_retryable(self):
attempts, sleeps, value, error = self._run(
[ValueError("bad invoice")]
)
self.assertEqual(attempts, 1)
self.assertEqual(sleeps, [])
self.assertIsNone(value)
self.assertIsInstance(error, ValueError)
if __name__ == "__main__":
unittest.main()
Commands
python -m unittest test_invoice_fetch_chars.py -v
A passing run means the current loop is pinned. It does not mean the loop is well designed. Do not refactor on a red characterization suite.
Numbered workflow
1. Copy the tangled function into a fixture module
Keep production imports out of the first pin. Reproduce the retry branch with a callable hook. The hook replaces HTTP for the test. Real sockets add noise and flakes.
Name the fixture after the production function. Do not rename delays during the copy. Copy-paste errors here become false goldens later.
2. Patch time.sleep on the module that calls it
Patch the name used by the messy module. Patching time.sleep on the wrong module yields a false green. Record every seconds argument in a list. Ignore the return value of sleep.
Confirm the spy with a deliberate wrong delay. If the test still passes, the patch target is wrong. Fix the target before writing more rows.
3. Drive one behavior per test
Do not fold five error classes into one case. Mixed assertions hide which table cell drifted. Name tests after the table rows above. Failures then point at one contract cell.
Keep loader outcomes in a short list. Pop one outcome per attempt. That matches how a real retry loop consumes failures.
4. Store goldens as literals
Keep expected sleep tuples in the test file. Do not rebuild delays with ** inside assertions. The test must show numbers a reviewer can read. Hidden formulas drift with the production formula.
Treat 0.1, 0.2 as data, not as a series. If production later uses 0.1, 0.3, the test must fail. Silent series generators hide that change.
5. Extract only the retry loop
Move the for loop into run_with_delays. Pass the retryable tuple and the delay tuple. Do not add jitter in this commit. Do not add logging in this commit.
Do not wrap exceptions in a new type. Callers already catch TimeoutError and ValueError. A new wrapper is a behavior change. Behavior changes need new table rows.
6. Re-run the same four fields
Attempt counts must match the table. Sleep lists must match the table. Escaping types must match the table. Any extra sleep means a failed extract.
If one row fails, revert the helper. Do not tune delays to soothe the test. The table wins over the new structure.
Smallest safe production edit
Proposed extract below. Still local example code. It keeps DELAYS and RETRYABLE unchanged.
from __future__ import annotations
import time
from typing import Callable, Iterable, TypeVar
T = TypeVar("T")
RETRYABLE = (TimeoutError, ConnectionError)
DELAYS = (0.1, 0.2, 0.4)
def run_with_delays(
load: Callable[[], T],
delays: Iterable[float],
retryable: tuple[type[BaseException], ...],
) -> T:
delay_list = list(delays)
last_exc: BaseException | None = None
for index, delay in enumerate(delay_list):
try:
return load()
except retryable as exc:
last_exc = exc
if index < len(delay_list) - 1:
time.sleep(delay)
continue
assert last_exc is not None
raise last_exc
def fetch_invoice(load: Callable[[], T]) -> T:
return run_with_delays(load, DELAYS, RETRYABLE)
Stop the commit after this single structural edit. Resist extra cleanups in the same diff. Header maps, URL joins, and JSON decoding stay put. Those cuts need their own pins.
False greens to reject
Patching time.sleep in the test module is a common miss. The messy module still sleeps for real. Tests look slow and still pass weak assertions. Always patch invoice_fetch.time.sleep.
Another false green is asserting len(sleeps) > 0 only. Length ignores 0.1 versus 0.5. A helper can double backoff and still pass. Compare the full argument list.
A third false green is catching Exception in the test. That hides a wrapper change. Assert type(error) against the table class. Wrappers are contract breaks.
After the suite is green
Models rewrite retry code with extra policy objects. That help is useful only after pins exist. Feed the unittest file and the messy function together. Ask for a loop extract that keeps every table cell.
Disclosure: This article was prepared as part of MonkeyCode's product outreach.
MonkeyCode provides free model access and a free server option. Those two availability facts are the product surface used here. A remote draft can propose run_with_delays against the unittest file. It cannot invent the sleep contract for you.
Reject any patch that introduces jitter or broader except Exception. If the characterization file already fails, skip the model session. Fix the pin first, or the model will optimize a lie.
Limitations
Sleep spies miss C extensions that delay without Python sleep. They also miss a local alias bound after the patch. Async code needs an asyncio.sleep spy instead. Threaded retries need a clock that is safe across threads.
Characterization freezes bugs as well as features. If attempt three never ran in production, the suite will protect that gap. Schedule a deliberate behavior change as a second commit. Do not hide it inside the extract.
Jitter, deadlines, and retry-after headers stay out of scope. Pin them with new rows before touching them. HTTP status retry rules are a different contract. Record status codes in a later suite.
Deadline loops that poll time.monotonic need a fake clock. This sleep-list method will not catch those. Choose the observable that the production code actually calls.
Who should not use this approach
Skip this method on greenfield services with no callers. Write an explicit retry policy first in that case. Skip it when delay math must change in the same patch. That mix is a behavior change, not an extract.
Skip it for latency checks that require real waits. These tests never sleep on the wall clock. They cannot detect a pause that bypasses time.sleep. Skip it when exception types cannot be frozen yet.
Broad except Exception retries are a product decision. Pins will only fossilize that decision. Do not use this workflow to launder a wider catch. Change the table in public review first.
Checklist before merge
- Table cells match the unittest method names.
-
python -m unittest test_invoice_fetch_chars.py -vis green. - Diff touches the loop and the new helper only.
- No jitter, logging, or HTTP edits in the same commit.
The extract is done when the table is unchanged. Anything else is a second change. Keep that second change behind a new golden row.
Top comments (0)