diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_bridge.py | 9 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_core.py | 42 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_device_map.py | 325 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_discovery.py | 214 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_harness.py | 144 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_mcp.py | 431 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py | 244 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py | 224 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_value_demos.py | 107 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_version_diff.py | 333 |
10 files changed, 2072 insertions, 1 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_bridge.py b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py index 472bc49..7eeb46c 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_bridge.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py @@ -9,11 +9,20 @@ import unittest import greeneye as ge from greeneye.gdbridge import GhidraBridge +from greeneye.ghidra import render_ghidra_command HERE = os.path.dirname(os.path.abspath(__file__)) DEMO_ELF = os.path.join(HERE, "..", "firmware", "uproc_demo", "build", "uproc_demo.elf") +class GhidraCommandTests(unittest.TestCase): + def test_render_command_names_project_program_and_hits(self): + cmd = render_ghidra_command("/proj", "P", "gsp.bin", "/tmp/hits.tsv") + self.assertIn("analyzeHeadless /proj P", cmd) + self.assertIn("-process gsp.bin", cmd) + self.assertIn("MarkGreenEyeHits.java /tmp/hits.tsv", cmd) + + def qemu_available() -> bool: return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64")) diff --git a/debug/risc-v-gsp-rm-vm/tests/test_core.py b/debug/risc-v-gsp-rm-vm/tests/test_core.py index 907a2eb..f02b2e6 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_core.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_core.py @@ -3,10 +3,12 @@ from __future__ import annotations import json import os +import struct import tempfile import unittest -from greeneye.elf import EM_RISCV, parse_elf, load_elf +from greeneye.elf import EM_RISCV, parse_elf, load_elf, write_relocated_elf +from greeneye.errors import GreenEyeError from greeneye.structs import Field, Schema, decode, render, schema_from_json from greeneye.symbols import Symbol, SymbolDB @@ -36,6 +38,19 @@ class ElfTests(unittest.TestCase): with self.assertRaises(Exception): parse_elf(b"not an elf") + def test_write_relocated_elf(self): + if not os.path.exists(DEMO_ELF): + self.skipTest("demo firmware not built") + image = load_elf(DEMO_ELF) + with tempfile.TemporaryDirectory(prefix="greeneye-reloc-") as tmp: + out = os.path.join(tmp, "relocated.elf") + path = write_relocated_elf(DEMO_ELF, out, 0x1000) + with open(path, "rb") as fh: + data = fh.read() + self.assertEqual(struct.unpack_from("<Q", data, 24)[0], image.entry + 0x1000) + moved = parse_elf(data) + self.assertEqual(moved.entry, image.entry + 0x1000) + class SymbolTests(unittest.TestCase): def setUp(self): @@ -50,6 +65,11 @@ class SymbolTests(unittest.TestCase): self.assertEqual(self.db.annotate(0x1010), "0x1010 alpha+0x10") self.assertEqual(self.db.lookup("beta"), 0x1200) + def test_by_subsystem(self): + self.db.add(Symbol(addr=0x2000, name="gamma", size=8, source="test", subsystem="core")) + self.assertEqual([s.name for s in self.db.by_subsystem("core")], ["gamma"]) + self.assertEqual(self.db.by_subsystem("nope"), []) + def test_names_tsv(self): with tempfile.NamedTemporaryFile("w", suffix=".tsv", delete=False) as fh: fh.write("# comment\n0x00279000\t_DT_INIT\tghidra\n0x00279020\tFUN_x\n") @@ -89,6 +109,26 @@ class SymbolTests(unittest.TestCase): finally: os.unlink(path) + def test_json_list_payload(self): + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + json.dump([{"addr": "0x2000", "name": "listed", "size": 4}], fh) + path = fh.name + try: + self.assertEqual(self.db.load_json(path), 1) + self.assertEqual(self.db.lookup("listed"), 0x2000) + finally: + os.unlink(path) + + def test_json_bad_input_raises_greeneye_error(self): + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh: + fh.write("not json") + path = fh.name + try: + with self.assertRaises(GreenEyeError): + self.db.load_json(path) + finally: + os.unlink(path) + class StructTests(unittest.TestCase): SCHEMA = { diff --git a/debug/risc-v-gsp-rm-vm/tests/test_device_map.py b/debug/risc-v-gsp-rm-vm/tests/test_device_map.py new file mode 100644 index 0000000..a38085a --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_device_map.py @@ -0,0 +1,325 @@ +"""Device-map (H2.1/H2.2 start) tests: harvest, merge, query, lenient MMIO. + +Unit tests use synthetic QEMU logs and tiny hand-assembled RV64 programs, so +they need neither QEMU nor a firmware build. The real-image test is bounded +(200k instructions, no QEMU) and skips when the TU104 570.144 fwimage is +absent. +""" +from __future__ import annotations + +import json +import os +import tempfile +import unittest + +from greeneye.device_map import ( + SCHEMA, + DeviceMap, + harvest_from_text, + harvest_probe_session, +) +from greeneye.errors import GreenEyeError +from sim.machine import R_BUDGET, R_EBREAK, R_FAULT, Machine + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.normpath(os.path.join(HERE, "..")) +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +REAL_IMAGE = os.path.normpath( + os.path.join( + HERE, "..", "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin" + ) +) + +BASE = 0x80000000 +TOUCH_A = 0x2000000000000A00 # QEMU probe first-touch address (570.144) +TOUCH_B = 0x2000000000000100 +BIAS = 0x80000000 + +# lui t0,0x20000; slli t0,t0,32; addi t0,t0,0x600; addi t0,t0,0x400; +# lw a0,0(t0); sw a0,0(t0); ebreak +PROGRAM_LOAD_STORE = bytes.fromhex( + "b702002093920202938202609382024003a5020023a0a20073001000" +) +# ... lw a0,0(t0); lui t1,0x20000; slli t1,t1,32; addi t1,t1,0x100; +# sd a0,0(t1); ebreak (two distinct unknown addresses) +PROGRAM_TWO_TOUCHES = bytes.fromhex( + "b702002093920202938202609382024003a502003703002013130302130303102330a30073001000" +) + +SAMPLE_LOG = """\ +Invalid read at addr 0x2000000000000A00, size 4, region '(null)', reason: rejected +riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000005, epc:0x0000000084009014, tval:0x2000000000000a00, desc=fault_load +Invalid read at addr 0x2000000000000A00, size 4, region '(null)', reason: rejected +riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000005, epc:0x0000000084009014, tval:0x2000000000000a00, desc=fault_load +Invalid write at addr 0x2000000000001000, size 8, region '(null)', reason: rejected +riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000007, epc:0x0000000084009020, tval:0x2000000000001000, desc=fault_store +""" + + +def _run_program(program: bytes, **kwargs): + machine = Machine(**kwargs) + machine.load_segments([(BASE, program)], BASE) + return machine, machine.run(100) + + +class HarvestTests(unittest.TestCase): + def test_harvest_from_text_fields(self): + dm = harvest_from_text(SAMPLE_LOG, bias=BIAS) + entries = {(e.address, e.width, e.kind): e for e in dm.entries()} + self.assertEqual(set(entries), {(TOUCH_A, 4, "r"), (0x2000000000001000, 8, "w")}) + read = entries[(TOUCH_A, 4, "r")] + self.assertEqual(read.hits, 2) + self.assertEqual(read.first_epc, 0x84009014) + self.assertEqual(read.first_epc_fw, 0x4009014) + self.assertEqual(read.source, "probe") + self.assertIn("fault_load", read.notes) + write = entries[(0x2000000000001000, 8, "w")] + self.assertEqual(write.hits, 1) + self.assertEqual(write.first_epc_fw, 0x4009020) + self.assertEqual(dm.meta["bias"], BIAS) + + def test_harvest_skips_fetch_faults(self): + log = ( + "riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000001, " + "epc:0x0000000000000000, tval:0x0, desc=fault_fetch\n" + ) + self.assertEqual(len(harvest_from_text(log, bias=BIAS)), 0) + + def test_harvest_probe_session_blob(self): + with tempfile.TemporaryDirectory() as root: + os.makedirs(os.path.join(root, "blobs")) + with open(os.path.join(root, "blobs", "qemu.log"), "w", encoding="utf-8") as fh: + fh.write(SAMPLE_LOG) + session = { + "format": "greeneye-session/1", + "bias": BIAS, + "entry": 0x4009000, + "firmware": "/nonexistent/gsp-570.144.fwimage.bin", + "qemu_log_blob": "blobs/qemu.log", + "qemu_fault_count": 3, + } + with open(os.path.join(root, "session.json"), "w", encoding="utf-8") as fh: + json.dump(session, fh) + dm = harvest_probe_session(root) + self.assertEqual(len(dm), 2) + self.assertEqual(dm.meta["bias"], BIAS) + self.assertEqual(dm.meta["qemu_fault_count"], 3) + self.assertNotIn("firmware_sha256", dm.meta) # firmware absent: honest to omit + + def test_harvest_probe_session_first_fault_fallback(self): + with tempfile.TemporaryDirectory() as root: + session = { + "bias": BIAS, + "qemu_first_fault": { + "epc": 0x84009014, + "cause": 5, + "desc": "fault_load", + "access": {"rw": "read", "access_addr": TOUCH_A, "size": 4, + "region": "(null)", "reason": "rejected"}, + }, + } + with open(os.path.join(root, "session.json"), "w", encoding="utf-8") as fh: + json.dump(session, fh) + dm = harvest_probe_session(root) + self.assertEqual(len(dm), 1) + self.assertEqual(dm.lookup(TOUCH_A, 4), "zero") # observed, no value recorded + + def test_harvest_bias_override(self): + with tempfile.TemporaryDirectory() as root: + session = { + "qemu_first_fault": { + "epc": 0x84009014, + "cause": 5, + "desc": "fault_load", + "access": {"rw": "read", "access_addr": TOUCH_A, "size": 4, + "region": "(null)", "reason": "rejected"}, + }, + } + with open(os.path.join(root, "session.json"), "w", encoding="utf-8") as fh: + json.dump(session, fh) + dm = harvest_probe_session(root, bias=BIAS) + entry = dm.entries()[0] + self.assertEqual(entry.first_epc, 0x84009014) + self.assertEqual(entry.first_epc_fw, 0x4009014) + self.assertEqual(dm.meta["bias"], BIAS) + + def test_harvest_missing_session_raises(self): + with tempfile.TemporaryDirectory() as root: + with self.assertRaises(GreenEyeError): + harvest_probe_session(os.path.join(root, "nope")) + + +class MapIoTests(unittest.TestCase): + def _map(self) -> DeviceMap: + dm = DeviceMap(meta={"firmware": "gsp.bin", "bias": BIAS}) + dm.record(TOUCH_A, 4, "r", pc=0x84009014, bias=BIAS, source="probe") + dm.record(0x2000, 8, "w", value=7, pc=0x84009020, bias=BIAS, source="manual") + return dm + + def test_roundtrip_is_byte_identical(self): + dm = self._map() + with tempfile.TemporaryDirectory() as root: + path = os.path.join(root, "map.json") + dm.save(path) + first = open(path, "rb").read() + again = DeviceMap.load(path) + again.save(path) + self.assertEqual(open(path, "rb").read(), first) + self.assertEqual(again.to_dict(), dm.to_dict()) + self.assertEqual(again.meta["bias"], BIAS) + + def test_entries_are_deterministically_ordered(self): + dm = DeviceMap() + for addr in (0x3000, 0x1000, 0x2000): + dm.record(addr, 4, "r") + self.assertEqual([e.address for e in dm.entries()], [0x1000, 0x2000, 0x3000]) + + def test_merge_sums_hits_and_unions_values(self): + left = DeviceMap() + left.record(0x1000, 4, "r", value=1) + left.record(0x1000, 4, "r", value=1) + right = DeviceMap(meta={"source": "manual"}) + right.record(0x1000, 4, "r", value=2) + right.record(0x2000, 8, "w") + merged = left.merge(right) + entries = {(e.address, e.kind): e for e in merged.entries()} + self.assertEqual(entries[(0x1000, "r")].hits, 3) + self.assertEqual(entries[(0x1000, "r")].values, (1, 2)) + self.assertEqual(entries[(0x2000, "w")].hits, 1) + self.assertEqual(merged.meta["source"], "manual") + + def test_rejects_unknown_schema(self): + with self.assertRaises(GreenEyeError): + DeviceMap.from_json(json.dumps({"schema": "greeneye-device-map/2", "entries": []})) + self.assertEqual(SCHEMA, "greeneye-device-map/1") + + +class LookupTests(unittest.TestCase): + def test_unimplemented_zero_and_value(self): + dm = DeviceMap() + self.assertEqual(dm.lookup(0x1234, 4), "unimplemented") + dm.record(0x2000, 4, "r", value=0xAB) + self.assertEqual(dm.lookup(0x2000, 4), "value:0xab") + self.assertEqual(dm.lookup(0x2000, 8), "value:0xab") # width fallback + dm.record(0x3000, 4, "r", value=None) + self.assertEqual(dm.lookup(0x3000, 4), "zero") + + def test_write_and_opposite_direction(self): + dm = DeviceMap() + dm.record(0x2000, 4, "r", value=5) + self.assertEqual(dm.lookup(0x2000, 4, write=True), "zero") # absorb + dm.record(0x3000, 4, "w", value=9) + self.assertEqual(dm.lookup(0x3000, 4, write=True), "zero") + self.assertEqual(dm.lookup(0x3000, 4), "zero") # write-only reads back zero + + +class LenientMachineTests(unittest.TestCase): + def test_strict_default_faults_without_logging(self): + machine, result = _run_program(PROGRAM_LOAD_STORE) + self.assertEqual(result[0], R_FAULT) + self.assertEqual(result[1], BASE + 16) # PC of the faulting load + self.assertEqual(result[2], 5) # load access fault + self.assertEqual(result[3], TOUCH_A) + self.assertEqual(machine.unknown_mmio, "fault") + self.assertEqual(machine.mmio_log, []) + self.assertEqual(machine.mmio_unknown, 0) + + def test_bad_policy_rejected(self): + with self.assertRaises(ValueError): + Machine(unknown_mmio="ignore") + + def test_lenient_returns_zero_and_logs(self): + machine, result = _run_program(PROGRAM_TWO_TOUCHES, unknown_mmio="lenient") + self.assertEqual(result[0], R_EBREAK) + self.assertEqual(machine.x[10], 0) # load returned 0 + self.assertEqual(machine.mmio_unknown, 2) + self.assertEqual( + machine.mmio_log, + [ + {"pc": BASE + 16, "addr": TOUCH_A, "size": 4, "rw": "r"}, + {"pc": BASE + 32, "addr": TOUCH_B, "size": 8, "rw": "w"}, + ], + ) + self.assertIsNone(machine.device_map) # no learning requested + + def test_lenient_learns_observed_values(self): + machine, result = _run_program( + PROGRAM_LOAD_STORE, unknown_mmio="lenient", learn_mmio=True + ) + self.assertEqual(result[0], R_EBREAK) + self.assertEqual(machine.mmio_unknown, 1) # the store is served once learned + entry = machine.device_map.entries()[0] + self.assertEqual((entry.address, entry.kind, entry.values), (TOUCH_A, "r", (0,))) + self.assertEqual(entry.first_epc_fw, BASE + 16) # bias 0: PC unchanged + self.assertEqual(machine.device_map.lookup(TOUCH_A, 4), "value:0x0") + + def test_map_serves_mapped_values_even_in_strict_mode(self): + dm = DeviceMap() + dm.record(TOUCH_A, 4, "r", value=0x12345678) + machine, result = _run_program(PROGRAM_LOAD_STORE, device_map=dm) + self.assertEqual(result[0], R_EBREAK) + self.assertEqual(machine.x[10], 0x12345678) + self.assertEqual(machine.mmio_unknown, 0) + + def test_ram_and_uart_are_unaffected(self): + # lui t0,0x10000; addi a0,zero,'A'; sb a0,0(t0); lbu a1,5(t0); ebreak + program = bytes.fromhex("b7020010130510042380a20083c5520073001000") + machine, result = _run_program(program, unknown_mmio="lenient") + self.assertEqual(result[0], R_EBREAK) + self.assertEqual(machine.uart.text(), "A") + self.assertEqual(machine.x[11], 0x60) # LSR: THRE | TEMT + self.assertEqual(machine.mmio_unknown, 0) + + +class LaunchNativeTests(unittest.TestCase): + @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware absent") + def test_launch_native_accepts_lenient_options(self): + from greeneye.sim_backend import launch_native + + with tempfile.TemporaryDirectory() as root: + map_path = os.path.join(root, "map.json") + DeviceMap(meta={"firmware": "demo"}).save(map_path) + session = launch_native( + kernel=DEMO_ELF, + device_map=map_path, + unknown_mmio="lenient", + name="unittest-device-map", + ) + try: + self.assertEqual(session.target.machine.unknown_mmio, "lenient") + self.assertEqual(session.target.mmio_unknown, 0) + self.assertEqual(session.target.mmio_log, []) + finally: + session.close() + + +@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "TU104 570.144 fwimage absent") +class RealImageLenientTests(unittest.TestCase): + def test_first_touch_matches_qemu_probe(self): + from greeneye.elf import load_elf + + image = load_elf(REAL_IMAGE) + dm = DeviceMap(meta={"bias": BIAS}) + machine = Machine( + device_map=dm, unknown_mmio="lenient", learn_mmio=True, bias=BIAS + ) + machine.load_segments(image.flat_load_plan(bias=BIAS), image.relocated_entry(bias=BIAS)) + self.assertEqual(machine.pc, 0x84009000) + reason, pc, _, _ = machine.run(200_000) + self.assertGreaterEqual(machine.mmio_unknown, 1) + self.assertEqual( + machine.mmio_log[0], + {"pc": 0x84009014, "addr": TOUCH_A, "size": 4, "rw": "r"}, + ) + # Lenient execution runs past the QEMU first fault to the budget. + self.assertEqual(reason, R_BUDGET) + self.assertEqual(machine.insns, 200_000) + self.assertEqual(pc, 0x84009070) + entry = dm.entries()[0] + self.assertEqual(entry.first_epc_fw, 0x4009014) + self.assertEqual(entry.values, (0,)) + self.assertEqual(dm.meta["bias"], BIAS) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_discovery.py b/debug/risc-v-gsp-rm-vm/tests/test_discovery.py new file mode 100644 index 0000000..4a1cfc2 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_discovery.py @@ -0,0 +1,214 @@ +"""Integration tests for unknown-call discovery and reachability (docs/19). + +The integration tests drive the two harness CLIs as subprocesses against the +bundled demo firmware and skip cleanly when qemu-system-riscv64 or the demo +ELF is absent. The pure-function tests need neither. +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest + +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") +DISCOVERY = os.path.join(TOOL_ROOT, "harness", "control_discovery.py") +REACH = os.path.join(TOOL_ROOT, "harness", "reach.py") +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.common import qemu_available # noqa: E402 +from harness.control_discovery import ( # noqa: E402 + DEMO_CASE_HANDLERS, + classify_outcome, + generate_probes, + pattern_args, +) + +MASK64 = (1 << 64) - 1 + + +def _run(args: list[str], timeout: float = 120.0) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable] + args, + capture_output=True, + text=True, + cwd=TOOL_ROOT, + timeout=timeout, + ) + + +def _read_findings(out: str) -> list[dict]: + with open(os.path.join(out, "findings.jsonl"), encoding="utf-8") as fh: + return [json.loads(line) for line in fh if line.strip()] + + +class DiscoveryUnitTests(unittest.TestCase): + def test_pattern_args_deterministic(self): + self.assertEqual(pattern_args("zero", 1, 3), (0, 0, 0)) + self.assertEqual(pattern_args("boundary", 1, 3), (MASK64, 1, 0x7FFF_FFFF_FFFF_FFFF)) + self.assertEqual(pattern_args("sentinel", 5, 3), pattern_args("sentinel", 5, 3)) + self.assertNotEqual(pattern_args("sentinel", 5, 3), pattern_args("sentinel", 5, 4)) + self.assertTrue(all(0 <= value <= MASK64 for value in pattern_args("sentinel", 9, 1))) + + def test_generate_probes_order_and_cap(self): + probes = generate_probes(4, ["zero", "sentinel"], seed=1) + self.assertEqual([probe.cmd for probe in probes], [1, 1, 2, 2, 3, 3, 4, 4]) + self.assertEqual([probe.pattern for probe in probes[0:2]], ["zero", "sentinel"]) + self.assertEqual(len(generate_probes(64, ["zero", "sentinel", "boundary"], seed=1, iterations=5)), 5) + + def test_classify_outcome_priority(self): + base = dict(cmd=6, is_injected=False, handler="rpc_fault", dispatch_seen=True, call_retval=None) + self.assertEqual(classify_outcome(faults=[{"cause": 3}], signals=[], settle_ok=True, **base), "trap") + self.assertEqual(classify_outcome(faults=[{"cause": 5}], signals=[], settle_ok=True, **base), "fault") + self.assertEqual(classify_outcome(faults=[], signals=[], settle_ok=True, **base), "known-handler") + self.assertEqual( + classify_outcome( + cmd=16, is_injected=True, handler="rpc_ping", dispatch_seen=True, call_retval=None, + faults=[], signals=[], settle_ok=True, + ), + "injected-slot", + ) + self.assertEqual( + classify_outcome( + cmd=7, is_injected=False, handler=None, dispatch_seen=True, call_retval=None, + faults=[], signals=[], settle_ok=True, + ), + "default-path", + ) + self.assertEqual( + classify_outcome( + cmd=7, is_injected=False, handler=None, dispatch_seen=False, call_retval=None, + faults=[], signals=[], settle_ok=True, + ), + "no-progress", + ) + self.assertEqual(classify_outcome(faults=[], signals=[], settle_ok=False, **base), "settle-failed") + + +@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") +@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") +class DiscoveryIntegrationTests(unittest.TestCase): + def test_sweep_classifies_known_and_unknown(self): + with tempfile.TemporaryDirectory(prefix="discovery-sweep-") as out: + result = _run( + [ + DISCOVERY, + "--out", out, + "--max-id", "32", + "--patterns", "zero", + "--seed", "5", + "--timeout", "0.25", + "--settle", "1.0", + ] + ) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + findings = _read_findings(out) + self.assertEqual(len(findings), 32) + by_cmd: dict[int, dict] = {finding["cmd"]: finding for finding in findings} + + # Known commands 1..6 are attributed to their handlers; zero args + # make ECHO fault (NULL string) and FAULT trap (ebreak). + for cmd, name in DEMO_CASE_HANDLERS.items(): + self.assertEqual(by_cmd[cmd]["handler"], name, f"cmd {cmd} handler") + for cmd in (1, 3, 4, 5): + self.assertEqual(by_cmd[cmd]["outcome"], "known-handler") + self.assertEqual(by_cmd[2]["outcome"], "fault") + self.assertEqual(by_cmd[6]["outcome"], "trap") + + # At least one default/unused id reaches the catch-all with ~id. + self.assertEqual(by_cmd[7]["outcome"], "default-path") + self.assertEqual(by_cmd[7]["effects"]["after"]["result"], (~7) & MASK64) + self.assertEqual(by_cmd[32]["outcome"], "default-path") + + # Injected-service dispatch: 16..21 alias handler slots 0..5. + self.assertEqual(by_cmd[16]["outcome"], "injected-slot") + self.assertEqual(by_cmd[16]["handler"], "rpc_ping") + self.assertEqual(by_cmd[21]["outcome"], "trap") + + report_path = os.path.join(out, "control-discovery.md") + self.assertTrue(os.path.exists(report_path)) + with open(report_path, encoding="utf-8") as fh: + report = fh.read() + self.assertIn("TOOL IN DEVELOPMENT - NOT READY YET!", report) + self.assertIn("Unknown / undiscovered ids", report) + self.assertIn("Injected-slot dispatch ids", report) + self.assertIn("rpc_fault", report) + self.assertTrue(os.path.isdir(os.path.join(out, "sessions"))) + + def test_sweep_deterministic(self): + base = [ + DISCOVERY, + "--max-id", "8", + "--patterns", "zero", + "--seed", "11", + "--timeout", "0.25", + "--settle", "1.0", + ] + with tempfile.TemporaryDirectory(prefix="discovery-det-a-") as out_a, tempfile.TemporaryDirectory( + prefix="discovery-det-b-" + ) as out_b: + first = _run(base + ["--out", out_a]) + second = _run(base + ["--out", out_b]) + self.assertEqual(first.returncode, 0, first.stderr[-2000:]) + self.assertEqual(second.returncode, 0, second.stderr[-2000:]) + with open(os.path.join(out_a, "findings.jsonl"), encoding="utf-8") as fa, open( + os.path.join(out_b, "findings.jsonl"), encoding="utf-8" + ) as fb: + self.assertEqual(fa.read(), fb.read(), "same seed/params must replay byte-for-byte") + + def test_rm_surface_exits_cleanly(self): + with tempfile.TemporaryDirectory(prefix="discovery-rm-") as out: + result = _run([DISCOVERY, "--surface", "rm", "--out", out]) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + self.assertIn("skipped", result.stdout) + self.assertFalse(os.path.exists(os.path.join(out, "findings.jsonl"))) + + def test_reach_rpc_ping_via_call(self): + with tempfile.TemporaryDirectory(prefix="reach-ping-") as out: + result = _run([REACH, "--target", "rpc_ping", "--method", "call", "--args", "0x41", "--out", out]) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + with open(os.path.join(out, "reach-report.json"), encoding="utf-8") as fh: + report = json.load(fh) + self.assertTrue(report["reachable"]) + self.assertEqual(report["method"], "call") + self.assertEqual(report["entry"]["symbol"], "rpc_ping") + self.assertEqual(report["effects"]["retval"], 0x41 ^ 0xBEEF) + self.assertTrue(os.path.exists(os.path.join(out, "probe.json"))) + with open(os.path.join(out, "probe.json"), encoding="utf-8") as fh: + probe = json.load(fh) + self.assertEqual(probe["method"], "call") + self.assertEqual(probe["args"], [0x41]) + + def test_reach_rpc_sum_via_run(self): + with tempfile.TemporaryDirectory(prefix="reach-sum-") as out: + result = _run([REACH, "--target", "rpc_sum", "--method", "run", "--out", out]) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + with open(os.path.join(out, "reach-report.json"), encoding="utf-8") as fh: + report = json.load(fh) + self.assertTrue(report["reachable"]) + self.assertEqual(report["method"], "run") + self.assertEqual(report["entry"]["symbol"], "rpc_sum") + self.assertTrue(report["prior_events"], "the self-test hook history must be recorded") + + def test_reach_rpc_fault_via_trigger(self): + with tempfile.TemporaryDirectory(prefix="reach-fault-") as out: + result = _run([REACH, "--target", "rpc_fault", "--method", "trigger", "--out", out]) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + with open(os.path.join(out, "reach-report.json"), encoding="utf-8") as fh: + report = json.load(fh) + self.assertTrue(report["reachable"]) + self.assertEqual(report["method"], "trigger") + self.assertEqual(report["trigger"]["id"], 6) + self.assertEqual(report["trigger"]["handler"], "rpc_fault") + self.assertEqual(report["required_args"]["cmd"], 6) + self.assertTrue(report["effects"]["trap"], "rpc_fault must show the ebreak trap") + self.assertTrue(os.path.exists(os.path.join(out, "probe.json"))) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_harness.py b/debug/risc-v-gsp-rm-vm/tests/test_harness.py new file mode 100644 index 0000000..55938ae --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_harness.py @@ -0,0 +1,144 @@ +"""Tests for the green-eye automation harness (roadmap R5/R6, H4.1/H4.3). + +The integration tests drive the two harness CLIs as subprocesses against the +bundled demo firmware and skip cleanly when qemu-system-riscv64 or the demo +ELF is absent. The pure-function tests need neither. +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest + +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") +MATRIX = os.path.join(TOOL_ROOT, "harness", "portability_matrix.py") +MUTATION = os.path.join(TOOL_ROOT, "harness", "mutation_harness.py") +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.common import qemu_available # noqa: E402 +from harness.mutation_harness import Mutation, expected_for, expected_ok, generate_mutations # noqa: E402 +from harness.portability_matrix import counter_delta # noqa: E402 + + +def _run(args: list[str], timeout: float = 600.0) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable] + args, + capture_output=True, + text=True, + cwd=TOOL_ROOT, + timeout=timeout, + ) + + +class HarnessUnitTests(unittest.TestCase): + def test_generate_mutations_deterministic(self): + first = generate_mutations(9, seed=5) + second = generate_mutations(9, seed=5) + self.assertEqual([m.to_json() for m in first], [m.to_json() for m in second]) + self.assertEqual( + {m.strategy for m in first}, + {"cmd-invalid", "arg-boundary", "bit-flip", "buffer-length"}, + ) + self.assertNotEqual([m.to_json() for m in first], [m.to_json() for m in generate_mutations(9, seed=6)]) + + def test_memory_region_strategy(self): + mutations = generate_mutations(5, seed=1, region=(0x80001000, 64)) + region = [m for m in mutations if m.strategy == "mem-region"] + self.assertEqual(len(region), 1) + self.assertEqual(region[0].region["addr"], 0x80001000) + self.assertTrue(0 <= region[0].region["offset"] < 64) + + def test_demo_oracle(self): + before = {"result": 0x30, "last_rpc": 5} + ping = Mutation(0, "bit-flip", cmd=1, arg0=0x1234) + self.assertEqual(expected_for(ping, before, 0x80000800)["value"], 0x1234 ^ 0xBEEF) + fill = Mutation(1, "arg-boundary", cmd=3, arg1=1 << 40) + self.assertEqual(expected_for(fill, before, 0)["value"], 256) + invalid = Mutation(2, "cmd-invalid", cmd=13) + after = {"result": (~13) & ((1 << 64) - 1), "last_rpc": 13} + self.assertTrue(expected_ok(expected_for(invalid, before, 0), invalid, before, after)) + idle = Mutation(3, "cmd-invalid", cmd=0) + unchanged = {"result": 0x30, "last_rpc": 5} + self.assertTrue(expected_ok(expected_for(idle, before, 0), idle, before, unchanged)) + + def test_counter_delta_zero_for_identical_profiles(self): + rows = counter_delta({"rpc_ping": 2}, {"rpc_ping": 2}) + self.assertTrue(all(row["delta"] == 0 for row in rows)) + + +@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") +@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") +class HarnessIntegrationTests(unittest.TestCase): + def test_portability_matrix_self_comparison(self): + with tempfile.TemporaryDirectory(prefix="harness-matrix-") as out: + result = _run( + [ + MATRIX, + "--target", + DEMO_ELF, + "--target", + DEMO_ELF, + "--out", + out, + "--probe-instructions", + "200", + "--scenario-duration", + "0.2", + ] + ) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + with open(os.path.join(out, "portability-report.json"), encoding="utf-8") as fh: + report = json.load(fh) + self.assertEqual([t["status"] for t in report["targets"]], ["ok", "ok"]) + self.assertEqual(len(report["comparisons"]), 1) + comparison = report["comparisons"][0] + self.assertTrue(comparison["all_zero"], comparison["deltas"]) + for rows in comparison["deltas"].values(): + for row in rows: + self.assertEqual(row["delta"], 0) + self.assertTrue(os.path.exists(os.path.join(out, "portability-report.md"))) + self.assertTrue(all(t["scenario"]["event_counts"] for t in report["targets"])) + + def test_mutation_run_and_seed_determinism(self): + with tempfile.TemporaryDirectory(prefix="harness-mut-a-") as out_a, tempfile.TemporaryDirectory( + prefix="harness-mut-b-" + ) as out_b: + base = [ + MUTATION, + "--kernel", + DEMO_ELF, + "--iterations", + "12", + "--seed", + "7", + "--timeout", + "0.3", + "--settle", + "0.1", + ] + first = _run(base + ["--out", out_a]) + self.assertEqual(first.returncode, 0, first.stderr[-2000:]) + second = _run(base + ["--out", out_b]) + self.assertEqual(second.returncode, 0, second.stderr[-2000:]) + findings_path = os.path.join(out_a, "findings.jsonl") + self.assertTrue(os.path.exists(findings_path)) + with open(findings_path, encoding="utf-8") as fh: + findings = [json.loads(line) for line in fh if line.strip()] + self.assertEqual(len(findings), 12) + self.assertTrue({finding["outcome"] for finding in findings}) + self.assertTrue(all("mutation" in finding and "detail" in finding for finding in findings)) + self.assertTrue(os.path.exists(os.path.join(out_a, "summary.md"))) + with open(findings_path, encoding="utf-8") as fa, open( + os.path.join(out_b, "findings.jsonl"), encoding="utf-8" + ) as fb: + self.assertEqual(fa.read(), fb.read(), "the same seed must replay byte-for-byte") + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py new file mode 100644 index 0000000..614d397 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py @@ -0,0 +1,431 @@ +"""Tests for the green-eye MCP server plugin (roadmap R3). + +The protocol and tool tests drive an in-memory Session over a stub RSP-shaped +client, so they need no QEMU and no simulator. A standalone subprocess test +launches the native backend end-to-end and skips when that backend is absent. +""" +from __future__ import annotations + +import io +import json +import os +import subprocess +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +PLUGINS_DIR = os.path.join(TOOL_ROOT, "plugins") +MCP_SERVER = os.path.join(PLUGINS_DIR, "mcp_server.py") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +for _path in (TOOL_ROOT, PLUGINS_DIR): + if _path not in sys.path: + sys.path.insert(0, _path) + +import mcp_server as mcp # noqa: E402 +from greeneye.rsp import StopReply # noqa: E402 +from greeneye.session import Session # noqa: E402 +from greeneye.symbols import Symbol, SymbolDB # noqa: E402 + +BASE = 0x80000000 + + +class StubClient: + """RSP-shaped in-memory client: enough for Session without an emulator.""" + + def __init__(self, size: int = 0x20000): + self.base = BASE + self.mem = bytearray(size) + self.regs = {"pc": BASE, "ra": 0, "sp": BASE + size, "a0": 0, "a1": 0} + self.sw_bps: set[int] = set() + self.hw_bps: set[int] = set() + self.watches: dict[tuple[int, int, str], None] = {} + self.closed = False + self.features = "swbreak+;hwbreak+" + + def _span(self, addr: int, size: int) -> int: + start = addr - self.base + if start < 0 or start + size > len(self.mem): + raise ValueError(f"out of range: {addr:#x}+{size}") + return start + + def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes: + start = self._span(addr, size) + return bytes(self.mem[start : start + size]) + + def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None: + start = self._span(addr, len(data)) + self.mem[start : start + len(data)] = data + + def read_registers(self) -> dict[str, int]: + return dict(self.regs) + + def read_reg(self, name: str) -> int: + return int(self.regs.get(str(name), 0)) + + def write_reg(self, name: str, value: int) -> None: + self.regs[str(name)] = int(value) + + @property + def pc(self) -> int: + return self.regs["pc"] + + def set_sw_break(self, addr: int, kind: int = 4) -> None: + self.sw_bps.add(addr) + + def clear_sw_break(self, addr: int, kind: int = 4) -> None: + self.sw_bps.discard(addr) + + def set_hw_break(self, addr: int, kind: int = 4) -> None: + self.hw_bps.add(addr) + + def clear_hw_break(self, addr: int, kind: int = 4) -> None: + self.hw_bps.discard(addr) + + def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: + self.watches[(addr, size, kind)] = None + + def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: + self.watches.pop((addr, size, kind), None) + + def continue_(self, timeout: float | None = None) -> StopReply: + return StopReply(raw="T05", signal=5) + + def step(self, timeout: float | None = None) -> StopReply: + return StopReply(raw="T05", signal=5) + + def interrupt(self, timeout: float | None = None) -> StopReply: + return StopReply(raw="T02", signal=2) + + def close(self) -> None: + self.closed = True + + +class McpTestCase(unittest.TestCase): + def setUp(self): + self.client = StubClient() + symbols = SymbolDB() + symbols.add(Symbol(addr=BASE + 0x100, name="rpc_ping", size=0x20, source="test")) + symbols.add(Symbol(addr=BASE + 0x200, name="gsp_ctrl", size=0x40, source="test")) + self.session = Session(client=self.client, symbols=symbols, record=False, name="mcp-test") + self.addCleanup(self.session.close) + + def make_server(self, **kw): + self.inp = io.StringIO() + self.out = io.StringIO() + self.log = io.StringIO() + options = dict(input_stream=self.inp, output_stream=self.out, log_stream=self.log) + options.update(kw) + return mcp.GreenEyeMCPServer(self.session, **options) + + def request(self, server, method, params=None, request_id=1): + message = {"jsonrpc": "2.0", "id": request_id, "method": method} + if params is not None: + message["params"] = params + return server.handle_message(message) + + def call_tool(self, server, name, **arguments): + response = self.request(server, "tools/call", {"name": name, "arguments": arguments}, request_id=9) + self.assertIn("result", response, response) + result = response["result"] + payload = json.loads(result["content"][0]["text"]) + return payload, result.get("isError", False) + + +class ProtocolTests(McpTestCase): + def test_initialize(self): + response = self.request( + self.make_server(), + "initialize", + {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "t", "version": "0"}}, + ) + result = response["result"] + self.assertEqual(response["jsonrpc"], "2.0") + self.assertEqual(result["protocolVersion"], "2024-11-05") + self.assertEqual(result["serverInfo"]["name"], "greeneye-mcp") + self.assertIn("tools", result["capabilities"]) + + def test_initialize_negotiation_and_fallback(self): + server = self.make_server() + self.assertEqual( + self.request(server, "initialize", {"protocolVersion": "2025-03-26"})["result"]["protocolVersion"], + "2025-03-26", + ) + self.assertEqual( + self.request(server, "initialize", {"protocolVersion": "1999-01-01"})["result"]["protocolVersion"], + mcp.DEFAULT_PROTOCOL, + ) + + def test_initialized_notification_has_no_response(self): + server = self.make_server() + self.assertIsNone(server.handle_message({"jsonrpc": "2.0", "method": "notifications/initialized"})) + + def test_ping(self): + self.assertEqual(self.request(self.make_server(), "ping")["result"], {}) + + def test_tools_list_catalogue(self): + tools = self.request(self.make_server(), "tools/list")["result"]["tools"] + names = [tool["name"] for tool in tools] + self.assertEqual(len(names), 24) + self.assertEqual(set(names), mcp._TOOL_NAMES) + for tool in tools: + self.assertEqual(tool["inputSchema"]["type"], "object") + self.assertTrue(tool["description"]) + + def test_unknown_method(self): + response = self.request(self.make_server(), "no/such") + self.assertEqual(response["error"]["code"], mcp.METHOD_NOT_FOUND) + self.assertEqual(response["id"], 1) + + def test_unknown_tool(self): + response = self.request(self.make_server(), "tools/call", {"name": "nope", "arguments": {}}) + self.assertEqual(response["error"]["code"], mcp.INVALID_PARAMS) + + def test_tools_call_before_initialize_is_allowed(self): + payload, is_error = self.call_tool(self.make_server(), "session_info") + self.assertFalse(is_error) + self.assertEqual(payload["name"], "mcp-test") + + def test_parse_error_is_reported(self): + server = self.make_server() + server.process_line("{not json") + line = self.out.getvalue().strip() + self.assertEqual(json.loads(line)["error"]["code"], mcp.PARSE_ERROR) + + def test_serve_stream_end_to_end_and_no_stdout_pollution(self): + server = self.make_server() + lines = [ + json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}), + json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}), + json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}), + json.dumps( + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "session_info", "arguments": {}}} + ), + ] + server.serve_stream(io.StringIO("\n".join(lines) + "\n"), self.out) + raw = self.out.getvalue() + responses = [json.loads(line) for line in raw.splitlines()] + self.assertEqual([response["id"] for response in responses], [1, 2, 3]) + self.assertTrue(all(response["jsonrpc"] == "2.0" for response in responses)) + self.assertNotIn("[greeneye-mcp]", raw) + self.assertNotIn("[greeneye-mcp]", self.inp.getvalue()) + + +class ReadOnlyToolTests(McpTestCase): + def test_symbols_resolve(self): + payload, is_error = self.call_tool(self.make_server(), "symbols_resolve", target="rpc_ping") + self.assertFalse(is_error) + self.assertEqual(payload["address"], BASE + 0x100) + self.assertEqual(payload["symbol"], "rpc_ping") + + def test_read_mem_and_hexdump(self): + self.client.mem[0x10:0x14] = b"\xde\xad\xbe\xef" + server = self.make_server() + payload, is_error = self.call_tool(server, "read_mem", addr=BASE + 0x10, size=4) + self.assertFalse(is_error) + self.assertEqual(payload["hex"], "deadbeef") + payload, is_error = self.call_tool(server, "hexdump", addr=BASE + 0x10, size=4) + self.assertFalse(is_error) + self.assertIn("de ad be ef", payload["text"]) + + def test_bp_set_and_clear(self): + server = self.make_server() + payload, is_error = self.call_tool(server, "bp_set", target="rpc_ping") + self.assertFalse(is_error) + self.assertIn(BASE + 0x100, self.client.sw_bps) + payload, is_error = self.call_tool(server, "bp_clear", target="rpc_ping") + self.assertFalse(is_error) + self.assertEqual(payload["cleared"], 1) + self.assertNotIn(BASE + 0x100, self.client.sw_bps) + + def test_watch_set(self): + payload, is_error = self.call_tool(self.make_server(), "watch_set", target="gsp_ctrl", size=8, kind="w") + self.assertFalse(is_error) + self.assertEqual(payload["addr"], BASE + 0x200) + self.assertIn((BASE + 0x200, 8, "w"), self.client.watches) + + def test_snapshot_digest(self): + payload, is_error = self.call_tool(self.make_server(), "snapshot", addr=BASE, size=16, label="unit") + self.assertFalse(is_error) + self.assertEqual(payload["size"], 16) + self.assertEqual(len(payload["sha256"]), 64) + + def test_session_info_capabilities(self): + payload, is_error = self.call_tool( + self.make_server(allow_write=True, allow_inject=True), "session_info" + ) + self.assertFalse(is_error) + self.assertEqual(payload["name"], "mcp-test") + self.assertTrue(payload["capabilities"]["write"]) + self.assertTrue(payload["capabilities"]["inject"]) + self.assertFalse(payload["capabilities"]["trace"]) + + def test_plugins_list(self): + payload, is_error = self.call_tool(self.make_server(), "plugins_list") + self.assertFalse(is_error) + self.assertIn("mcp_server.py", payload["bundled"]) + + def test_analyze_current_reads_active_recording(self): + with tempfile.TemporaryDirectory(prefix="mcp-rec-") as directory: + session = Session(client=StubClient(), symbols=SymbolDB(), record=directory, name="active") + try: + server = mcp.GreenEyeMCPServer( + session, + input_stream=io.StringIO(), + output_stream=io.StringIO(), + log_stream=io.StringIO(), + ) + payload, is_error = self.call_tool(server, "analyze_current") + self.assertFalse(is_error) + self.assertIn("# green-eye session summary", payload["report"]) + finally: + session.close() + + +class GatingTests(McpTestCase): + def test_write_mem_refused_without_allow_write(self): + server = self.make_server() + payload, is_error = self.call_tool(server, "write_mem", addr=BASE + 0x10, hex="0102") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "write_disabled") + self.assertEqual(bytes(self.client.mem[0x10:0x12]), b"\x00\x00") + + def test_write_mem_enabled(self): + server = self.make_server(allow_write=True) + payload, is_error = self.call_tool(server, "write_mem", addr=BASE + 0x10, hex="0x01 02") + self.assertFalse(is_error) + self.assertEqual(payload["written"], 2) + self.assertEqual(bytes(self.client.mem[0x10:0x12]), b"\x01\x02") + + def test_call_function_refused_without_allow_write(self): + payload, is_error = self.call_tool(self.make_server(), "call_function", target="rpc_ping", args=[1]) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "write_disabled") + + def test_inject_refused_without_allow_inject(self): + payload, is_error = self.call_tool(self.make_server(), "inject_service", service="/nonexistent.elf") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "inject_disabled") + + def test_inject_missing_file_after_gate(self): + payload, is_error = self.call_tool( + self.make_server(allow_inject=True), "inject_service", service="/nonexistent.elf" + ) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + +class AvailabilityTests(McpTestCase): + def test_analyze_current_without_recording(self): + payload, is_error = self.call_tool(self.make_server(), "analyze_current") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "not_available") + + def test_trace_start_on_non_native_backend(self): + payload, is_error = self.call_tool(self.make_server(), "trace_start") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "not_available") + + def test_trace_stop_without_active_trace(self): + payload, is_error = self.call_tool(self.make_server(), "trace_stop") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_mutation_missing_script_is_structured(self): + server = self.make_server() + original = mcp.MUTATION_SCRIPT + mcp.MUTATION_SCRIPT = "/nonexistent/mutation_harness.py" + try: + payload, is_error = self.call_tool(server, "mutation_run") + finally: + mcp.MUTATION_SCRIPT = original + self.assertTrue(is_error) + self.assertEqual(payload["error"], "not_available") + + def test_portability_missing_script_is_structured(self): + server = self.make_server() + original = mcp.MATRIX_SCRIPT + mcp.MATRIX_SCRIPT = "/nonexistent/portability_matrix.py" + try: + payload, is_error = self.call_tool(server, "portability_matrix_run", targets=["/x.elf"]) + finally: + mcp.MATRIX_SCRIPT = original + self.assertTrue(is_error) + self.assertEqual(payload["error"], "not_available") + + def test_probe_missing_elf(self): + payload, is_error = self.call_tool(self.make_server(), "probe_firmware", elf="/nonexistent/fw.elf") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + +class CloseAndTransportTests(McpTestCase): + def test_close_stops_serving_but_leaves_plugin_session(self): + server = self.make_server(owns_session=False) + payload, is_error = self.call_tool(server, "close") + self.assertFalse(is_error) + self.assertFalse(payload["session_closed"]) + self.assertTrue(server._stop) + self.assertFalse(self.client.closed) + + def test_close_closes_owned_session(self): + server = self.make_server(owns_session=True) + payload, is_error = self.call_tool(server, "close") + self.assertFalse(is_error) + self.assertTrue(payload["session_closed"]) + self.assertTrue(self.client.closed) + + def test_parse_tcp_is_loopback_only(self): + self.assertEqual(mcp.parse_tcp("127.0.0.1:8765"), ("127.0.0.1", 8765)) + self.assertEqual(mcp.parse_tcp("localhost:0")[0], "localhost") + with self.assertRaises(Exception): + mcp.parse_tcp("0.0.0.0:8765") + + +class StandaloneEntryTests(unittest.TestCase): + @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing") + def test_help(self): + result = subprocess.run( + [sys.executable, MCP_SERVER, "--help"], capture_output=True, text=True, cwd=TOOL_ROOT + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("--stdio", result.stdout) + self.assertIn("--allow-write", result.stdout) + + @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing") + def test_stdio_native_session(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + with tempfile.TemporaryDirectory(prefix="mcp-native-") as dumps: + env = dict(os.environ, GREENEYE_DUMPS=dumps) + requests = [ + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2024-11-05"}}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "session_info", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "close", "arguments": {}}}, + ] + payload = "".join(json.dumps(request) + "\n" for request in requests) + result = subprocess.run( + [sys.executable, MCP_SERVER, "--stdio", "--backend", "native", "--record"], + input=payload, + capture_output=True, + text=True, + cwd=TOOL_ROOT, + env=env, + timeout=180, + ) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + responses = [json.loads(line) for line in result.stdout.splitlines() if line.strip()] + self.assertEqual(len(responses), 3) + by_id = {response.get("id"): response for response in responses} + info = json.loads(by_id[2]["result"]["content"][0]["text"]) + self.assertEqual(info["backend"], "native") + self.assertFalse(by_id[3]["result"].get("isError", False)) + self.assertTrue(any(name.endswith("-mcp") for name in os.listdir(dumps))) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py new file mode 100644 index 0000000..a0635ab --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py @@ -0,0 +1,244 @@ +"""Native (M1) backend tests: demo parity with QEMU and engine determinism. + +The native simulator is pure Python and always available; every test still +skips cleanly when the demo ELF cannot be loaded. The differential class +additionally requires qemu-system-riscv64 and compares the scripted +hook/watch/leave profile of both backends event for event. +""" +from __future__ import annotations + +import os +import shutil +import tempfile +import unittest + +import greeneye as ge +from greeneye.errors import RspTimeout + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.join(HERE, "..") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +RPC_HANDLERS = ["rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state"] + + +def native_available() -> bool: + try: + import greeneye.sim_backend # noqa: F401 + except Exception: + return False + return True + + +def qemu_available() -> bool: + return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64")) + + +def demo_available() -> bool: + return os.path.exists(DEMO_ELF) and native_available() + + +def launch(backend: str, record=None, name: str = "sim"): + return ge.launch(kernel=DEMO_ELF, backend=backend, record=record, name=name) + + +@unittest.skipUnless(native_available(), "native simulator unavailable") +class NativeIsaTests(unittest.TestCase): + """Exercise RV64M/A/C paths the demo firmware does not reach.""" + + # Assembled with llvm-mc -triple=riscv64 -mattr=+m,+a,+c (encodings are + # data, so this runs without a toolchain): + # li a0,6; li a1,7; mul a2,a0,a1; div a5,a1,a0; rem a6,a1,a0; + # srai a4,a1,1; subw a3,a0,a1; lui s0,0x12345; addi sp,sp,-16; + # sd a0,8(sp); ld s1,8(sp); lh s2,8(sp); lbu s3,8(sp); + # amoadd.d s4,a0,(sp); lr.d s5,(sp); sc.d s6,a0,(sp); + # li t0,1; c.addi t0,3; c.mv a2,t0; c.andi a2,15; ebreak + PROGRAM = bytes.fromhex( + "".join( + """ + 19 45 9d 45 33 06 b5 02 b3 c7 a5 02 33 e8 a5 02 13 d7 15 40 + bb 06 b5 40 37 54 34 12 41 11 2a e4 a2 64 03 19 81 00 83 49 81 00 + 2f 3a a1 00 af 3a 01 10 2f 3b a1 18 85 42 8d 02 16 86 3d 8a 02 90 + """.split() + ) + ) + BASE = 0x80000000 + STACK = 0x80004000 + + def test_m_a_c_execution(self): + from sim.machine import Machine + + machine = Machine() + machine.load_segments([(self.BASE, self.PROGRAM)], self.BASE) + machine.set_reg(2, self.STACK) # sp + reason, pc, _, _ = machine.run(1000) + self.assertEqual(reason, 3) # R_EBREAK: stops before executing it + self.assertEqual(pc, self.BASE + len(self.PROGRAM) - 2) + expected = { + 5: 4, # t0: c.addi + 8: 0x12345000, # s0: lui + 9: 6, # s1: c.sdsp/c.ldsp round-trip + 10: 6, # a0 + 11: 7, # a1 + 12: 4, # a2: mul then c.mv/c.andi + 13: 0xFFFFFFFFFFFFFFFF, # a3: subw sign extension + 14: 3, # a4: srai + 15: 1, # a5: div + 16: 1, # a6: rem + 18: 6, # s2: lh + 19: 6, # s3: lbu + 20: 0, # s4: amoadd old value + 21: 6, # s5: lr + 22: 0, # s6: sc success + 2: self.STACK - 16, # c.addi16sp + } + got = {index: machine.x[index] for index in expected} + self.assertEqual(got, expected) + self.assertEqual(machine.read_mem(self.STACK - 16, 8), (6).to_bytes(8, "little")) + + +@unittest.skipUnless(demo_available(), "demo firmware or native simulator unavailable") +class NativeSmokeTests(unittest.TestCase): + def test_start_sequence_sets_stack_and_enters_main(self): + session = launch("native", name="unittest-sim-start") + try: + session.step(4) # auipc sp / addi sp / auipc ra / jalr main + stack_top = next(s.value for s in session.image.symbols if s.name == "_stack_top") + self.assertEqual(session.pc, session.symbols.lookup("main")) + self.assertEqual(session.read_regs()["sp"], stack_top) + finally: + session.close() + + def test_boot_hook_watch_snapshot_analyze(self): + root = tempfile.mkdtemp(prefix="greeneye-sim-") + session = launch("native", record=root, name="unittest-sim-smoke") + hits = [] + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.hook("rpc_sum", on_enter=lambda hit: hits.append(hit)) + watch_hits = [] + session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit)) + session.run(duration=0.6) + self.assertTrue(session.target.wait_for_uart("GREEN-EYE UPROC DEMO", timeout=1.0)) + self.assertGreaterEqual(len(hits), 1, "rpc_sum hook never fired") + self.assertGreaterEqual(len(watch_hits), 7, "rpc_count watchpoint never fired") + self.assertEqual(watch_hits[0].addr, ctrl + 16) + self.assertEqual(watch_hits[0].watch.kind, "w") + after = session.snapshot(ctrl, 0x188, "ctrl") + self.assertEqual(int.from_bytes(after[0:4], "little"), 0x47535031) + self.assertGreaterEqual(int.from_bytes(after[16:24], "little"), 7) + finally: + directory = getattr(session.recorder, "dir", None) + session.close() + from greeneye.analyze import render_summary + + report = render_summary(directory) + self.assertIn("rpc_sum", report) + self.assertIn("watch", report) + + def test_hook_args_and_leave_value(self): + session = launch("native", name="unittest-sim-hookargs") + events = [] + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.hook( + "rpc_sum", + on_enter=lambda hit: events.append(("enter", hit.arg(0), hit.arg(1))), + on_leave=lambda hit: events.append(("leave", hit.retval)), + ) + session.run(duration=0.2) + # Self-test: rpc_dispatch(4, ctl->buf, 16, 0) after fill(0x5a, 32). + self.assertIn(("enter", ctrl + 0x48, 16), events) + self.assertIn(("leave", 16 * 0x5A), events) + finally: + session.close() + + def test_call_and_run_until(self): + session = launch("native", name="unittest-sim-call") + try: + before = session.pc + self.assertEqual(session.call("rpc_ping", (0x1111,)), 0x1111 ^ 0xBEEF) + self.assertEqual(session.pc, before) + hit = session.run_until("rpc_sum", timeout=2.0) + self.assertEqual(hit.kind, "breakpoint") + self.assertEqual(hit.addr, session.symbols.lookup("rpc_sum")) + finally: + session.close() + + def test_run_until_timeout(self): + session = launch("native", name="unittest-sim-timeout") + try: + with self.assertRaises(RspTimeout): + session.run_until("rpc_fault", timeout=0.05) # self-test never calls it + finally: + session.close() + + +@unittest.skipUnless(demo_available(), "demo firmware or native simulator unavailable") +class NativeDeterminismTests(unittest.TestCase): + def _scripted_profile(self): + """Drive fixed instruction budgets: no host clock in the result.""" + session = ge.launch(kernel=DEMO_ELF, backend="native", name="unittest-sim-determinism") + events = [] + stops = [] + try: + session.hook( + "rpc_sum", + on_enter=lambda hit: events.append(("enter", hit.arg(0), hit.arg(1))), + on_leave=lambda hit: events.append(("leave", hit.retval)), + ) + ctrl = session.symbols.lookup("gsp_ctrl") + session.watch(ctrl + 16, size=8, kind="w") + for _ in range(20): + try: + stop = session.client.continue_(timeout=0.02) # exactly 20k insns + except RspTimeout: + stops.append(("budget", session.pc)) + continue + session._dispatch_stop(stop) + stops.append(("stop", session.pc)) + ctrl_image = session.read_regs() + snapshot = session.client.read_mem(ctrl, 0x188) + finally: + session.close() + return events, stops, ctrl_image["pc"], snapshot + + def test_two_runs_identical(self): + first = self._scripted_profile() + second = self._scripted_profile() + self.assertEqual(first, second) + + +@unittest.skipUnless(demo_available() and qemu_available(), "native simulator and QEMU required") +class DifferentialTests(unittest.TestCase): + """Run one scripted scenario on both backends and compare the profiles.""" + + def _profile(self, backend: str): + session = launch(backend, name=f"unittest-diff-{backend}") + events = [] + watch_hits = [] + try: + for name in RPC_HANDLERS: + session.hook( + name, + on_enter=lambda hit, n=name: events.append(("enter", n, hit.arg(0), hit.arg(1))), + on_leave=lambda hit, n=name: events.append(("leave", n, hit.retval)), + ) + ctrl = session.symbols.lookup("gsp_ctrl") + session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit.addr)) + session.run(duration=0.3) + uart = session.target.read_uart() + bp_counts = [handle.hits for handle in session._bps.values()] + finally: + session.close() + return events, len(watch_hits), bp_counts, uart.splitlines()[0] if uart else "" + + def test_hook_watch_profile_parity(self): + qemu = self._profile("qemu") + native = self._profile("native") + self.assertEqual(qemu, native) + self.assertEqual(native[1], 8) # 7 self-test dispatches + ctrl init + self.assertEqual(native[2], [1, 1, 1, 1, 3]) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py new file mode 100644 index 0000000..96934e9 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_trace.py @@ -0,0 +1,224 @@ +"""Native R4 trace tests: collection, filters, session blob, analyzer, CLI. + +The engine tests are pure (no QEMU, no demo build needed); the session/CLI +tests skip cleanly when the demo ELF is absent. +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import tempfile +import unittest + +import greeneye as ge +from greeneye.analyze import render_summary, render_summary_from +from greeneye.errors import RspTimeout +from sim.machine import Machine +from sim.trace import Trace + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.join(HERE, "..") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") + +# lui a0,0x80000; addi a1,zero,0x2a; sw a1,0(a0); lw a2,0(a0); ebreak +PROGRAM = bytes.fromhex("37 05 00 80 93 05 a0 02 23 20 b5 00 03 26 05 00 73 00 10 00") +BASE = 0x80000000 + + +def demo_available() -> bool: + return os.path.exists(DEMO_ELF) + + +def records(payload: bytes) -> list[dict]: + return [json.loads(line) for line in payload.decode().splitlines() if line.strip()] + + +def read_blob(session_dir: str) -> bytes: + with open(os.path.join(session_dir, "blobs", "trace.jsonl"), "rb") as fh: + return fh.read() + + +class TraceEngineTests(unittest.TestCase): + def _machine(self): + machine = Machine() + machine.load_segments([(BASE, PROGRAM)], BASE) + return machine + + def test_exec_and_access_lines(self): + trace = Trace() + reason, _, _, _ = self._machine().run(100, trace=trace) + self.assertEqual(reason, 3) # R_EBREAK, stopped at the trap + self.assertEqual(trace.counts(), { + "exec": 5, "read": 1, "write": 1, "events": 7, "dropped": 0, + }) + lines = records(trace.payload()) + self.assertEqual(lines[-1]["t"], "summary") + self.assertEqual(lines[-1]["insns"], 5) + self.assertEqual(lines[-1]["mem"], 2) + self.assertEqual( + [(r["t"], r["pc"]) for r in lines[:-1] if r["t"] == "exec"], + [("exec", "0x80000000"), ("exec", "0x80000004"), ("exec", "0x80000008"), + ("exec", "0x8000000c"), ("exec", "0x80000010")], + ) + write = next(r for r in lines if r["t"] == "write") + read = next(r for r in lines if r["t"] == "read") + self.assertEqual((write["addr"], write["size"], write["pc"], write["vt"]), ("0x80000000", 4, "0x80000008", 3)) + self.assertEqual((read["addr"], read["size"], read["pc"], read["vt"]), ("0x80000000", 4, "0x8000000c", 4)) + + def test_address_filters_and_exec_off(self): + trace = Trace(exec_enabled=False, start=BASE + 8, end=BASE + 8) + self._machine().run(100, trace=trace) + lines = records(trace.payload()) + self.assertEqual([r["t"] for r in lines], ["write", "summary"]) + self.assertEqual(lines[0]["pc"], "0x80000008") # the in-range store only + self.assertFalse(lines[1]["exec"]) + self.assertEqual(lines[1]["start"], "0x80000008") + + def test_ring_budget_is_deterministic(self): + first = Trace(max_events=3) + second = Trace(max_events=3) + self._machine().run(100, trace=first) + self._machine().run(100, trace=second) + self.assertEqual(first.payload(), second.payload()) + self.assertEqual(first.counts()["events"], 7) + self.assertEqual(first.counts()["dropped"], 4) + + def test_trace_none_path_unchanged(self): + machine = self._machine() + machine.run(100) # the default fast path still works untouched + self.assertEqual(machine.x[10], 0x80000000) + self.assertEqual(machine.x[11], 0x2A) + self.assertEqual(machine.x[12], 0x2A) + + +@unittest.skipUnless(demo_available(), "demo firmware unavailable") +class SessionTraceTests(unittest.TestCase): + def _recorded_trace(self, trace) -> tuple[str, bytes]: + session_dir = tempfile.mkdtemp(prefix="greeneye-trace-") + session = ge.launch(kernel=DEMO_ELF, backend="native", record=session_dir, + name="unittest-sim-trace", trace=trace) + try: + try: + session.client.continue_(timeout=0.02) # exactly 20k instructions + except RspTimeout: + pass + finally: + session.close() + return session_dir, read_blob(session_dir) + + def test_blob_recorded_and_analyzed(self): + session_dir, blob = self._recorded_trace({"exec": True}) + meta = json.load(open(os.path.join(session_dir, "session.json"), encoding="utf-8")) + self.assertEqual(meta["trace_blob"], "blobs/trace.jsonl") + self.assertTrue(meta["trace"]["exec"]) + lines = blob.decode().splitlines() + self.assertTrue(lines[0].startswith('{"t":"exec"')) + self.assertIn('"t":"summary"', lines[-1]) + events = [json.loads(line) for line in open(os.path.join(session_dir, "events.jsonl"), encoding="utf-8")] + self.assertTrue(any(e.get("type") == "blob_saved" for e in events)) + self.assertFalse(any(e.get("type") in ("exec", "read", "write") for e in events), + "trace volume must stay out of events.jsonl") + report = render_summary(session_dir) + self.assertIn("## Execution trace", report) + self.assertIn("unique PCs", report) + self.assertIn("### Accessed pages", report) + + def test_trace_is_deterministic(self): + _, first = self._recorded_trace(True) + _, second = self._recorded_trace(True) + self.assertEqual(first, second) + + def test_exec_off_records_accesses_only(self): + session_dir, blob = self._recorded_trace({"exec": False}) + lines = blob.decode().splitlines() + self.assertFalse(any('"t":"exec"' in line for line in lines)) + self.assertTrue(any('"t":"read"' in line or '"t":"write"' in line for line in lines)) + report = render_summary(session_dir) + self.assertIn("exec=off", report) + + def test_launch_trace_implies_recording(self): + old = os.environ.get("GREENEYE_DUMPS") + os.environ["GREENEYE_DUMPS"] = tempfile.mkdtemp(prefix="greeneye-trace-dumps-") + try: + session = ge.launch(kernel=DEMO_ELF, backend="native", name="unittest-sim-trace-auto", trace=True) + try: + directory = session.recorder.dir + self.assertIsNotNone(directory) + finally: + session.close() + self.assertTrue(os.path.exists(os.path.join(directory, "blobs", "trace.jsonl"))) + finally: + if old is None: + os.environ.pop("GREENEYE_DUMPS", None) + else: + os.environ["GREENEYE_DUMPS"] = old + + def test_render_from_folded_events(self): + meta = {"name": "synthetic", "format": "greeneye-session/1", "trace": {"exec": True}} + events = [ + {"type": "exec", "pc": 0x80000010, "vt": 1}, + {"type": "exec", "pc": 0x80000010, "vt": 2}, + {"type": "read", "pc": 0x80000010, "addr": 0x80000100, "size": 4}, + {"type": "write", "pc": 0x80000014, "addr": 0x80000200, "size": 8}, + {"type": "access", "pc": 0x80000018, "addr": 0x80000200, "size": 4, "rw": "w"}, + ] + report = render_summary_from(meta, events) + self.assertIn("## Execution trace", report) + self.assertIn("unique PCs: 1", report) + self.assertIn("1 reads (1 pages)", report) + self.assertIn("2 writes (1 pages)", report) + self.assertIn("| `0x80000000` | 1 | 2 |", report) + + +@unittest.skipUnless(demo_available(), "demo firmware unavailable") +class CliTraceTests(unittest.TestCase): + def test_info_reports_native(self): + from greeneye import cli + + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = cli.main(["info"]) + self.assertEqual(code, 0) + self.assertIn("native simulator: available", out.getvalue()) + self.assertIn("native trace: available", out.getvalue()) + + def test_record_native_trace_cli(self): + from greeneye import cli + + session_dir = tempfile.mkdtemp(prefix="greeneye-trace-cli-") + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = cli.main([ + "record", "--kernel", DEMO_ELF, "--backend", "native", + "--duration", "0.05", "--trace", "--record", session_dir, + ]) + self.assertEqual(code, 0) + summary = open(os.path.join(session_dir, "summary.md"), encoding="utf-8").read() + self.assertIn("## Execution trace", summary) + lines = read_blob(session_dir).decode().splitlines() + self.assertTrue(any('"t":"exec"' in line for line in lines)) + + def test_record_trace_address_filter_cli(self): + from greeneye import cli + + session_dir = tempfile.mkdtemp(prefix="greeneye-trace-cli-filter-") + with contextlib.redirect_stdout(io.StringIO()): + code = cli.main([ + "record", "--kernel", DEMO_ELF, "--backend", "native", + "--duration", "0.05", "--trace", "--trace-exec", "off", + "--trace-start", "0x80000000", "--trace-end", "0x800000ff", + "--record", session_dir, + ]) + self.assertEqual(code, 0) + for line in read_blob(session_dir).decode().splitlines(): + record = json.loads(line) + if record["t"] in ("read", "write"): + pc = int(record["pc"], 16) + self.assertLessEqual(pc, 0x800000FF) + self.assertNotEqual(record["t"], "exec") + + +if __name__ == "__main__": + unittest.main() 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() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py b/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py new file mode 100644 index 0000000..df704df --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py @@ -0,0 +1,333 @@ +"""Tests for version_diff / ci_release (roadmap H4.1, planned docs/21). + +Pure-function tests need neither QEMU nor the demo; the self-diff integration +test boots the demo twice and skips cleanly when qemu-system-riscv64 or the +demo ELF is absent. +""" +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, "..")) +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.ci_release import main as ci_release_main # noqa: E402 +from harness.common import SymbolDB, qemu_available # noqa: E402 +from harness.version_diff import ( # noqa: E402 + LOAD_ADDR, + Rm610GateError, + VersionDiffOptions, + check_rm610_gate, + corpus_diff, + corpus_section, + image_diff, + inventory_diff, + inventory_records, + is_rm610_label, + load_any_target, + load_symbol_sources, + load_three_column_export, + looks_like_three_column_export, + main as version_diff_main, + rm610_approved, + run_version_diff, + shape_payload, +) + +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 _write(path: str, text: str) -> str: + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +class Rm610GateTests(unittest.TestCase): + def test_label_detection(self): + for label in ("rm610", "RM610", "rm-610-fw.bin", "gsp_rm_610.elf", "release-RM610"): + self.assertTrue(is_rm610_label(label), label) + for label in ("570.144", "uproc_demo", "gsp-570.211.fwimage.bin", ""): + self.assertFalse(is_rm610_label(label), label) + + def test_refused_by_default(self): + with self.assertRaises(Rm610GateError) as ctx: + check_rm610_gate(["rm610"], environ={}) + message = str(ctx.exception) + self.assertIn("docs/23", message) + self.assertIn("RM610-APPROVED", message) + self.assertIn("no target file was opened", message) + + def test_token_and_env_approve(self): + check_rm610_gate(["rm610"], approval_token="RM610-APPROVED", environ={}) + check_rm610_gate(["rm_610"], environ={"GREENEYE_RM610_APPROVED": "1"}) + self.assertTrue(rm610_approved("RM610-APPROVED", {})) + self.assertTrue(rm610_approved(None, {"GREENEYE_RM610_APPROVED": "1"})) + with self.assertRaises(Rm610GateError): + check_rm610_gate(["rm610"], approval_token="rm610-approved", environ={}) + with self.assertRaises(Rm610GateError): + check_rm610_gate(["rm610"], environ={"GREENEYE_RM610_APPROVED": "0"}) + + def test_other_labels_are_not_gated(self): + check_rm610_gate(["570.144", "uproc_demo"], environ={}) + + +class InventoryDiffTests(unittest.TestCase): + def test_matching_discipline(self): + records_a = [ + {"name": "keep", "addr": 0x100, "size": 16, "source": "elf"}, + {"name": "gone", "addr": 0x180, "size": 8, "source": "elf"}, + {"name": "FUN_00000200", "addr": 0x200, "size": 12, "source": "elf"}, + ] + records_b = [ + {"name": "keep", "addr": 0x110, "size": 16, "source": "elf"}, + {"name": "newfn", "addr": 0x190, "size": 8, "source": "elf"}, + {"name": "FUN_00000210", "addr": 0x210, "size": 12, "source": "elf"}, + ] + diff = inventory_diff(records_a, records_b) + self.assertEqual(diff["shift"], 0x10) + self.assertEqual([row["name"] for row in diff["moved"]], ["keep"]) + self.assertEqual(diff["matched_shifted"], 1) + self.assertEqual(diff["totals"]["added"], 1) + self.assertEqual(diff["totals"]["removed"], 1) + self.assertEqual(diff["added"][0]["name"], "newfn") + self.assertEqual(diff["removed"][0]["name"], "gone") + + def test_resize_and_self_diff(self): + records = [ + {"name": "a", "addr": 0x10, "size": 4, "source": "elf"}, + {"name": "FUN_00000020", "addr": 0x20, "size": 4, "source": "elf"}, + ] + resized = [dict(row) for row in records] + resized[1]["size"] = 8 + diff = inventory_diff(records, resized) + self.assertEqual(diff["resized_anonymous"][0]["new_size"], 8) + zero = inventory_diff(records, records) + self.assertEqual(zero["totals"]["added"], 0) + self.assertEqual(zero["totals"]["removed"], 0) + self.assertEqual(zero["totals"]["moved"], 0) + self.assertEqual(zero["totals"]["resized_named"], 0) + + +class SymbolSourceTests(unittest.TestCase): + def test_three_column_export(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write( + os.path.join(tmp, "symbols.tsv"), + "addr\tname\tsize\n0x80000010\trpc_ping\t16\n0x80000020\trpc_sum\t32\n", + ) + self.assertTrue(looks_like_three_column_export(path)) + db = SymbolDB() + self.assertEqual(load_three_column_export(db, path, bias=0), 2) + self.assertEqual(db.get("rpc_ping").size, 16) + self.assertEqual(db.get("rpc_sum").addr, 0x80000020) + + def test_names_tsv_is_not_mistaken_for_export(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "names.tsv"), "0x80000010\trpc_ping\tghidra\n") + self.assertFalse(looks_like_three_column_export(path)) + + @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") + def test_sources_merge_on_demo(self): + loaded = load_any_target(DEMO_ELF) + self.assertTrue(loaded.ok) + db, sources, _ = load_symbol_sources(loaded) + self.assertTrue(any(source.startswith("elf:") for source in sources)) + self.assertTrue(len(inventory_records(db)) > 0) + + +class TargetShapeTests(unittest.TestCase): + def test_missing_and_incompatible(self): + missing = load_any_target("/nonexistent/nope.elf") + self.assertEqual(missing.kind, "missing") + self.assertFalse(missing.ok) + with tempfile.TemporaryDirectory() as tmp: + bad = os.path.join(tmp, "bad.elf") + with open(bad, "wb") as fh: + fh.write(b"\x7fELF" + b"\x00" * 48) + loaded = load_any_target(bad) + self.assertEqual(loaded.kind, "unsupported") + self.assertIn("EI_CLASS", loaded.reason or "") + + def test_raw_target_shape(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "raw.bin") + with open(path, "wb") as fh: + fh.write(NOP * 16) + loaded = load_any_target(path) + self.assertEqual(loaded.kind, "raw") + self.assertEqual(loaded.bias, LOAD_ADDR) + payload = shape_payload(loaded) + self.assertEqual(payload["span"], 64) + self.assertEqual(payload["sections"], []) + self.assertEqual(payload["loads"][0]["vaddr"], 0) + + @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") + def test_self_shape_and_inventory_zero(self): + old, new = load_any_target(DEMO_ELF), load_any_target(DEMO_ELF) + image = image_diff(old, new, max_rows=40) + self.assertEqual(image["status"], "ok") + self.assertTrue(image["same"]) + old_db, _, _ = load_symbol_sources(old) + new_db, _, _ = load_symbol_sources(new) + diff = inventory_diff(inventory_records(old_db), inventory_records(new_db)) + self.assertEqual(diff["shift"], 0) + self.assertEqual(diff["totals"]["added"], 0) + self.assertEqual(diff["totals"]["removed"], 0) + self.assertEqual(diff["totals"]["moved"], 0) + self.assertEqual(diff["totals"]["resized_named"], 0) + + +class CorpusTests(unittest.TestCase): + def test_corpus_diff(self): + with tempfile.TemporaryDirectory() as tmp: + a = _write( + os.path.join(tmp, "a.tsv"), + "addr\tfile_addr\tsize\tname\tbucket\tline\tstatus\n" + "0x1000\t0x80000000\t16\trpc_ping\tb\t0\tok\n", + ) + b = _write( + os.path.join(tmp, "b.tsv"), + "addr\tfile_addr\tsize\tname\tbucket\tline\tstatus\n" + "0x1000\t0x80000010\t16\trpc_ping\tb\t0\tok\n", + ) + diff = corpus_diff(a, b) + self.assertEqual(diff["shift"], 0x10) + self.assertEqual(diff["matched_named"], 1) + + def test_not_supplied(self): + options = VersionDiffOptions(old="old.elf", new="new.elf", out="/tmp/unused") + section = corpus_section(options, options.out) + self.assertEqual(section["status"], "skipped") + self.assertIn("not supplied", section["reason"]) + + +class RunVersionDiffTests(unittest.TestCase): + def test_missing_target_skips_cleanly(self): + with tempfile.TemporaryDirectory() as tmp: + new = os.path.join(tmp, "new.bin") + with open(new, "wb") as fh: + fh.write(NOP * 16) + out = os.path.join(tmp, "out") + options = VersionDiffOptions(old=os.path.join(tmp, "missing.elf"), new=new, out=out, + probe_instructions=8) + report = run_version_diff(options) + for name in ("image", "symbols", "probe", "behavior"): + self.assertEqual(report["sections"][name]["status"], "skipped", name) + self.assertTrue(report["sections"][name]["reason"], name) + self.assertIn("not supplied", report["sections"]["corpus"]["reason"]) + self.assertTrue(os.path.exists(os.path.join(out, "version-diff.json"))) + self.assertTrue(os.path.exists(os.path.join(out, "version-diff.md"))) + + def test_gate_refused_before_any_file_open(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, _, err = _run_main( + version_diff_main, + ["--old", os.path.join(tmp, "rm610.elf"), "--new", os.path.join(tmp, "new.elf"), + "--out", out], + ) + self.assertEqual(rc, 3) + self.assertIn("docs/23", err) + self.assertFalse(os.path.exists(out), "the gate must fire before --out is created") + + def test_gate_passes_with_token(self): + with tempfile.TemporaryDirectory() as tmp: + new = os.path.join(tmp, "new.bin") + with open(new, "wb") as fh: + fh.write(NOP * 16) + out = os.path.join(tmp, "out") + rc, _, _ = _run_main( + version_diff_main, + ["--old", os.path.join(tmp, "rm610.bin"), "--new", new, "--release", "rm610", + "--out", out, "--approval-token", "RM610-APPROVED", "--probe-instructions", "4"], + ) + self.assertEqual(rc, 0) + self.assertTrue(os.path.exists(os.path.join(out, "version-diff.json"))) + + +class CiReleaseTests(unittest.TestCase): + def test_dry_run_exits_zero_and_touches_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "bundle") + rc, stdout, _ = _run_main( + ci_release_main, + ["--old", os.path.join(tmp, "old.elf"), "--new", os.path.join(tmp, "new.elf"), + "--out", out, "--dry-run"], + ) + self.assertEqual(rc, 0) + self.assertIn("version_diff", stdout) + self.assertIn("portability_matrix", stdout) + self.assertIn("mutation_smoke", stdout) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_refused(self): + rc, _, err = _run_main( + ci_release_main, + ["--old", "rm610.elf", "--new", "new.elf", "--dry-run"], + ) + self.assertEqual(rc, 3) + self.assertIn("docs/23", err) + + def test_help_is_clean(self): + for main_fn in (ci_release_main, version_diff_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()) + + +@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") +@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") +class VersionDiffQemuTests(unittest.TestCase): + def test_self_diff_zero_inventory_and_behavior(self): + with tempfile.TemporaryDirectory(prefix="vdiff-self-") as tmp: + options = VersionDiffOptions( + old=DEMO_ELF, + new=DEMO_ELF, + out=tmp, + probe_instructions=200, + behavior_duration=0.2, + max_rows=20, + ) + report = run_version_diff(options) + self.assertTrue(report["sections"]["image"]["same"]) + symbols = report["sections"]["symbols"] + self.assertEqual(symbols["status"], "ok") + self.assertEqual(symbols["shift"], 0) + self.assertEqual(symbols["totals"]["added"], 0) + self.assertEqual(symbols["totals"]["removed"], 0) + self.assertEqual(symbols["totals"]["moved"], 0) + self.assertEqual(symbols["totals"]["resized_named"], 0) + probe = report["sections"]["probe"] + self.assertEqual(probe["status"], "ok") + self.assertFalse(probe["changed"]) + behavior = report["sections"]["behavior"] + self.assertEqual(behavior["status"], "ok") + self.assertTrue(behavior["all_zero"], behavior["deltas"]) + for rows in behavior["deltas"].values(): + for row in rows: + self.assertEqual(row["delta"], 0) + self.assertTrue(os.path.exists(os.path.join(tmp, "version-diff.md"))) + with open(os.path.join(tmp, "version-diff.json"), encoding="utf-8") as fh: + self.assertEqual(json.load(fh)["format"], "greeneye-version-diff/1") + + +if __name__ == "__main__": + unittest.main() |