diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_triage.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_triage.py | 264 |
1 files changed, 264 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_triage.py b/debug/risc-v-gsp-rm-vm/tests/test_triage.py new file mode 100644 index 0000000..35ac128 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_triage.py @@ -0,0 +1,264 @@ +"""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() |