import argparse
import copy
import importlib.util
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
spec = importlib.util.spec_from_file_location(
"bench_short", Path(__file__).with_name("bench-short.py")
)
bench = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bench)
def result(backend="kache", ms=10000):
phase = {
"wall_ms": ms,
"hits": 10,
"cache_hits": 10,
"misses": 0,
"storage": {"restored_bytes": 1000},
"event_log": {"passed_through": 0},
}
return {
"git_ref": "subject-sha",
"cache_tool_version": backend + " 1.0",
"verdict": {"ok": True},
"warm_same_tree_verdict": {"ok": True},
**{p: copy.deepcopy(phase) for p in bench.PHASES},
}
def record(arm, sample, ms=10000):
return {
"arm": arm,
"sample": sample,
"cold_reused": sample % 3 != 0,
"result": result(ms=ms),
}
class BenchTests(unittest.TestCase):
def test_process_exit_and_timeout_cleanup(self):
bench.run_measurement([sys.executable, "-c", "pass"])
with self.assertRaises(bench.subprocess.CalledProcessError):
bench.run_measurement([sys.executable, "-c", "raise SystemExit(4)"])
process = MagicMock()
process.pid = 1234
process.wait.side_effect = [bench.subprocess.TimeoutExpired("engine", 1200), 0]
with (
patch.object(bench.subprocess, "Popen") as spawn,
patch.object(bench.os, "killpg") as kill,
):
spawn.return_value.__enter__.return_value = process
with self.assertRaises(bench.subprocess.TimeoutExpired):
bench.run_measurement(["engine"])
kill.assert_called_once_with(1234, bench.signal.SIGKILL)
def test_refuses_invalid_measurements(self):
for backend in ("kache", "sccache", "mbx"):
bench.validate(result(backend), backend)
for phase in bench.PHASES:
for value in (None, 0, -1, float("nan"), True):
r = result(backend)
r[phase]["wall_ms"] = value
with self.assertRaises(ValueError):
bench.validate(r, backend)
for phase in bench.PHASES[1:]:
r = result(backend)
r[phase]["hits"] = r[phase]["cache_hits"] = 0
with self.assertRaises(ValueError):
bench.validate(r, backend)
r = result()
r["warm_same_tree_verdict"]["ok"] = False
with self.assertRaises(ValueError):
bench.validate(r, "kache")
r = result()
r["warm"]["invalid_reasons"] = ["store error"]
with self.assertRaises(ValueError):
bench.validate(r, "kache")
def test_cold_reuse_does_not_inflate_samples(self):
summary = bench.summarize([record("kache", i) for i in range(6)])
self.assertEqual([s["n"] for s in summary["statistics"]], [2, 6, 6])
def test_paired_regression_and_noise(self):
self.assertEqual(
bench.paired_change([10000] * 6, [12000] * 6)["outcome"], "regression"
)
self.assertEqual(
bench.paired_change([10000] * 6, [8000] * 6)["outcome"], "improvement"
)
self.assertEqual(
bench.paired_change([10000] * 6, [9000, 12000] * 3)["outcome"],
"inconclusive",
)
self.assertEqual(
bench.paired_change([10000], [15000])["outcome"], "inconclusive"
)
self.assertEqual(
bench.paired_change([1000] * 6, [1100] * 6)["outcome"], "inconclusive"
)
def test_pair_and_identity_validation(self):
with self.assertRaises(ValueError):
bench.summarize([record("base", 0)])
records = [record("base", 0), record("head", 0)]
records[1]["result"]["git_ref"] = "different"
with self.assertRaises(ValueError):
bench.summarize(records)
records = [record("kache", 0), record("kache", 1)]
records[1]["result"]["cache_tool_version"] = "changed"
with self.assertRaises(ValueError):
bench.summarize(records)
def test_count_regression_cannot_hide_in_faster_timing(self):
records = [record("base", 0), record("head", 0, 5000)]
records[1]["result"]["warm"]["misses"] = 1
self.assertIn("misses rose", bench.summarize(records)["failures"][0])
def test_tool_path_sees_through_mise_shims(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
install = root / "installs" / "sccache-1.0"
install.mkdir(parents=True)
real = install / "sccache"
real.write_text("#!/bin/sh\necho real\n")
real.chmod(0o755)
bin_dir = root / "bin"
bin_dir.mkdir()
mise = bin_dir / "mise"
mise.write_text(
'#!/bin/sh\ncase "$1 $2" in "which sccache") echo "%s" ;; *) exit 1 ;; esac\n'
% real
)
mise.chmod(0o755)
shims = root / "shims"
shims.mkdir()
(shims / "sccache").symlink_to(mise)
(shims / "unknown").symlink_to(mise)
plain = root / "plain"
plain.mkdir()
link = plain / "kache"
link.symlink_to(real)
with patch.dict(
"os.environ",
{"PATH": os.pathsep.join(map(str, (shims, plain, bin_dir)))},
):
self.assertEqual(bench.tool_path("sccache"), str(real))
self.assertEqual(
bench.tool_path("unknown"),
str(shims / "unknown"),
"a shim mise cannot locate is kept as the shim itself",
)
self.assertEqual(
bench.tool_path("kache"),
os.path.realpath(real),
"an ordinary symlink is followed as before",
)
self.assertEqual(
bench.tool_path(str(root / "absent")),
os.path.realpath(root / "absent"),
"a missing tool keeps its name so the failure names it",
)
def test_driver_alternates_arms_and_saves_samples(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
args = argparse.Namespace(
output=root / "output",
project="hk",
engine=root / "engine",
scenarios=root / "scenarios",
kache="/kache",
base="/base",
sccache="/sccache",
mbx="/mbx",
samples=6,
order_seed=0,
skip_contention=False,
)
calls = []
def invoke(command, **kwargs):
calls.append(command)
work = Path(command[command.index("--work-dir") + 1])
work.mkdir(parents=True, exist_ok=True)
scenario = command[command.index("--profile") + 1]
backend = command[command.index("--cache-backend") + 1]
(work / f"{scenario}.json").write_text(json.dumps(result(backend)))
points = [
{
"attributes": [
{
"key": "kache.bench.phase",
"value": {"stringValue": phase},
}
]
}
for phase in ("cold", "warm")
]
(work / "metrics.otlp.json").write_text(
json.dumps(
{
"resourceMetrics": [
{
"scopeMetrics": [
{"metrics": [{"gauge": {"dataPoints": points}}]}
]
}
]
}
)
)
if backend == "kache":
for phase in ("cold", "warm-same-tree", "warm"):
dest = work / f"cache-otlp-{phase}"
dest.mkdir(exist_ok=True)
(dest / "metrics.otlp.json").write_text(
(work / "metrics.otlp.json").read_text()
)
def contention(args, arms):
self.assertFalse((args.output / "scratch").exists())
self.assertEqual(
[arm[0] for arm in arms], ["base", "head", "sccache", "mbx"]
)
output = args.output / "contention"
output.mkdir()
(output / "samples.json").write_text(
json.dumps({"revision": "subject-sha"})
)
(output / "report.md").write_text("## Contention: hk\n")
(output / "metrics.otlp.json").write_text(
json.dumps({"resourceMetrics": []})
)
return {"statistics": [], "comparisons": [], "failures": []}
with (
patch.object(bench, "run_measurement", invoke),
patch.object(bench, "run_contention", contention),
):
self.assertEqual(bench.run(args), 0)
self.assertEqual(len(calls), 24)
self.assertTrue(calls[0][calls[0].index("--kache") + 1].endswith("/base"))
self.assertEqual(calls[4][calls[4].index("--cache-backend") + 1], "mbx")
self.assertNotIn("--retry", calls[0])
self.assertIn("--retry", calls[4])
self.assertNotIn("--retry", calls[12])
self.assertIn("--skip-clone", calls[12])
payload = json.loads((args.output / "samples.json").read_text())
self.assertEqual(len(payload["records"]), 24)
self.assertEqual(payload["contention_samples"], "contention/samples.json")
self.assertIn(
"## Contention: hk", (args.output / "perf-gate.md").read_text()
)
self.assertFalse((args.output / "scratch").exists())
metrics = json.loads((args.output / "metrics.otlp.json").read_text())
points = [
point
for resource in metrics["resourceMetrics"]
for scope in resource["scopeMetrics"]
for metric in scope["metrics"]
for point in metric["gauge"]["dataPoints"]
]
cold = [
point
for point in points
if point["attributes"][0]["value"]["stringValue"] == "cold"
]
self.assertEqual(len(cold), 8)
self.assertIn("inconclusive", (args.output / "perf-gate.md").read_text())
for phase, expected in (("cold", 4), ("warm", 12)):
cached = json.loads(
(
args.output / f"cache-otlp-{phase}" / "metrics.otlp.json"
).read_text()
)
self.assertEqual(len(cached["resourceMetrics"]), expected)
def test_contention_compares_paired_work_and_rejects_incomplete_pairs(self):
records = []
for sample in range(6):
for arm, ms in (("base", 10000), ("head", 12000)):
for phase in ("cold", "warm"):
records.append(
{
"sample": sample,
"arm": arm,
"phase": phase,
"wall_ms": ms,
"events": {
"duplicate_key_compiles": 0,
"results": {"miss": int(phase == "cold")},
},
}
)
comparisons, failures = bench.contention_comparison(records)
self.assertEqual(len(comparisons), 2)
self.assertEqual(len(failures), 2)
for row in records:
row["wall_ms"] = 10000
records[-1]["events"]["results"]["miss"] = 1
self.assertIn(
"miss count increased", bench.contention_comparison(records)[1][0]
)
with self.assertRaisesRegex(ValueError, "incomplete"):
bench.contention_comparison(records[:-1])
def test_contention_driver_requires_all_tool_phase_samples(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
args = argparse.Namespace(
output=root,
project="eza",
scenarios=root / "scenarios",
samples=6,
order_seed=1,
)
arms = [
("kache", "kache", "/kache"),
("sccache", "sccache", "/sccache"),
("mbx", "mbx", "/mbx"),
]
records = [
{"sample": sample, "arm": arm, "phase": phase}
for sample in range(6)
for arm, _, _ in arms
for phase in ("cold", "warm")
if phase == "warm" or sample % 3 == 0
]
calls = []
def run(command, **kwargs):
calls.append(command)
output = root / "contention"
output.mkdir(exist_ok=True)
(output / "samples.json").write_text(json.dumps({"records": records}))
(output / "summary.json").write_text("[]")
with patch.object(bench.subprocess, "run", run):
self.assertEqual(bench.run_contention(args, arms)["failures"], [])
self.assertIn("kache=/kache,1", calls[0])
self.assertIn("--sccache", calls[0])
self.assertIn("--mbx", calls[0])
self.assertEqual(calls[0][calls[0].index("--samples") + 1], "6")
records[-1] = records[0]
with self.assertRaisesRegex(ValueError, "incomplete contention"):
bench.run_contention(args, arms)
def test_subprocess_failure_keeps_logs_and_fails(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
args = argparse.Namespace(
output=root / "output",
project="eza",
engine=root / "engine",
scenarios=root / "scenarios",
kache="/kache",
base=None,
sccache="/sccache",
mbx="/mbx",
samples=1,
order_seed=0,
skip_contention=True,
)
with patch.object(
bench,
"run_measurement",
side_effect=bench.subprocess.CalledProcessError(1, ["engine"]),
):
self.assertEqual(bench.run(args), 1)
self.assertIn(
"INVALID MEASUREMENT", (args.output / "perf-gate.md").read_text()
)
self.assertTrue((args.output / "logs/00-kache/engine.log").exists())
if __name__ == "__main__":
unittest.main()