Karti Tripathi
357 lines
14 KiB
Python
357 lines
14 KiB
Python
"""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)
|