Testing a Retry Counter with Fixed Attempt Sequences

The fake tool is the seam. Give it three jobs: return outcomes in order, count calls, and record the payload of each call so you can check the wrapper passes the same arguments on the retry.

from dataclasses import dataclass, field

class RetryableError(Exception):
    def __init__(self, reason):
        super().__init__(reason)
        self.reason = reason

@dataclass
class FakeTool:
    outcomes: list
    calls: int = 0
    seen_args: list = field(default_factory=list)

    def call(self, payload):
        self.calls += 1
        self.seen_args.append(payload)
        outcome = self.outcomes[min(self.calls - 1, len(self.outcomes) - 1)]
        if isinstance(outcome, Exception):
            raise outcome
        return outcome

Three design choices matter here.

The min(...) clamp means a regression that produces one extra call does not blow up with an IndexError. It returns the last outcome again, so your assertion on calls fails with a readable message (expected 3, got 4) instead of a confusing traceback. The cost is that you must always assert the count, which you were going to do anyway.

Storing seen_args costs one line and lets you check that the wrapper forwards the same payload on each attempt. That matters when the wrapper is meant to be a transparent pass-through, and it is the assertion that catches a retry loop that re-serializes the payload and loses a field.

Keeping exceptions as the outcomes, rather than magic strings, means a regression shows up as a type or attribute mismatch rather than a string comparison that silently drifts.

The wrapper under test

For reference, here is a wrapper the suite is designed to hold still. Yours will differ; the tests should not have to.

def with_retries(tool, payload, max_attempts=3):
    last = None
    for _ in range(max_attempts):
        try:
            return tool.call(payload)
        except RetryableError as exc:
            last = exc
    raise last

Note the last line: the wrapper re-raises the error object it caught, not a freshly constructed one. That single choice is what the third test verifies, and it is the choice most likely to be quietly reversed when somebody later "improves" the error message.

This test looks almost too simple to write. Write it anyway, because it is the only one that catches a wrapper that retries unconditionally.

def test_success_first_try():
    tool = FakeTool(outcomes=["ok"])
    result = with_retries(tool, {"id": 42})
    assert tool.calls == 1
    assert result == "ok"

The count assertion is the whole point. A wrapper that loops until success, with no check for whether the first attempt already succeeded, would still return "ok" and would still fail this test with calls == 2 or more. That bug is easy to introduce when the retry loop and the actual call get refactored into separate functions.

If the wrapper returns a structured result instead of raising, assert on the outcome value and on any attempt counter attached to it. If the wrapper decorates the result with metering metadata (an attempt number, a timestamp, a cost estimate), assert that number directly: result.attempts == 1 is clearer than inferring it from the fake.

This is the test that proves the wrapper actually retries, as opposed to swallowing a failure and moving on. The failure-then-success sequence is the canonical shape; the Quickmock guide describes exactly this pattern as the one retry tests need, with the mock walking the sequence per call in a fixed order and then repeating the Quickmock retry guide.

def test_retry_then_success():
    tool = FakeTool(outcomes=[RetryableError("429"), "ok"])
    result = with_retries(tool, {"id": 42})
    assert tool.calls == 2
    assert result == "ok"

Two assertions, two different failures. calls == 2 catches a wrapper that gave up after the first error. result == "ok" catches one that gave up but returned a stale or empty value. Assert only one of them and you get a test that passes for the wrong reason roughly half the time.

The retryable error should be your own type, not the raw HTTP status. Retrying everything is as wrong as retrying nothing: a 500 or 503 is worth another attempt, a 400 or 422 will fail identically forever, and a 429 should honor the server's Retry-After header rather than follow your own schedule the Quickmock retry guide. So the fake raises a RetryableError that your wrapper's classifier maps from a status code, and a separate test outside this suite's three covers the classification table itself.

The third test is the one people get wrong, because it checks two things at once: that the wrapper stopped, and that it stopped for the right reason.

import pytest

def test_cap_exhausted_preserves_cause():
    original = RetryableError("upstream 503 after 12s")
    tool = FakeTool(outcomes=[original])
    with pytest.raises(RetryableError) as caught:
        with_retries(tool, {"id": 42}, max_attempts=3)
    assert tool.calls == 3
    assert caught.value is original
    assert "503" in str(caught.value)

caught.value is original is an identity check, not an equality check. It fails if the wrapper catches the original error and raises a fresh RetryableError("retries exhausted"), which destroys the diagnostic detail the on-call engineer needs. That is the "cause is preserved rather than replaced" requirement, and identity is the strictest way to express it. If your service has to re-wrap, assert on the chain instead: in Python, caught.value.__cause__ is original; in Java, assertSame(original, thrown.getCause()).

For the mechanics, wrapping in a try/catch that expects the exact type, fails if no exception is thrown, and re-fails on any other exception is more precise than the annotation form, because you can check other state after the catch the exception assertion thread.

