| 1 | from __future__ import annotations
|
| 2 |
|
| 3 | import json
|
| 4 | import subprocess
|
| 5 | import sys
|
| 6 | import tempfile
|
| 7 | import unittest
|
| 8 | from pathlib import Path
|
| 9 |
|
| 10 | ROOT = Path(__file__).resolve().parents[1]
|
| 11 | SCRIPT = ROOT / "scripts" / "runner_calibration.py"
|
| 12 |
|
| 13 | EXPECTED_KEYS = {
|
| 14 | "schema_version",
|
| 15 | "runner_id",
|
| 16 | "runner_class",
|
| 17 | "cpu_sha256_mib_per_s",
|
| 18 | "io_write_read_mib_per_s",
|
| 19 | "spawn_ms_median",
|
| 20 | "elapsed_s",
|
| 21 | }
|
| 22 |
|
| 23 |
|
| 24 | class RunnerCalibrationCliTest(unittest.TestCase):
|
| 25 | def test_quick_run_emits_schema_with_runner_identity(self) -> None:
|
| 26 | with tempfile.TemporaryDirectory() as workdir:
|
| 27 | completed = subprocess.run(
|
| 28 | [sys.executable, str(SCRIPT), "--quick", "--path", workdir],
|
| 29 | capture_output=True,
|
| 30 | text=True,
|
| 31 | timeout=60,
|
| 32 | )
|
| 33 | self.assertEqual(completed.returncode, 0, completed.stderr)
|
| 34 | row = json.loads(completed.stdout)
|
| 35 | self.assertEqual(set(row), EXPECTED_KEYS)
|
| 36 | self.assertEqual(row["schema_version"], 1)
|
| 37 | self.assertIsNotNone(row["runner_id"])
|
| 38 | self.assertGreater(row["elapsed_s"], 0)
|
| 39 | for key in ("cpu_sha256_mib_per_s", "io_write_read_mib_per_s", "spawn_ms_median"):
|
| 40 | if row[key] is not None:
|
| 41 | self.assertGreater(row[key], 0)
|
| 42 |
|
| 43 |
|
| 44 | if __name__ == "__main__":
|
| 45 | unittest.main()
|