logan
e18bd3ccbb
xdotool chat injection silently fails under KDE Wayland/Xwayland, so the v1 harness recorded runs whose scene commands never executed. v2 drives scenes from a generated world datapack, detects crashes, aligns MangoHud CSVs by file birth time, flags vsync/backup/low-n windows, and records full run metadata to JSONL; analyze.py applies the BENCH-MATRIX win rule.
242 lines
10 KiB
Python
242 lines
10 KiB
Python
"""Offline tests for bench.py / analyze.py (no game, no display).
|
|
|
|
python3 -m unittest discover -s analysis/client-perf/harness/tests
|
|
"""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(HERE.parent))
|
|
|
|
import analyze # noqa: E402
|
|
import bench # noqa: E402
|
|
|
|
FIX = HERE / "fixtures"
|
|
CSV = FIX / "mangohud_java_2026-10-15_15-25-26.csv"
|
|
|
|
|
|
def lines(name):
|
|
return (FIX / name).read_text().splitlines()
|
|
|
|
|
|
class CsvParsing(unittest.TestCase):
|
|
def test_header_and_rows(self):
|
|
meta, rows = bench.parse_mangohud_csv(CSV)
|
|
self.assertIn("RX 7800 XT", meta["gpu"])
|
|
self.assertEqual(len(rows), 200) # malformed trailing row skipped
|
|
self.assertAlmostEqual(rows[0][0], 0.1) # elapsed ns -> s
|
|
self.assertEqual(rows[0][1], 4.0)
|
|
|
|
def test_name_time(self):
|
|
import time
|
|
t = bench.csv_name_time("java_2026-10-15_15-25-26.csv")
|
|
self.assertEqual(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t)),
|
|
"2026-10-15 15:25:26")
|
|
self.assertIsNone(bench.csv_name_time("weird.csv"))
|
|
|
|
def test_file_birth_method(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = Path(d) / CSV.name
|
|
shutil.copyfile(CSV, p)
|
|
t, method = bench.file_birth(p)
|
|
self.assertIn(method, {"st_birthtime", "stat-W", "filename", "mtime"})
|
|
self.assertGreater(t, 0)
|
|
|
|
|
|
class Math(unittest.TestCase):
|
|
def test_pct(self):
|
|
self.assertEqual(bench.pct([1, 2, 3, 4, 5], 0.5), 3)
|
|
self.assertAlmostEqual(bench.pct(list(range(1, 101)), 0.99), 99.01)
|
|
self.assertNotEqual(bench.pct([], 0.5), bench.pct([], 0.5)) # nan
|
|
|
|
def test_one_low_from_frametimes(self):
|
|
# 200 frames: 198 x 5 ms, 2 x 25 ms -> worst 1 % (2 frames) mean 25 ms -> 40 fps
|
|
rows = [(i * 0.005, 25.0 if i in (50, 150) else 5.0, 200.0) for i in range(200)]
|
|
m = bench.window_metrics(rows, 1000.0, 1000.0, 1001.0)
|
|
self.assertEqual(m["n"], 200)
|
|
self.assertEqual(m["one_low_fps"], 40.0)
|
|
self.assertEqual(m["p50_ms"], 5.0)
|
|
self.assertEqual(m["peak_ms"], 25.0)
|
|
self.assertAlmostEqual(m["avg_fps"], 1000 / (1040 / 200), places=1)
|
|
self.assertEqual(m["one_low_fps_legacy"], 200.0) # fps column was flat
|
|
|
|
def test_window_slicing_alignment(self):
|
|
_, rows = bench.parse_mangohud_csv(CSV)
|
|
origin = 5000.0
|
|
m = bench.window_metrics(rows, origin, origin + 10.0, origin + 20.0)
|
|
self.assertEqual(m["n"], 101)
|
|
self.assertEqual(m["p50_ms"], 4.0)
|
|
self.assertEqual(m["p99_ms"], 20.0)
|
|
self.assertEqual(m["one_low_fps"], 50.0)
|
|
self.assertEqual(m["one_low_fps_legacy"], 50.0)
|
|
self.assertIsNone(bench.window_metrics(rows, origin, origin + 30, origin + 40))
|
|
# first 10 s contain no 20 ms frames
|
|
early = bench.window_metrics(rows, origin, origin, origin + 9.9)
|
|
self.assertEqual(early["p99_ms"], 4.0)
|
|
|
|
|
|
class Markers(unittest.TestCase):
|
|
def test_parse(self):
|
|
marks, done = bench.parse_markers(lines("latest.log"))
|
|
st, en = marks[("start", "s1", 1)], marks[("end", "s1", 1)]
|
|
self.assertAlmostEqual(en - st, 30.0, places=3)
|
|
self.assertAlmostEqual(marks[("warm", "s1", 1)] % 1, 0.0, places=3) # server line, not CHAT
|
|
self.assertAlmostEqual(done, en)
|
|
|
|
def test_events_and_alignment(self):
|
|
ln = lines("latest.log")
|
|
ev = bench.scan_events(ln)
|
|
self.assertEqual(ev["game_took_s"], 28.366)
|
|
self.assertEqual(ev["atlas"], "4096x4096x2")
|
|
self.assertIsNone(ev["crash"])
|
|
self.assertAlmostEqual(ev["joined_t"] - ev["title_t"], 0.764, places=3)
|
|
# CSV born 10 s before the start marker -> window covers elapsed 10..40 s,
|
|
# CSV ends at 20 s -> 101 frames -> low_n
|
|
marks, _ = bench.parse_markers(ln)
|
|
ts, te = marks[("start", "s1", 1)], marks[("end", "s1", 1)]
|
|
_, rows = bench.parse_mangohud_csv(CSV)
|
|
m = bench.window_metrics(rows, ts - 10.0, ts, te)
|
|
self.assertEqual(m["n"], 101)
|
|
self.assertIn("low_n", bench.validity_flags(m, te - ts, [], bench.backup_times(ln), ts, te))
|
|
|
|
def test_backup_flag(self):
|
|
ln = lines("latest.log")
|
|
bk = bench.backup_times(ln)
|
|
self.assertEqual(len(bk), 1)
|
|
m = {"n": 6000, "p50_ms": 4.0, "avg_fps": 200.0}
|
|
self.assertEqual(bench.validity_flags(m, 30, [], bk, bk[0] + 30, bk[0] + 60), [])
|
|
self.assertIn("backup_in_window", bench.validity_flags(m, 30, [], bk, bk[0] - 1, bk[0] + 29))
|
|
self.assertIn("backup_in_window", bench.validity_flags(m, 30, [], bk, bk[0] + 3, bk[0] + 33))
|
|
|
|
def test_vsync_flag(self):
|
|
m = {"n": 1800, "p50_ms": 16.7, "avg_fps": 59.8}
|
|
self.assertIn("vsync_suspect", bench.validity_flags(m, 30))
|
|
m144 = {"n": 4320, "p50_ms": 6.95, "avg_fps": 144}
|
|
self.assertIn("vsync_suspect", bench.validity_flags(m144, 30, [144.0]))
|
|
self.assertNotIn("vsync_suspect", bench.validity_flags(m144, 30, [60.0]))
|
|
self.assertEqual(bench.validity_flags(None, 30), ["no_data"])
|
|
|
|
|
|
class Crash(unittest.TestCase):
|
|
def test_classify(self):
|
|
got = [bench.classify_line(l) for l in lines("crash.log")]
|
|
self.assertEqual(got, ["mod_instance", "fatal", "broken_mod_state", None])
|
|
self.assertIsNone(bench.classify_line(lines("latest.log")[3])) # backup line
|
|
self.assertEqual(bench.scan_events(lines("crash.log"))["crash"], "mod_instance")
|
|
|
|
|
|
class Datapack(unittest.TestCase):
|
|
def test_generate(self):
|
|
files = bench.generate_datapack("p1", ["s1", "s2s", "s5"], 2, 15, 30, 10)
|
|
tag = json.loads(files["data/minecraft/tags/function/load.json"])
|
|
self.assertEqual(tag["values"], ["bench:autoplan/p1/load"])
|
|
self.assertEqual(json.loads(files["pack.mcmeta"])["pack"]["pack_format"], 48)
|
|
load = files["data/bench/function/autoplan/p1/load.mcfunction"]
|
|
self.assertIn("unless score #p1 bench_plan matches 1..", load)
|
|
steps = sorted((k for k in files if "/step" in k),
|
|
key=lambda k: int(k.rsplit("step", 1)[1].split(".")[0]))
|
|
body = "".join(files[k] for k in steps)
|
|
for kind in ("warm", "start", "end"):
|
|
for sc in ("s1", "s2s", "s5"):
|
|
for rep in (1, 2):
|
|
self.assertEqual(body.count(f"say BENCH_MARK {kind} {sc} {rep}\n"), 1)
|
|
self.assertIn("function bench:tour_start", body)
|
|
self.assertIn("function bench:storm_on", body)
|
|
self.assertIn("function bench:storm_off", body)
|
|
self.assertIn("schedule function bench:autoplan/p1/step", body)
|
|
self.assertNotIn(" 0t", body)
|
|
self.assertTrue(body.rstrip().endswith("say BENCH_DONE p1"))
|
|
|
|
def test_timeline_ticks(self):
|
|
tl = bench.build_timeline(["s1", "s5"], 1, 15, 30, "x")
|
|
delays = [d for _, d in tl]
|
|
self.assertEqual(delays, [300, 600, 200, 2400, 0])
|
|
self.assertEqual(tl[1][0], ["say BENCH_MARK start s1 1"])
|
|
self.assertIn("say BENCH_MARK end s1 1", tl[2][0]) # merged into next scene's step
|
|
self.assertEqual(bench.plan_seconds(["s1", "s5"], 1, 15, 30), 45 + 130)
|
|
|
|
def test_order_rule(self):
|
|
class A:
|
|
plan, any_order = "s2,s1", False
|
|
with self.assertRaises(SystemExit):
|
|
bench.session_plan(A)
|
|
A.any_order = True
|
|
self.assertEqual(bench.session_plan(A), ["s2", "s1"])
|
|
A.plan, A.any_order = "s1,s6", False
|
|
with self.assertRaises(SystemExit):
|
|
bench.session_plan(A)
|
|
|
|
def test_pins(self):
|
|
out = bench.merge_pins("version:3955\nmaxFps:120\nenableVsync:true\nfov:0.5\n")
|
|
self.assertIn("maxFps:260", out)
|
|
self.assertIn("enableVsync:false", out)
|
|
self.assertIn("fov:0.5", out)
|
|
self.assertIn("pauseOnLostFocus:false", out)
|
|
self.assertEqual(out.count("maxFps"), 1)
|
|
|
|
|
|
def row(label, variant, session, p99, low, scene="s1", flags=()):
|
|
return {"kind": "window", "label": label, "variant": variant, "scene": scene,
|
|
"session": session, "flags": list(flags),
|
|
"metrics": {"p99_ms": p99, "one_low_fps": low, "avg_fps": 200.0}}
|
|
|
|
|
|
class Analyze(unittest.TestCase):
|
|
def setUp(self):
|
|
self.rows = [
|
|
row("B0", "C0", "a", 10.0, 100), row("B0", "C0", "b", 10.2, 98),
|
|
row("B0", "C0", "c", 10.4, 99),
|
|
row("B2", "C0", "d", 10.2, 99),
|
|
row("B2", "WIN", "e", 9.0, 100), row("B2", "TIE", "f", 10.0, 98),
|
|
row("B2", "LOSS", "g", 9.0, 90),
|
|
row("B2", "BAD", "h", 1.0, 999, flags=["crashed"]),
|
|
]
|
|
|
|
def test_summary_spread(self):
|
|
s = analyze.summarise(self.rows[:3])
|
|
g = s[("B0", "C0", "s1")]
|
|
self.assertEqual(g["sessions"], 3)
|
|
self.assertAlmostEqual(g["p99"], 10.2)
|
|
self.assertAlmostEqual(g["p99_spread"], 0.4)
|
|
self.assertAlmostEqual(g["low_spread"], 2)
|
|
|
|
def test_verdicts(self):
|
|
valid = [r for r in self.rows if analyze.is_valid(r)]
|
|
self.assertEqual(len(valid), 7)
|
|
s = analyze.summarise(valid)
|
|
sp = analyze.noise_spread(s, "B0")
|
|
base = s[("B2", "C0", "s1")]
|
|
self.assertEqual(analyze.verdict(base, s[("B2", "WIN", "s1")], *sp), "win")
|
|
self.assertEqual(analyze.verdict(base, s[("B2", "TIE", "s1")], *sp), "tie")
|
|
self.assertEqual(analyze.verdict(base, s[("B2", "LOSS", "s1")], *sp), "loss")
|
|
t = analyze.table(s, valid, "C0", "B0")
|
|
self.assertIn("| B2 | WIN | s1 |", t)
|
|
self.assertIn("| win |", t)
|
|
self.assertNotIn("BAD", t)
|
|
|
|
def test_pooled_baseline_and_cli(self):
|
|
rows = [r for r in self.rows if not (r["label"] == "B2" and r["variant"] == "C0")]
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = Path(d) / "s.jsonl"
|
|
p.write_text("".join(json.dumps(r) + "\n" for r in rows))
|
|
loaded = analyze.load_rows(p)
|
|
s = analyze.summarise([r for r in loaded if analyze.is_valid(r)])
|
|
b, lab = analyze.baseline_for(s, loaded, "B2", "s1", "C0")
|
|
self.assertEqual(lab, "*")
|
|
self.assertAlmostEqual(b["p99"], 10.2)
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = analyze.main(["--file", str(p), "--baseline", "C0", "--noise", "B0"])
|
|
self.assertEqual(rc, 0)
|
|
self.assertIn("| B2 | WIN | s1 |", buf.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|