mirror of
https://github.com/lexmount/moli.git
synced 2026-09-23 16:01:25 +00:00
276 lines
12 KiB
Python
276 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from moli_benchmark.cli import (
|
|
_amiibo_limit,
|
|
_amiibo_modes,
|
|
_amiibo_pools,
|
|
_finish_report,
|
|
_report_date_output_dir,
|
|
_report_output_dir,
|
|
_run_runs,
|
|
_run_suites,
|
|
_startup_idle_seconds,
|
|
_startup_include_cdp_first_page,
|
|
_startup_include_cdp_warm_pages,
|
|
_startup_exit_code,
|
|
_startup_runs,
|
|
_startup_warm_pages,
|
|
build_parser,
|
|
)
|
|
from moli_benchmark.config import FORMAL_RESULTS_ROOT
|
|
from moli_benchmark.startup import FORMAL_STARTUP_IDLE_SECONDS, FORMAL_STARTUP_RUNS
|
|
|
|
|
|
class CliAmiiboProfileTests(unittest.TestCase):
|
|
def test_startup_formal_profile_defaults_to_full_workflow(self) -> None:
|
|
args = argparse.Namespace(
|
|
startup_profile="formal",
|
|
runs=None,
|
|
include_cdp_first_page=False,
|
|
include_cdp_warm_pages=False,
|
|
cdp_warm_pages=10,
|
|
idle_seconds=None,
|
|
)
|
|
|
|
self.assertEqual(_startup_runs(args), FORMAL_STARTUP_RUNS)
|
|
self.assertTrue(_startup_include_cdp_first_page(args))
|
|
self.assertTrue(_startup_include_cdp_warm_pages(args))
|
|
self.assertEqual(_startup_warm_pages(args), 10)
|
|
self.assertEqual(_startup_idle_seconds(args), FORMAL_STARTUP_IDLE_SECONDS)
|
|
|
|
def test_startup_explicit_values_override_formal_defaults(self) -> None:
|
|
args = argparse.Namespace(
|
|
startup_profile="formal",
|
|
runs=3,
|
|
include_cdp_first_page=False,
|
|
include_cdp_warm_pages=False,
|
|
cdp_warm_pages=4,
|
|
idle_seconds=[5.0, 1.0, 5.0],
|
|
)
|
|
|
|
self.assertEqual(_startup_runs(args), 3)
|
|
self.assertTrue(_startup_include_cdp_first_page(args))
|
|
self.assertTrue(_startup_include_cdp_warm_pages(args))
|
|
self.assertEqual(_startup_warm_pages(args), 4)
|
|
self.assertEqual(_startup_idle_seconds(args), (5.0, 1.0))
|
|
|
|
def test_startup_exit_uses_formal_gate_failures_only_for_formal_profile(self) -> None:
|
|
self.assertEqual(_startup_exit_code({"profile": "smoke", "total_failures": 0, "gate_failures": 3}), 0)
|
|
self.assertEqual(_startup_exit_code({"profile": "formal", "total_failures": 0, "gate_failures": 1}), 1)
|
|
|
|
def test_smoke_profile_defaults_to_small_amiibo_workload(self) -> None:
|
|
args = argparse.Namespace(amiibo_profile="smoke", pool=None, amiibo_mode=None, limit=None)
|
|
|
|
self.assertEqual(_amiibo_pools(args), (1,))
|
|
self.assertEqual(_amiibo_modes(args), ("session",))
|
|
self.assertEqual(_amiibo_limit(args), 5)
|
|
|
|
def test_formal_profile_defaults_to_full_amiibo_matrix(self) -> None:
|
|
args = argparse.Namespace(amiibo_profile="formal", pool=None, amiibo_mode=None, limit=None)
|
|
|
|
self.assertEqual(_amiibo_pools(args), (1, 2, 5, 10, 25, 100))
|
|
self.assertEqual(_amiibo_modes(args), ("session", "process"))
|
|
self.assertEqual(_amiibo_limit(args), 0)
|
|
|
|
def test_explicit_amiibo_values_override_profile_defaults(self) -> None:
|
|
args = argparse.Namespace(amiibo_profile="formal", pool=[5, 1, 5], amiibo_mode=["process"], limit=7)
|
|
|
|
self.assertEqual(_amiibo_pools(args), (5, 1))
|
|
self.assertEqual(_amiibo_modes(args), ("process",))
|
|
self.assertEqual(_amiibo_limit(args), 7)
|
|
|
|
def test_report_date_uses_formal_results_directory(self) -> None:
|
|
self.assertEqual(_report_date_output_dir("2026-05-07"), FORMAL_RESULTS_ROOT / "2026-05-07")
|
|
|
|
def test_report_date_rejects_non_iso_date(self) -> None:
|
|
with self.assertRaisesRegex(RuntimeError, "expected YYYY-MM-DD"):
|
|
_report_date_output_dir("05-07-2026")
|
|
with self.assertRaisesRegex(RuntimeError, "expected YYYY-MM-DD"):
|
|
_report_date_output_dir("20260507")
|
|
|
|
def test_report_output_dir_keeps_explicit_output_without_report_date(self) -> None:
|
|
output_dir = Path("/tmp/moli-benchmark-smoke")
|
|
args = argparse.Namespace(output_dir=output_dir, report_date=None)
|
|
|
|
self.assertEqual(_report_output_dir(args), output_dir)
|
|
|
|
def test_report_output_dir_prefers_report_date_for_formal_artifact(self) -> None:
|
|
args = argparse.Namespace(output_dir=Path("/tmp/ignored"), report_date="2026-05-07")
|
|
|
|
self.assertEqual(_report_output_dir(args), FORMAL_RESULTS_ROOT / "2026-05-07")
|
|
|
|
def test_run_smoke_profile_keeps_existing_default_suites(self) -> None:
|
|
args = argparse.Namespace(profile="smoke", suite=None, runs=None)
|
|
|
|
self.assertEqual(_run_suites(args), ("startup", "synthetic"))
|
|
self.assertEqual(_run_runs(args), 5)
|
|
|
|
def test_run_horizontal_profile_defaults_to_compare_and_cdp_session(self) -> None:
|
|
args = argparse.Namespace(profile="horizontal", suite=None, runs=None)
|
|
|
|
self.assertEqual(_run_suites(args), ("synthetic-compare", "cdp-session"))
|
|
self.assertEqual(_run_runs(args), 10)
|
|
|
|
def test_run_explicit_suite_and_runs_override_profile_defaults(self) -> None:
|
|
args = argparse.Namespace(profile="horizontal", suite=["synthetic-compare"], runs=3)
|
|
|
|
self.assertEqual(_run_suites(args), ("synthetic-compare",))
|
|
self.assertEqual(_run_runs(args), 3)
|
|
|
|
def test_wpt_runner_arguments_are_nextest_only(self) -> None:
|
|
parser = build_parser()
|
|
|
|
args = parser.parse_args(["wpt", "--runner", "nextest", "--no-run"])
|
|
self.assertEqual(args.runner, "nextest")
|
|
|
|
run_args = parser.parse_args(["run", "--suite", "wpt", "--wpt-runner", "nextest", "--wpt-no-run"])
|
|
self.assertEqual(run_args.wpt_runner, "nextest")
|
|
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as wpt_error:
|
|
parser.parse_args(["wpt", "--runner", "cargo", "--no-run"])
|
|
self.assertEqual(wpt_error.exception.code, 2)
|
|
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as run_error:
|
|
parser.parse_args(["run", "--suite", "wpt", "--wpt-runner", "cargo", "--wpt-no-run"])
|
|
self.assertEqual(run_error.exception.code, 2)
|
|
|
|
def test_agent_episode_defaults_match_the_canonical_local_benchmark(self) -> None:
|
|
args = build_parser().parse_args(["agent-episode"])
|
|
|
|
self.assertIsNone(args.target)
|
|
self.assertEqual(args.runs, 1)
|
|
self.assertEqual(args.workers, 1)
|
|
self.assertEqual(args.parallelism, 1)
|
|
self.assertEqual(args.step_dwell_ms, 14_000)
|
|
self.assertEqual(args.sample_interval_ms, 500)
|
|
|
|
def test_agent_episode_has_no_stress_or_live_profile(self) -> None:
|
|
parser = build_parser()
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
parser.parse_args(["agent-episode", "--profile", "stress"])
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
parser.parse_args(["agent-episode", "--live"])
|
|
|
|
def test_finish_report_builds_readiness_after_top_level_artifacts_exist(self) -> None:
|
|
summaries = [
|
|
{
|
|
"suite": "synthetic-matrix",
|
|
"profile": "formal",
|
|
"gate_failures": 0,
|
|
"formal_gate_rows": [
|
|
{"gate": "profile", "ok": True},
|
|
{"gate": "runs", "ok": True},
|
|
{"gate": "repeats", "ok": True},
|
|
{"gate": "concurrency_levels", "ok": True},
|
|
{"gate": "cases", "ok": True},
|
|
{"gate": "workload_failures", "ok": True},
|
|
{"gate": "stability_failures", "ok": True},
|
|
],
|
|
"targets": {"moli": {}, "lightpanda": {}, "chrome": {}, "obscura": {}},
|
|
},
|
|
{
|
|
"suite": "startup",
|
|
"profile": "formal",
|
|
"total_failures": 0,
|
|
"gate_failures": 0,
|
|
"formal_gate_rows": [
|
|
{"gate": "profile", "ok": True},
|
|
{"gate": "runs", "ok": True},
|
|
{"gate": "cdp-first-page", "ok": True},
|
|
{"gate": "cdp-warm-pages", "ok": True},
|
|
{"gate": "idle-footprint", "ok": True},
|
|
{"gate": "workload-failures", "ok": True},
|
|
],
|
|
},
|
|
{
|
|
"suite": "wpt",
|
|
"cases": 1,
|
|
"total_failures": 0,
|
|
"summary": {"unexpected_fail": 0, "skip": 0, "known_fail": 0},
|
|
},
|
|
{
|
|
"suite": "cdp-smoke",
|
|
"profile": "formal",
|
|
"total_failures": 0,
|
|
"gate_failures": 0,
|
|
"client_coverage": {"raw_cdp": True, "playwright": True, "puppeteer": True},
|
|
"client_rows": [
|
|
{"client": "raw_cdp", "gate_ok": True},
|
|
{"client": "playwright", "gate_ok": True},
|
|
{"client": "puppeteer", "gate_ok": True},
|
|
],
|
|
},
|
|
{"suite": "amiibo-crawler", "profile": "formal", "gate_failures": 0},
|
|
{
|
|
"suite": "wild-web",
|
|
"gate_failures": 0,
|
|
"seeds": ["zhihu-home", "toutiao-home"],
|
|
"targets": {"moli": {"extraction_failures": 0}},
|
|
},
|
|
]
|
|
target_matrix = {
|
|
"moli": {"available": True},
|
|
"lightpanda": {"available": True},
|
|
"chrome": {"available": True},
|
|
"obscura": {"available": True},
|
|
}
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
output_dir = Path(temp_dir)
|
|
(output_dir / "environment.json").write_text("{}\n", encoding="utf-8")
|
|
(output_dir / "versions.json").write_text("{}\n", encoding="utf-8")
|
|
_finish_report(output_dir=output_dir, moli_bin=None, target_matrix=target_matrix, summaries=summaries)
|
|
readiness = json.loads((output_dir / "publish-readiness.json").read_text(encoding="utf-8"))
|
|
|
|
top_level = next(check for check in readiness["checks"] if check["name"] == "top-level artifacts")
|
|
self.assertTrue(top_level["ok"])
|
|
self.assertEqual(readiness["status"], "publishable")
|
|
|
|
def test_finish_report_writes_top_level_diff_when_baseline_is_provided(self) -> None:
|
|
target_matrix = {"moli": {"available": True}}
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
baseline_dir = root / "baseline"
|
|
output_dir = root / "current"
|
|
baseline_dir.mkdir()
|
|
output_dir.mkdir()
|
|
(output_dir / "environment.json").write_text("{}\n", encoding="utf-8")
|
|
(output_dir / "versions.json").write_text("{}\n", encoding="utf-8")
|
|
(baseline_dir / "summary.json").write_text(
|
|
json.dumps({"suites": [{"suite": "startup", "total_failures": 2}]}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
_finish_report(
|
|
output_dir=output_dir,
|
|
moli_bin=None,
|
|
target_matrix=target_matrix,
|
|
summaries=[{"suite": "startup", "total_failures": 1}],
|
|
baseline_report=baseline_dir,
|
|
)
|
|
|
|
diff = json.loads((output_dir / "report-diff.json").read_text(encoding="utf-8"))
|
|
report_data = json.loads((output_dir / "report-data.json").read_text(encoding="utf-8"))
|
|
summary_md = (output_dir / "summary.md").read_text(encoding="utf-8")
|
|
index_html = (output_dir / "index.html").read_text(encoding="utf-8")
|
|
diff_csv_exists = (output_dir / "report-diff.csv").exists()
|
|
|
|
self.assertEqual(diff["summary"]["total_failures_delta"], -1)
|
|
self.assertEqual(report_data["report_diff"]["summary"]["total_failures_delta"], -1)
|
|
self.assertTrue(diff_csv_exists)
|
|
self.assertIn("Previous report diff", summary_md)
|
|
self.assertIn("Previous Report Diff", index_html)
|
|
self.assertIn("cdn.jsdelivr.net/npm/chart.js", index_html)
|
|
self.assertIn("report-data", index_html)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|