"""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()