Gitea-Tools/tests/test_retry_backoff.py

"""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()