"""Stdlib-only tests (no pytest): plain asserts in functions called from a __main__ block. Run: `python3 test_circuit_breaker.py`. The headline tests pin the demo trace and the arithmetic the commentary derives from it (see commentary.html, sections 6 and 7). The rest are unit checks on the state machine and on each dependency in isolation. """ from circuit_breaker import (CircuitBreaker, DeadDependency, NoBreaker, OverloadedDependency, drained_at, simulate, totals) BASE, SPIKE, CAP, SURGE = 5, 40, 20, (10, 20) def trace(ticks, base=BASE, spike=SPIKE): return [spike if SURGE[0] <= t < SURGE[1] else base for t in range(ticks)] def breaker(**overrides): """The demo's configuration, with named overrides for counterfactuals.""" kw = dict(window=20, min_calls=10, error_rate=0.5, cooldown=5, probes=1) kw.update(overrides) return CircuitBreaker(**kw) # --- the two dependencies' defining properties ------------------------- def test_dead_dependency_ignores_load(): """Its verdict is a function of `now` alone: 1 call and 10_000 calls get the same answer, and the answer changes only when the clock does.""" dep = DeadDependency(10, 20) assert dep.call(15, 1) == (0, 1) assert dep.call(15, 10_000) == (0, 10_000) assert dep.call(20, 1) == (1, 0) assert dep.call(20, 10_000) == (10_000, 0) def test_overloaded_dependency_ignores_time(): """Its verdict is a function of `k` alone. It is never 'down': at or below capacity it is perfect, at any tick.""" dep = OverloadedDependency(20) for now in (0, 15, 999): assert dep.call(now, 1) == (1, 0) assert dep.call(now, 20) == (20, 0) assert dep.call(now, 40) == (20, 20) # --- the state machine -------------------------------------------------- def test_trips_on_error_rate_not_before_min_calls(): """9 failures do not trip a breaker with min_calls=10; the 10th does.""" b = breaker() for _ in range(9): b.record(False, 0) assert b.state == "closed" b.record(False, 0) assert b.state == "open" def test_open_rejects_until_exactly_cooldown(): """`now - opened_at < cooldown` rejects, so a cooldown of 5 opened at t=0 rejects through t=4 and probes at t=5.""" b = breaker() for _ in range(10): b.record(False, 0) assert b.state == "open" assert [b.allow(t) for t in range(1, 5)] == [False, False, False, False] assert b.allow(5) is True assert b.state == "half_open" def test_half_open_admits_exactly_probes_per_tick(): """One probe per tick, and the budget is refreshed on the next tick — which is why a half-open breaker still sheds a whole herd.""" b = breaker() for _ in range(10): b.record(False, 0) assert [b.allow(5) for _ in range(4)] == [True, False, False, False] assert [b.allow(6) for _ in range(2)] == [True, False] def test_one_probe_decides_everything(): """A single success closes; a single failure re-opens and restarts the cooldown. The breaker never asks for a second opinion.""" b = breaker() for _ in range(10): b.record(False, 0) b.allow(5) b.record(True, 5) assert b.state == "closed" b = breaker() for _ in range(10): b.record(False, 0) b.allow(5) b.record(False, 5) assert b.state == "open" assert b.opened_at == 5 # --- the aha ------------------------------------------------------------ def test_probe_verdict_generalizes_only_for_the_dead_dependency(): """Same trace, same breaker. Every probe against the dead dependency is a correct verdict; every probe against the overloaded one succeeds and is contradicted by the very next tick. """ rows = trace(45) dead = simulate(DeadDependency(*SURGE), breaker(), rows) over = simulate(OverloadedDependency(CAP), breaker(), rows) dead_probes = [(r.now, r.ok, r.failed) for r in dead if r.state == "half_open"] # t=15 the dependency is still down and the probe correctly fails; # t=20 it is back and the probe correctly succeeds. assert dead_probes == [(15, 0, 1), (20, 1, 0)] # The tick after the successful probe serves the entire 409-deep herd. assert (dead[21].admitted, dead[21].ok, dead[21].failed) == (409, 409, 0) over_probes = [(r.now, r.ok, r.failed) for r in over if r.state == "half_open"] assert over_probes == [(15, 1, 0), (21, 1, 0), (27, 1, 0), (33, 1, 0), (39, 1, 0)] # all five succeed after = [(over[n + 1].admitted, over[n + 1].ok, over[n + 1].failed) for n, _, _ in over_probes] assert after == [(259, 20, 239), (373, 20, 353), (382, 20, 362), (391, 20, 371), (400, 20, 380)] # all five refuted def test_headline_the_incident_never_ends(): """The dead dependency's incident is over one tick after it recovers. The overloaded one's 10-tick surge becomes an unbounded outage.""" rows = trace(400) assert drained_at(simulate(DeadDependency(*SURGE), NoBreaker(), rows), SURGE[1]) == 20 assert drained_at(simulate(DeadDependency(*SURGE), breaker(), rows), SURGE[1]) == 21 assert drained_at(simulate(OverloadedDependency(CAP), NoBreaker(), rows), SURGE[1]) == 33 assert drained_at(simulate(OverloadedDependency(CAP), breaker(), rows), SURGE[1]) is None # ...and the breaker costs the dead dependency nothing in throughput. assert totals(simulate(DeadDependency(*SURGE), NoBreaker(), rows))[ "completed"] == 2350 assert totals(simulate(DeadDependency(*SURGE), breaker(), rows))[ "completed"] == 2350 def test_limit_cycle_arithmetic(): """The cycle is 6 ticks long and completes 21 requests: one closed tick at capacity (20) plus one probe (1). Against 5 arrivals/tick that is a deficit of 30 - 21 = 9 requests per cycle, forever. """ rows = simulate(OverloadedDependency(CAP), breaker(), trace(400)) a, b = rows[300:306], rows[306:312] assert sum(r.ok for r in a) == 21 assert sum(r.ok for r in b) == 21 assert sum(r.arrivals for r in a) == 30 assert rows[306].pending - rows[300].pending == 9 assert rows[312].pending - rows[306].pending == 9 def test_cooldown_is_the_load_bearing_knob(): """throughput = (capacity + probes) / (cooldown + 1), stable iff it exceeds the 5/tick baseline. cooldown=3 gives 5.25 and drains (slowly); cooldown=4 gives 4.20 and grows forever. The error-rate threshold, the knob everyone reaches for, changes nothing at all. """ long = trace(3000) assert drained_at(simulate(OverloadedDependency(CAP), breaker(cooldown=3), long), SURGE[1]) == 1315 assert drained_at(simulate(OverloadedDependency(CAP), breaker(cooldown=4), long), SURGE[1]) is None rows = trace(400) baseline = totals(simulate(OverloadedDependency(CAP), breaker(), rows)) for rate in (0.30, 0.50, 0.70, 0.90, 0.99): assert totals(simulate(OverloadedDependency(CAP), breaker(error_rate=rate), rows)) == baseline def test_boundary_no_overload_no_divergence(): """Where the effect vanishes: if the surge never exceeds capacity the breaker never trips, and the two arms are identical.""" rows = trace(400, spike=CAP) with_breaker = simulate(OverloadedDependency(CAP), breaker(), rows) without = simulate(OverloadedDependency(CAP), NoBreaker(), rows) assert totals(with_breaker)["trips"] == 0 assert totals(with_breaker) == totals(without) def test_bounded_retries_are_a_co_author(): """Capping attempts at 3 ends the limit cycle — so the unbounded retry policy is half the cause. The breaker still owns its share: 358 dropped against no-breaker's 160 on the overloaded dependency, and roughly nothing extra on the dead one. """ rows = trace(400) over_nb = totals(simulate(OverloadedDependency(CAP), NoBreaker(), rows, max_attempts=3)) over_br = totals(simulate(OverloadedDependency(CAP), breaker(), rows, max_attempts=3)) assert over_nb["dropped"] == 160 assert over_br["dropped"] == 358 # 2.2x, but finite assert over_br["backlog"] == 0 dead_nb = totals(simulate(DeadDependency(*SURGE), NoBreaker(), rows, max_attempts=3)) dead_br = totals(simulate(DeadDependency(*SURGE), breaker(), rows, max_attempts=3)) assert dead_nb["dropped"] == 320 assert dead_br["dropped"] == 359 # 39 extra, all from one probe TESTS = [ test_dead_dependency_ignores_load, test_overloaded_dependency_ignores_time, test_trips_on_error_rate_not_before_min_calls, test_open_rejects_until_exactly_cooldown, test_half_open_admits_exactly_probes_per_tick, test_one_probe_decides_everything, test_probe_verdict_generalizes_only_for_the_dead_dependency, test_headline_the_incident_never_ends, test_limit_cycle_arithmetic, test_cooldown_is_the_load_bearing_knob, test_boundary_no_overload_no_divergence, test_bounded_retries_are_a_co_author, ] if __name__ == "__main__": failed = 0 for test in TESTS: try: test() except AssertionError as exc: failed += 1 print(f"FAIL {test.__name__}: {exc}") else: print(f"PASS {test.__name__}") if failed: print(f"\n{failed} of {len(TESTS)} tests FAILED") raise SystemExit(1) print(f"\nAll {len(TESTS)} tests PASSED")