Unit test a Sume 429 retry loop in Python with a fake opener

A retry loop is the code you only exercise on a bad day. If the first time it runs is a real 429 during a paid batch, a bug in it costs you either a stalled job queue or a second charge. You can test it without the network and without spending anything, if the loop takes its HTTP call as an argument instead of reaching for a global.

The Sume API sends retry-after on its 429 responses, and the error body carries code: rate_limited with details describing which budget you spent. A loop that honors that header is simple to state: wait what the server says, then try again, and stop after a fixed number of tries. Each of those clauses is one test.

Two seams make it testable

• opener is the function that performs the request. In production it is urllib.request.urlopen with a prepared request. In a test it is a Mock whose side_effect is a list: an exception to raise, or a response to return, one per call.
• sleep is injected too and defaults to a no-op here. In production pass time.sleep. In a test pass list.append and assert on what was recorded, so the suite runs in milliseconds and still proves the wait was 7 seconds.

The loop and its two tests

Everything below is in the standard library and runs as one file. HTTPError is built with the same shape the API returns: status 429, a retry-after header and a JSON body that json.load can read. The ok() helper stands in for a response object used as a context manager.

import io, json, unittest, urllib.error
from unittest import mock
def get_json(opener, url, tries=3, sleep=lambda s: None):
    for attempt in range(tries):
        try:
            with opener(url) as r:
                return json.load(r)
        except urllib.error.HTTPError as e:
            if e.code != 429 or attempt == tries - 1:
                raise
            sleep(float(e.headers.get("retry-after") or 2**attempt))

def too_many(wait):
    body = io.BytesIO(b'{"error": {"code": "rate_limited"}}')
    return urllib.error.HTTPError("u", 429, "Too Many", {"retry-after": wait}, body)

ok = mock.MagicMock()
ok.__enter__.side_effect = lambda: io.BytesIO(b'{"data": {"terminal": true}}')
class RetryTests(unittest.TestCase):
    def test_waits_the_hinted_time(self):
        opener, waits = mock.Mock(side_effect=[too_many("7"), ok]), []
        self.assertTrue(get_json(opener, "u", sleep=waits.append)["data"]["terminal"])
        self.assertEqual(waits, [7.0])
    def test_gives_up_after_the_last_try(self):
        opener = mock.Mock(side_effect=[too_many("0")] * 3)
        with self.assertRaises(urllib.error.HTTPError):
            get_json(opener, "u")
        self.assertEqual(opener.call_count, 3)
unittest.main(argv=["t"], exit=False)

What each test pins down

Keep the test honest

• Use a real error body in the fixtures. Copy one from a failed call of your own, so a change in the envelope breaks the test and not production.
• Test the policy and not the transport. This suite proves when to retry. It cannot prove that a POST is replay-safe, which depends on sending the same Idempotency-Key on every attempt.
• Add one test per status you decided not to retry. For Sume a 402 insufficient_credits is not retryable and needs funds, so it should raise on the first call.

The header and budget rules are in the authentication guide. The mocking tools used here are described in the unittest.mock docs.