diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py | 43 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_injection.py | 61 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_mcp.py | 422 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py | 299 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py | 458 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py | 323 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_triage.py | 264 |
7 files changed, 1869 insertions, 1 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py b/debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py new file mode 100644 index 0000000..c52f75f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_gpgpu_sim_setup.py @@ -0,0 +1,43 @@ +"""Regression guard for ``gpgpu-sim/setup.sh`` clone-source policy. + +The script must default to the upstream URL and use a local mirror only via +the explicit ``GPGPUSIM_CLONE_SOURCE`` environment variable; no personal host +path may be hardcoded. These checks are static (plus ``bash -n``/``--help``) +so they need neither a network nor a GPGPU-Sim checkout. +""" +import os +import subprocess +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SETUP = os.path.join(TOOL_ROOT, "gpgpu-sim", "setup.sh") +PIN = "90ec3399763d7c8512cfe7dc193473086c38ca38" +UPSTREAM_URL = "https://github.com/gpgpu-sim/gpgpu-sim_distribution.git" + + +class SetupShTests(unittest.TestCase): + def test_clone_source_policy(self): + with open(SETUP, "r", encoding="utf-8") as fh: + text = fh.read() + self.assertNotIn("LOCAL_SRC=", text) + self.assertNotIn("/home/rmcore", text) + self.assertIn("GPGPUSIM_CLONE_SOURCE", text) + self.assertIn(UPSTREAM_URL, text) + self.assertIn(PIN, text) # pin enforcement stays in the script + + def test_script_syntax_and_help(self): + result = subprocess.run(["bash", "-n", SETUP], + capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + result = subprocess.run(["bash", SETUP, "--help"], + capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("GPGPUSIM_CLONE_SOURCE", result.stdout) + self.assertIn("upstream URL", result.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_injection.py b/debug/risc-v-gsp-rm-vm/tests/test_injection.py index 5dca9cb..33cc092 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_injection.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_injection.py @@ -1,6 +1,9 @@ """Integration tests for Frida-like remote calls, run-until and service injection. Skipped cleanly when QEMU, the demo firmware, clang/lld or the SDK are absent. +The native simulator is a first-class injection target too: ``NativeInjectionTests`` +proves the demo service blob loads and that direct and firmware-mailbox +invocations agree without QEMU. """ from __future__ import annotations @@ -24,12 +27,28 @@ def qemu_available() -> bool: return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64")) +def native_available() -> bool: + try: + import greeneye.sim_backend # noqa: F401 + + return True + except Exception: + return False + + def toolchain_available() -> bool: return bool(shutil.which("clang")) and ( shutil.which("ld.lld") or os.path.exists("/usr/lib/llvm-18/bin/ld.lld") ) +def build_sdk() -> bool: + if os.path.exists(SERVICE_ELF): + return True + result = subprocess.run(["make", "-C", os.path.join(TOOL_ROOT, "sdk")], capture_output=True, text=True) + return result.returncode == 0 and os.path.exists(SERVICE_ELF) + + @unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") class CallTests(unittest.TestCase): @@ -117,5 +136,47 @@ class InjectionTests(unittest.TestCase): session.close() +@unittest.skipUnless(native_available(), "native backend unavailable") +@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") +@unittest.skipUnless(toolchain_available(), "clang/lld not available") +class NativeInjectionTests(unittest.TestCase): + """Injection SDK against the native simulator (no QEMU required).""" + + @classmethod + def setUpClass(cls): + if not build_sdk(): + raise unittest.SkipTest("SDK build failed") + + def test_native_direct_and_firmware_invocation(self): + from greeneye.inject import invoke_remote, invoke_via_firmware, load_service, services + + session = ge.launch(kernel=DEMO_ELF, backend="native", record=True, name="unittest-native-inject") + try: + service = load_service(session, SERVICE_ELF, slot=6) + self.assertEqual(services(session), [service]) + self.assertEqual(invoke_remote(session, service, 1, 0x41), 0x41 ^ 0xBEEF) + self.assertEqual(invoke_remote(session, service, 2, 0x10, 0x20), 0x30) + session.run(duration=0.3) + self.assertEqual(invoke_via_firmware(session, service, 1, 0x41), 0x41 ^ 0xBEEF) + self.assertEqual(invoke_via_firmware(session, service, 2, 0x10, 0x20), 0x30) + finally: + session.close() + + def test_native_injection_replay_is_deterministic(self): + from greeneye.inject import invoke_remote, load_service + + results = [] + for _ in range(2): + session = ge.launch(kernel=DEMO_ELF, backend="native", record=False, name="unittest-native-inject-replay") + try: + service = load_service(session, SERVICE_ELF, slot=6) + results.append(invoke_remote(session, service, 2, 0x10, 0x20)) + results.append(session.pc) + finally: + session.close() + self.assertEqual(results[0], 0x30) + self.assertEqual(results, results[:2] + results[:2]) + + 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 index 2df8a4a..87a85a8 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py @@ -19,11 +19,14 @@ 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") +SERVICE_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_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 +import greeneye as ge # noqa: E402 +from greeneye.inject import SERVICE_REGISTRY_ATTR, InjectedService # noqa: E402 from greeneye.rsp import StopReply # noqa: E402 from greeneye.session import Session # noqa: E402 from greeneye.symbols import Symbol, SymbolDB # noqa: E402 @@ -167,8 +170,19 @@ class ProtocolTests(McpTestCase): 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.assertGreaterEqual(len(names), 24) + self.assertEqual(len(names), len(mcp.TOOLS)) self.assertEqual(set(names), mcp._TOOL_NAMES) + for expected in ( + "backends_list", + "session_launch", + "service_list", + "sim_cycles", + "sim_device_state", + "sim_profile", + "sim_regions", + ): + self.assertIn(expected, names) for tool in tools: self.assertEqual(tool["inputSchema"]["type"], "object") self.assertTrue(tool["description"]) @@ -283,6 +297,328 @@ class ReadOnlyToolTests(McpTestCase): session.close() +class StubCycleModel: + """Minimal CycleModel shape for MCP sim-tool tests (no sim/ import needed).""" + + name = "rv64imac-modeled/1" + + def describe(self) -> dict: + return {"name": self.name, "alu": 1, "memory_latency": 3, "fidelity": "modeled"} + + +class StubMachine: + """Machine-shaped object exposing the public surface the sim tools read.""" + + def __init__(self): + self.cycles = 120 + self.insns = 100 + self.cycle_model = StubCycleModel() + self.unknown_mmio = "lenient" + self.bias = 0x80000000 + self.ram_base = BASE + self.ram_size = 0x20000 + self.mmio_log = [{"pc": BASE + 4, "addr": 0x2000000000000A00, "size": 8, "rw": "r"}] + self.mmio_unknown = 1 + self.device_map = None + + +class StubTarget: + """SimTarget-shaped handle so the server classifies the session as native.""" + + keep_alive = True + + def __init__(self, machine): + self.machine = machine + self.mmio_log = machine.mmio_log + self.mmio_unknown = machine.mmio_unknown + self.ram_base = BASE + self.memory_mb = 256 + + def kill(self): + pass + + +class NativeSimToolTests(McpTestCase): + def setUp(self): + super().setUp() + self.client.trace = None # backend detection: native + self.machine = StubMachine() + self.session.target = StubTarget(self.machine) + + def test_backends_list(self): + payload, is_error = self.call_tool(self.make_server(owns_session=True), "backends_list") + self.assertFalse(is_error) + self.assertEqual(payload["active"], "native") + self.assertTrue(payload["launchable"]) + names = {entry["name"] for entry in payload["backends"]} + self.assertEqual(names, {"native", "qemu"}) + + def test_sim_cycles(self): + payload, is_error = self.call_tool(self.make_server(), "sim_cycles") + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["insns"], 100) + self.assertEqual(payload["cycles"], 120) + self.assertEqual(payload["cycles_per_instruction"], 1.2) + self.assertEqual(payload["cycle_model"]["name"], "rv64imac-modeled/1") + self.assertIn("modeled", payload["fidelity"]) + + def test_sim_device_state_reports_absent_device_model(self): + payload, is_error = self.call_tool(self.make_server(), "sim_device_state", limit=4) + self.assertFalse(is_error) + self.assertEqual(payload["mmio_unknown"], 1) + self.assertEqual(len(payload["mmio_log"]), 1) + self.assertEqual(payload["unknown_mmio"], "lenient") + self.assertIn("device_model", payload) + if "error" in payload["device_model"]: + self.assertEqual(payload["device_model"]["error"], "not_available") + + def test_sim_device_state_serialises_attached_registry(self): + class Registry: + def state(self): + return {"schema": "greeneye-sim-devices/1", "devices": []} + + self.machine.devices = Registry() + payload, is_error = self.call_tool(self.make_server(), "sim_device_state") + self.assertFalse(is_error) + self.assertEqual(payload["device_model"]["source"], "machine.devices.state") + self.assertEqual(payload["device_model"]["state"]["schema"], "greeneye-sim-devices/1") + json.dumps(payload) # the tool result must stay JSON-serialisable + + def test_sim_profile(self): + payload, is_error = self.call_tool(self.make_server(), "sim_profile") + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["bias"], 0x80000000) + self.assertEqual(payload["cycles"], 120) + self.assertIn("session_meta", payload) + + def test_sim_regions(self): + payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=4) + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["bias"], 0x80000000) + self.assertEqual(payload["unknown_mmio"], "lenient") + regions = {region["name"]: region for region in payload["regions"]} + self.assertIn("ram", regions) + self.assertIn("uart", regions) + self.assertEqual(regions["ram"]["base"], BASE) + self.assertEqual(regions["ram"]["size"], 0x20000) + self.assertEqual(regions["ram"]["end"], BASE + 0x20000) + self.assertEqual(regions["uart"]["base"], 0x10000000) + self.assertEqual(regions["uart"]["size"], 0x100) + self.assertIsNone(payload["device_map_entries"]) + self.assertEqual(payload["device_map"], []) + self.assertEqual(len(payload["observed_mmio_pages"]), 1) + page = payload["observed_mmio_pages"][0] + self.assertEqual(page["page"], 0x2000000000000A00 & ~0xFFF) + self.assertEqual(page["reads"], 1) + self.assertEqual(page["writes"], 0) + self.assertEqual(page["addresses"], [0x2000000000000A00]) + self.assertEqual(page["pcs"], [BASE + 4]) + + def test_sim_regions_device_map_entries(self): + class Entry: + address = 0x2000000000000A00 + width = 8 + kind = "r" + hits = 3 + first_epc_fw = 0x4009014 + source = "probe" + + class Map: + def entries(self): + return [Entry()] + + self.machine.device_map = Map() + payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=8) + self.assertFalse(is_error) + self.assertEqual(payload["device_map_entries"], 1) + self.assertEqual(payload["device_map"][0]["address"], 0x2000000000000A00) + self.assertEqual(payload["device_map"][0]["first_epc_fw"], 0x4009014) + self.assertEqual(payload["device_map"][0]["source"], "probe") + + def test_sim_tools_refuse_non_native(self): + # a fresh stub client without .trace is not a native session + client = StubClient() + session = Session(client=client, symbols=self.session.symbols, record=False, name="stub-qemu") + server = mcp.GreenEyeMCPServer( + session, input_stream=io.StringIO(), output_stream=io.StringIO(), log_stream=io.StringIO() + ) + try: + for name in ("sim_cycles", "sim_device_state", "sim_profile", "sim_regions"): + payload, is_error = self.call_tool(server, name) + self.assertTrue(is_error, name) + self.assertEqual(payload["error"], "not_available", name) + finally: + session.close() + + +class ServiceListTests(McpTestCase): + def test_service_list_empty_with_slots(self): + self.session.memory.write_u64(BASE + 0x200 + 0x150 + 6 * 8, 0x81000000) + payload, is_error = self.call_tool(self.make_server(), "service_list") + self.assertFalse(is_error) + self.assertEqual(payload["count"], 0) + self.assertEqual(payload["mailbox_cmd_base"], 16) + self.assertEqual(payload["slots"][6]["entry"], 0x81000000) + self.assertEqual(payload["slots"][6]["cmd"], 22) + + def test_service_list_reports_registry(self): + service = InjectedService( + path="/tmp/svc.elf", + name="svc.elf", + load_addr=0x81000000, + entry_addr=0x81000100, + dispatch_offset=0x100, + slot=6, + size=64, + ) + setattr(self.session, SERVICE_REGISTRY_ATTR, [service]) + payload, is_error = self.call_tool(self.make_server(), "service_list") + self.assertFalse(is_error) + self.assertEqual(payload["count"], 1) + self.assertEqual(payload["services"][0]["entry_addr"], 0x81000100) + self.assertEqual(payload["services"][0]["slot"], 6) + + +class SessionLaunchTests(McpTestCase): + def test_refused_in_plugin_mode(self): + payload, is_error = self.call_tool(self.make_server(owns_session=False), "session_launch", backend="native") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "not_available") + + def test_invalid_backend(self): + payload, is_error = self.call_tool(self.make_server(owns_session=True), "session_launch", backend="mips") + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_missing_kernel(self): + payload, is_error = self.call_tool( + self.make_server(owns_session=True), "session_launch", backend="native", kernel="/nonexistent.elf" + ) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_qemu_trace_argument_refused(self): + payload, is_error = self.call_tool( + self.make_server(owns_session=True), "session_launch", backend="qemu", trace=True + ) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)") + def test_launch_native_replaces_owned_session(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + server = self.make_server(owns_session=True) + # Contract (docs/20 §5/§7): session_launch is session-class and allowed + # without --allow-write; it only replaces the local served session. + self.assertFalse(server.allow_write) + with tempfile.TemporaryDirectory(prefix="mcp-launch-") as directory: + payload, is_error = self.call_tool( + server, "session_launch", backend="native", kernel=DEMO_ELF, record=directory + ) + self.assertFalse(is_error, payload) + self.assertEqual(payload["backend"], "native") + self.assertEqual(payload["session"]["kernel"], DEMO_ELF) + self.assertTrue(self.client.closed) + cycles, is_error = self.call_tool(server, "sim_cycles") + self.assertFalse(is_error, cycles) + self.assertEqual(cycles["backend"], "native") + self.assertIsNotNone(cycles["cycle_model"]) + regions, is_error = self.call_tool(server, "sim_regions") + self.assertFalse(is_error, regions) + self.assertIn("ram", {region["name"] for region in regions["regions"]}) + server.session.close() + + +@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)") +class NativeSessionToolTests(unittest.TestCase): + """End-to-end tool checks against a real native session (demo firmware).""" + + def setUp(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + self.session = ge.launch(kernel=DEMO_ELF, backend="native", record=None, name="native-mcp") + self.machine = self.session.target.machine + self.entry = self.machine.pc + self.addCleanup(self.session.close) + self.inp = io.StringIO() + self.out = io.StringIO() + self.log = io.StringIO() + + def make_server(self, **kw): + return mcp.GreenEyeMCPServer( + self.session, input_stream=self.inp, output_stream=self.out, log_stream=self.log, **kw + ) + + def call_tool(self, server, name, **arguments): + response = server.handle_message( + {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": name, "arguments": arguments}} + ) + self.assertIn("result", response, response) + result = response["result"] + return json.loads(result["content"][0]["text"]), result.get("isError", False) + + def test_native_step_budget(self): + server = self.make_server() + self.assertEqual(self.machine.insns, 0) + payload, is_error = self.call_tool(server, "step", count=3) + self.assertFalse(is_error, payload) + self.assertEqual(payload["steps"], 3) + self.assertEqual(self.machine.insns, 3) + self.assertNotEqual(self.machine.pc, self.entry) + self.assertEqual(payload["pc"], self.machine.pc) + + def test_native_step_budget_cap(self): + payload, is_error = self.call_tool(self.make_server(), "step", count=10001) + self.assertTrue(is_error) + self.assertEqual(payload["error"], "invalid_arguments") + + def test_native_run_budget_and_cycles(self): + server = self.make_server() + budget = 0.02 + before = self.machine.insns + payload, is_error = self.call_tool(server, "run", duration=budget) + self.assertFalse(is_error, payload) + self.assertGreaterEqual(payload["stops"], 1) + insns = self.machine.insns + self.assertGreater(insns, before) + self.assertLessEqual(insns, int(budget * 1_000_000) + 2000) + cycles, is_error = self.call_tool(server, "sim_cycles") + self.assertFalse(is_error, cycles) + self.assertEqual(cycles["backend"], "native") + self.assertEqual(cycles["insns"], insns) + self.assertGreater(cycles["cycles"], 0) + self.assertEqual(cycles["accounting_enabled"], True) + + def test_native_sim_regions(self): + payload, is_error = self.call_tool(self.make_server(), "sim_regions", limit=8) + self.assertFalse(is_error, payload) + self.assertEqual(payload["backend"], "native") + regions = {region["name"]: region for region in payload["regions"]} + self.assertEqual(regions["ram"]["base"], 0x80000000) + self.assertEqual(regions["ram"]["size"], 256 * 1024 * 1024) + self.assertEqual(regions["uart"]["base"], 0x10000000) + image_regions = [region for region in payload["regions"] if region["kind"] == "firmware"] + self.assertTrue(image_regions) + self.assertGreaterEqual(image_regions[0]["base"], 0x80000000) + self.assertTrue(image_regions[0]["flags"]["exec"]) + self.assertIsNone(payload["device_map_entries"]) + self.assertEqual(payload["observed_mmio_pages"], []) + + def test_native_session_info_reports_native(self): + payload, is_error = self.call_tool(self.make_server(), "session_info") + self.assertFalse(is_error) + self.assertEqual(payload["backend"], "native") + self.assertTrue(payload["capabilities"]["trace"]) + + class GatingTests(McpTestCase): def test_write_mem_refused_without_allow_write(self): server = self.make_server() @@ -426,6 +762,90 @@ class StandaloneEntryTests(unittest.TestCase): self.assertFalse(by_id[3]["result"].get("isError", False)) self.assertTrue(any(name.endswith("-mcp") for name in os.listdir(dumps))) + @unittest.skipUnless(os.path.isfile(MCP_SERVER), "mcp_server.py missing") + @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built (make demo)") + def test_stdio_native_sim_tools_injection_and_trace(self): + try: + import greeneye.sim_backend # noqa: F401 + except Exception as exc: + self.skipTest(f"native backend unavailable: {exc}") + if not os.path.isfile(SERVICE_ELF): + build = subprocess.run( + ["make", "-C", os.path.join(TOOL_ROOT, "sdk")], capture_output=True, text=True + ) + if build.returncode != 0 or not os.path.isfile(SERVICE_ELF): + self.skipTest(f"SDK build failed: {build.stderr[-200:]}") + with tempfile.TemporaryDirectory(prefix="mcp-native-sim-") 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": "backends_list", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "sim_cycles", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "sim_profile", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "sim_device_state", "arguments": {"limit": 2}}}, + {"jsonrpc": "2.0", "id": 6, "method": "tools/call", "params": {"name": "trace_start", "arguments": {"exec": False, "max_events": 10000}}}, + {"jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "trace_stop", "arguments": {}}}, + { + "jsonrpc": "2.0", + "id": 8, + "method": "tools/call", + "params": { + "name": "inject_service", + "arguments": {"service": SERVICE_ELF, "slot": 6, "cmd": 1, "arg0": 65, "path": "both"}, + }, + }, + {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": "service_list", "arguments": {}}}, + {"jsonrpc": "2.0", "id": 10, "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", "--allow-inject"], + input=payload, + capture_output=True, + text=True, + cwd=TOOL_ROOT, + env=env, + timeout=300, + ) + self.assertEqual(result.returncode, 0, result.stderr[-2000:]) + responses = [json.loads(line) for line in result.stdout.splitlines() if line.strip()] + by_id = {response.get("id"): response for response in responses} + self.assertEqual(sorted(by_id), [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + + def call_result(request_id): + response = by_id[request_id] + self.assertIn("result", response, response) + body = response["result"] + self.assertFalse(body.get("isError", False), body) + return json.loads(body["content"][0]["text"]) + + backends = call_result(2) + self.assertEqual(backends["active"], "native") + cycles = call_result(3) + self.assertEqual(cycles["backend"], "native") + self.assertIsNotNone(cycles["cycle_model"]) + profile = call_result(4) + self.assertEqual(profile["backend"], "native") + self.assertEqual(profile["kernel"], DEMO_ELF) + device_state = call_result(5) + self.assertIn("device_model", device_state) + trace_start = call_result(6) + self.assertTrue(trace_start["tracing"]) + trace_stop = call_result(7) + self.assertFalse(trace_stop["tracing"]) + self.assertTrue(str(trace_stop["blob"]).endswith("trace.jsonl")) + injected = call_result(8) + self.assertEqual(injected["remote"], 0x41 ^ 0xBEEF) + self.assertEqual(injected["firmware"], 0x41 ^ 0xBEEF) + services = call_result(9) + self.assertEqual(services["count"], 1) + self.assertEqual(services["services"][0]["entry_addr"], injected["service"]["entry_addr"]) + slots = services["slots"] + if slots: + self.assertEqual(slots[6]["entry"], injected["service"]["entry_addr"]) + self.assertFalse(by_id[10]["result"].get("isError", False)) + if __name__ == "__main__": unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py new file mode 100644 index 0000000..9bb69ad --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_cycles.py @@ -0,0 +1,299 @@ +"""Cycle model (sim/cycles.py) and GSP cycle profile (sim/gsp_profile.py) tests. + +All assertions here are about a *modeled* cycle count and its determinism, not +about silicon timings. Everything is stdlib-only and offline: no hardware, +no GPU, no QEMU, no network. The real-image tests are bounded (200k +instructions), read-only, and skip cleanly when the 570.144 fwimage is absent. +""" +from __future__ import annotations + +import json +import os +import tempfile +import unittest + +from greeneye.device_map import QEMU_REFERENCE +from sim.cycles import ( + CLASS_ALU, + CLASS_DIV, + CLASS_LOAD, + CLASS_MUL, + CYCLE_FIDELITY, + DEFAULT_MODEL, + NOTE, + CycleModel, +) +from sim.decode import ( + A_ADD, + A_DIV, + A_MUL, + A_SUB, + K_AMO, + K_BRANCH, + K_CSR, + K_EBREAK, + K_ECALL, + K_ILLEGAL, + K_JAL, + K_JALR, + K_LOAD, + K_OP, + K_OP32, + decode32, +) +from sim.gsp_profile import DEFAULT_IMAGE, run_profile +from sim.machine import R_EBREAK, 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 = QEMU_REFERENCE["access_addr"] # 0x2000000000000A00 (570.144 first touch) + +# addi a0,a0,1 x3; ebreak +PROGRAM_ALU = bytes.fromhex("13051500" * 3 + "73001000") +# lui t0,0x80000; lw a0,0(t0); ebreak (one completed RAM load) +PROGRAM_RAM_LOAD = bytes.fromhex("b702008003a5020073001000") +# lui t0,0x20000; slli t0,t0,32; addi t0,t0,0x600; addi t0,t0,0x400; +# lw a0,0(t0); sw a0,0(t0); ebreak (one unknown-MMIO read + one write) +PROGRAM_MMIO = bytes.fromhex("b702002093920202938202609382024003a5020023a0a20073001000") + + +def decoded_cost(program: bytes, model: CycleModel, pc0: int = BASE) -> int: + """Instruction cost of a straight-line 4-byte-encoded program (no memory).""" + total = 0 + pc = pc0 + while pc < pc0 + len(program): + word = int.from_bytes(program[pc - pc0 : pc - pc0 + 4], "little") + kind, sub, _rd, _r1, _r2, _imm, _nxt = decode32(pc, word) + total += model.instruction_cost(kind, sub) + pc += 4 + return total + + +class CycleTableTests(unittest.TestCase): + def test_every_decoded_kind_has_a_positive_cost(self): + subs = {K_OP: range(18), K_OP32: range(18)} + for kind in range(18): + for sub in subs.get(kind, (0,)): + cost = DEFAULT_MODEL.instruction_cost(kind, sub) + self.assertIsInstance(cost, int) + self.assertGreaterEqual(cost, 1, f"kind={kind} sub={sub}") + + def test_alu_branch_and_jump_are_unit_cost(self): + model = DEFAULT_MODEL + self.assertEqual(model.instruction_cost(K_OP, A_ADD), model.alu) + self.assertEqual(model.instruction_cost(K_OP, A_SUB), model.alu) + self.assertEqual(model.instruction_cost(K_BRANCH), model.branch) + self.assertEqual(model.instruction_cost(K_JAL), model.jump) + self.assertEqual(model.instruction_cost(K_JALR), model.jump) + + def test_m_extension_is_conservative(self): + model = DEFAULT_MODEL + mul = model.instruction_cost(K_OP, A_MUL) + div = model.instruction_cost(K_OP, A_DIV) + self.assertGreater(mul, model.alu) + self.assertGreater(div, mul) + # Same table is used for the 32-bit M forms. + self.assertEqual(model.instruction_cost(K_OP32, A_MUL), mul) + self.assertEqual(model.instruction_cost(K_OP32, A_DIV), div) + + def test_memory_latency_is_a_parameter(self): + model = DEFAULT_MODEL + self.assertGreaterEqual(model.memory_cost(False), 1) + self.assertGreaterEqual(model.memory_cost(True), model.memory_cost(False)) + custom = CycleModel(memory_latency=9, mmio_latency=40) + self.assertEqual(custom.memory_cost(False), 9) + self.assertEqual(custom.memory_cost(True), 40) + self.assertEqual(custom.instruction_cost(K_LOAD), model.instruction_cost(K_LOAD)) + + def test_csr_fence_amo_and_trap_are_costs(self): + model = DEFAULT_MODEL + for kind in (K_CSR, K_AMO, K_EBREAK, K_ECALL, K_ILLEGAL): + self.assertGreaterEqual(model.instruction_cost(kind), 1) + + def test_classify_reports_classes(self): + self.assertEqual(CycleModel.classify(K_OP, A_ADD), CLASS_ALU) + self.assertEqual(CycleModel.classify(K_OP, A_MUL), CLASS_MUL) + self.assertEqual(CycleModel.classify(K_OP, A_DIV), CLASS_DIV) + self.assertEqual(CycleModel.classify(K_LOAD), CLASS_LOAD) + + def test_model_is_labelled_modeled_not_measured(self): + self.assertIn("modeled", NOTE.lower()) + self.assertIn("not measured", NOTE.lower()) + self.assertIn("L2", CYCLE_FIDELITY) + self.assertIn("modeled", DEFAULT_MODEL.name) + described = DEFAULT_MODEL.describe() + self.assertEqual(described["name"], DEFAULT_MODEL.name) + self.assertEqual(described["note"], NOTE) + self.assertIn("memory_latency", described) + + def test_invalid_parameters_rejected(self): + with self.assertRaises(ValueError): + CycleModel(alu=0) + with self.assertRaises(ValueError): + CycleModel(memory_latency=-1) + with self.assertRaises(ValueError): + CycleModel(div=True) + + +class MachineCycleTests(unittest.TestCase): + def _machine(self, program: bytes, model=None, **kwargs) -> Machine: + machine = Machine(cycle_model=model, **kwargs) + machine.load_segments([(BASE, program)], BASE) + return machine + + def test_counting_enabled_by_default(self): + machine = self._machine(PROGRAM_ALU) + machine.run(3) # three addi, budget stop + self.assertEqual(machine.insns, 3) + self.assertEqual(machine.cycles, 3) + + def test_ram_load_adds_the_modelled_memory_latency(self): + machine = self._machine(PROGRAM_RAM_LOAD) + reason, _pc, _a, _b = machine.run(100) + self.assertEqual(reason, R_EBREAK) + expected = decoded_cost(PROGRAM_RAM_LOAD, DEFAULT_MODEL) + DEFAULT_MODEL.memory_latency + self.assertEqual(machine.cycles, expected) + + def test_mmio_accesses_add_the_modelled_mmio_latency(self): + machine = self._machine(PROGRAM_MMIO, unknown_mmio="lenient") + reason, _pc, _a, _b = machine.run(100) + self.assertEqual(reason, R_EBREAK) + self.assertEqual(machine.mmio_unknown, 2) + self.assertEqual(machine.mmio_log[0]["addr"], TOUCH_A) + expected = decoded_cost(PROGRAM_MMIO, DEFAULT_MODEL) + 2 * DEFAULT_MODEL.mmio_latency + self.assertEqual(machine.cycles, expected) + + def test_variable_memory_latency_changes_only_the_counter(self): + slow = CycleModel(memory_latency=9) + quick = self._machine(PROGRAM_RAM_LOAD, model=DEFAULT_MODEL) + slower = self._machine(PROGRAM_RAM_LOAD, model=slow) + quick.run(100) + slower.run(100) + self.assertEqual(slower.insns, quick.insns) + self.assertEqual(slower.x, quick.x) + self.assertEqual(slower.cycles - quick.cycles, 9 - DEFAULT_MODEL.memory_latency) + + def test_disabled_model_leaves_the_counter_at_zero(self): + machine = self._machine(PROGRAM_ALU, model=False) + machine.run(3) + self.assertIsNone(machine.cycle_model) + self.assertEqual(machine.insns, 3) + self.assertEqual(machine.cycles, 0) + + def test_model_does_not_change_execution(self): + without = self._machine(PROGRAM_MMIO, model=False, unknown_mmio="lenient") + with_model = self._machine(PROGRAM_MMIO, unknown_mmio="lenient") + result_without = without.run(100) + result_with_model = with_model.run(100) + self.assertEqual(result_without, result_with_model) + self.assertEqual(without.insns, with_model.insns) + self.assertEqual(without.x, with_model.x) + self.assertEqual(without.mmio_log, with_model.mmio_log) + self.assertGreater(with_model.cycles, 0) + + def test_double_run_identical(self): + def profile(): + machine = self._machine(PROGRAM_MMIO, unknown_mmio="lenient") + result = machine.run(100) + return result, machine.insns, machine.cycles, machine.pc, tuple(machine.x) + + self.assertEqual(profile(), profile()) + + @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware absent") + def test_demo_execution_identical_with_and_without_model(self): + from greeneye.elf import load_elf + + image = load_elf(DEMO_ELF) + plan = image.flat_load_plan(bias=0) + entry = image.relocated_entry(bias=0) + + def run(model): + machine = Machine(cycle_model=model) + machine.load_segments(plan, entry) + result = machine.run(20_000) + return result, machine.insns, tuple(machine.x), machine.uart.text() + + self.assertEqual(run(False), run(None)) + + +class GspProfileTests(unittest.TestCase): + def test_default_image_path_matches_workspace_firmware(self): + self.assertEqual( + os.path.normpath(DEFAULT_IMAGE), + os.path.normpath(REAL_IMAGE), + ) + + def test_absent_image_skips_cleanly(self): + with tempfile.TemporaryDirectory() as root: + missing = os.path.join(root, "no-such-image.bin") + result = run_profile(image=missing, instructions=1000) + self.assertFalse(result.ran) + self.assertIn("absent", result.reason) + self.assertIsNone(result.session_dir) + + def test_bad_budget_rejected(self): + from greeneye.errors import GreenEyeError + + with self.assertRaises(GreenEyeError): + run_profile(image=REAL_IMAGE, instructions=0) + + +@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "TU104 570.144 fwimage absent") +class RealImageCycleProfileTests(unittest.TestCase): + def test_bounded_profile_first_touch_and_metadata(self): + with tempfile.TemporaryDirectory() as root: + result = run_profile( + image=REAL_IMAGE, instructions=200_000, record=root, name="unittest-cycle0" + ) + self.assertTrue(result.ran, result.reason) + self.assertEqual(result.instructions, 200_000) + self.assertEqual(result.stop_reason, "budget") + self.assertGreater(result.cycles, result.instructions) + self.assertEqual(result.cycle_model, DEFAULT_MODEL.name) + self.assertEqual(result.first_unknown_mmio["addr"], TOUCH_A) + self.assertEqual(result.first_unknown_mmio["rw"], "r") + self.assertEqual(result.first_unknown_mmio["pc"], 0x84009014) + self.assertEqual(result.stop_pc, 0x84009070) + self.assertEqual(result.stop_pc_fw, 0x4009070) + + with open(os.path.join(result.session_dir, "session.json"), "r", encoding="utf-8") as fh: + meta = json.load(fh) + self.assertEqual(meta["cycles"], result.cycles) + self.assertEqual(meta["instructions_executed"], 200_000) + self.assertEqual(meta["cycle_model"], DEFAULT_MODEL.name) + self.assertEqual(meta["firmware_sha256"], result.image_sha256) + self.assertIn("modeled", meta["cycle_fidelity"]) + self.assertEqual(meta["first_unknown_mmio"]["addr"], TOUCH_A) + self.assertEqual(meta["cycle_model_table"]["name"], DEFAULT_MODEL.name) + + def test_two_runs_identical_cycles(self): + with tempfile.TemporaryDirectory() as root: + first = run_profile( + image=REAL_IMAGE, + instructions=200_000, + record=os.path.join(root, "a"), + name="unittest-cycle-a", + ) + second = run_profile( + image=REAL_IMAGE, + instructions=200_000, + record=os.path.join(root, "b"), + name="unittest-cycle-b", + ) + self.assertTrue(first.ran and second.ran) + self.assertEqual(first.instructions, second.instructions) + self.assertEqual(first.cycles, second.cycles) + self.assertEqual(first.stop_pc, second.stop_pc) + self.assertEqual(first.image_sha256, second.image_sha256) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py new file mode 100644 index 0000000..8063ee9 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_devices.py @@ -0,0 +1,458 @@ +"""Device-model tests: registry, per-block semantics, profile and boot ladder. + +Every modelled register assertion here is tied to the evidence documented in +``sim/devices/*.py`` and ``docs/36-GSP-DEVICE-MODEL.md``. The real-firmware +tests are bounded (<= 20k instructions) and assert *observed* rungs only. +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sim.devices import ( # noqa: E402 + CsbWindow, + DmaTranslator, + FalconBlock, + IrqRouter, + PmcBlock, + PmuStub, + PriscvBlock, + PtimerBlock, + Wpr2Block, + load_profile, + registry_from_profile, +) +from sim.devices.base import DeviceRegistry # noqa: E402 +from sim.devices import falcon as falcon_mod # noqa: E402 +from sim.machine import Machine # noqa: E402 + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +DEFAULT_IMAGE = os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", + "gsp-570.144.fwimage.bin")) +DEFAULT_PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") + + +class RegistryTests(unittest.TestCase): + def test_containment_and_unhandled_log(self): + reg = DeviceRegistry() + dev = PmcBlock(base=0x1000, size=0x1000) + reg.add(dev) + ok, value = reg.read(0x1A00, 4) + self.assertTrue(ok) + self.assertEqual(value, dev.boot_42) + # unmodelled offset inside the window: unhandled but window-matched + ok, _ = reg.read(0x1A04, 4) + self.assertFalse(ok) + self.assertEqual(reg.unhandled, 1) + # outside every window + ok, _ = reg.read(0x2000, 4) + self.assertFalse(ok) + self.assertEqual(reg.unhandled, 2) + # Only window-matched accesses are recorded; outside-window accesses + # are counted but left to the machine's unknown log. + self.assertEqual(len(reg.accesses), 2) + self.assertFalse(reg.accesses[1]["handled"]) + + def test_overlapping_windows_fall_back_on_refusal(self): + # A later window shadows an earlier one, but an offset the overlay does + # not model (here the PMU stub refuses everything) falls back to the + # wider device instead of being logged as unknown. + reg = DeviceRegistry() + fal = FalconBlock(base=0x1000, size=0x1000) + reg.add(fal) + reg.add(PmuStub(base=0x1000, size=0x1000)) + self.assertTrue(reg.write(0x1040, 4, 0xCA)) # MAILBOX0 via fallback + self.assertEqual(fal.regs[falcon_mod.MAILBOX0], 0xCA) + ok, value = reg.read(0x1040, 4) + self.assertTrue(ok) + self.assertEqual(value, 0xCA) + self.assertEqual(reg.unhandled, 0) + self.assertTrue(all(a["handled"] for a in reg.accesses)) + + def test_overlapping_windows_all_refuse_records_once(self): + reg = DeviceRegistry() + reg.add(PmuStub(base=0x1000, size=0x1000)) + reg.add(PmuStub(base=0x1000, size=0x1000)) + ok, _ = reg.read(0x1000, 4) + self.assertFalse(ok) + self.assertEqual(reg.unhandled, 1) + self.assertEqual(len(reg.accesses), 1) + self.assertFalse(reg.accesses[0]["handled"]) + self.assertFalse(reg.write(0x1000, 4, 1)) + self.assertEqual(reg.unhandled, 2) + self.assertEqual(len(reg.accesses), 2) + + def test_straddling_access_is_not_served(self): + reg = DeviceRegistry() + reg.add(PmcBlock(base=0x1000, size=0x10)) + ok, _ = reg.read(0x1008, 8) # ends at 0x1010, outside the window + self.assertFalse(ok) + + def test_state_round_trip(self): + reg = DeviceRegistry() + fal = FalconBlock(base=0x1000, size=0x1000) + reg.add(fal) + reg.write(0x1040, 4, 0xCA) + snap = reg.state() + reg2 = DeviceRegistry() + fal2 = FalconBlock(base=0x1000, size=0x1000) + reg2.add(fal2) + reg2.load_state(snap) + self.assertEqual(fal2.regs[falcon_mod.MAILBOX0], 0xCA) + self.assertEqual(fal2.state(), fal.state()) + + +class PmcTests(unittest.TestCase): + def test_tu104_value_fields(self): + dev = PmcBlock(chip_id=0x164) + value = dev.read(0xA00, 4) + self.assertEqual((value >> 20) & 0x1FF, 0x164) + self.assertEqual((value >> 24) & 0x1F, 0x16) # ARCHITECTURE TU100 + self.assertEqual((value >> 20) & 0xF, 0x4) # IMPLEMENTATION + self.assertEqual(dev.read(0xA04, 4), None) + self.assertFalse(dev.write(0xA00, 4, 0)) + + def test_profile_profile_value_override(self): + dev = PmcBlock(boot_42=0x00123456) + self.assertEqual(dev.read(0xA00, 4), 0x00123456) + + def test_chip_id_constants_are_distinct(self): + from sim.devices import pmc as pmc_mod + ids = [pmc_mod.TU102, pmc_mod.TU104, pmc_mod.TU105, pmc_mod.TU106, + pmc_mod.TU116, pmc_mod.TU117] + self.assertEqual(len(ids), len(set(ids))) + self.assertEqual(pmc_mod.TU117, 0x167) + self.assertFalse(hasattr(pmc_mod, "TU107")) + + +class FalconTests(unittest.TestCase): + def test_irq_set_clear_status(self): + dev = FalconBlock() + dev.write(falcon_mod.IRQSSET, 4, falcon_mod.IRQ_HALT) + self.assertEqual(dev.read(falcon_mod.IRQSTAT, 4), falcon_mod.IRQ_HALT) + dev.write(falcon_mod.IRQSCLR, 4, falcon_mod.IRQ_HALT) + self.assertEqual(dev.read(falcon_mod.IRQSTAT, 4), 0) + # write-only set/clear registers have no read-back + self.assertIsNone(dev.read(falcon_mod.IRQSSET, 4)) + + def test_mailbox_rw(self): + dev = FalconBlock() + dev.write(falcon_mod.MAILBOX0, 4, 0xCA) + dev.write(falcon_mod.MAILBOX1, 4, 0x11DEAD) + self.assertEqual(dev.read(falcon_mod.MAILBOX0, 4), 0xCA) + self.assertEqual(dev.read(falcon_mod.MAILBOX1, 4), 0x11DEAD) + + def test_debuginfo_storage_without_shim(self): + dev = FalconBlock(debuginfo_host_consume=False) + dev.write(falcon_mod.DEBUGINFO, 4, 0x11DEAD) + self.assertEqual(dev.read(falcon_mod.DEBUGINFO, 4), 0x11DEAD) + + def test_debuginfo_host_consume_shim(self): + dev = FalconBlock(debuginfo_host_consume=True) + dev.write(falcon_mod.DEBUGINFO, 4, 0x11DEAD) + value = dev.read(falcon_mod.DEBUGINFO, 4) + self.assertEqual(value, 0x11DEAD) # value preserved for the reader + self.assertEqual(dev.read(falcon_mod.DEBUGINFO, 4), 0x1DEAD) # stage cleared + self.assertEqual(dev.debuginfo_consumed, 1) + # debugger peek never consumes + dev.write(falcon_mod.DEBUGINFO, 4, 0x21DEAD) + self.assertEqual(dev.peek(falcon_mod.DEBUGINFO, 4), 0x21DEAD) + self.assertEqual(dev.debuginfo_consumed, 1) + + def test_dma_transfer_through_translator(self): + ram = bytearray(0x10000) + ram[0x100:0x108] = b"DEADBEEF" + dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram), + windows=[{"pa": 0x0, "sim": 0, "size": 0x10000}]) + dev = FalconBlock(dma=dma) + # Program: destination = 0x200, source = IMEM + 0x100. The register + # encodes ((base1 << 32) | base) << 8; the profile's DMA windows decide + # which aperture that address resolves to (aperture selection itself is + # not proven yet and is documented as an unknown). + dev.write(falcon_mod.DMATRFBASE1, 4, 0) + dev.write(falcon_mod.DMATRFBASE, 4, 0) + dev.write(falcon_mod.DMATRFFBOFFS, 4, 0x200) + dev.write(falcon_mod.DMATRFMOFFS, 4, 0x100) + cmd = (falcon_mod.CMD_IMEM | falcon_mod.CMD_WRITE + | (2 << falcon_mod.CMD_SIZE_SHIFT)) # IMEM -> FB, 16 bytes + dev.write(falcon_mod.DMATRFCMD, 4, cmd) + self.assertEqual(ram[0x200:0x210], b"DEADBEEF" + bytes(8)) + self.assertTrue(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_IDLE) + self.assertEqual(dev.dma_transfers, 1) + + def test_dma_refused_when_unmapped(self): + ram = bytearray(0x1000) + dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram), windows=[]) + dev = FalconBlock(dma=dma) + dev.write(falcon_mod.DMATRFBASE, 4, 0) + dev.write(falcon_mod.DMATRFCMD, 4, falcon_mod.CMD_IMEM) + self.assertFalse(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_IDLE) + self.assertTrue(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_ERROR) + self.assertEqual(dev.dma_refused, 1) + + +class PriscvTests(unittest.TestCase): + def test_irq_mask_semantics(self): + dev = PriscvBlock() + dev.write(0x2AC, 4, 0x5) + dev.write(0x2B0, 4, 0x1) + self.assertEqual(dev.read(0x2B4, 4), 0x4) + self.assertFalse(dev.write(0x2B4, 4, 0xFFFF)) # read-only + + def test_trace_and_error_registers(self): + dev = PriscvBlock() + dev.write(0x344, 4, 0x1234) + self.assertEqual(dev.read(0x344, 4), 0x1234) + # observed boot panic reads (R-I4R): read-back is zero, writes refused + self.assertEqual(dev.read(0x364, 4), 0) + self.assertEqual(dev.read(0x368, 4), 0) + self.assertFalse(dev.write(0x368, 4, 0x1)) + + +class TimerTests(unittest.TestCase): + def test_modeled_time_reads(self): + clock = {"insns": 0} + dev = PtimerBlock(ns_per_insn=2.0, clock=lambda: clock["insns"]) + self.assertEqual(dev.read(0x400, 4), 0) + self.assertEqual(dev.read(0x410, 4), 0) + clock["insns"] = 16 + # 32 ns -> TIME_0 NSEC field (bits 31:5) = 1 + self.assertEqual(dev.read(0x400, 4), 1) + self.assertEqual(dev.read(0x410, 4), 32) + + def test_profile_window_reaches_time_registers(self): + # Regression: the profile window is PRIV+0x9000, so TIME_0/TIME_1 must + # be block-relative (0x400/0x410); aperture-absolute 0x9400/0x9410 fall + # outside the declared window and serve nothing. + profile = { + "schema": "greeneye-sim-profile/1", + "devices": { + "timer": { + "ns_per_insn": 1.0, + "windows": [{"base": "0x9000", "size": "0x1000"}], + } + }, + } + machine = Machine(memory_mb=1, ram_base=0) + reg = registry_from_profile(profile, machine=machine) + machine.insns = 32 + ok, value = reg.read(0x9000 + 0x400, 4, pc=0) + self.assertTrue(ok) + self.assertEqual(value, 1) # (32 ns >> 5) + ok, value = reg.read(0x9000 + 0x410, 4, pc=0) + self.assertTrue(ok) + self.assertEqual(value, 32) + + def test_timer_registration_is_opt_in(self): + machine = Machine(memory_mb=1, ram_base=0) + no_section = { + "schema": "greeneye-sim-profile/1", + "devices": {"pmc": {"windows": [{"base": "0x1000", "size": "0x1000"}]}}, + } + reg = registry_from_profile(no_section, machine=machine) + self.assertIsNone(reg.device_by_name("timer")) + disabled = { + "schema": "greeneye-sim-profile/1", + "devices": {"timer": {"enabled": False, + "windows": [{"base": "0x9000", "size": "0x1000"}]}}, + } + reg = registry_from_profile(disabled, machine=machine) + self.assertIsNone(reg.device_by_name("timer")) + enabled = { + "schema": "greeneye-sim-profile/1", + "devices": {"timer": {"ns_per_insn": 1.0, + "windows": [{"base": "0x9000", "size": "0x1000"}]}}, + } + reg = registry_from_profile(enabled, machine=machine) + self.assertIsNotNone(reg.device_by_name("timer")) + + +class CsbTests(unittest.TestCase): + def test_undeclared_window_serves_nothing(self): + dev = CsbWindow(base=0x1000, size=0x1000) + self.assertIsNone(dev.read(0x0, 4)) + self.assertFalse(dev.write(0x0, 4, 1)) + + def test_declared_semantics(self): + dev = CsbWindow(base=0x1000, size=0x1000, registers={ + 0x00: {"width": 4, "value": 0xF, "semantics": "w1c"}, + 0x04: {"width": 4, "value": 0, "semantics": "rw"}, + 0x08: {"width": 4, "value": 0, "semantics": "ro"}, + }) + dev.write(0x00, 4, 0x3) + self.assertEqual(dev.read(0x00, 4), 0xC) + dev.write(0x04, 4, 0x77) + self.assertEqual(dev.read(0x04, 4), 0x77) + self.assertFalse(dev.write(0x08, 4, 1)) + + +class Wpr2Tests(unittest.TestCase): + def test_range_encoding(self): + dev = Wpr2Block() + dev.set_range(0x2000000, 0x6000000) + lo = dev.read(0x1FA824, 4) + hi = dev.read(0x1FA828, 4) + self.assertEqual((lo << 12), 0x2000000) + self.assertEqual((hi << 12) + 0x20000, 0x6000000) + self.assertIsNone(dev.read(0x1FA800, 4)) + + def test_state_derives_range_after_writes(self): + # Regression: writes update regs only; state() must derive start/end + # from the registers rather than reporting a stale set_range range. + dev = Wpr2Block() + dev.write(0x1FA824, 4, 0x2000000 >> 12) + dev.write(0x1FA828, 4, (0x6000000 - 0x20000) >> 12) + state = dev.state() + self.assertEqual(state["start"], 0x2000000) + self.assertEqual(state["end"], 0x6000000) + restored = Wpr2Block() + restored.load_state(state) + self.assertEqual(restored.start, 0x2000000) + self.assertEqual(restored.end, 0x6000000) + # A stale serialised start/end never wins over the registers. + stale = Wpr2Block() + stale.load_state({"regs": {str(0x1FA824): 0x2000}, + "start": 0xDEAD, "end": 0xBEEF}) + self.assertEqual(stale.start, 0x2000000) + self.assertIsNone(stale.end) + + +class PmuTests(unittest.TestCase): + def test_stub_never_fakes(self): + dev = PmuStub(base=0x1000, size=0x1000) + self.assertIsNone(dev.read(0x0, 4)) + self.assertFalse(dev.write(0x0, 4, 1)) + self.assertEqual(dev.seen["r:0x0:4"], 1) + + +class IrqRouterTests(unittest.TestCase): + def test_aggregation(self): + fal = FalconBlock() + pri = PriscvBlock() + router = IrqRouter(falcon=fal, priscv=pri) + fal.write(falcon_mod.IRQSSET, 4, falcon_mod.IRQ_HALT) + pri.write(0x2AC, 4, 0x2) + state = router.pending() + self.assertTrue(state["falcon_halt"]) + self.assertEqual(state["riscv_irq_mask"], 0x2) + + +class DmaTranslatorTests(unittest.TestCase): + def test_translate_and_refuse(self): + ram = bytearray(0x100) + dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram), + windows=[{"pa": 0x1000, "sim": 0, "size": 0x100}]) + self.assertEqual(dma.translate(0x1000), 0) + self.assertEqual(dma.translate(0x10FF), 0xFF) + self.assertIsNone(dma.translate(0x2000)) + self.assertTrue(dma.write(0x1000, b"AB")) + self.assertEqual(dma.read(0x1000, 2), b"AB") + self.assertIsNone(dma.read(0x2000, 2)) + self.assertEqual(dma.refused, 1) + + def test_copy_counts_only_completed_transfers_and_exports_events(self): + ram = bytearray(0x100) + ram[0:4] = b"ABCD" + dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram), + windows=[{"pa": 0x1000, "sim": 0, "size": 0x100}]) + # Destination outside every window: refused, not a transfer. + self.assertFalse(dma.copy(0x1000, 0x2000, 4)) + self.assertEqual(dma.transfers, 0) + self.assertEqual(dma.refused, 1) + # Both mapped: one completed transfer. + self.assertTrue(dma.copy(0x1000, 0x1004, 4)) + self.assertEqual(dma.transfers, 1) + self.assertEqual(ram[0x4:0x8], b"ABCD") + state = dma.state() + self.assertIn("events", state) + self.assertEqual(state["events"][0]["kind"], "w") + self.assertEqual(len(state["events"]), 1) + + +class MachineDeviceIntegrationTests(unittest.TestCase): + def _machine(self): + reg = DeviceRegistry() + reg.add(PmcBlock(base=0x2000000000000000, size=0x1000)) + m = Machine(memory_mb=1, ram_base=0, unknown_mmio="lenient", devices=reg) + return m, reg + + def test_modeled_device_takes_precedence_over_lenient(self): + m, reg = self._machine() + self.assertEqual(m._mmio_load(0x2000000000000A00, 4), PmcBlock().boot_42) + self.assertEqual(m.mmio_unknown, 0) + self.assertEqual(len(reg.accesses), 1) + + def test_unknown_inside_window_is_logged(self): + m, reg = self._machine() + self.assertEqual(m._mmio_load(0x2000000000000A04, 4), 0) + self.assertEqual(m.mmio_unknown, 1) + self.assertFalse(reg.accesses[0]["handled"]) + + def test_debugger_peek_and_device_state(self): + m, reg = self._machine() + self.assertEqual(m.read_mem(0x2000000000000A00, 4), + PmcBlock().boot_42.to_bytes(4, "little")) + self.assertIsInstance(m.device_state(), dict) + self.assertEqual(m.device_state()["schema"], "greeneye-sim-devices/1") + + def test_reset_regs_names(self): + m = Machine(memory_mb=1, ram_base=0, reset_regs={"a0": 0x1234, "sp": 0x8000}) + self.assertEqual(m.x[10], 0x1234) + self.assertEqual(m.x[2], 0x8000) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent") +class BootLadderTests(unittest.TestCase): + """Bounded real-firmware runs: observed rungs only.""" + + def test_profile_loads_and_builds_windows(self): + profile = load_profile(DEFAULT_PROFILE) + self.assertEqual(profile["schema"], "greeneye-sim-profile/1") + m = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient") + reg = registry_from_profile(profile, machine=m) + names = {d.name for d in reg.devices} + self.assertIn("pmc", names) + self.assertIn("falcon", names) + self.assertIn("priscv", names) + self.assertIsNotNone(reg.dma) + + def test_no_shim_b1_device_model_serves_boot(self): + from sim.gsp_boot import run_boot + r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=False) + self.assertTrue(r.ran) + self.assertEqual(r.rung, "B1") + self.assertIn("bootargs_consumer", r.milestones) + self.assertEqual(r.mmio_unknown, 0) + self.assertEqual(r.device_unhandled, 0) + + def test_shim_reaches_relocation_b2(self): + from sim.gsp_boot import run_boot + r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=True) + self.assertTrue(r.ran) + self.assertEqual(r.rung, "B2") + self.assertIn("relocation_copy", r.milestones) + self.assertEqual(r.mmio_unknown, 0) + # Known unknown: the DMEM runspace content the app relocates from is + # not reconstructed; the run stops there rather than faking it. + self.assertEqual(r.stop_reason, "fault") + self.assertEqual(r.stop_pc, 0x400EA0C) + + def test_two_runs_identical(self): + from sim.gsp_boot import run_boot + kwargs = dict(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE, + instructions=20_000, record=False, use_shim=True) + a = run_boot(**kwargs) + b = run_boot(**kwargs) + self.assertEqual(a.instructions, b.instructions) + self.assertEqual(a.stop_pc, b.stop_pc) + self.assertEqual(a.stop_reason, b.stop_reason) + self.assertEqual(a.device_state_sha256, b.device_state_sha256) + self.assertEqual(a.registers, b.registers) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py new file mode 100644 index 0000000..5a798f2 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_qemu_line.py @@ -0,0 +1,323 @@ +"""Tests for the S4 QEMU-line evaluation harness (harness/sim_qemu_line.py). + +The pure-function tests need no firmware. The integration tests use the real +TU104 570.144 fwimage read-only and **skip cleanly** when it (or +qemu-system-riscv64, or the bundled demo firmware) is absent. All runs are +bounded (<= 20k native instructions in tests, <= 1 s QEMU) and touch no +hardware, GPU, module, guest or network. + +Run:: + + python3 -m unittest tests.test_sim_qemu_line -v +""" +from __future__ import annotations + +import json +import os +import shutil +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +REAL_IMAGE = os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin") +) +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 greeneye.device_map import QEMU_REFERENCE, firmware_sha256 # noqa: E402 +from harness.common import qemu_available # noqa: E402 +from harness.sim_qemu_line import FROZEN_THRESHOLDS # noqa: E402 + + +class PureUnitTests(unittest.TestCase): + """No firmware required: rung maths, thresholds and report rendering.""" + + def test_rung_ordinal_and_above(self): + from harness.sim_qemu_line import above_qemu_line, rung_ordinal + + self.assertEqual(rung_ordinal("pre-B0"), 0) + self.assertEqual(rung_ordinal("B0"), 1) + self.assertIsNone(rung_ordinal("B9")) + self.assertTrue(above_qemu_line("pre-B0", "B0")) + self.assertFalse(above_qemu_line("pre-B0", "pre-B0")) + self.assertFalse(above_qemu_line("B1", "B0")) + self.assertFalse(above_qemu_line(None, "B0")) + + def test_threshold_checks_require_device_model(self): + from harness.sim_qemu_line import threshold_checks + + qemu = {"ran": True, "first_touch_match": True, "rung": "pre-B0"} + native = { + "ran": True, + "rung": "B0", + "repeat": 2, + "identical": True, + "pair_wall_s": 1.0, + "device_model_used": False, + } + checks = threshold_checks({"qemu": qemu, "native": native}) + self.assertTrue(checks["qemu_first_fault_reproduced"]) + self.assertTrue(checks["native_above_qemu"]) + self.assertTrue(checks["identical_runs"]) + self.assertFalse(checks["device_model_documented"]) + self.assertFalse(checks["all"]) + native["device_model_used"] = True + self.assertTrue(threshold_checks({"qemu": qemu, "native": native})["all"]) + + def test_build_report_verdicts(self): + from harness.sim_qemu_line import build_report + + qemu = {"ran": True, "first_touch_match": True, "rung": "pre-B0"} + native_ok = { + "ran": True, + "rung": "B1", + "repeat": 2, + "identical": True, + "pair_wall_s": 1.0, + "device_model_used": True, + "runs": [], + } + report = build_report( + image=os.path.join(TOOL_ROOT, "does-not-exist.bin"), + qemu=qemu, + native=native_ok, + features=None, + s1={"source": None, "declared_rung": None, "notes": []}, + thresholds=dict(FROZEN_THRESHOLDS), + ) + self.assertEqual(report["verdict"], "above") + self.assertTrue(report["above_qemu_line"]) + self.assertTrue(report["threshold_checks"]["all"]) + + lenient = dict(native_ok, device_model_used=False) + report = build_report( + image="missing.bin", + qemu=qemu, + native=lenient, + features=None, + s1={"source": None, "declared_rung": None, "notes": []}, + thresholds=dict(FROZEN_THRESHOLDS), + ) + self.assertEqual(report["verdict"], "above_lenient_only") + self.assertFalse(report["above_qemu_line"]) + + report = build_report( + image="missing.bin", + qemu={"ran": False, "rung": None}, + native=native_ok, + features=None, + s1={"source": None, "declared_rung": None, "notes": []}, + thresholds=dict(FROZEN_THRESHOLDS), + ) + self.assertEqual(report["verdict"], "unverified_qemu_absent") + + report = build_report( + image="missing.bin", + qemu={"ran": True, "first_touch_match": False, "rung": "pre-B0"}, + native=native_ok, + features=None, + s1={"source": None, "declared_rung": None, "notes": []}, + thresholds=dict(FROZEN_THRESHOLDS), + ) + self.assertEqual(report["verdict"], "unverified_qemu_mismatch") + + def test_render_markdown_shape(self): + from harness.sim_qemu_line import build_report, render_markdown + + report = build_report( + image="missing.bin", + qemu={ + "ran": True, + "first_touch_match": True, + "rung": "pre-B0", + "wall_s": 1.0, + "first_fault": {"desc": "fault_load", "epc_fw": 0x4009014, + "access_addr": QEMU_REFERENCE["access_addr"], + "size": 4, "rw": "read"}, + "session_hashes": {"session.json": "ab" * 32}, + "session": "harness/qemu_line/sessions/qemu-probe", + }, + native={ + "ran": True, + "rung": "B0", + "rung_basis": "test", + "repeat": 2, + "identical": True, + "pair_wall_s": 1.0, + "device_model_used": False, + "runs": [ + { + "rung": "B0", + "instructions": 10, + "cycles": 12, + "wall_s": 0.1, + "stop": {"reason": "budget", "pc_fw": 0x4009070}, + "session": "harness/qemu_line/sessions/native-1", + "session_hashes": {"session.json": "cd" * 32}, + } + ], + "compare": {"ran": True, "headers": []}, + }, + features={"features": {"mcp": {"status": "not_run", "reason": "S3"}}}, + s1={"source": None, "declared_rung": None, "notes": []}, + thresholds=dict(FROZEN_THRESHOLDS), + ) + text = render_markdown(report) + self.assertIn("Rung table", text) + self.assertIn("Session hashes", text) + self.assertIn("Frozen thresholds", text) + self.assertIn("above_lenient_only", text) + self.assertIn("native run 1", text) + + +@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "TU104 570.144 fwimage absent") +class NativeIntegrationTests(unittest.TestCase): + """Native-only run on the real image: shape + two-run determinism.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="sim-qemu-line-test-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_profile_discovery_when_s1_present(self): + from harness.sim_qemu_line import discover_s1_device_model + + profile = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") + if not os.path.isfile(profile): + self.skipTest("S1 profile absent") + s1 = discover_s1_device_model(REAL_IMAGE, firmware_sha256(REAL_IMAGE)) + self.assertEqual(os.path.abspath(s1.get("profile") or ""), profile) + self.assertTrue(any("greeneye-sim-profile" in note for note in s1["notes"])) + + def test_native_report_shape_and_determinism(self): + from harness.sim_qemu_line import run_harness + + report = run_harness( + image=REAL_IMAGE, + out_dir=self.tmp, + instructions=20_000, + repeat=2, + features=False, + run_qemu=False, + ) + self.assertEqual(report["schema"], "greeneye-sim-qemu-line/1") + self.assertTrue(report["firmware"]["sha256"]) + native = report["native"] + self.assertTrue(native["ran"]) + self.assertEqual(native["repeat"], 2) + self.assertTrue(native["identical"]) + runs = native["runs"] + self.assertEqual(len(runs), 2) + keys = ( + "instructions", "cycles", "stop", "first_unknown_mmio", "mmio_unknown", + "rung", "device_state_sha256", "milestones", + ) + self.assertEqual( + [{k: r.get(k) for k in keys} for r in runs], + [{k: runs[0].get(k) for k in keys}] * 2, + ) + for run in runs: + self.assertGreater(run["instructions"], 0) + self.assertLessEqual(run["instructions"], 20_000) + self.assertGreater(run["cycles"], 0) + self.assertTrue(run["session_hashes"]["session.json"]) + if native.get("device_model_used"): + # S1 semantic path: device model + recorded shim, B2 observed. + self.assertEqual(native["rung"], "B2") + self.assertIsNone(runs[0]["first_unknown_mmio"]) + self.assertEqual(runs[0]["mmio_unknown"], 0) + self.assertEqual(runs[0]["milestones"], ["bootargs_consumer", "relocation_copy"]) + self.assertTrue(runs[0]["device_state_sha256"]) + else: + # Lenient path (no S1 in the tree): B0 observation baseline. + self.assertEqual(native["rung"], "B0") + self.assertEqual(runs[0]["first_unknown_mmio"]["addr"], QEMU_REFERENCE["access_addr"]) + self.assertEqual(report["verdict"], "unverified_qemu_absent") + for name in ("sim-qemu-line.json", "sim-qemu-line.md"): + path = os.path.join(self.tmp, name) + self.assertTrue(os.path.isfile(path), path) + with open(os.path.join(self.tmp, "sim-qemu-line.json"), encoding="utf-8") as fh: + payload = json.load(fh) + self.assertEqual(payload["schema"], report["schema"]) + with open(os.path.join(self.tmp, "sim-qemu-line.md"), encoding="utf-8") as fh: + md = fh.read() + self.assertIn("| native run 1 |", md) + self.assertIn("Session hashes", md) + + def test_native_lenient_path_still_available(self): + from harness.sim_qemu_line import run_native_baseline + + native = run_native_baseline( + image=REAL_IMAGE, + out_dir=self.tmp, + instructions=20_000, + repeat=1, + profile=None, + device_map=None, + ) + self.assertTrue(native["ran"]) + self.assertFalse(native["device_model_used"]) + self.assertEqual(native["rung"], "B0") + self.assertEqual( + native["runs"][0]["first_unknown_mmio"]["addr"], QEMU_REFERENCE["access_addr"] + ) + + +@unittest.skipUnless( + os.path.isfile(REAL_IMAGE) and qemu_available(), + "TU104 570.144 fwimage or qemu-system-riscv64 absent", +) +class QemuIntegrationTests(unittest.TestCase): + """QEMU probe on the real image must reproduce its documented first fault.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="sim-qemu-line-test-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_qemu_first_fault_reproduced(self): + from harness.sim_qemu_line import run_qemu_baseline + + result = run_qemu_baseline(image=REAL_IMAGE, out_dir=self.tmp, duration=0.5) + self.assertTrue(result["ran"], result.get("reason")) + self.assertEqual(result["rung"], "pre-B0") + self.assertTrue(result["first_touch_match"], "documented first fault not reproduced") + first = result["first_fault"] + self.assertEqual(first["epc_fw"], QEMU_REFERENCE["firmware_vaddr"]) + self.assertEqual(first["access_addr"], QEMU_REFERENCE["access_addr"]) + self.assertTrue(result["session_hashes"]["session.json"]) + self.assertTrue(os.path.isfile(os.path.join(self.tmp, "sessions", "qemu-probe", "session.json"))) + + +@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware absent (run make demo)") +class FeatureMatrixTests(unittest.TestCase): + """The native feature matrix runs on the bundled demo firmware.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="sim-qemu-line-test-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_feature_matrix_shape(self): + from harness.sim_qemu_line import run_feature_matrix + + result = run_feature_matrix(self.tmp) + self.assertTrue(result["ran"], result.get("reason")) + features = result["features"] + for name in ("hooks", "breakpoints", "watch", "memory", "struct", "sessions", "trace"): + self.assertIn(name, features) + self.assertEqual(features[name]["status"], "pass", features[name]) + self.assertEqual(features["mcp"]["status"], "not_run") + self.assertIn(features["injection"]["status"], ("pass", "not_run")) + self.assertIn(features["analyze_compare"]["status"], ("pass", "not_run")) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_triage.py b/debug/risc-v-gsp-rm-vm/tests/test_triage.py new file mode 100644 index 0000000..35ac128 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_triage.py @@ -0,0 +1,264 @@ +"""Tests for the new-version triage workflow (harness/triage.py, docs/40). + +The orchestration tests use injected fake step runners: no QEMU, no native +boot, no firmware. The integration test uses the real TU104 570.144 fwimage +read-only with a tiny budget and skips cleanly when it (or +qemu-system-riscv64) is absent. Everything is bounded and touches no +hardware, GPU, module, guest or network. + +Run:: + + python3 -m unittest tests.test_triage -v +""" +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, "..")) +REAL_IMAGE = os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin") +) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.common import qemu_available # noqa: E402 +from harness.triage import ( # noqa: E402 + DEFAULT_RUNNERS, + FORMAT, + PLAN_FORMAT, + TriageOptions, + build_plan, + extract_version, + main, + run_triage, +) + + +def _run_main(argv: list[str]) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main(argv) + return rc, out.getvalue(), err.getvalue() + + +def _fake_runners(fail: str | None = None, session_dir: str | None = None): + """Deterministic step runners; optionally one raises.""" + calls: list[str] = [] + + def make(name: str, **detail): + def fn(options, out): + calls.append(name) + payload = {"status": "ok"} + payload.update(detail) + if session_dir and name in ("probe", "native_boot"): + os.makedirs(session_dir, exist_ok=True) + with open(os.path.join(session_dir, "session.json"), "w", encoding="utf-8") as fh: + fh.write("{}\n") + payload["sessions"] = [{"name": name, "dir": session_dir}] + payload["session_hashes"] = {"session.json": "deadbeef"} + return payload + + return fn + + def boom(options, out): + calls.append(fail) + raise RuntimeError("injected failure") + + runners = [ + ("identity", boom if fail == "identity" else make("identity")), + ("probe", boom if fail == "probe" else make("probe", rung="pre-B0")), + ("native_boot", boom if fail == "native_boot" else make("native_boot", rung="B2")), + ("ab_eval", boom if fail == "ab_eval" else make("ab_eval", verdict="above_lenient_only", above_qemu_line=False)), + ("version_diff", boom if fail == "version_diff" else make("version_diff", sections={"image": "ok"})), + ("features", boom if fail == "features" else make("features", unavailable=[])), + ] + return runners, calls + + +class VersionExtractionTests(unittest.TestCase): + def test_explicit_label_wins(self): + version, source = extract_version("/does/not/matter.bin", "999.1") + self.assertEqual((version, source), ("999.1", "explicit --version")) + + def test_sidecar_convention(self): + with tempfile.TemporaryDirectory() as tmp: + fw = os.path.join(tmp, "gsp-x.fwimage.bin") + open(fw, "wb").close() + with open(os.path.join(tmp, "gsp_version.txt"), "w", encoding="utf-8") as fh: + fh.write("570.211W\n") + version, source = extract_version(fw) + self.assertEqual(version, "570.211W") + self.assertEqual(source, "gsp_version.txt sidecar") + + def test_parent_directory_fallback(self): + with tempfile.TemporaryDirectory() as tmp: + parent = os.path.join(tmp, "580.10") + os.makedirs(parent) + fw = os.path.join(parent, "gsp-580.10.fwimage.bin") + open(fw, "wb").close() + self.assertEqual(extract_version(fw)[0], "580.10") + + def test_stem_fallback(self): + with tempfile.TemporaryDirectory() as tmp: + fw = os.path.join(tmp, "mystery-image.bin") + open(fw, "wb").close() + version, source = extract_version(fw) + self.assertEqual(version, "mystery-image") + self.assertEqual(source, "file stem") + + +class OrchestrationTests(unittest.TestCase): + def _options(self, tmp: str, **overrides) -> TriageOptions: + fw = os.path.join(tmp, "new.fwimage.bin") + if not os.path.exists(fw): + with open(fw, "wb") as fh: + fh.write(b"\x00" * 16) + values = dict(firmware=fw, out=os.path.join(tmp, "triage-out"), budget=1000) + values.update(overrides) + return TriageOptions(**values) + + def test_report_shape_and_artifacts(self): + with tempfile.TemporaryDirectory() as tmp: + session_dir = os.path.join(tmp, "session") + runners, calls = _fake_runners(session_dir=session_dir) + report = run_triage(self._options(tmp), runners=runners) + self.assertEqual(report["format"], FORMAT) + self.assertEqual([s["name"] for s in report["steps"]], [n for n, _ in DEFAULT_RUNNERS]) + self.assertTrue(all(s["status"] == "ok" for s in report["steps"])) + self.assertEqual(report["errors"], []) + self.assertTrue(report["sessions"]) + self.assertEqual(report["sessions"][0]["hashes"]["session.json"], "deadbeef") + self.assertEqual(report["support_readiness"]["verdict"], "sim-observable") + self.assertFalse(report["support_readiness"]["baseline_compared"]) + self.assertEqual(report["threshold_checks"]["required_steps_missing"], []) + self.assertTrue(report["threshold_checks"]["beats_manual_baseline"]) + for key in ("report_json", "report_md"): + self.assertTrue(os.path.isfile(report["artifacts"][key]), key) + with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh: + loaded = json.load(fh) + self.assertEqual(loaded["format"], FORMAT) + self.assertEqual(len(loaded["steps"]), len(DEFAULT_RUNNERS)) + with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh: + md = fh.read() + self.assertIn("New-version triage", md) + self.assertIn("Support readiness", md) + + def test_partial_failure_does_not_crash(self): + with tempfile.TemporaryDirectory() as tmp: + runners, _calls = _fake_runners(fail="probe") + report = run_triage(self._options(tmp), runners=runners) + self.assertEqual(report["errors"], ["probe"]) + step = next(s for s in report["steps"] if s["name"] == "probe") + self.assertEqual(step["status"], "error") + self.assertIn("injected failure", step["reason"]) + self.assertEqual( + next(s for s in report["steps"] if s["name"] == "native_boot")["status"], "ok" + ) + self.assertTrue(any("QEMU probe failed" in b for b in report["support_readiness"]["blockers"])) + self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) + + def test_missing_firmware_is_reported_not_raised(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + report = run_triage(TriageOptions(firmware=os.path.join(tmp, "nope.bin"), out=out)) + step = report["steps"][0] + self.assertEqual(step["name"], "identity") + self.assertEqual(step["status"], "error") + self.assertEqual(report["support_readiness"]["verdict"], "blocked") + self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) + + def test_budget_bounds(self): + with tempfile.TemporaryDirectory() as tmp: + options = self._options(tmp, budget=0) + with self.assertRaises(ValueError): + run_triage(options) + with self.assertRaises(ValueError): + run_triage(self._options(tmp, budget=3_000_000)) + + def test_plan_shape(self): + with tempfile.TemporaryDirectory() as tmp: + plan = build_plan(self._options(tmp)) + self.assertEqual(plan["format"], PLAN_FORMAT) + self.assertEqual([s["name"] for s in plan["steps"]], [n for n, _ in DEFAULT_RUNNERS]) + self.assertEqual(plan["side_effects"]["hardware"], "none") + + +class CliTests(unittest.TestCase): + def test_rm610_refusal(self): + with tempfile.TemporaryDirectory() as tmp: + fw = os.path.join(tmp, "gsp-rm610.fwimage.bin") + open(fw, "wb").close() + out = os.path.join(tmp, "out") + rc, _stdout, stderr = _run_main( + ["--firmware", fw, "--version", "RM-610", "--out", out] + ) + self.assertEqual(rc, 3) + self.assertIn("RM610", stderr) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_has_no_side_effects(self): + with tempfile.TemporaryDirectory() as tmp: + fw = os.path.join(tmp, "new.fwimage.bin") + open(fw, "wb").close() + out = os.path.join(tmp, "out") + rc, stdout, _stderr = _run_main(["--firmware", fw, "--out", out, "--dry-run"]) + self.assertEqual(rc, 0) + plan = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1]) + self.assertEqual(plan["format"], PLAN_FORMAT) + self.assertFalse(os.path.exists(out)) + + def test_missing_firmware_exits_2(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, _stdout, stderr = _run_main(["--firmware", os.path.join(tmp, "nope.bin"), "--out", out]) + self.assertEqual(rc, 2) + self.assertIn("firmware absent", stderr) + self.assertFalse(os.path.exists(out)) + + def test_bad_budget_is_a_parser_error(self): + with tempfile.TemporaryDirectory() as tmp: + fw = os.path.join(tmp, "new.fwimage.bin") + open(fw, "wb").close() + with self.assertRaises(SystemExit): + _run_main(["--firmware", fw, "--out", os.path.join(tmp, "out"), "--budget", "9999999"]) + + +@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "real 570.144 fwimage absent") +class RealImageIntegrationTests(unittest.TestCase): + def test_triage_real_image_tiny_budget(self): + if not qemu_available(): + self.skipTest("qemu-system-riscv64 not installed") + with tempfile.TemporaryDirectory() as tmp: + report = run_triage( + TriageOptions( + firmware=REAL_IMAGE, + out=os.path.join(tmp, "out"), + budget=3000, + qemu_duration=0.2, + step_timeout=120.0, + ) + ) + identity = next(s for s in report["steps"] if s["name"] == "identity") + self.assertEqual(identity["status"], "ok") + self.assertIn("570.144", identity["firmware"]["version"]) + probe = next(s for s in report["steps"] if s["name"] == "probe") + self.assertEqual(probe["status"], "ok") + native = next(s for s in report["steps"] if s["name"] == "native_boot") + self.assertIn(native["status"], ("ok", "skipped")) + self.assertIn( + report["support_readiness"]["verdict"], ("sim-observable", "first-touch-only", "blocked") + ) + version_diff = next(s for s in report["steps"] if s["name"] == "version_diff") + self.assertEqual(version_diff["status"], "skipped") + self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) + + +if __name__ == "__main__": + unittest.main() |