1 files changed, 107 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py b/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
new file mode 100644
index 0000000..7640e13
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
@@ -0,0 +1,107 @@
+"""End-to-end tests for the value-validation demos (docs/25).
+
+Runs the fastest demos as subprocesses exactly as a developer would, parses
+their machine-readable ``RESULT_JSON`` and asserts the answers, then checks
+that ``run_all.py`` emits its aggregate results JSON.
+
+Skipped cleanly when qemu-system-riscv64 or the bundled demo firmware is
+absent; the demos that use the native backend still need the demo ELF.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.dirname(HERE)
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+DEMO_DIR = os.path.join(TOOL_ROOT, "harness", "value_demos")
+RESULTS_JSON = os.path.join(DEMO_DIR, "artifacts", "results.json")
+
+
+def qemu_available() -> bool:
+    return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
+
+
+def _run_demo(script: str, backend: str, timeout: float = 60.0) -> dict:
+    env = dict(os.environ)
+    env["VALUE_DEMO_BACKEND"] = backend
+    proc = subprocess.run(
+        [sys.executable, os.path.join(DEMO_DIR, script)],
+        cwd=TOOL_ROOT,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        env=env,
+    )
+    for line in reversed(proc.stdout.splitlines()):
+        if line.startswith("RESULT_JSON: "):
+            return json.loads(line[len("RESULT_JSON: ") :])
+    raise AssertionError(
+        f"{script}: no RESULT_JSON (exit {proc.returncode})\n"
+        f"stdout tail:\n{proc.stdout[-1200:]}\nstderr tail:\n{proc.stderr[-1200:]}"
+    )
+
+
+@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
+@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)")
+class ValueDemoTests(unittest.TestCase):
+    def test_s01_reply_consumer(self):
+        result = _run_demo("s01_reply_consumer.py", "qemu")
+        self.assertTrue(result["success"], result)
+        replies = {r["cmd"]: r for r in result["replies"]}
+        self.assertEqual(replies[1]["result"], 0x1234 ^ 0xBEEF)
+        self.assertTrue(replies[1]["consumed"])
+        self.assertEqual(replies[7]["result"], (~7) & 0xFFFFFFFFFFFFFFFF)
+        self.assertFalse(replies[6]["consumed"], "FAULT must trap before the reply store")
+        self.assertGreater(result["elapsed_s"], 0)
+
+    def test_s04_run_until_function(self):
+        result = _run_demo("s04_run_until_function.py", "native")
+        self.assertTrue(result["success"], result)
+        self.assertTrue(result["natural_path"]["reachable"])
+        self.assertTrue(result["before_trigger"]["timeout"])
+        self.assertTrue(result["after_trigger"]["reachable"])
+        self.assertEqual(result["after_trigger"]["dispatch_args"][0]["cmd"], 6)
+        self.assertEqual(result["direct_call"]["result"], 0x41 ^ 0xBEEF)
+
+    def test_s05_snapshot_diff(self):
+        result = _run_demo("s05_snapshot_diff.py", "native")
+        self.assertTrue(result["success"], result)
+        self.assertTrue({"result", "rpc_count", "last_rpc", "buf"}.issubset(set(result["changed_field_names"])))
+        self.assertEqual(result["after"]["result"], 15)
+
+    def test_s09_ghidra_join(self):
+        result = _run_demo("s09_grabbing_a_ghidra_address.py", "qemu")
+        self.assertTrue(result["success"], result)
+        self.assertEqual(result["hits"], [int(result["runtime_addr"], 16)])
+        self.assertEqual(int(result["ghidra_addr"], 16) + result["bias"], int(result["runtime_addr"], 16))
+        self.assertIn("rpc_fault", result["greeneye_hits_tsv"])
+
+    def test_run_all_emits_results_json(self):
+        proc = subprocess.run(
+            [sys.executable, os.path.join(DEMO_DIR, "run_all.py"), "--only", "s01,s04,s05"],
+            cwd=TOOL_ROOT,
+            capture_output=True,
+            text=True,
+            timeout=120,
+            env={**os.environ, "VALUE_DEMO_BACKEND": "native"},
+        )
+        self.assertEqual(proc.returncode, 0, proc.stdout[-1500:] + proc.stderr[-1500:])
+        self.assertTrue(os.path.exists(RESULTS_JSON), "run_all.py did not write results.json")
+        with open(RESULTS_JSON, "r", encoding="utf-8") as fh:
+            payload = json.load(fh)
+        self.assertEqual(payload["format"], "greeneye-value-demos-results/1")
+        self.assertEqual(payload["demos_selected"], ["s01", "s04", "s05"])
+        self.assertEqual(payload["demos_failed"], [])
+        self.assertEqual(len(payload["demos_passed"]), 3)
+        for result in payload["results"]:
+            self.assertGreater(result["runner_elapsed_s"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()