"""Every number the commentary claims, pinned. Plain asserts, no pytest. python3 test_detector.py """ import statistics from detector import ( HEARTBEAT_INTERVAL_MS, FixedDetector, PhiDetector, evaluate, heartbeat_trace, time_to_suspect, ) CALM = (10.0, 3.0) CONGESTED = (80.0, 40.0) PHASES = [(40, *CALM), (30, *CONGESTED), (30, *CALM)] TRACES = 200 tests = [] def test(fn): tests.append(fn) return fn # ---- the fixed detector -------------------------------------------------- @test def fixed_suspects_strictly_after_the_timeout(): d = FixedDetector(300.0) d.heartbeat(1000.0) assert d.suspects(1300.0) is False, "at exactly the timeout it is still alive" assert d.suspects(1300.001) is True @test def fixed_detects_at_last_arrival_plus_timeout(): d = FixedDetector(300.0) d.heartbeat(1000.0) assert abs(time_to_suspect(d, 1000.0) - 1300.0) < 1e-6 # ---- phi ----------------------------------------------------------------- def jittery(n=50, seed=1): """A detector fed a stream with real variance, so sigma is fitted rather than clamped to the floor.""" import random rng = random.Random(seed) d = PhiDetector() t = 0.0 for _ in range(n): t += rng.gauss(100.0, 12.0) d.heartbeat(t) return d, t @test def phi_is_small_at_the_mean_and_grows_with_silence(): d, t = jittery() at_mean = d.phi(t + 100.0) assert at_mean < 1.0, at_mean assert d.phi(t + 140.0) > at_mean assert d.phi(t + 180.0) > d.phi(t + 140.0) @test def phi_saturates_at_300_because_the_normal_tail_underflows(): """The 1e-300 clamp caps phi. Past that point phi cannot tell a 200ms silence from a 4-second one — which never matters for a convict/acquit decision at threshold 8, but does mean phi is not a distance.""" d = PhiDetector() for k in range(1, 51): d.heartbeat(k * 100.0) # perfectly regular: sigma clamps to the floor assert d.phi(5000.0 + 200.0) == 300.0 assert d.phi(5000.0 + 4000.0) == 300.0 @test def phi_is_monotonic_in_now_which_is_what_makes_bisection_valid(): d = PhiDetector() for k in range(1, 51): d.heartbeat(k * 100.0) last = -1.0 for step in range(0, 800, 5): now = 5000.0 + step assert d.phi(now) >= last - 1e-9 last = d.phi(now) @test def the_first_heartbeat_records_no_interval(): d = PhiDetector(bootstrap=3) assert list(d.intervals) == [HEARTBEAT_INTERVAL_MS] * 3 d.heartbeat(4321.0) # a join at an arbitrary wall-clock time assert list(d.intervals) == [HEARTBEAT_INTERVAL_MS] * 3 d.heartbeat(4421.0) assert list(d.intervals) == [HEARTBEAT_INTERVAL_MS] * 3 + [100.0] @test def the_sigma_floor_clamps_a_perfectly_regular_stream(): """With zero variance the fitted sigma is 0; the floor is what stops a division by zero from deciding whether the cluster stays up.""" d = PhiDetector(min_stddev_ms=5.0, bootstrap=10) for k in range(1, 31): d.heartbeat(k * 100.0) assert statistics.pstdev(d.intervals) == 0.0 assert d.phi(3000.0 + 100.0) < 1.0 # at the mean: unremarkable assert 1.0 < d.phi(3000.0 + 115.0) < 8.0 # 3 floors out: suspicious assert d.phi(3000.0 + 150.0) > 8.0 # 10 floors out: convicted # ---- the trace ----------------------------------------------------------- @test def the_trace_is_monotonic_and_the_right_length(): arrivals, crash_due = heartbeat_trace(0, PHASES) assert len(arrivals) == 100 assert all(b > a for a, b in zip(arrivals, arrivals[1:])) assert crash_due == 100 * HEARTBEAT_INTERVAL_MS @test def the_trace_is_reproducible_and_seed_dependent(): assert heartbeat_trace(7, PHASES)[0] == heartbeat_trace(7, PHASES)[0] assert heartbeat_trace(7, PHASES)[0] != heartbeat_trace(8, PHASES)[0] # ---- the headline -------------------------------------------------------- def census(make_detector): alarms, latencies = 0, [] for seed in range(TRACES): arrivals, crash_due = heartbeat_trace(seed, PHASES) fa, latency = evaluate(make_detector(), arrivals, crash_due) alarms += fa latencies.append(latency) return alarms, statistics.fmean(latencies) @test def phi_at_the_standard_threshold_raises_384_false_alarms(): alarms, latency = census(lambda: PhiDetector(threshold=8.0)) assert alarms == 384, alarms assert abs(latency - 179.7) < 0.05, latency @test def a_fixed_300ms_timeout_raises_none(): alarms, latency = census(lambda: FixedDetector(300.0)) assert alarms == 0, alarms assert abs(latency - 209.8) < 0.05, latency @test def fixed_250ms_dominates_phi_8_on_both_axes(): p_alarms, p_latency = census(lambda: PhiDetector(threshold=8.0)) f_alarms, f_latency = census(lambda: FixedDetector(250.0)) assert f_alarms < p_alarms, (f_alarms, p_alarms) assert f_latency < p_latency, (f_latency, p_latency) assert (f_alarms, round(f_latency, 1)) == (20, 159.8) @test def raising_the_threshold_never_lets_phi_escape_domination(): """The whole point of the sweep: there is no threshold that rescues it.""" f_alarms, f_latency = census(lambda: FixedDetector(250.0)) for th in (8, 10, 12, 16, 20): alarms, latency = census(lambda th=th: PhiDetector(threshold=float(th))) assert f_alarms <= alarms and f_latency <= latency, (th, alarms, latency) @test def seed_4_convicts_on_a_shorter_silence_than_it_later_tolerates(): arrivals, _ = heartbeat_trace(4, PHASES) d = PhiDetector() d.heartbeat(arrivals[0]) verdicts = [] for i in range(1, len(arrivals)): verdicts.append((i, arrivals[i] - arrivals[i - 1], d.suspects(arrivals[i]))) d.heartbeat(arrivals[i]) convicted = [(i, g) for i, g, hit in verdicts if hit][0] assert convicted[0] == 40 assert abs(convicted[1] - 132.6) < 0.05, convicted # within the ten beats the demo prints, beat 44 is the visible inversion in_table = max(g for i, g, hit in verdicts if 40 < i <= 46 and not hit) assert abs(in_table - 142.8) < 0.05, in_table # across the whole trace it goes further tolerated = max(g for i, g, hit in verdicts if i > 40 and not hit) assert abs(tolerated - 193.1) < 0.05, tolerated assert tolerated > convicted[1], "the inversion is the whole point" @test def four_fifths_of_traces_show_the_inversion(): inversions = fired = 0 for seed in range(TRACES): arrivals, _ = heartbeat_trace(seed, PHASES) d = PhiDetector() d.heartbeat(arrivals[0]) first, later_ok = None, [] for i in range(1, len(arrivals)): gap = arrivals[i] - arrivals[i - 1] hit = d.suspects(arrivals[i]) if hit and first is None: first = gap elif not hit and first is not None: later_ok.append(gap) d.heartbeat(arrivals[i]) if first is not None and later_ok: fired += 1 if max(later_ok) > first: inversions += 1 assert (fired, inversions) == (200, 161), (fired, inversions) @test def the_sigma_floor_is_the_load_bearing_line(): assert census(lambda: PhiDetector(min_stddev_ms=1.0))[0] == 384 assert census(lambda: PhiDetector(min_stddev_ms=30.0))[0] == 5 zero_alarms, slow = census(lambda: PhiDetector(min_stddev_ms=50.0)) assert zero_alarms == 0 assert abs(slow - 290.4) < 0.05, slow # and it costs 110ms of detection @test def a_gradual_ramp_does_not_rescue_phi(): """If suddenness were the cause, ramping would fix it. It does not.""" ramped = [(40, *CALM)] for k in range(30): f = (k + 1) / 30 ramped.append((1, CALM[0] + f * (CONGESTED[0] - CALM[0]), CALM[1] + f * (CONGESTED[1] - CALM[1]))) ramped.append((30, *CALM)) alarms = 0 for seed in range(TRACES): arrivals, crash_due = heartbeat_trace(seed, ramped) alarms += evaluate(PhiDetector(), arrivals, crash_due)[0] assert alarms == 322, alarms if __name__ == "__main__": for fn in tests: fn() print(f" ok {fn.__name__}") print(f"\n{len(tests)} tests passed")