diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_package_sim.py | 216 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py | 301 |
2 files changed, 517 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_package_sim.py b/debug/risc-v-gsp-rm-vm/tests/test_package_sim.py new file mode 100644 index 0000000..c6df626 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_package_sim.py @@ -0,0 +1,216 @@ +"""Tests for the packaging/trialing feature (docs/29, harness/package_sim.py). + +Builds real packages from the bundled demo plus the SDK service, checks the +layout and manifest hashes, runs ``run.sh`` in a bounded subprocess and asserts +the injected call results, round-trips the ``--pack-image`` combined artifact, +and checks manifest/archive determinism. Skips cleanly when the demo firmware +or the SDK blob has not been built (``make demo && make sdk``); the real-image +test additionally skips when the 570.144 fwimage is absent. +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.normpath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +import harness.package_sim as pk # noqa: E402 +from greeneye.elf import load_elf # noqa: E402 + +DEMO_ELF = pk.DEMO_ELF +SERVICE_ELF = pk.DEFAULT_SERVICE +REAL_IMAGE = pk.REAL_IMAGE +EXPECTED_PING = 0x41 ^ pk.PING_XOR +EXPECTED_SUM = 0x30 + +def artifacts_available() -> bool: + return os.path.isfile(DEMO_ELF) and os.path.isfile(SERVICE_ELF) + + +def run_package(out: str, args: tuple = (), timeout: float = 120.0): + env = dict(os.environ) + env["GREENEYE_HOME"] = TOOL_ROOT + return subprocess.run( + ["/bin/sh", os.path.join(out, "run.sh"), *args], + cwd=out, + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def load_results(out: str) -> dict: + with open(os.path.join(out, "results", "results.json"), "r", encoding="utf-8") as fh: + return json.load(fh) + + +@unittest.skipUnless(artifacts_available(), "demo firmware or SDK service not built (make demo && make sdk)") +class PackageTests(unittest.TestCase): + def test_layout_and_manifest_hashes(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out)) + + for relpath in ( + "README.md", + "run.sh", + "manifest.json", + "scenario.py", + "payload/uproc_demo.elf", + "payload/injected_demo.elf", + "results/.gitkeep", + ): + self.assertTrue(os.path.isfile(os.path.join(out, relpath)), relpath) + + self.assertEqual(manifest["format"], pk.FORMAT) + self.assertEqual(manifest["target"]["kind"], "demo") + self.assertTrue(manifest["target"]["runnable"]) + self.assertEqual(manifest["target"]["sha256"], pk.sha256_file(DEMO_ELF)) + self.assertEqual(manifest["service"]["sha256"], pk.sha256_file(SERVICE_ELF)) + self.assertEqual(manifest["service"]["entry_symbol"], "ge_demo_service_dispatch") + + paths = [comp["path"] for comp in manifest["components"]] + self.assertEqual(paths, sorted(paths), "component order must be stable") + for comp in manifest["components"]: + full = os.path.join(out, comp["path"]) + self.assertEqual(comp["sha256"], pk.sha256_file(full), comp["path"]) + self.assertEqual(comp["bytes"], os.path.getsize(full), comp["path"]) + script = next(comp for comp in manifest["components"] if comp["path"] == "scenario.py") + self.assertEqual(script["sha256"], pk.sha256_bytes(pk.BUILTIN_SCENARIO.encode("utf-8"))) + + def test_run_sh_writes_passing_results(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out)) + proc = run_package(out) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + + results = load_results(out) + self.assertEqual(results["status"], "ok") + self.assertTrue(results["success"]) + self.assertEqual(results["calls"]["remote"]["ping"], EXPECTED_PING) + self.assertEqual(results["calls"]["remote"]["sum"], EXPECTED_SUM) + self.assertEqual(results["calls"]["via_firmware"], results["calls"]["remote"]) + self.assertTrue(results["calls"]["agree"]) + self.assertEqual(results["assertions"]["failed"], []) + self.assertGreaterEqual(results["hooks"]["rpc_dispatch"]["hits"], 1) + self.assertGreaterEqual(results["watches"]["gsp_ctrl.result"]["hits"], 1) + self.assertTrue(results["assertions"]["checks"]["uart_banner"]) + self.assertFalse(results["service"]["preinjected"]) + # Non-preinjected packages use greeneye.inject's default load address. + self.assertEqual(results["handlers_slot"], 0x81000000 + manifest["service"]["entry"]) + self.assertTrue(os.path.isfile(os.path.join(out, "results", "session", "session.json"))) + self.assertTrue(os.path.isfile(os.path.join(out, "results", "results.md"))) + + def test_pack_image_round_trip(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out, pack_image=True)) + self.assertIsNotNone(manifest["preinject"]) + self.assertTrue(manifest["preinject"]["runnable"]) + combined = os.path.join(out, "payload", "combined.elf") + preinject = os.path.join(out, "payload", "preinject.json") + self.assertTrue(os.path.isfile(combined)) + self.assertTrue(os.path.isfile(preinject)) + + demo = load_elf(DEMO_ELF) + service = load_elf(SERVICE_ELF) + merged = load_elf(combined) + self.assertEqual(merged.entry, demo.entry) + self.assertEqual( + len(merged.load_segments), + len(demo.load_segments) + len(service.load_segments), + ) + + proc = run_package(out) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + results = load_results(out) + self.assertEqual(results["status"], "ok") + self.assertTrue(results["service"]["preinjected"]) + self.assertEqual(results["calls"]["remote"], results["calls"]["via_firmware"]) + self.assertEqual(results["calls"]["remote"]["ping"], EXPECTED_PING) + self.assertEqual(results["calls"]["remote"]["sum"], EXPECTED_SUM) + + def test_manifest_and_tar_deterministic(self): + with mock.patch.dict(os.environ, {"SOURCE_DATE_EPOCH": "1700000000"}): + with tempfile.TemporaryDirectory() as tmp_a, tempfile.TemporaryDirectory() as tmp_b: + out_a = os.path.join(tmp_a, "trial") + out_b = os.path.join(tmp_b, "trial") + pk.build_package(pk.PackageRequest(out=out_a, pack_image=True, tar=True)) + pk.build_package(pk.PackageRequest(out=out_b, pack_image=True, tar=True)) + with open(os.path.join(out_a, "manifest.json"), "rb") as fh_a: + manifest_a = fh_a.read() + with open(os.path.join(out_b, "manifest.json"), "rb") as fh_b: + manifest_b = fh_b.read() + self.assertEqual(manifest_a, manifest_b) + with open(out_a + ".tar.gz", "rb") as fh_a: + tar_a = fh_a.read() + with open(out_b + ".tar.gz", "rb") as fh_b: + tar_b = fh_b.read() + self.assertEqual(tar_a, tar_b) + + def test_cli_build_and_run(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + proc = subprocess.run( + [ + sys.executable, + os.path.join(TOOL_ROOT, "harness", "package_sim.py"), + "--pack-image", + "--tar", + "--run", + "--out", + out, + ], + cwd=TOOL_ROOT, + capture_output=True, + text=True, + timeout=180.0, + ) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertTrue(os.path.isfile(out + ".tar.gz")) + self.assertEqual(load_results(out)["status"], "ok") + + +@unittest.skipUnless(artifacts_available(), "demo firmware or SDK service not built") +@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "570.144 fwimage absent") +class ImagePackageTests(unittest.TestCase): + def test_image_package_is_bounded_and_honest(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out, target="image")) + self.assertEqual(manifest["target"]["kind"], "image") + self.assertFalse(manifest["target"]["runnable"]) + self.assertTrue(os.path.isfile(os.path.join(out, "payload", os.path.basename(REAL_IMAGE)))) + self.assertTrue(os.path.isfile(os.path.join(out, "payload", "injected_demo.elf"))) + self.assertTrue(os.path.isfile(os.path.join(out, "payload", "device_map.json"))) + self.assertTrue(manifest["device_map"]["learn"]) + with open(os.path.join(out, "README.md"), "r", encoding="utf-8") as fh: + self.assertIn("cannot boot", fh.read()) + + proc = run_package(out, args=("--budget", "20000")) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + results = load_results(out) + self.assertEqual(results["status"], "blocked") + self.assertTrue(results["blocked_on_device_model"]) + self.assertFalse(results["success"]) + self.assertGreater(results["instructions_executed"], 0) + self.assertGreaterEqual(results["mmio_unknown"], 1) + self.assertTrue(results["first_unknown_mmio"]) + self.assertFalse(results["service"]["loaded"]) + self.assertTrue(os.path.isfile(os.path.join(out, "results", "learned_device_map.json"))) + with open(os.path.join(out, "results", "results.md"), "r", encoding="utf-8") as fh: + self.assertIn("cannot boot", fh.read()) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py b/debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py new file mode 100644 index 0000000..5c7e795 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py @@ -0,0 +1,301 @@ +"""Tests for the porting automation chain (docs/28, roadmap H4.1/R6). + +Pure/CLI tests need neither QEMU nor the demo; the one-version sweep test +boots the demo under QEMU and skips cleanly when qemu-system-riscv64 or the +demo ELF is absent. Everything else runs from a temp directory and is +side-effect free. +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +from unittest import mock + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness import port_driver_versions as pdv # noqa: E402 +from harness import port_new_device as pnd # noqa: E402 +from harness.common import qemu_available # noqa: E402 +from harness.version_diff import check_rm610_gate # noqa: E402 + +NOP = b"\x13\x00\x00\x00" # addi x0, x0, 0 + + +def _run_main(main_fn, argv: list[str]) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main_fn(argv) + return rc, out.getvalue(), err.getvalue() + + +def _plan_from_stdout(stdout: str) -> dict: + """The dry-run prints one JSON plan followed by [tool] status lines.""" + marker = stdout.find("\n[sweep]") + if marker < 0: + marker = stdout.find("\n[device]") + return json.loads(stdout[:marker] if marker >= 0 else stdout) + + +class VersionHelperTests(unittest.TestCase): + def test_parse_versions(self): + self.assertEqual(pdv.parse_versions("540,550,560,570"), ["540", "550", "560", "570"]) + self.assertEqual(pdv.parse_versions(" 570.144 , 570.144 "), ["570.144"]) + self.assertEqual(pdv.parse_versions(""), []) + with self.assertRaises(ValueError): + pdv.parse_versions("540$") + + def test_version_ge_610(self): + self.assertTrue(pdv.version_ge_610("610")) + self.assertTrue(pdv.version_ge_610("610.1")) + self.assertTrue(pdv.version_ge_610("620")) + self.assertFalse(pdv.version_ge_610("570.144")) + self.assertFalse(pdv.version_ge_610("609.99")) + + def test_default_cadence_is_documented_and_not_invented(self): + self.assertIn("540", pdv.DEFAULT_VERSIONS) + self.assertIn("570", pdv.DEFAULT_VERSIONS) + self.assertNotIn("610", pdv.DEFAULT_VERSIONS) + self.assertIn("DEFAULT_VERSIONS", pdv.__doc__ or "") + + def test_version_sort_key(self): + values = sorted(["560", "570.144", "540", "550"], key=pdv.version_sort_key) + self.assertEqual(values, ["540", "550", "560", "570.144"]) + + def test_non_rm610_labels_pass_the_gate(self): + check_rm610_gate(["570.144", "uproc_demo"], environ={}) + + +class SweepDryRunTests(unittest.TestCase): + def test_help_is_clean(self): + for main_fn in (pdv.main, pnd.main): + with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit) as ctx: + main_fn(["--help"]) + self.assertEqual(ctx.exception.code, 0) + self.assertIn("usage", buf.getvalue()) + + def test_help_documents_cron_and_agent_invocation(self): + with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit): + pdv.main(["--help"]) + text = buf.getvalue() + self.assertIn("cron", text) + self.assertIn("agent", text) + self.assertIn("RM610-APPROVED", text) + + def test_dry_run_explicit_target_exits_zero_and_creates_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + image = os.path.join(tmp, "gsp-570.144.bin") + with open(image, "wb") as fh: + fh.write(NOP * 4) + out = os.path.join(tmp, "out") + rc, stdout, _ = _run_main( + pdv.main, + ["--versions", "570.144", "--target", f"570.144={image}", "--out", out], + ) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["format"], pdv.PLAN_FORMAT) + self.assertEqual(plan["found"], 1) + self.assertEqual(plan["versions"][0]["status"], "found") + self.assertFalse(os.path.exists(out), "dry-run must not create --out") + + def test_dry_run_missing_input_dir_is_clean(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, stdout, _ = _run_main( + pdv.main, + ["--versions", "540,550", "--input-dir", os.path.join(tmp, "absent"), "--out", out], + ) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["missing"], 2) + self.assertTrue(all(row["status"] == "no-input" for row in plan["versions"])) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_refused_before_out(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, _, err = _run_main(pdv.main, ["--versions", "610", "--out", out]) + self.assertEqual(rc, 3) + self.assertIn("docs/23", err) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_passes_with_token(self): + rc, stdout, _ = _run_main( + pdv.main, ["--versions", "610", "--approval-token", "RM610-APPROVED"] + ) + self.assertEqual(rc, 0) + self.assertEqual(_plan_from_stdout(stdout)["rm610_gate"]["status"], "passed") + + def test_config_file_drives_the_plan(self): + with tempfile.TemporaryDirectory() as tmp: + image = os.path.join(tmp, "gsp-550.bin") + with open(image, "wb") as fh: + fh.write(NOP * 4) + config = os.path.join(tmp, "sweep.json") + with open(config, "w", encoding="utf-8") as fh: + json.dump({ + "format": pdv.CONFIG_FORMAT, + "versions": ["550"], + "baseline": "550", + "targets": {"550": [image]}, + }, fh) + rc, stdout, _ = _run_main(pdv.main, ["--config", config]) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["baseline"], "550") + self.assertEqual(plan["found"], 1) + + def test_config_rejects_unknown_keys(self): + with tempfile.TemporaryDirectory() as tmp: + config = os.path.join(tmp, "bad.json") + with open(config, "w", encoding="utf-8") as fh: + fh.write('{"versons": ["540"]}') + with self.assertRaises(ValueError): + pdv.load_config(config) + + +class SweepRunTests(unittest.TestCase): + def test_help_lists_exit_interpretation(self): + with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit): + pdv.main(["--help"]) + self.assertIn("--run", buf.getvalue()) + + def test_run_without_inputs_fails_cleanly(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, stdout, err = _run_main( + pdv.main, + ["--run", "--versions", "540,550", "--input-dir", os.path.join(tmp, "absent"), + "--out", out], + ) + self.assertEqual(rc, 1) + self.assertIn("no baseline target", stdout) + self.assertNotIn("Traceback", err) + self.assertTrue(os.path.exists(os.path.join(out, "sweep-summary.json"))) + + def test_completed_status_only_accepts_resumable_states(self): + with tempfile.TemporaryDirectory() as tmp: + for state, expected in (("ok", "ok"), ("partial", "partial"), + ("error", None), ("no-input", None)): + version_dir = os.path.join(tmp, state) + os.makedirs(version_dir, exist_ok=True) + with open(os.path.join(version_dir, "version.json"), "w", encoding="utf-8") as fh: + json.dump({"status": state}, fh) + self.assertEqual(pdv.completed_status(version_dir), expected, state) + + def test_run_resumes_completed_versions(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + for version in ("1", "2"): + version_dir = os.path.join(out, version) + os.makedirs(version_dir) + with open(os.path.join(version_dir, "version.json"), "w", encoding="utf-8") as fh: + json.dump({"status": "ok"}, fh) + fake = { + version: pdv.SweepTarget(version=version, paths=[os.path.join(tmp, f"{version}.bin")], + kind="raw", source="explicit") + for version in ("1", "2") + } + options = pdv.SweepOptions(versions=["1", "2"], out=out, run=True, step_timeout=5) + with mock.patch.object(pdv, "resolve_targets", return_value=fake): + report = pdv.run_sweep(options) + self.assertEqual(report["failed"], []) + self.assertEqual(report["ran"], 2) + self.assertTrue(all(row.get("resumed") for row in report["versions"])) + self.assertTrue(os.path.exists(os.path.join(out, "bundle-manifest.json"))) + + +class DevicePortDryRunTests(unittest.TestCase): + def test_dry_run_exits_zero_and_creates_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + firmware = os.path.join(tmp, "gsp-570.144.bin") + with open(firmware, "wb") as fh: + fh.write(NOP * 4) + out = os.path.join(tmp, "out") + rc, stdout, _ = _run_main( + pnd.main, ["--device", "rtx4090", "--firmware", firmware, "--out", out] + ) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["format"], pnd.PLAN_FORMAT) + self.assertEqual(plan["profile"]["action"], "would-create-scaffold") + self.assertTrue(plan["firmware"]["found"]) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_refused(self): + rc, _, err = _run_main(pnd.main, ["--device", "rm610-box"]) + self.assertEqual(rc, 3) + self.assertIn("docs/23", err) + + def test_dry_run_rm610_passes_with_env(self): + with mock.patch.dict(os.environ, {"GREENEYE_RM610_APPROVED": "1"}): + rc, stdout, _ = _run_main(pnd.main, ["--device", "rm610-box"]) + self.assertEqual(rc, 0) + self.assertEqual(_plan_from_stdout(stdout)["rm610_gate"]["status"], "passed") + + def test_run_without_firmware_fails_cleanly(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, _, err = _run_main(pnd.main, ["--run", "--device", "testdev", "--out", out]) + self.assertEqual(rc, 2) + self.assertIn("no device firmware", err) + self.assertNotIn("Traceback", err) + + def test_scaffold_profile_has_all_sections(self): + profile = pnd.scaffold_profile("testdev") + self.assertEqual(profile["format"], pnd.PROFILE_FORMAT) + for key in ("pci", "bars", "fb", "engines", "interrupts", "rom", "physical_validation"): + self.assertIn(key, profile) + self.assertEqual(sorted(pnd.profile_missing_fields(profile)), + ["bars", "engines", "fb", "interrupts", "rom"]) + + +@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") +@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") +class OneVersionSweepQemuTests(unittest.TestCase): + def test_one_version_sweep_writes_the_evidence_bundle(self): + with tempfile.TemporaryDirectory(prefix="port-sweep-") as tmp: + out = os.path.join(tmp, "out") + rc, stdout, err = _run_main( + pdv.main, + ["--run", "--versions", "demo", "--target", f"demo={DEMO_ELF}", + "--baseline", "demo", "--out", out, + "--probe-instructions", "16", "--probe-duration", "0.05", + "--scenario-duration", "0.05", "--max-stops", "50", + "--mutation-iterations", "1", "--mutation-seed", "1", + "--mutation-timeout", "0.05", "--mutation-settle", "0.01", + "--step-timeout", "180"], + ) + self.assertEqual(rc, 0, err + stdout) + version_dir = os.path.join(out, "demo") + with open(os.path.join(version_dir, "version.json"), encoding="utf-8") as fh: + version = json.load(fh) + self.assertEqual(version["status"], "ok", version) + self.assertEqual(pdv.completed_status(version_dir), "ok") + for rel in ("version-diff/version-diff.json", + "portability-matrix/portability-report.json", + "mutation-smoke/findings.jsonl", + "bundle-manifest.json"): + self.assertTrue(os.path.exists(os.path.join(version_dir, rel)), rel) + with open(os.path.join(out, "sweep-summary.json"), encoding="utf-8") as fh: + summary = json.load(fh) + self.assertEqual(summary["failed"], []) + self.assertEqual(summary["ran"], 1) + self.assertTrue(summary["interpretation"].startswith("pass")) + with open(os.path.join(version_dir, "bundle-manifest.json"), encoding="utf-8") as fh: + manifest = json.load(fh) + paths = {row["path"] for row in manifest["files"]} + self.assertIn("version-diff/version-diff.json", paths) + + +if __name__ == "__main__": + unittest.main() |