"""Tests for the run diff. Written against `unittest` rather than pytest so they run with a bare interpreter — this repo has no dev dependency group, and a test you cannot run is not a test. pytest collects TestCase subclasses too, so adding it later changes nothing. python3 -m unittest discover -s tests -v """ from __future__ import annotations import json import sys import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from kbench.compare import PERF_NOISE_FLOOR_PCT, compare, direction # noqa: E402 RESULTS = Path(__file__).resolve().parent.parent / "results" def run( *, run_id="r", model="m", quant="nvfp4", checkpoint=None, serving=None, host="your-node", quality=None, perf=None, ): return { "run_id": run_id, "target": { "model": model, "quantization": quant, "checkpoint": checkpoint, "serving_config": serving or {"gpu-memory-utilization": "0.55"}, "engine": "vllm", }, "host": {"id": host}, "quality": quality or [], "perf": {"points": perf} if perf is not None else None, } def sample(sid, score, passed): return {"sample_id": sid, "score": score, "passed": passed} def point(concurrency, tps, ttft=None, accept=None, natural_stop=False): return { "concurrency": concurrency, "output_tps_total": tps, "output_tps_per_stream": tps / concurrency, "ttft_p50_ms": ttft, "spec_acceptance_rate": accept, "natural_stop": natural_stop, } class TargetComparability(unittest.TestCase): def test_identical_targets_are_flagged_as_variance_not_change(self): c = compare(run(), run()) self.assertFalse(c.target_deltas) self.assertTrue(any("variance" in w for w in c.warnings)) def test_one_changed_axis_is_attributable(self): c = compare(run(quant="nvfp4"), run(quant="fp8")) self.assertEqual([d.field for d in c.target_deltas], ["quantization"]) self.assertFalse(c.confounded) def test_two_changed_axes_are_called_confounded(self): # The failure this exists to prevent: changing quant AND host, then attributing the # throughput difference to the quant. c = compare(run(quant="nvfp4", host="your-node"), run(quant="fp8", host="other")) self.assertTrue(c.confounded) self.assertTrue(any("attributed" in w for w in c.warnings)) def test_reordered_serving_config_is_not_a_change(self): a = run(serving={"a": "1", "b": "2"}) b = run(serving={"b": "2", "a": "1"}) self.assertFalse(compare(a, b).target_deltas) class QualityDiff(unittest.TestCase): def _with(self, samples, metrics=None): return run( quality=[{"task": "t", "metrics": metrics or {}, "samples": samples}] ) def test_regressions_are_listed_before_fixes(self): before = self._with([sample("a", 1.0, True), sample("b", 0.0, False)]) after = self._with([sample("a", 0.0, False), sample("b", 1.0, True)]) flips = compare(before, after).flips self.assertEqual([f.became for f in flips], ["fail", "pass"]) self.assertEqual(flips[0].sample_id, "a") def test_unchanged_samples_are_not_reported(self): same = self._with([sample("a", 1.0, True)]) self.assertEqual(compare(same, same).flips, []) def test_a_changed_eval_set_is_a_warning_not_a_flip(self): # Silently comparing aggregates across different sample sets is the quiet way to be # wrong, so it must never look like a normal diff. before = self._with([sample("a", 1.0, True)]) after = self._with([sample("b", 1.0, True)]) c = compare(before, after) self.assertEqual(c.flips, []) self.assertEqual(c.only_in_before, ["t/a"]) self.assertEqual(c.only_in_after, ["t/b"]) self.assertTrue(any("eval set changed" in w for w in c.warnings)) def test_a_perf_only_baseline_says_so_instead_of_claiming_the_set_changed(self): c = compare(run(), self._with([sample("a", 1.0, True)])) self.assertTrue(any("no quality results" in w for w in c.warnings)) self.assertFalse(any("eval set changed" in w for w in c.warnings)) def test_sample_ids_are_scoped_to_their_task(self): # The same id in two tasks is two different samples; collapsing them would diff # unrelated things against each other. before = run(quality=[ {"task": "x", "metrics": {}, "samples": [sample("1", 1.0, True)]}, {"task": "y", "metrics": {}, "samples": [sample("1", 0.0, False)]}, ]) self.assertEqual(compare(before, before).flips, []) class PerfDiff(unittest.TestCase): def test_direction_accounts_for_metrics_where_lower_is_better(self): c = compare( run(perf=[point(1, 30.0, ttft=300.0)]), run(perf=[point(1, 30.0, ttft=200.0)]), ) deltas = {d.name: d for d in c.perf[1]} # Latency fell, which is an improvement — the sign alone would say otherwise. self.assertEqual(direction("ttft_p50_ms", deltas["ttft_p50_ms"]), "better") def test_more_throughput_is_better(self): c = compare(run(perf=[point(1, 20.0)]), run(perf=[point(1, 30.0)])) deltas = {d.name: d for d in c.perf[1]} self.assertEqual(direction("output_tps_total", deltas["output_tps_total"]), "better") def test_movement_under_the_noise_floor_reads_as_flat(self): small = 100.0 * (1 + (PERF_NOISE_FLOOR_PCT - 1) / 100) c = compare(run(perf=[point(1, 100.0)]), run(perf=[point(1, small)])) deltas = {d.name: d for d in c.perf[1]} self.assertEqual(direction("output_tps_total", deltas["output_tps_total"]), "flat") def test_a_different_sweep_is_warned_about(self): c = compare( run(perf=[point(1, 30.0), point(8, 180.0)]), run(perf=[point(1, 30.0), point(32, 400.0)]), ) self.assertEqual(sorted(c.perf), [1]) self.assertTrue(any("sweep differs" in w for w in c.warnings)) class SpeculativeDecoding(unittest.TestCase): def test_acceptance_rate_is_diffed_and_higher_is_better(self): # Throughput moves for many reasons; acceptance moves only because drafting got better # or worse, which is the number an MTP comparison is actually about. c = compare( run(perf=[point(1, 30.0, accept=0.55)]), run(perf=[point(1, 31.0, accept=0.80)]), ) deltas = {d.name: d for d in c.perf[1]} self.assertIn("spec_acceptance_rate", deltas) self.assertEqual(direction("spec_acceptance_rate", deltas["spec_acceptance_rate"]), "better") def test_a_target_without_drafting_yields_no_acceptance_delta(self): c = compare(run(perf=[point(1, 30.0)]), run(perf=[point(1, 30.0, accept=0.8)])) deltas = {d.name: d for d in c.perf[1]} self.assertIsNone(deltas["spec_acceptance_rate"].absolute) def test_mixing_pinned_and_natural_stop_is_warned_about(self): # Forced continuation past EOS and natural stopping produce different token # distributions; comparing their throughput directly is comparing two things. c = compare( run(perf=[point(1, 30.0)]), run(perf=[point(1, 45.0, natural_stop=True)]), ) self.assertTrue(any("not comparable across those two modes" in w for w in c.warnings)) def test_two_natural_stop_runs_are_not_warned_about(self): c = compare( run(perf=[point(1, 30.0, natural_stop=True)]), run(perf=[point(1, 45.0, natural_stop=True)]), ) self.assertFalse(any("two modes" in w for w in c.warnings)) class AgainstCommittedResults(unittest.TestCase): """The two real score cards in results/ — the diff must survive real data.""" def setUp(self): files = sorted(RESULTS.glob("*.json")) if len(files) < 2: self.skipTest("needs two committed results") self.a, self.b = (json.loads(f.read_text()) for f in files[:2]) def test_diffs_real_runs_without_error_and_finds_the_shared_sweep(self): c = compare(self.a, self.b) self.assertEqual(sorted(c.perf), [1, 8, 32]) # Same target, same host, same day: the tool must say so rather than imply a change. self.assertTrue(any("variance" in w for w in c.warnings)) def test_neither_run_is_mutated(self): before = json.dumps(self.a, sort_keys=True) compare(self.a, self.b) self.assertEqual(json.dumps(self.a, sort_keys=True), before) if __name__ == "__main__": unittest.main() class EditedSamplesAreDetected(unittest.TestCase): """An edited sample keeps its id, so the add/remove diff sees nothing. This is the most dangerous shape of eval drift: loosen a regex, and every id still lines up while the scores now answer a different question. """ def _run(self, fingerprint, score=1.0): return { "target_id": "m@h", "quality": [ { "task": "agent_ops", "tier": "signal", "dataset_fingerprint": fingerprint, "metrics": {"accuracy": score}, "samples": [ {"sample_id": "agent_ops/0001", "score": score, "passed": True} ], } ], "perf": None, "verdict": {"contamination": "clean"}, } def test_same_fingerprint_does_not_warn(self): r = compare(self._run("sha256:aaaa"), self._run("sha256:aaaa")) self.assertFalse([w for w in r.warnings if "edited" in w or "fingerprint" in w]) def test_changed_fingerprint_warns_even_when_ids_match(self): r = compare(self._run("sha256:aaaa"), self._run("sha256:bbbb")) self.assertFalse(r.only_in_before) self.assertFalse(r.only_in_after) self.assertTrue( any("was edited between runs" in w for w in r.warnings), f"an edit with matching ids went unreported: {r.warnings}", ) def test_missing_fingerprint_warns(self): r = compare(self._run(None), self._run("sha256:bbbb")) self.assertTrue(any("no dataset fingerprint" in w for w in r.warnings)) class SamplingChangesAreDetected(unittest.TestCase): """Sampling is part of a quality result's identity. Unpinned sampling moved this bench by 27% of its samples between identical runs -- more than most real regressions. Diffing across different sampling reports noise as signal, so it has to be called out rather than silently compared. """ def _run(self, sampling, score=1.0): return { "target_id": "m@h", "sampling": sampling, "quality": [ { "task": "agent_ops", "tier": "signal", "dataset_fingerprint": "sha256:aaaa", "metrics": {"accuracy": score}, "samples": [ {"sample_id": "agent_ops/0001", "score": score, "passed": True} ], } ], "perf": None, "verdict": {"contamination": "clean"}, } PINNED = {"temperature": 0.6, "top_p": 0.95, "seed": 1} def test_identical_sampling_does_not_warn(self): r = compare(self._run(self.PINNED), self._run(self.PINNED)) self.assertFalse([w for w in r.warnings if "ampling" in w], r.warnings) def test_changed_temperature_warns(self): other = {**self.PINNED, "temperature": 1.0} r = compare(self._run(self.PINNED), self._run(other)) self.assertTrue( any("Sampling changed between runs" in w for w in r.warnings), r.warnings ) self.assertTrue(any("temperature" in w for w in r.warnings), r.warnings) def test_missing_sampling_block_warns(self): r = compare(self._run(None), self._run(self.PINNED)) self.assertTrue( any("does not record how the model was sampled" in w for w in r.warnings), r.warnings, ) class EpochsAreNotASamplingChange(unittest.TestCase): """More epochs is better precision on the same estimand, not a different measurement.""" def _run(self, sampling): return { "target_id": "m@h", "sampling": sampling, "quality": [ { "task": "agent_ops", "tier": "signal", "dataset_fingerprint": "sha256:aaaa", "metrics": {"accuracy": 1.0}, "samples": [ {"sample_id": "agent_ops/0001", "score": 1.0, "passed": True} ], } ], "perf": None, "verdict": {"contamination": "clean"}, } BASE = {"temperature": 0.6, "top_p": 0.95, "seed": 1, "epochs": 1} def test_differing_epochs_does_not_claim_incomparability(self): r = compare(self._run(self.BASE), self._run({**self.BASE, "epochs": 5})) self.assertFalse( any("not comparable across different sampling" in w for w in r.warnings), f"epoch count reported as a sampling change: {r.warnings}", ) self.assertTrue(any("Epochs differ" in w for w in r.warnings), r.warnings) def test_a_real_sampling_change_alongside_epochs_still_warns(self): other = {**self.BASE, "epochs": 5, "temperature": 1.0} r = compare(self._run(self.BASE), self._run(other)) self.assertTrue( any("not comparable across different sampling" in w for w in r.warnings), r.warnings, ) self.assertTrue(any("Epochs differ" in w for w in r.warnings), r.warnings) # the incomparability warning must name temperature, not epochs sampling_warn = next(w for w in r.warnings if "not comparable" in w) self.assertIn("temperature", sampling_warn) self.assertNotIn("epochs", sampling_warn)