From a9bc1f7080429ded8ff174d5ef9aba6e9b627b98 Mon Sep 17 00:00:00 2001 From: beam Date: Thu, 29 Oct 2026 14:38:19 +0000 Subject: [PATCH] beam: convert retry_jitter tests to TestCase + mutant-killing assertions (Q59) --- wizard-work/beam/test_retry_jitter.py | 192 ++++++++++++++++++-------- 1 file changed, 136 insertions(+), 56 deletions(-) diff --git a/wizard-work/beam/test_retry_jitter.py b/wizard-work/beam/test_retry_jitter.py index 9634302..a4ee1b7 100644 --- a/wizard-work/beam/test_retry_jitter.py +++ b/wizard-work/beam/test_retry_jitter.py @@ -1,64 +1,144 @@ -import sys, os, time +import sys, os, time, random, unittest sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from retry_jitter import retry -def test_success_no_sleep(): - calls = {"n":0} - sleeps = [] - def fn(): - calls["n"]+=1 - return 42 - assert retry(fn, sleep=sleeps.append) == 42 - assert calls["n"]==1 and sleeps==[] -def test_retries_then_succeeds(): - calls={"n":0} - def fn(): - calls["n"]+=1 - if calls["n"]<3: raise ValueError("boom") - return "ok" - assert retry(fn, max_attempts=3, base_delay=0, sleep=lambda x:None)=="ok" - assert calls["n"]==3 +class FakeRng: + """Deterministic rng whose uniform(a, b) returns (a+b)/2.""" + def uniform(self, a, b): + return (a + b) / 2.0 -def test_exhausted_raises_original(): - class E(Exception): pass - def fn(): - raise E("x") - try: - retry(fn, max_attempts=2, base_delay=0, sleep=lambda x:None) - except E as e: - assert str(e)=="x" - else: - raise AssertionError("no raise") -def test_nonretryable_propagates(): - class A(Exception): pass - class B(Exception): pass - def fn(): - raise B("no") - try: - retry(fn, retryable=(A,), max_attempts=3, base_delay=0, sleep=lambda x:None) - except B: - pass - else: - raise AssertionError("not propagated") +class TestRetryJitter(unittest.TestCase): + def test_success_no_sleep(self): + calls = {"n": 0} + sleeps = [] + def fn(): + calls["n"] += 1 + return 42 + self.assertEqual(retry(fn, sleep=sleeps.append), 42) + self.assertEqual(calls["n"], 1) + self.assertEqual(sleeps, []) -def test_max_attempts_validation(): - try: - retry(lambda: 1, max_attempts=0) - except ValueError: - pass - else: - raise AssertionError("no validation") + def test_retries_then_succeeds(self): + calls = {"n": 0} + def fn(): + calls["n"] += 1 + if calls["n"] < 3: + raise ValueError("boom") + return "ok" + self.assertEqual(retry(fn, max_attempts=3, base_delay=0, sleep=lambda x: None), "ok") + self.assertEqual(calls["n"], 3) -def test_jitter_bounded(): - rng=__import__("random").Random(1) - sleeps=[] - def fn(): - raise ValueError("e") - try: - retry(fn, max_attempts=4, base_delay=1.0, max_delay=2.0, rng=rng, sleep=sleeps.append) - except ValueError: - pass - for d in sleeps: - assert 0.0 <= d <= 2.0 + def test_exhausted_raises_original(self): + class E(Exception): + pass + def fn(): + raise E("x") + with self.assertRaises(E) as ctx: + retry(fn, max_attempts=2, base_delay=0, sleep=lambda x: None) + self.assertEqual(str(ctx.exception), "x") + + def test_nonretryable_propagates(self): + class A(Exception): + pass + class B(Exception): + pass + calls = {"n": 0} + def fn(): + calls["n"] += 1 + raise B("no") + with self.assertRaises(B): + retry(fn, retryable=(A,), max_attempts=3, base_delay=0, sleep=lambda x: None) + # non-retryable must not be retried + self.assertEqual(calls["n"], 1) + + def test_max_attempts_validation(self): + with self.assertRaises(ValueError): + retry(lambda: 1, max_attempts=0) + + def test_jitter_bounded(self): + rng = random.Random(1) + sleeps = [] + def fn(): + raise ValueError("e") + with self.assertRaises(ValueError): + retry(fn, max_attempts=4, base_delay=1.0, max_delay=2.0, rng=rng, sleep=sleeps.append) + for d in sleeps: + self.assertGreaterEqual(d, 0.0) + self.assertLessEqual(d, 2.0) + self.assertEqual(len(sleeps), 3) + + # --- mutant-killing additions --- + + def test_jitter_is_full_not_constant(self): + # full jitter: delay uniform over [0, exp]; with FakeRng midpoint => 0.5*exp + sleeps = [] + def fn(): + raise ValueError("e") + with self.assertRaises(ValueError): + retry(fn, max_attempts=2, base_delay=4.0, max_delay=10.0, + rng=FakeRng(), sleep=sleeps.append) + self.assertEqual(len(sleeps), 1) + self.assertAlmostEqual(sleeps[0], 2.0, places=9) # 0.5 * 4.0 + + def test_sleep_is_called_on_retry(self): + sleeps = [] + def fn(): + raise ValueError("e") + with self.assertRaises(ValueError): + retry(fn, max_attempts=2, base_delay=1.0, max_delay=10.0, + rng=FakeRng(), sleep=sleeps.append) + self.assertEqual(len(sleeps), 1) + + def test_exponential_growth(self): + sleeps = [] + def fn(): + raise ValueError("e") + with self.assertRaises(ValueError): + retry(fn, max_attempts=4, base_delay=1.0, max_delay=100.0, + rng=FakeRng(), sleep=sleeps.append) + # exp doubles each attempt: 1,2,4 -> delays 0.5,1.0,2.0 + self.assertEqual(len(sleeps), 3) + for i in range(1, len(sleeps)): + self.assertAlmostEqual(sleeps[i] / sleeps[i - 1], 2.0, places=9) + + def test_negative_jitter_clamped(self): + class NegRng: + def uniform(self, a, b): + return -5.0 + sleeps = [] + def fn(): + raise ValueError("e") + with self.assertRaises(ValueError): + retry(fn, max_attempts=2, base_delay=1.0, max_delay=10.0, + rng=NegRng(), sleep=sleeps.append) + self.assertEqual(len(sleeps), 1) + self.assertGreaterEqual(sleeps[0], 0.0) + + def test_original_exception_instance_preserved(self): + class E(Exception): + pass + exc = E("x") + def fn(): + raise exc + try: + retry(fn, max_attempts=1, base_delay=0, sleep=lambda x: None) + except E as e: + self.assertIs(e, exc) + else: + self.fail("no raise") + + def test_cap_respected(self): + sleeps = [] + def fn(): + raise ValueError("e") + with self.assertRaises(ValueError): + retry(fn, max_attempts=2, base_delay=100.0, max_delay=2.0, + rng=FakeRng(), sleep=sleeps.append) + self.assertEqual(len(sleeps), 1) + self.assertLessEqual(sleeps[0], 2.0) + + +if __name__ == "__main__": + unittest.main()