1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
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"], len(b"hello-risc-v-gsp-rm-debugger"))
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()
|
