"""Offline standard-library checks for the illustrative timing model.""" import csv import json from pathlib import Path import tempfile import unittest from timing_model import generate, nearest_rank, simulate, state_area_ms, summarize, timeline class TimingBoundaryTests(unittest.TestCase): def test_four_request_timestamps_and_latencies(self): rows = simulate(4, 100) self.assertEqual([(r.intended_schedule_ms, r.credit_issued_ms, r.request_start_ms, r.response_end_ms) for r in rows], [(0, 0, 0, 400), (100, 100, 400, 800), (200, 200, 800, 1200), (300, 300, 1200, 1600)]) self.assertEqual([r.request_latency_ms for r in rows], [400] * 4) self.assertEqual([r.effective_latency_ms for r in rows], [400, 700, 1000, 1300]) self.assertEqual([r.credit_to_start_wait_ms for r in rows], [0, 300, 600, 900]) self.assertEqual(summarize(rows, 100)["effective_mean_ms"], 850) def test_fifo_recurrence_and_no_overlap(self): for interval in (100, 200, 400, 800): with self.subTest(interval=interval): prior_end = 0 for row in simulate(20, interval): self.assertEqual(row.request_start_ms, max(row.credit_issued_ms, prior_end)) self.assertEqual(row.response_end_ms, row.request_start_ms + 400) self.assertGreaterEqual(row.request_start_ms, prior_end) prior_end = row.response_end_ms def test_closed_form_for_zero_issuance_delay(self): for interval in (100, 200, 400, 800): for i, row in enumerate(simulate(20, interval)): self.assertEqual(row.credit_to_start_wait_ms, i * max(400 - interval, 0)) self.assertEqual(row.effective_latency_ms, 400 + i * max(400 - interval, 0)) def test_boundary_equalities(self): scenarios = [simulate(20, interval) for interval in (100, 200, 400, 800)] scenarios += [simulate(4, 100, issuance_delays_ms=[300, 300, 300, 300])] for rows in scenarios: for row in rows: self.assertEqual(row.effective_latency_ms, row.request_latency_ms + row.credit_to_start_wait_ms) self.assertEqual(row.scheduled_latency_ms, row.effective_latency_ms + row.issuance_delay_ms) def test_sweep_means_and_maxima(self): for interval, expected_mean, expected_max in ( (100, 3250, 6100), (200, 2300, 4200), (400, 400, 400), (800, 400, 400) ): s = summarize(simulate(20, interval), interval) self.assertEqual(s["request_mean_ms"], 400) self.assertEqual(s["effective_mean_ms"], expected_mean) self.assertEqual(s["scheduled_mean_ms"], expected_mean) self.assertEqual(s["effective_max_ms"], expected_max) def test_one_second_slo_includes_equality(self): four = summarize(simulate(4, 100), 100) self.assertEqual(four["request_slo_pass_fraction"], 1) self.assertEqual(four["effective_slo_pass_count"], 3) self.assertEqual(four["effective_slo_pass_fraction"], .75) for interval, count in ((100, 3), (200, 4), (400, 20), (800, 20)): s = summarize(simulate(20, interval), interval) self.assertEqual(s["request_slo_pass_fraction"], 1) self.assertEqual(s["effective_slo_pass_count"], count) self.assertEqual(s["effective_slo_pass_fraction"], count / 20) def test_delayed_issuance_omits_precredit_delay(self): row = simulate(1, 100, issuance_delays_ms=[300])[0] self.assertEqual((row.intended_schedule_ms, row.credit_issued_ms, row.request_start_ms, row.response_end_ms), (0, 300, 300, 700)) self.assertEqual(row.request_latency_ms, 400) self.assertEqual(row.effective_latency_ms, 400) self.assertEqual(row.scheduled_latency_ms, 700) self.assertEqual(row.issuance_delay_ms, 300) self.assertEqual(row.credit_to_start_wait_ms, 0) def test_delayed_issuance_and_wait_are_distinct(self): rows = simulate(4, 100, issuance_delays_ms=[300] * 4) self.assertEqual([r.effective_latency_ms for r in rows], [400, 700, 1000, 1300]) self.assertEqual([r.scheduled_latency_ms for r in rows], [700, 1000, 1300, 1600]) def test_queue_conservation_at_all_event_boundaries(self): for interval in (100, 200, 400, 800): states = timeline(simulate(20, interval)) for state in states: self.assertEqual(state["issued_cumulative"], state["waiting"] + state["in_service"] + state["completed_cumulative"]) self.assertEqual(state["started_cumulative"], state["in_service"] + state["completed_cumulative"]) self.assertGreaterEqual(state["waiting"], 0) self.assertIn(state["in_service"], (0, 1)) self.assertEqual(states[-1]["completed_cumulative"], 20) self.assertEqual(states[-1]["in_system"], 0) def test_queue_areas_equal_summed_individual_durations(self): scenarios = [simulate(20, interval) for interval in (100, 200, 400, 800)] scenarios += [simulate(4, 100, issuance_delays_ms=[300] * 4)] for rows in scenarios: states = timeline(rows) self.assertEqual(state_area_ms(states, "waiting"), sum(r.credit_to_start_wait_ms for r in rows)) self.assertEqual(state_area_ms(states, "in_service"), sum(r.request_latency_ms for r in rows)) self.assertEqual(state_area_ms(states, "in_system"), sum(r.effective_latency_ms for r in rows)) def test_tied_events_do_not_add_spurious_waiting(self): states = timeline(simulate(20, 400)) self.assertTrue(all(s["waiting"] == 0 for s in states)) at_400 = next(s for s in states if s["time_ms"] == 400) self.assertEqual((at_400["issued_cumulative"], at_400["started_cumulative"], at_400["completed_cumulative"], at_400["in_service"]), (2, 2, 1, 1)) def test_queue_sweep_peaks_and_areas(self): for interval, peak, area in ((100, 15, 57000), (200, 10, 38000), (400, 0, 0), (800, 0, 0)): s = summarize(simulate(20, interval), interval) self.assertEqual(s["max_waiting_requests"], peak) self.assertEqual(s["queue_area_request_ms"], area) self.assertEqual(s["in_service_area_request_ms"], 8000) self.assertEqual(s["in_system_area_request_ms"], area + 8000) def test_nearest_rank_is_not_interpolated(self): values = [6100 - 300 * i for i in range(20)] # Unsorted input. self.assertEqual(nearest_rank(values, 95), 5800) self.assertEqual(nearest_rank(values, 99), 6100) self.assertEqual(nearest_rank(values, 100), 6100) self.assertEqual(nearest_rank(values, 50), 3100) self.assertEqual(nearest_rank(values, .1), 400) self.assertEqual(nearest_rank([400, 700, 1000, 1300], 95), 1300) self.assertEqual(nearest_rank([400], 99), 400) self.assertEqual(nearest_rank(range(1, 101), 7), 7) def test_nearest_rank_rejects_undefined_inputs(self): for values, p in (([], 95), ([1], 0), ([1], -1), ([1], 101)): with self.assertRaises(ValueError): nearest_rank(values, p) def test_invalid_model_inputs(self): for kwargs in ({"n": 0, "interval_ms": 100}, {"n": 1, "interval_ms": 0}, {"n": 1, "interval_ms": 100, "service_ms": 0}, {"n": 2, "interval_ms": 100, "issuance_delays_ms": [0]}, {"n": 1, "interval_ms": 100, "issuance_delays_ms": [-1]}, {"n": 2, "interval_ms": 100, "issuance_delays_ms": [500, 0]}): with self.assertRaises(ValueError): simulate(**kwargs) def test_generated_files_round_trip_and_are_deterministic(self): with tempfile.TemporaryDirectory(dir=Path(__file__).parent) as temp: root = Path(temp) result = generate(root) self.assertEqual(result, json.loads((root / "results.json").read_text())) with (root / "requests.csv").open(newline="") as stream: records = list(csv.DictReader(stream)) self.assertEqual(len(records), 80) self.assertEqual({int(r["target_interval_ms"]) for r in records}, {100, 200, 400, 800}) self.assertTrue(all(int(r["request_latency_ms"]) == 400 for r in records)) first = {p.name: p.read_bytes() for p in root.iterdir()} generate(root) self.assertEqual(first, {p.name: p.read_bytes() for p in root.iterdir()}) self.assertEqual(len(first), 6) if __name__ == "__main__": unittest.main(verbosity=2)