"""Synthetic measurement-boundary checks, never a serving benchmark.""" from copy import deepcopy import json from pathlib import Path import unittest from latency_boundaries import summarize_latency class LatencyBoundaryTests(unittest.TestCase): def setUp(self): self.fixture = json.loads(Path(__file__).with_name("synthetic-latency-boundaries.json").read_text()) self.plan = self.fixture["schedule"] self.rows = self.fixture["observations"] self.cutoff = self.fixture["observation_cutoff_ms"] def report(self, plan=None, rows=None, cutoff=None): return summarize_latency(self.plan if plan is None else plan, self.rows if rows is None else rows, self.cutoff if cutoff is None else cutoff) def test_fixture_preserves_author_and_synthetic_status(self): self.assertIs(self.fixture["synthetic"], True) self.assertEqual(self.fixture["status"], "synthetic_arithmetic_fixture_not_a_run") self.assertEqual(self.fixture["author"], "Anant Mishra") self.assertEqual(self.fixture["study_id"], "AM-RP-001") def test_frozen_schedule_outcomes_reconcile(self): report = self.report() self.assertEqual(report["counts"], self.fixture["expected_counts"]) self.assertEqual(report["counts"]["scheduled"], sum(v for k, v in report["counts"].items() if k != "scheduled")) def test_first_example_has_two_distinct_timing_boundaries(self): row = self.report()["per_request"][0] self.assertEqual(row["request_id"], "L01") self.assertEqual([row[k] for k in ("generator_wait_ms", "client_ttft_ms", "scheduled_ttft_ms", "client_e2e_ms", "scheduled_e2e_ms")], [1800, 600, 2400, 1200, 3000]) def test_completed_statistics_match_declared_population_and_method(self): report = self.report() self.assertEqual(report["latency_population"], "completed_requests_only") self.assertEqual(report["completed_latency"], self.fixture["expected_completed_latency"]) self.assertEqual(report["percentile_method"], "nearest_rank_ceil_p_times_n_no_interpolation") def test_partial_timeout_keeps_output_observation_without_inventing_completion(self): row = self.report()["per_request"][3] self.assertEqual(row["status"], "timeout") self.assertEqual(row["client_ttft_ms"], 400) self.assertEqual(row["client_observed_duration_ms"], 3000) self.assertIsNone(row["client_e2e_ms"]) self.assertIsNone(row["scheduled_e2e_ms"]) self.assertTrue(all(s["n"] == 3 for s in self.report()["completed_latency"].values())) def test_non_dispatched_work_has_no_fabricated_zero_latency(self): row = self.report()["per_request"][5] self.assertEqual(row["status"], "not_dispatched") self.assertTrue(all(v is None for k, v in row.items() if k not in ("request_id", "status"))) def test_duplicate_schedule_or_observation_ids_are_rejected(self): for plan, rows in ((self.plan + [self.plan[0]], self.rows), (self.plan, self.rows + [self.rows[0]])): with self.subTest(plan=len(plan), rows=len(rows)), self.assertRaisesRegex(ValueError, "Duplicate"): self.report(plan, rows) def test_missing_and_unplanned_observations_are_rejected(self): for rows in (self.rows[:-1], self.rows + [{**self.rows[0], "request_id": "UNPLANNED"}]): with self.subTest(length=len(rows)), self.assertRaisesRegex(ValueError, "match exactly"): self.report(rows=rows) def test_invalid_numeric_offsets_are_rejected(self): for value in (-1, float("nan"), float("inf"), True, "1800", 10 ** 1000): with self.subTest(value_type=type(value).__name__), self.assertRaises(ValueError): self.report(rows=[{**self.rows[0], "dispatch_ms": value}] + self.rows[1:]) def test_temporal_order_and_cutoff_are_enforced(self): for change in ({"dispatch_ms": -1}, {"dispatch_ms": 2500}, {"first_output_ms": 1700}, {"first_output_ms": 3100}, {"end_ms": 7000}): with self.subTest(change=change), self.assertRaises(ValueError): self.report(rows=[{**self.rows[0], **change}] + self.rows[1:]) with self.assertRaises(ValueError): self.report(plan=[{**self.plan[0], "scheduled_ms": 6100}] + self.plan[1:]) def test_completed_response_requires_dispatch_and_first_output(self): for field in ("dispatch_ms", "first_output_ms"): with self.subTest(field=field), self.assertRaises(ValueError): self.report(rows=[{**self.rows[0], field: None}] + self.rows[1:]) def test_rejection_and_non_dispatch_cannot_claim_output(self): for index, change in ((4, {"first_output_ms": 2820}), (5, {"dispatch_ms": 1100}), (5, {"first_output_ms": 1200})): rows = deepcopy(self.rows) rows[index].update(change) with self.subTest(index=index, change=change), self.assertRaises(ValueError): self.report(rows=rows) def test_in_flight_work_is_observed_at_cutoff_not_called_a_timeout(self): rows = deepcopy(self.rows) rows[3].update(status="in_flight", end_ms=self.cutoff) report = self.report(rows=rows) self.assertEqual(report["counts"]["in_flight"], 1) self.assertEqual(report["counts"]["timeout"], 0) self.assertIsNone(report["per_request"][3]["client_e2e_ms"]) rows[3]["end_ms"] -= 1 with self.assertRaisesRegex(ValueError, "cutoff"): self.report(rows=rows) def test_empty_or_all_non_completed_populations_have_null_statistics(self): for report in (self.report([], [], 0), self.report(self.plan[3:], self.rows[3:])): self.assertTrue(all(s == {"n": 0, "mean": None, "p50": None, "p95": None} for s in report["completed_latency"].values())) def test_nearest_rank_does_not_interpolate_even_sized_samples(self): plan = [{"request_id": str(i), "scheduled_ms": 0} for i in range(1, 21)] rows = [{"request_id": str(i), "status": "completed", "dispatch_ms": 0, "first_output_ms": 0, "end_ms": i} for i in range(1, 21)] summary = self.report(plan, rows, 20)["completed_latency"]["client_e2e_ms"] self.assertEqual(summary, {"n": 20, "mean": 10.5, "p50": 10, "p95": 19}) def test_shared_clock_origin_shift_preserves_intervals(self): shift = 10000 plan = [{**r, "scheduled_ms": r["scheduled_ms"] + shift} for r in self.plan] rows = [{k: (v + shift if k.endswith("_ms") and v is not None else v) for k, v in r.items()} for r in self.rows] self.assertEqual(self.report(plan, rows, self.cutoff + shift), self.report()) def test_input_order_and_data_are_preserved(self): original = deepcopy((self.plan, self.rows)) self.assertEqual(self.report(list(reversed(self.plan)), list(reversed(self.rows))), self.report()) self.assertEqual((self.plan, self.rows), original) def test_schema_identity_and_state_errors_are_rejected(self): for change in ({"extra": 1}, {"request_id": " "}, {"status": "probably_done"}, {"status": []}): with self.subTest(change=change), self.assertRaises(ValueError): self.report(rows=[{**self.rows[0], **change}] + self.rows[1:]) with self.assertRaises(ValueError): self.report(plan={}) with self.assertRaises(ValueError): self.report(cutoff=True) if __name__ == "__main__": unittest.main()