The cap itself should come from a named constant, not a literal in the test. If the test hardcodes three while production moves to five, the test keeps passing while production misbehaves.

What the three tests cover, side by side

That table doubles as the specification. When somebody proposes a change to the retry policy, the rows tell you which test should move.

Keep the sequences as named fixtures so a policy change makes exactly one test fail. Here is what that looks like:

import pytest

SEQUENCES = {
    "clean_success": ["ok"],
    "one_retry": [RetryableError("429"), "ok"],
    "cap_exhausted": [RetryableError("503")],
}

EXPECTED_CALLS = {
    "clean_success": 1,
    "one_retry": 2,
    "cap_exhausted": 3,
}

@pytest.fixture(params=SEQUENCES.items(), ids=SEQUENCES.keys())
def scenario(request):
    name, outcomes = request.param
    return name, FakeTool(outcomes=outcomes), EXPECTED_CALLS[name]

The parametrized ids show up as three separate cases in the report, so a CI failure names the scenario. The expected count travels with the sequence, which is the coupling that matters: nobody can change the outcome list without deciding whether the count should change too.

Now if someone changes the wrapper to stop after two attempts, cap_exhausted fails on the count and the other two keep passing. That is the whole value of the fixture: the failure points at the policy, not at the harness. This is the same idea as the ordering guarantee in a response-sequence mock, where each call consumes one step of the sequence deterministically the Quickmock retry guide.

Do not stop at ordering, though. A fixed sequence proves the count; it does not prove the delay. Record a timestamp alongside each call if backoff is part of the contract, and assert that the gaps grow the way the policy says they should, because counting attempts proves you retried but not that you waited the Quickmock retry guide. Mock frameworks that only return canned values will not do this for you; the fake has to record times itself.

Keeping the suite honest

Two habits keep this suite useful over time.

First, assert the count before the result. In a test with two assertions, the first failure message is the one people read. expected 2 calls, got 1 tells the reviewer the retry did not happen; result was None sends them to the wrong layer.

Second, do not let the suite grow a fourth sequence without a reason. Each sequence should map to a decision the wrapper makes: succeed, retry, give up. If you catch yourself adding "two retryable errors then success" as a separate test, check whether the cap test's count assertion already covers it. Usually it does, and the extra test just duplicates the fixture with a different number.

If the wrapper sits in front of a model call rather than a flaky HTTP tool, the same three tests apply almost unchanged, but the attempt cap becomes a cost control as well as a reliability control: a turn that retries twelve times on a misconfigured tool can spend many times what a single request should have cost, which is why a per-turn retry budget is treated as non-optional in production the MotoMtech agent retry guide. When the tool is a model endpoint reached through a multi-provider gateway, pointing the fake at a recorded response and swapping in a real client later is the cheap way to keep the suite deterministic; Quanta Inference exposes text, image, video, audio and embedding models through one OpenAI-compatible API key, so the client your fake replaces keeps the same shape across providers (https://quantaintelligence.ai/inference).

The three tests prove the wrapper's control flow. They do not prove that retrying is safe, and that distinction matters more with every wrapper you add.

A timeout after dispatch does not establish whether a tool's side effect committed: a payment service may have accepted a charge even though the caller never received the response, so retrying because the call looked unsuccessful can repeat work that already happened the Formation retry and recovery guide. Your sequence tests will pass while the wrapper double-charges, because the fake has no side effect to duplicate.

The fix lives outside the wrapper's loop. The identity of the operation has to be generated by the orchestration layer, derived from the durable execution step, persisted, and reused across transport retries, model-requested retries and recovery after a restart, rather than synthesized fresh on each attempt the Formation retry and recovery guide. The same guide separates three retry triggers (timeout or transport loss, malformed output, and model judgment) and notes that only the first leaves the outcome genuinely unknown; a model asking for another attempt is not evidence that the previous one failed.

So add a fourth kind of test, separate from the three: run the same payload through the wrapper twice under the same operation identity and assert that the downstream deduplication store records one effect, not two. That needs a fake that records writes, which is a bigger fake than the one above, but it is the only one that catches a duplicate side effect.

Per-attempt telemetry is the other gap. Counting calls in-process proves behavior in the test; it does not tell you how many attempts production traffic makes. If the wrapper is instrumented, emit one span per attempt with the operation name and the outcome, so the attempts-per-operation distribution is queryable afterwards. OpenTelemetry's GenAI and MCP semantic conventions remain in Development status, so pin the attribute names you rely on and treat silent renames as a breaking change to your dashboards the Greptime conventions walkthrough. The agent span specification defines a tool execution span with a required tool name, which is the natural place to attach an attempt counter the GenAI agent span spec. Keeping attribute names consistent across providers is the same problem the conventions exist to solve: without a shared gen_ai.* vocabulary, two engineers instrument the same call with three different field names and nothing composes into one dashboard the OpenObserve practical guide.