"""Tests for HTTP 429 Retry-After handling and jittered backoff in gitea_auth. All timing is made deterministic by injecting ``sleep_func``, ``rand_func`` and ``now_func`` — no real sleeping or randomness, and no real network calls. """ import io import sys import email.message import unittest import urllib.error from email.utils import formatdate from unittest.mock import MagicMock, patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import gitea_auth # noqa: E402 FIXED_NOW = 1_000_000.0 def _http_error(code, retry_after=None, body=b"error"): """Build a urllib HTTPError with an optional Retry-After header.""" hdrs = email.message.Message() if retry_after is not None: hdrs["Retry-After"] = retry_after return urllib.error.HTTPError("http://example.test", code, "msg", hdrs, io.BytesIO(body)) class _FakeResponse: """Minimal context-manager stand-in for a urlopen() success.""" def __init__(self, body=b""): self._body = body def __enter__(self): return self def __exit__(self, *exc): return False def read(self): return self._body def _call(side_effect, **kwargs): """Run api_request with urlopen patched to the given side_effect list.""" defaults = dict( sleep_func=MagicMock(), rand_func=lambda: 0.5, now_func=lambda: FIXED_NOW, ) defaults.update(kwargs) with patch("gitea_auth.urllib.request.urlopen", side_effect=side_effect): result = gitea_auth.api_request("GET", "http://example.test", "token x", **defaults) return result, defaults["sleep_func"] class TestParseRetryAfter(unittest.TestCase): def test_seconds_form(self): self.assertEqual(gitea_auth.parse_retry_after("120"), 120) def test_zero_seconds(self): self.assertEqual(gitea_auth.parse_retry_after("0"), 0) def test_negative_seconds_clamped(self): self.assertEqual(gitea_auth.parse_retry_after("-5"), 0) def test_http_date_future(self): header = formatdate(FIXED_NOW + 30, usegmt=True) delay = gitea_auth.parse_retry_after(header, now=FIXED_NOW) self.assertAlmostEqual(delay, 30.0, delta=1.0) def test_http_date_past_clamped_to_zero(self): header = formatdate(FIXED_NOW - 30, usegmt=True) delay = gitea_auth.parse_retry_after(header, now=FIXED_NOW) self.assertEqual(delay, 0.0) def test_none_returns_none(self): self.assertIsNone(gitea_auth.parse_retry_after(None)) def test_blank_returns_none(self): self.assertIsNone(gitea_auth.parse_retry_after(" ")) def test_garbage_returns_none(self): self.assertIsNone(gitea_auth.parse_retry_after("soon")) def test_float_string_returns_none(self): # Non-integer numerics are not valid seconds and are not HTTP-dates. self.assertIsNone(gitea_auth.parse_retry_after("1.5")) class TestBackoffDelay(unittest.TestCase): def test_zero_jitter_gives_zero(self): self.assertEqual(gitea_auth.backoff_delay(0, base=1.0, cap=60.0, rand=lambda: 0.0), 0.0) def test_full_jitter_gives_ceiling(self): # attempt=3 -> base*2**3 = 8, below cap self.assertEqual(gitea_auth.backoff_delay(3, base=1.0, cap=60.0, rand=lambda: 1.0), 8.0) def test_delay_capped(self): # base*2**10 = 1024, capped at 60 self.assertEqual(gitea_auth.backoff_delay(10, base=1.0, cap=60.0, rand=lambda: 1.0), 60.0) def test_delay_within_bounds(self): for attempt in range(6): for r in (0.0, 0.25, 0.5, 0.75, 1.0): delay = gitea_auth.backoff_delay(attempt, base=1.0, cap=30.0, rand=lambda: r) ceiling = min(30.0, 1.0 * (2 ** attempt)) self.assertGreaterEqual(delay, 0.0) self.assertLessEqual(delay, ceiling) class TestApiRequestRetry(unittest.TestCase): def test_success_no_retry(self): result, sleep = _call([_FakeResponse(b'{"ok": true}')]) self.assertEqual(result, {"ok": True}) sleep.assert_not_called() def test_non_429_error_raises_immediately(self): with self.assertRaises(RuntimeError) as ctx: _call([_http_error(500, body=b"boom")]) self.assertIn("HTTP 500", str(ctx.exception)) def test_non_429_error_does_not_sleep(self): sleep = MagicMock() with self.assertRaises(RuntimeError): _call([_http_error(404)], sleep_func=sleep) sleep.assert_not_called() def test_honors_retry_after_seconds(self): result, sleep = _call([ _http_error(429, retry_after="5"), _FakeResponse(b'{"ok": true}'), ]) self.assertEqual(result, {"ok": True}) sleep.assert_called_once_with(5) def test_honors_retry_after_http_date(self): header = formatdate(FIXED_NOW + 42, usegmt=True) result, sleep = _call([ _http_error(429, retry_after=header), _FakeResponse(b'{"ok": true}'), ]) self.assertEqual(result, {"ok": True}) self.assertEqual(sleep.call_count, 1) self.assertAlmostEqual(sleep.call_args[0][0], 42.0, delta=1.0) def test_invalid_retry_after_falls_back_to_backoff(self): # rand=1.0, base=2, attempt 0 -> ceiling min(100, 2*1) = 2.0 result, sleep = _call( [_http_error(429, retry_after="whenever"), _FakeResponse(b"{}")], rand_func=lambda: 1.0, base_delay=2.0, max_delay=100.0, ) self.assertEqual(result, {}) sleep.assert_called_once_with(2.0) def test_missing_retry_after_falls_back_to_backoff(self): result, sleep = _call( [_http_error(429), _FakeResponse(b"{}")], rand_func=lambda: 1.0, base_delay=1.0, max_delay=100.0, ) self.assertEqual(result, {}) sleep.assert_called_once_with(1.0) def test_gives_up_after_max_retries(self): # max_retries=3 -> 3 sleeps, then the 4th failure raises. errors = [_http_error(429, retry_after="1") for _ in range(4)] sleep = MagicMock() with self.assertRaises(RuntimeError) as ctx: _call(errors, sleep_func=sleep, max_retries=3) self.assertIn("HTTP 429", str(ctx.exception)) self.assertEqual(sleep.call_count, 3) def test_no_infinite_loop_when_always_429(self): # Far more failures than retries available; must still terminate. errors = [_http_error(429) for _ in range(20)] sleep = MagicMock() with self.assertRaises(RuntimeError): _call(errors, sleep_func=sleep, max_retries=2) self.assertEqual(sleep.call_count, 2) def test_recovers_after_several_429s(self): result, sleep = _call([ _http_error(429, retry_after="1"), _http_error(429, retry_after="1"), _FakeResponse(b'{"done": 1}'), ], max_retries=5) self.assertEqual(result, {"done": 1}) self.assertEqual(sleep.call_count, 2) if __name__ == "__main__": unittest.main()