beam: convert retry_jitter tests to TestCase + mutant-killing assertions (Q59)
This commit is contained in:
beam committed
1 parent
b2a4d1634f
commit
a9bc1f7080
1 file changed
+136
-56
@@ -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()
|
||||
Reference in new issue
Block a user