diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_discovery.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_discovery.py | 214 |
1 files changed, 214 insertions, 0 deletions
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() |