145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
import sys, os, time, random, unittest
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from retry_jitter import retry
|
|
|
|
|
|
class FakeRng:
|
|
"""Deterministic rng whose uniform(a, b) returns (a+b)/2."""
|
|
def uniform(self, a, b):
|
|
return (a + b) / 2.0
|
|
|
|
|
|
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_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_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()
|