"""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("Status: implemented, documented and tested", 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()