"""Tests for the new-version triage workflow (harness/triage.py, docs/40). The orchestration tests use injected fake step runners: no QEMU, no native boot, no firmware. The integration test uses the real TU104 570.144 fwimage read-only with a tiny budget and skips cleanly when it (or qemu-system-riscv64) is absent. Everything is bounded and touches no hardware, GPU, module, guest or network. Run:: python3 -m unittest tests.test_triage -v """ from __future__ import annotations import contextlib import io import json import os import sys import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) REAL_IMAGE = os.path.normpath( os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin") ) if TOOL_ROOT not in sys.path: sys.path.insert(0, TOOL_ROOT) from harness.common import qemu_available # noqa: E402 from harness.triage import ( # noqa: E402 DEFAULT_RUNNERS, FORMAT, PLAN_FORMAT, TriageOptions, build_plan, extract_version, main, run_triage, ) def _run_main(argv: list[str]) -> tuple[int, str, str]: out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): rc = main(argv) return rc, out.getvalue(), err.getvalue() def _fake_runners(fail: str | None = None, session_dir: str | None = None): """Deterministic step runners; optionally one raises.""" calls: list[str] = [] def make(name: str, **detail): def fn(options, out): calls.append(name) payload = {"status": "ok"} payload.update(detail) if session_dir and name in ("probe", "native_boot"): os.makedirs(session_dir, exist_ok=True) with open(os.path.join(session_dir, "session.json"), "w", encoding="utf-8") as fh: fh.write("{}\n") payload["sessions"] = [{"name": name, "dir": session_dir}] payload["session_hashes"] = {"session.json": "deadbeef"} return payload return fn def boom(options, out): calls.append(fail) raise RuntimeError("injected failure") runners = [ ("identity", boom if fail == "identity" else make("identity")), ("probe", boom if fail == "probe" else make("probe", rung="pre-B0")), ("native_boot", boom if fail == "native_boot" else make("native_boot", rung="B2")), ("ab_eval", boom if fail == "ab_eval" else make("ab_eval", verdict="above_lenient_only", above_qemu_line=False)), ("version_diff", boom if fail == "version_diff" else make("version_diff", sections={"image": "ok"})), ("features", boom if fail == "features" else make("features", unavailable=[])), ] return runners, calls class VersionExtractionTests(unittest.TestCase): def test_explicit_label_wins(self): version, source = extract_version("/does/not/matter.bin", "999.1") self.assertEqual((version, source), ("999.1", "explicit --version")) def test_sidecar_convention(self): with tempfile.TemporaryDirectory() as tmp: fw = os.path.join(tmp, "gsp-x.fwimage.bin") open(fw, "wb").close() with open(os.path.join(tmp, "gsp_version.txt"), "w", encoding="utf-8") as fh: fh.write("570.211W\n") version, source = extract_version(fw) self.assertEqual(version, "570.211W") self.assertEqual(source, "gsp_version.txt sidecar") def test_parent_directory_fallback(self): with tempfile.TemporaryDirectory() as tmp: parent = os.path.join(tmp, "580.10") os.makedirs(parent) fw = os.path.join(parent, "gsp-580.10.fwimage.bin") open(fw, "wb").close() self.assertEqual(extract_version(fw)[0], "580.10") def test_stem_fallback(self): with tempfile.TemporaryDirectory() as tmp: fw = os.path.join(tmp, "mystery-image.bin") open(fw, "wb").close() version, source = extract_version(fw) self.assertEqual(version, "mystery-image") self.assertEqual(source, "file stem") class OrchestrationTests(unittest.TestCase): def _options(self, tmp: str, **overrides) -> TriageOptions: fw = os.path.join(tmp, "new.fwimage.bin") if not os.path.exists(fw): with open(fw, "wb") as fh: fh.write(b"\x00" * 16) values = dict(firmware=fw, out=os.path.join(tmp, "triage-out"), budget=1000) values.update(overrides) return TriageOptions(**values) def test_report_shape_and_artifacts(self): with tempfile.TemporaryDirectory() as tmp: session_dir = os.path.join(tmp, "session") runners, calls = _fake_runners(session_dir=session_dir) report = run_triage(self._options(tmp), runners=runners) self.assertEqual(report["format"], FORMAT) self.assertEqual([s["name"] for s in report["steps"]], [n for n, _ in DEFAULT_RUNNERS]) self.assertTrue(all(s["status"] == "ok" for s in report["steps"])) self.assertEqual(report["errors"], []) self.assertTrue(report["sessions"]) self.assertEqual(report["sessions"][0]["hashes"]["session.json"], "deadbeef") self.assertEqual(report["support_readiness"]["verdict"], "sim-observable") self.assertFalse(report["support_readiness"]["baseline_compared"]) self.assertEqual(report["threshold_checks"]["required_steps_missing"], []) self.assertTrue(report["threshold_checks"]["beats_manual_baseline"]) for key in ("report_json", "report_md"): self.assertTrue(os.path.isfile(report["artifacts"][key]), key) with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh: loaded = json.load(fh) self.assertEqual(loaded["format"], FORMAT) self.assertEqual(len(loaded["steps"]), len(DEFAULT_RUNNERS)) with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh: md = fh.read() self.assertIn("New-version triage", md) self.assertIn("Support readiness", md) def test_partial_failure_does_not_crash(self): with tempfile.TemporaryDirectory() as tmp: runners, _calls = _fake_runners(fail="probe") report = run_triage(self._options(tmp), runners=runners) self.assertEqual(report["errors"], ["probe"]) step = next(s for s in report["steps"] if s["name"] == "probe") self.assertEqual(step["status"], "error") self.assertIn("injected failure", step["reason"]) self.assertEqual( next(s for s in report["steps"] if s["name"] == "native_boot")["status"], "ok" ) self.assertTrue(any("QEMU probe failed" in b for b in report["support_readiness"]["blockers"])) self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) def test_missing_firmware_is_reported_not_raised(self): with tempfile.TemporaryDirectory() as tmp: out = os.path.join(tmp, "out") report = run_triage(TriageOptions(firmware=os.path.join(tmp, "nope.bin"), out=out)) step = report["steps"][0] self.assertEqual(step["name"], "identity") self.assertEqual(step["status"], "error") self.assertEqual(report["support_readiness"]["verdict"], "blocked") self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) def test_budget_bounds(self): with tempfile.TemporaryDirectory() as tmp: options = self._options(tmp, budget=0) with self.assertRaises(ValueError): run_triage(options) with self.assertRaises(ValueError): run_triage(self._options(tmp, budget=3_000_000)) def test_plan_shape(self): with tempfile.TemporaryDirectory() as tmp: plan = build_plan(self._options(tmp)) self.assertEqual(plan["format"], PLAN_FORMAT) self.assertEqual([s["name"] for s in plan["steps"]], [n for n, _ in DEFAULT_RUNNERS]) self.assertEqual(plan["side_effects"]["hardware"], "none") class CliTests(unittest.TestCase): def test_rm610_refusal(self): with tempfile.TemporaryDirectory() as tmp: fw = os.path.join(tmp, "gsp-rm610.fwimage.bin") open(fw, "wb").close() out = os.path.join(tmp, "out") rc, _stdout, stderr = _run_main( ["--firmware", fw, "--version", "RM-610", "--out", out] ) self.assertEqual(rc, 3) self.assertIn("RM610", stderr) self.assertFalse(os.path.exists(out)) def test_dry_run_has_no_side_effects(self): with tempfile.TemporaryDirectory() as tmp: fw = os.path.join(tmp, "new.fwimage.bin") open(fw, "wb").close() out = os.path.join(tmp, "out") rc, stdout, _stderr = _run_main(["--firmware", fw, "--out", out, "--dry-run"]) self.assertEqual(rc, 0) plan = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1]) self.assertEqual(plan["format"], PLAN_FORMAT) self.assertFalse(os.path.exists(out)) def test_missing_firmware_exits_2(self): with tempfile.TemporaryDirectory() as tmp: out = os.path.join(tmp, "out") rc, _stdout, stderr = _run_main(["--firmware", os.path.join(tmp, "nope.bin"), "--out", out]) self.assertEqual(rc, 2) self.assertIn("firmware absent", stderr) self.assertFalse(os.path.exists(out)) def test_bad_budget_is_a_parser_error(self): with tempfile.TemporaryDirectory() as tmp: fw = os.path.join(tmp, "new.fwimage.bin") open(fw, "wb").close() with self.assertRaises(SystemExit): _run_main(["--firmware", fw, "--out", os.path.join(tmp, "out"), "--budget", "9999999"]) @unittest.skipUnless(os.path.isfile(REAL_IMAGE), "real 570.144 fwimage absent") class RealImageIntegrationTests(unittest.TestCase): def test_triage_real_image_tiny_budget(self): if not qemu_available(): self.skipTest("qemu-system-riscv64 not installed") with tempfile.TemporaryDirectory() as tmp: report = run_triage( TriageOptions( firmware=REAL_IMAGE, out=os.path.join(tmp, "out"), budget=3000, qemu_duration=0.2, step_timeout=120.0, ) ) identity = next(s for s in report["steps"] if s["name"] == "identity") self.assertEqual(identity["status"], "ok") self.assertIn("570.144", identity["firmware"]["version"]) probe = next(s for s in report["steps"] if s["name"] == "probe") self.assertEqual(probe["status"], "ok") native = next(s for s in report["steps"] if s["name"] == "native_boot") self.assertIn(native["status"], ("ok", "skipped")) self.assertIn( report["support_readiness"]["verdict"], ("sim-observable", "first-touch-only", "blocked") ) version_diff = next(s for s in report["steps"] if s["name"] == "version_diff") self.assertEqual(version_diff["status"], "skipped") self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) if __name__ == "__main__": unittest.main()