From cd1f2528e8243fed3ce80d5170ef93928e4e7284 Mon Sep 17 00:00:00 2001 From: Arthur Hanson Rasmusson Date: Tue, 6 Oct 2026 19:28:39 -0400 Subject: risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests rm_capture (chunked decode), launch_audit, state_seed/state_decode, valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk, msgq_attach, libos_port, log_decode, server_snapshot; doc index updates in the tool overview, operations pages and the repo README. --- debug/risc-v-gsp-rm-vm/tests/test_contracts.py | 171 ++++++ debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py | 243 ++++++++ debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py | 639 +++++++++++++++++++++ debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py | 477 +++++++++++++++ debug/risc-v-gsp-rm-vm/tests/test_libos_port.py | 291 ++++++++++ debug/risc-v-gsp-rm-vm/tests/test_log_decode.py | 335 +++++++++++ debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py | 342 +++++++++++ debug/risc-v-gsp-rm-vm/tests/test_rehost.py | 135 +++++ debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py | 409 +++++++++++++ .../risc-v-gsp-rm-vm/tests/test_server_snapshot.py | 364 ++++++++++++ debug/risc-v-gsp-rm-vm/tests/test_state_decode.py | 518 +++++++++++++++++ debug/risc-v-gsp-rm-vm/tests/test_state_seed.py | 311 ++++++++++ debug/risc-v-gsp-rm-vm/tests/test_valprobe.py | 200 +++++++ 13 files changed, 4435 insertions(+) create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_contracts.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_host_traffic.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_libos_port.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_log_decode.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_rehost.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_state_decode.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_state_seed.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_valprobe.py (limited to 'debug/risc-v-gsp-rm-vm/tests') diff --git a/debug/risc-v-gsp-rm-vm/tests/test_contracts.py b/debug/risc-v-gsp-rm-vm/tests/test_contracts.py new file mode 100644 index 0000000..3199b09 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_contracts.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""Tests for the wall-contract registry and runner (``harness/contracts.py``). + +The fast contracts execute the proven offline entry points (valprobe matrix, +GMMU golden vector, launch audit, rehost demo, RM-capture demo) and are +asserted to pass; prerequisites that are genuinely absent are reported as +skips, never as passes. No hardware, no module, no guest, no network. + +Run:: + + python3 -m unittest tests.test_contracts -v +""" +from __future__ import annotations + +import contextlib +import importlib +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, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness import run_contracts as runner # noqa: E402 +from harness.contracts import ( # noqa: E402 + FAST, + SLOW, + WALL_CLASSES, + Contract, + ContractSkip, + all_contracts, + fast_contracts, + run_contracts, + summarize, +) + +MISSING_PATH = os.path.join(TOOL_ROOT, "contract-test-missing-prerequisite") + + +def _entry_target(entry: str) -> tuple[str, str]: + module, _, rest = entry.partition(":") + return module, rest.split("(", 1)[0].strip() + + +def _raise(exc: Exception): + def _run(): + raise exc + + return _run + + +def _fake(name: str, **overrides) -> Contract: + fields = { + "name": name, + "wall_class": "a", + "doc": "docs/50", + "entry": "tests:fake", + "speed": FAST, + "expect": "test expectation", + "run": lambda: {"ok": True}, + "check": lambda report: (report.get("ok") is True, "ok"), + "requires": (), + } + fields.update(overrides) + return Contract(**fields) + + +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): + try: + rc = runner.main(argv) + except SystemExit as exc: + rc = int(exc.code or 0) + return rc, out.getvalue(), err.getvalue() + + +class RegistryShapeTests(unittest.TestCase): + def test_registry_shape(self): + contracts = all_contracts() + names = [contract.name for contract in contracts] + self.assertEqual(len(names), len(set(names)), "contract names must be unique") + self.assertGreaterEqual(len(contracts), 3, "keep at least the proven contracts") + for contract in contracts: + with self.subTest(contract=contract.name): + self.assertIn(contract.wall_class, WALL_CLASSES) + self.assertIn(contract.speed, (FAST, SLOW)) + self.assertTrue(contract.doc.startswith("docs/")) + self.assertTrue(contract.entry) + self.assertTrue(contract.expect) + self.assertTrue(fast_contracts(), "registry must have fast contracts") + + def test_entries_resolve_to_real_callables(self): + for contract in all_contracts(): + with self.subTest(contract=contract.name): + module_name, func_name = _entry_target(contract.entry) + module = importlib.import_module(module_name) + self.assertTrue(callable(getattr(module, func_name, None)), contract.entry) + + +class RunnerSemanticsTests(unittest.TestCase): + def test_classifies_pass_fail_skip(self): + results = run_contracts( + [ + _fake("good"), + _fake("raising", run=_raise(RuntimeError("boom"))), + _fake("missing-prereq", requires=(MISSING_PATH,)), + _fake("declared-skip", run=_raise(ContractSkip("not applicable here"))), + _fake("failed-check", check=lambda report: (False, "checker rejected")), + ] + ) + by_name = {result.name: result for result in results} + self.assertEqual(by_name["good"].status, "pass") + self.assertEqual(by_name["missing-prereq"].status, "skip") + self.assertEqual(by_name["declared-skip"].status, "skip") + self.assertEqual(by_name["raising"].status, "fail") + self.assertIn("boom", by_name["raising"].detail) + self.assertEqual(by_name["failed-check"].status, "fail") + + summary = summarize(results) + self.assertFalse(summary["ok"]) + self.assertEqual(summary["counts"], {"pass": 1, "fail": 2, "skip": 2}) + domains = {domain["wall_class"]: domain for domain in summary["domains"]} + self.assertEqual(set(domains), set(WALL_CLASSES)) + self.assertEqual(domains["a"]["total"], 5) + self.assertEqual(domains["d"]["total"], 0) + + def test_fast_contracts_pass(self): + contracts = fast_contracts() + self.assertGreaterEqual(len(contracts), 3) + results = run_contracts(contracts) + for result in results: + with self.subTest(contract=result.name): + self.assertNotEqual(result.status, "fail", result.detail) + if result.status == "skip": + self.skipTest(result.detail) + self.assertTrue(any(result.status == "pass" for result in results)) + + +class RunnerCliTests(unittest.TestCase): + def test_single_contract_json_summary(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "contracts.json") + rc, out, err = _run_main( + ["--fast", "--name", "gmmu-golden-vector", "--json", path] + ) + self.assertEqual(rc, 0, err) + self.assertIn("PASS", out) + with open(path, "r", encoding="utf-8") as fh: + summary = json.load(fh) + self.assertTrue(summary["ok"]) + self.assertEqual(summary["counts"]["pass"], 1) + self.assertEqual( + [contract["name"] for contract in summary["contracts"]], + ["gmmu-golden-vector"], + ) + self.assertEqual(len(summary["domains"]), len(WALL_CLASSES)) + + def test_unknown_contract_name_is_usage_error(self): + rc, _out, err = _run_main(["--fast", "--name", "no-such-contract"]) + self.assertEqual(rc, 2) + self.assertIn("unknown contract", err) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py new file mode 100644 index 0000000..f438068 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_gmmu_walk.py @@ -0,0 +1,243 @@ +"""Tests for the functional GMMU walker (sim/gmmu_walk.py, docs/48). + +The golden vector is a captured channel PDB (numeric entries); the synthetic +trees are built with the module's own encoders and must round-trip. No +hardware, no modules, no guest, no network. + +Run:: + + python3 -m unittest tests.test_gmmu_walk -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import struct +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.gmmu_walk import ( # noqa: E402 + FAULT_INVALID_APERTURE, + FAULT_PTE_INVALID, + FAULT_TABLE_MISSING, + FAULT_UNSUPPORTED, + FAULT_OUT_OF_RANGE, + WALK_FORMAT, + MemoryView, + WalkError, + channel_pdb_fixture, + encode_dual_slot, + encode_pde_pointer, + encode_pte, + main, + walk, +) + + +def _tree(root_pa: int, va: int, target_pa: int | None, *, base: int = 0x50000000, + read_only: bool = False) -> MemoryView: + i0 = (va >> 47) & 0x3 + i1 = (va >> 38) & 0x1FF + i2 = (va >> 29) & 0x1FF + i3 = (va >> 21) & 0xFF + i4 = (va >> 12) & 0x1FF + pd2, pd1, pd0, pt = base + 0x1000, base + 0x2000, base + 0x3000, base + 0x4000 + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into("> 47) & 0x3 + i1 = (va >> 38) & 0x1FF + i2 = (va >> 29) & 0x1FF + i3 = (va >> 21) & 0xFF + pd2, pd1, pd0 = 0x51001000, 0x51002000, 0x51003000 + page = bytearray(0x1000) + struct.pack_into("> 47) & 0x3 + i1 = (va >> 38) & 0x1FF + i2 = (va >> 29) & 0x1FF + i3 = (va >> 21) & 0xFF + pd2, pd1, pd0, pt = 0x52001000, 0x52002000, 0x52003000, 0x52004000 + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into("> 16) & 0x1FF) * 8, + encode_pte(pa, page_size="64k")) + mem.add(pt, bytes(page)) + report = walk(va, root_pa=0x52000000, mem=mem, page_mode="auto") + self.assertEqual(report["status"], "ok") + self.assertEqual(report["page_mode_used"], "64k") + self.assertEqual(report["evidence"], "modeled") + + +class FaultTests(unittest.TestCase): + def test_missing_table_page(self): + root_pa, mem = channel_pdb_fixture() + report = walk(0x100000, root_pa=root_pa, mem=MemoryView()) + self.assertEqual(report["status"], "fault") + self.assertEqual(report["fault"], FAULT_TABLE_MISSING) + + def test_invalid_leaf(self): + mem = _tree(0x60000000, 0x204608000, None) # leaf entry stays zero + report = walk(0x204608000, root_pa=0x60000000, mem=mem) + self.assertEqual(report["status"], "fault") + self.assertEqual(report["fault"], FAULT_PTE_INVALID) + + def test_invalid_pde_aperture(self): + root_pa = 0x60000000 + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into(" str: + payload = ( + (0xC2000006).to_bytes(4, "little") + + (0xABCD0080).to_bytes(4, "little") + + (0x2080012B).to_bytes(4, "little") + + bytes(4) + ) + return "\n".join( + [ + "[ 1.0] NVRM: GE-GOLDEN-BEGIN", + "[ 1.1] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 " + "hObject=0xabcd0080 size=0x8 paramp=0x0 ret=0x0", + "[ 1.1] NVRM: GE-CTRL 000:" + "".join(f" {b:02x}" for b in bytes(8)), + "[ 1.2] NVRM: GE-SEND fn=0x0000004c len=0x00000028 ret=0x0", + "[ 1.2] NVRM: GE-SEND 000:" + "".join(f" {b:02x}" for b in payload), + "[ 1.3] NVRM: GE-GOLDEN-END status=0x0", + ] + ) + "\n" + + +class FramingTests(unittest.TestCase): + def test_rpc_constants_match_capture_tool(self): + from harness.rm_capture import ( + GSP_RM_ALLOC as capture_alloc, + GSP_RM_CONTROL as capture_control, + RPC_HEADER_BYTES as capture_header_bytes, + ) + + self.assertEqual(RPC_HEADER_BYTES, capture_header_bytes) + self.assertEqual(GSP_RM_CONTROL, capture_control) + self.assertEqual(GSP_RM_ALLOC, capture_alloc) + self.assertEqual(RPC_HEADER_BYTES, 32) + + def test_rpc_message_roundtrip(self): + params = bytes(range(64)) + raw = build_rpc_message(RPC_FN_ECHO, params, sequence=7) + self.assertEqual(len(raw), RPC_HEADER_BYTES + len(params)) + message = parse_rpc_message(raw) + self.assertEqual(message.header_version, RPC_HEADER_VERSION) + self.assertEqual(message.signature, RPC_SIGNATURE) + self.assertEqual(message.function, RPC_FN_ECHO) + self.assertEqual(message.sequence, 7) + self.assertEqual(message.rpc_result, NV_OK) + self.assertEqual(message.length, len(raw)) + self.assertEqual(message.params, params) + self.assertEqual(message.to_bytes(), raw) + + def test_rpc_bad_signature_is_refused(self): + raw = bytearray(build_rpc_message(RPC_FN_ECHO, b"\x01", sequence=1)) + raw[4:8] = (0xDEADBEEF).to_bytes(4, "little") + with self.assertRaises(HostTrafficError): + parse_rpc_message(bytes(raw)) + + def test_rpc_short_and_overlong_length_is_refused(self): + raw = build_rpc_message(RPC_FN_ECHO, b"\x01\x02", sequence=1) + with self.assertRaises(HostTrafficError): + parse_rpc_message(raw[: RPC_HEADER_BYTES - 1]) + overlong = bytearray(raw) + overlong[8:12] = (len(raw) + 16).to_bytes(4, "little") + with self.assertRaises(HostTrafficError): + parse_rpc_message(bytes(overlong)) + + def test_queue_element_layout(self): + params = bytes(range(16)) + rpc = build_rpc_message(RPC_FN_ECHO, params, sequence=3) + element = build_queue_element(rpc, sequence=9) + self.assertEqual(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, 0x30) + self.assertEqual(element[0x30:0x34], rpc[:4]) + self.assertEqual(element[0x38:0x3C], len(rpc).to_bytes(4, "little")) + self.assertEqual(element[0x34:0x38], RPC_SIGNATURE.to_bytes(4, "little")) + self.assertEqual(element[0x24:0x28], (9).to_bytes(4, "little")) + self.assertEqual(element[0x28:0x2C], (1).to_bytes(4, "little")) + self.assertEqual(element[0x2C:0x30], bytes(4)) + self.assertEqual(element[0x50:], params) + prefix, parsed = parse_queue_element(element) + self.assertEqual(prefix["seq_num"], 9) + self.assertEqual(prefix["elem_count"], 1) + self.assertEqual(parsed.function, RPC_FN_ECHO) + self.assertEqual(parsed.params, params) + + def test_queue_element_checksum_verifies_and_detects_tamper(self): + rpc = build_rpc_message(RPC_FN_ECHO, bytes(range(32)), sequence=1) + element = build_queue_element(rpc, sequence=1) + self.assertEqual(compute_message_checksum(element), 0) + tampered = bytearray(element) + tampered[0x50] ^= 0x01 + self.assertNotEqual(compute_message_checksum(bytes(tampered)), 0) + with self.assertRaises(HostTrafficError): + parse_queue_element(bytes(tampered)) + + def test_multi_element_message_roundtrip(self): + params = bytes((i * 7) & 0xFF for i in range(5000)) + rpc = build_rpc_message(RPC_FN_ECHO, params, sequence=1) + element = build_queue_element(rpc, sequence=1) + self.assertEqual(element[0x28:0x2C], (2).to_bytes(4, "little")) + prefix, parsed = parse_queue_element(element) + self.assertEqual(prefix["elem_count"], 2) + self.assertEqual(parsed.params, params) + host = HostTraffic() + host.submit_rpc(RPC_FN_ECHO, params, sequence=1) + self.assertEqual(host.poll(), 1) + reply = host.read_reply() + self.assertEqual(reply["params"], params) + self.assertEqual(host.queues.status.header().write_ptr, 2) + host.close() + + +class MsgqTests(unittest.TestCase): + def test_queue_header_layout_matches_msgq_arithmetic(self): + mem = RegionMemory(SHARED_MEM_BASE, 0x40000) + queue = MsgqQueue(mem, SHARED_MEM_BASE, 0x40000) + queue.format() + header = queue.header() + self.assertEqual(header.version, 0) + self.assertEqual(header.size, 0x40000) + self.assertEqual(header.msg_size, GSP_MSG_QUEUE_ELEMENT_SIZE_MIN) + self.assertEqual(header.msg_count, (0x40000 - MSGQ_ENTRY_OFF) // 0x1000) + self.assertEqual(header.write_ptr, 0) + self.assertEqual(header.flags, MSGQ_FLAGS_SWAP_RX) + self.assertEqual(header.rx_hdr_off, MSGQ_RX_HDR_OFF) + self.assertEqual(header.entry_off, MSGQ_ENTRY_OFF) + self.assertEqual(MSGQ_RX_HDR_OFF, 32) + self.assertEqual(MSGQ_ENTRY_OFF, 48) + + def test_write_publish_and_read_back_entry(self): + mem = RegionMemory(SHARED_MEM_BASE, 0x40000) + queue = MsgqQueue(mem, SHARED_MEM_BASE, 0x40000) + queue.format() + payload = bytes(range(256)) + queue.write_entry(0, payload) + self.assertEqual(queue.read_entry(0, 256), payload) + queue.publish(1) + self.assertEqual(queue.write_ptr, 1) + queue.set_read(1) + self.assertEqual(queue.read_ptr, 1) + + def test_shared_queues_cross_link_rx_headers(self): + queues = SharedQueues(RegionMemory(SHARED_MEM_BASE, 0x80000)) + queues.format() + self.assertEqual(queues.command.rx_hdr_addr, queues.status.tx_base + MSGQ_RX_HDR_OFF) + self.assertEqual(queues.status.rx_hdr_addr, queues.command.tx_base + MSGQ_RX_HDR_OFF) + queues.command.set_read(5) + raw = queues.mem.read(queues.status.tx_base + MSGQ_RX_HDR_OFF, 4) + self.assertEqual(int.from_bytes(raw, "little"), 5) + queues.status.set_read(2) + raw = queues.mem.read(queues.command.tx_base + MSGQ_RX_HDR_OFF, 4) + self.assertEqual(int.from_bytes(raw, "little"), 2) + + def test_submit_and_take_roundtrip(self): + queues = SharedQueues(RegionMemory(SHARED_MEM_BASE, 0x80000)) + queues.format() + element = build_queue_element( + build_rpc_message(RPC_FN_ECHO, b"\xaa\xbb", sequence=4), sequence=0 + ) + first = queues.submit_element(element) + self.assertEqual(first["index"], 0) + self.assertEqual(queues.command.write_ptr, 1) + taken = queues.take_command() + self.assertIsNotNone(taken) + self.assertEqual(taken["element"], element) + self.assertEqual(queues.take_command(), None) + published = queues.publish_element(element) + self.assertEqual(published["index"], 0) + status = queues.take_status() + self.assertEqual(status["element"], element) + self.assertEqual(queues.take_status(), None) + + +class DoorbellTests(unittest.TestCase): + def test_model_records_writes(self): + host = HostTraffic() + host.doorbell.ring(0, 0) + host.doorbell.ring(0, 0x11) + host.doorbell.ring(2, 0x22) + self.assertEqual(host.doorbell.head(0), 0x11) + self.assertEqual(host.doorbell.head(2), 0x22) + self.assertEqual(host.doorbell.kicks, 3) + state = host.doorbell.state() + self.assertEqual(len(state["writes"]), 3) + self.assertEqual(state["writes"][-1]["queue"], 2) + with self.assertRaises(ValueError): + host.doorbell.ring(8, 0) + host.close() + + def test_block_serves_guest_read(self): + host = HostTraffic() + host.doorbell.ring(1, 0xCAFEF00D) + block = host.doorbell_block() + self.assertEqual(block.read(0, 4), 0) + self.assertEqual(block.read(8, 4), 0xCAFEF00D) + self.assertIsNone(block.read(8 * 8, 4)) + self.assertEqual(block.base, DOORBELL_BASE) + self.assertEqual(block.size, DOORBELL_SIZE) + host.close() + + def test_registry_overlay_beats_wider_falcon_window(self): + registry = DeviceRegistry() + registry.add(FalconBlock(), FALCON_BASE, 0x1000) + host = HostTraffic() + host.doorbell.ring(0, 0x1234) + registry.add(host.doorbell_block()) + handled, value = registry.read(DOORBELL_BASE, 4) + self.assertTrue(handled) + self.assertEqual(value, 0x1234) + handled, value = registry.read(FALCON_BASE + FALCON_DEBUGINFO, 4) + self.assertTrue(handled) + self.assertEqual(value, 0) + host.close() + + def test_guest_load_reads_doorbell(self): + machine = Machine(memory_mb=8, ram_base=0) + machine.load_segments([(0, PROGRAM_READ_DOORBELL)], entry=0) + host = HostTraffic() + host.doorbell.ring(0, 0x5A5A5A5A) + registry = DeviceRegistry() + registry.add(host.doorbell_block()) + machine.devices = registry + reason, _pc, _a, _b = machine.run(100) + self.assertEqual(reason, R_EBREAK) + self.assertEqual(machine.x[10], 0x5A5A5A5A) + host.close() + + def test_guest_store_is_recorded(self): + machine = Machine(memory_mb=8, ram_base=0) + machine.load_segments([(0, PROGRAM_WRITE_DOORBELL)], entry=0) + machine.apply_reset_regs({"a1": 0x77}) + host = HostTraffic() + registry = DeviceRegistry() + registry.add(host.doorbell_block()) + machine.devices = registry + reason, _pc, _a, _b = machine.run(100) + self.assertEqual(reason, R_EBREAK) + self.assertEqual(host.doorbell.head(2), 0x77) + host.close() + + +class TrafficTests(unittest.TestCase): + def test_submit_writes_ring_and_rings_doorbell(self): + host = HostTraffic() + raw = build_rpc_message(RPC_FN_ECHO, b"\x01\x02\x03", sequence=5) + result = host.submit(raw) + self.assertEqual(result["function"], RPC_FN_ECHO) + self.assertEqual(result["sequence"], 5) + self.assertEqual(result["index"], 0) + self.assertEqual(result["write_ptr"], 1) + self.assertEqual(host.queues.command.write_ptr, 1) + self.assertEqual(host.doorbell.kicks, 1) + self.assertEqual(host.doorbell.head(0), 0) + taken = host.queues.take_command() + self.assertIsNotNone(taken) + _prefix, parsed = parse_queue_element(taken["element"]) + self.assertEqual(parsed.params, b"\x01\x02\x03") + host.close() + + def test_poll_dispatches_echo(self): + host = HostTraffic() + host.submit_rpc(RPC_FN_ECHO, b"\xaa\xbb\xcc", sequence=1) + self.assertEqual(host.poll(), 1) + self.assertEqual(host.poll(), 0) + reply = host.read_reply() + self.assertIsNotNone(reply) + self.assertEqual(reply["function"], RPC_FN_ECHO) + self.assertEqual(reply["sequence"], 1) + self.assertEqual(reply["result"], NV_OK) + self.assertEqual(reply["params"], b"\xaa\xbb\xcc") + self.assertEqual(host.read_reply(), None) + host.close() + + def test_unknown_function_gets_not_supported(self): + host = HostTraffic() + host.submit_rpc(0x999, b"\x01", sequence=2) + self.assertEqual(host.poll(), 1) + reply = host.read_reply() + self.assertEqual(reply["function"], 0x999) + self.assertEqual(reply["result"], NV_ERR_NOT_SUPPORTED) + self.assertEqual(reply["params"], b"") + dispatched = [ + event for event in host.transcript()["events"] + if event["type"] == "dispatch" + ] + self.assertEqual(dispatched[0]["handler"], "stub:unsupported") + host.close() + + def test_status_query_reply_is_deterministic(self): + replies = [] + for _ in range(2): + host = HostTraffic() + host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1) + host.submit_rpc(RPC_FN_STATUS_QUERY, b"", sequence=2) + host.poll() + first = host.read_reply() + second = host.read_reply() + replies.append((first["params"], second["params"])) + host.close() + self.assertEqual(replies[0], replies[1]) + self.assertEqual(replies[0][1][:4], STATUS_MAGIC) + + def test_empty_poll_and_read_reply(self): + host = HostTraffic() + self.assertEqual(host.poll(), 0) + self.assertIsNone(host.read_reply()) + host.close() + + def test_run_cycles_is_bounded(self): + machine = Machine(memory_mb=32) + host = HostTraffic(machine=machine) + host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1) + result = host.run_cycles(64) + self.assertEqual(result["budget"], 64) + self.assertEqual(result["insns"], machine.insns) + self.assertEqual(result["dispatched"], 1) + self.assertEqual(result["replies"], 1) + with self.assertRaises(ValueError): + host.run_cycles(0) + with self.assertRaises(ValueError): + host.run_cycles(-1) + host.close() + + def test_oversized_message_is_refused(self): + host = HostTraffic() + with self.assertRaises(HostTrafficError): + host.submit_rpc(RPC_FN_ECHO, bytes(GSP_MSG_QUEUE_ELEMENT_SIZE_MIN * 16), + sequence=1) + host.close() + + def test_transcript_is_deterministic(self): + def run(): + host = HostTraffic() + host.submit_rpc(RPC_FN_ECHO, b"\x10\x20", sequence=1) + host.submit_rpc(RPC_FN_STATUS_QUERY, b"", sequence=2) + host.submit_rpc(0x999, b"", sequence=3) + host.poll() + while host.read_reply() is not None: + pass + transcript = host.transcript() + host.close() + return transcript + + first, second = run(), run() + self.assertEqual(first, second) + self.assertEqual(first["format"], "greeneye-host-traffic-log/1") + self.assertEqual( + [event["type"] for event in first["events"]], + ["submit", "submit", "submit", "dispatch", "dispatch", "dispatch", + "reply", "reply", "reply"], + ) + + +class ReplayTests(unittest.TestCase): + def test_replay_stimulus_in_order(self): + stimulus = { + "format": "greeneye-rm-rpc-stimulus/1", + "messages": [ + { + "seq": 0, + "kind": "send", + "fn": RPC_FN_ECHO, + "params_hex": "0102", + "truncated": False, + }, + { + "seq": 1, + "kind": "send", + "fn": RPC_FN_ECHO, + "params_hex": "0304", + "truncated": False, + }, + ], + } + host = HostTraffic() + report = replay_stimulus(host, stimulus) + self.assertEqual(report["format"], REPLAY_FORMAT) + self.assertEqual(report["submitted"], 2) + self.assertEqual(report["skipped"], []) + self.assertEqual([pair["reply"]["params"] for pair in report["pairs"]], + [b"\x01\x02", b"\x03\x04"]) + host.close() + + def test_replay_strict_rejects_truncated(self): + stimulus = { + "format": "greeneye-rm-rpc-stimulus/1", + "messages": [ + { + "seq": 0, + "kind": "control", + "params_hex": "00", + "truncated": True, + } + ], + } + host = HostTraffic() + with self.assertRaises(HostTrafficError): + replay_stimulus(host, stimulus, strict=True) + host.close() + + def test_replay_lenient_skips_truncated(self): + stimulus = { + "format": "greeneye-rm-rpc-stimulus/1", + "messages": [ + { + "seq": 0, + "kind": "send", + "fn": RPC_FN_ECHO, + "params_hex": "01", + "truncated": False, + }, + { + "seq": 1, + "kind": "control", + "params_hex": "00", + "truncated": True, + }, + ], + } + host = HostTraffic() + report = replay_stimulus(host, stimulus, strict=False) + self.assertEqual(report["submitted"], 1) + self.assertEqual(len(report["skipped"]), 1) + self.assertIn("truncated", report["skipped"][0]["reason"]) + host.close() + + def test_control_stimulus_maps_to_control_function(self): + stimulus = { + "format": "greeneye-rm-rpc-stimulus/1", + "messages": [ + { + "seq": 0, + "kind": "control", + "params_hex": "01020304", + "truncated": False, + } + ], + } + host = HostTraffic() + report = replay_stimulus(host, stimulus) + self.assertEqual(report["pairs"][0]["reply"]["function"], GSP_RM_CONTROL) + self.assertEqual(report["pairs"][0]["reply"]["result"], NV_ERR_NOT_SUPPORTED) + host.close() + + def test_load_stimulus_from_capture_document(self): + from harness.rm_capture import ( + capture_to_dict, + decode_text, + stimulus_from_capture, + ) + + capture = decode_text(_sample_capture_log(), source="fixture") + self.assertIsNotNone(stimulus_from_capture(capture)) + with tempfile.TemporaryDirectory() as tmp: + capture_path = os.path.join(tmp, "capture.json") + with open(capture_path, "w", encoding="utf-8") as fh: + json.dump(capture_to_dict(capture), fh) + loaded = load_stimulus(capture_path) + self.assertEqual(loaded["format"], "greeneye-rm-rpc-stimulus/1") + self.assertEqual(len(loaded["messages"]), 2) + + def test_load_stimulus_accepts_inline_document(self): + stimulus = {"format": "greeneye-rm-rpc-stimulus/1", "messages": []} + self.assertEqual(load_stimulus(stimulus), stimulus) + with self.assertRaises(HostTrafficError): + load_stimulus({"format": "not-a-stimulus"}) + + +class SessionTests(unittest.TestCase): + def test_recorder_events_and_blob(self): + with tempfile.TemporaryDirectory() as tmp: + recorder = Recorder(tmp, name="host-traffic-test", meta={"kind": "test"}) + host = HostTraffic(recorder=recorder) + host.submit_rpc(RPC_FN_ECHO, b"\x01", sequence=1) + host.poll() + host.read_reply() + host.close() + recorder.close() + events = load_events(tmp) + types = [event["type"] for event in events] + self.assertIn("host_traffic_config", types) + self.assertIn("doorbell", types) + self.assertIn("host_submit", types) + self.assertIn("host_dispatch", types) + self.assertIn("host_reply_read", types) + blob_path = os.path.join(tmp, "blobs", "host-traffic.json") + self.assertTrue(os.path.isfile(blob_path)) + with open(blob_path, "r", encoding="utf-8") as fh: + payload = json.load(fh) + self.assertEqual(payload["format"], "greeneye-host-traffic-log/1") + + +class DemoTests(unittest.TestCase): + def test_demo_end_to_end_and_deterministic(self): + first, second = run_demo(), run_demo() + self.assertTrue(first["ok"], first.get("checks")) + self.assertEqual(first["format"], DEMO_FORMAT) + self.assertEqual(first["replies"], second["replies"]) + self.assertEqual(first["transcript"], second["transcript"]) + self.assertEqual(first["control"]["bytes"], 512) + self.assertEqual(first["control"]["reply_result"], NV_ERR_NOT_SUPPORTED) + self.assertTrue(all(reply["read"] for reply in first["replies"])) + + def test_demo_writes_artifacts(self): + with tempfile.TemporaryDirectory() as tmp: + result = run_demo(out_dir=tmp) + self.assertTrue(result["ok"], result.get("checks")) + self.assertTrue(os.path.isfile(os.path.join(tmp, "blobs", "host-traffic.json"))) + self.assertTrue(os.path.isfile(os.path.join(tmp, "host-traffic-demo.json"))) + self.assertTrue(os.path.isfile(os.path.join(tmp, "session.json"))) + + def test_cli_demo(self): + import contextlib + import io + + with tempfile.TemporaryDirectory() as tmp: + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = main(["demo", "--out", tmp]) + self.assertEqual(code, 0) + payload = json.loads(out.getvalue()) + self.assertEqual(payload["format"], DEMO_FORMAT) + self.assertTrue(payload["ok"]) + + def test_cli_replay(self): + import contextlib + import io + + stimulus = { + "format": "greeneye-rm-rpc-stimulus/1", + "messages": [ + { + "seq": 0, + "kind": "send", + "fn": RPC_FN_ECHO, + "params_hex": "0102", + "truncated": False, + } + ], + } + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "stimulus.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(stimulus, fh) + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = main(["replay", "--stimulus", path, "--out", tmp]) + self.assertEqual(code, 0) + payload = json.loads(out.getvalue()) + self.assertEqual(payload["format"], REPLAY_FORMAT) + + def test_cli_requires_a_subcommand(self): + import contextlib + import io + + err = io.StringIO() + with contextlib.redirect_stderr(err): + code = main([]) + self.assertEqual(code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py b/debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py new file mode 100644 index 0000000..55dce3e --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_launch_audit.py @@ -0,0 +1,477 @@ +"""Tests for the launch memory-contract audit (harness/launch_audit.py, docs/48). + +Offline only: synthetic page-table trees plus the captured channel-PDB vector. + +Run:: + + python3 -m unittest tests.test_launch_audit -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import struct +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.launch_audit import ( # noqa: E402 + AUDIT_FORMAT, + CHECK_FORMAT, + SPEC_FORMAT, + VERDICT_ACCESS_DENIED, + VERDICT_OK, + VERDICT_PA_MISMATCH, + VERDICT_UNMAPPED, + AuditError, + _tree_for_va, + audit, + check, + main, + run_demo, +) +from sim.gmmu_walk import ( # noqa: E402 + MemoryView, + encode_dual_slot, + encode_pde_pointer, + encode_pte, +) + + +def _spec(root_pa: int, mem: MemoryView, contract: list, **extra) -> dict: + spec = { + "format": SPEC_FORMAT, + "name": "test", + "gmmu": "ver2", + "page_mode": "4k", + "pdb_pa": f"{root_pa:#x}", + "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _blobs(mem)], + "contract": contract, + } + spec.update(extra) + return spec + + +def _blobs(mem: MemoryView): + return [(int(item["pa"], 16), mem.read(int(item["pa"], 16), item["size"])) + for item in mem.spans()] + + +def _drop(mem: MemoryView, drop_pas) -> MemoryView: + """Copy a memory view without the given table pages (a partial dump).""" + out = MemoryView() + for pa, blob in _blobs(mem): + if pa not in drop_pas: + out.add(pa, blob) + return out + + +def _run_main(argv): + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main(argv) + return rc, out.getvalue(), err.getvalue() + + +class AuditTests(unittest.TestCase): + def test_shadow_maps_to_expected_pa_is_ok(self): + va, shadow = 0x204608000, 0x15E8FA000 + root = 0x60000000 + report = audit(_spec(root, _tree_for_va(root, va, shadow), [ + {"label": "qmd-dest", "va": f"{va:#x}", "access": "write", + "expect_pa": f"{shadow:#x}"}, + ])) + self.assertEqual(report["format"], AUDIT_FORMAT) + row = report["contract"][0] + self.assertEqual(row["verdict"], VERDICT_OK) + self.assertTrue(report["ok"]) + + def test_host_checks_guest_page_is_pa_mismatch(self): + va, shadow = 0x204608000, 0x15E8FA000 + root = 0x60000000 + report = audit(_spec(root, _tree_for_va(root, va, shadow), [ + {"label": "qmd-dest", "va": f"{va:#x}", "access": "write", + "expect_pa": f"{va:#x}"}, + ])) + row = report["contract"][0] + self.assertEqual(row["verdict"], VERDICT_PA_MISMATCH) + self.assertFalse(report["ok"]) + self.assertIn("0x15e8fa000", row["detail"]) + + def test_unmapped_leaf_is_unmapped(self): + va = 0x204608000 + root = 0x60000000 + report = audit(_spec(root, _tree_for_va(root, va, None), [ + {"label": "qmd-dest", "va": f"{va:#x}", "access": "write"}, + ])) + self.assertEqual(report["contract"][0]["verdict"], VERDICT_UNMAPPED) + self.assertFalse(report["ok"]) + + def test_read_only_leaf_for_write_is_access_denied(self): + va, target = 0x204608000, 0x15E8FA000 + # build the same tree as _tree_for_va but with a read-only leaf + i0, i1 = (va >> 47) & 0x3, (va >> 38) & 0x1FF + i2, i3 = (va >> 29) & 0x1FF, (va >> 21) & 0xFF + i4 = (va >> 12) & 0x1FF + base = 0x50000000 + pd2, pd1, pd0, pt = base + 0x1000, base + 0x2000, base + 0x3000, base + 0x4000 + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into("> 29) & 0x1FF + self.assertEqual(missing["pa"], f"{pde1:#x}") + self.assertEqual(missing["table_pa"], f"{pde1:#x}") + self.assertEqual(missing["level"], "pde1") + self.assertEqual(missing["depth"], 2) + self.assertEqual(missing["index"], index) + self.assertEqual(missing["entry_pa"], f"{pde1 + index * 8:#x}") + self.assertEqual([hop["level"] for hop in row["path"]], + ["root", "pde2", "pde1"]) + self.assertEqual(row["path"][0]["table_pa"], f"{root:#x}") + self.assertTrue(row["path"][0]["present"]) + self.assertFalse(row["path"][-1]["present"]) + self.assertEqual(report["needed_pages"][0]["pa"], f"{pde1:#x}") + self.assertEqual(report["needed_pages"][0]["legs"], ["qmd"]) + + def test_check_partial_memory_marks_complete_and_missing_legs(self): + root = 0x60000000 + base = 0x51000000 + pd2, pd1 = base + 0x1000, base + 0x2000 + pd0_done, pt_done = base + 0x3000, base + 0x4000 + pd0_missing = base + 0x5000 + target = 0x12345000 + va_done, va_miss = 0x100000, 0x204608000 + + def idx(va): + return ((va >> 47) & 3, (va >> 38) & 0x1FF, (va >> 29) & 0x1FF, + (va >> 21) & 0xFF, (va >> 12) & 0x1FF) + + i0d, i1d, i2d, i3d, i4d = idx(va_done) + i0m, i1m, i2m, _i3m, _i4m = idx(va_miss) + self.assertEqual((i0d, i1d), (i0m, i1m)) # share root[i0] and pd2[i1] + self.assertNotEqual(i2d, i2m) + + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into("> 47) & 3, 0) + self.assertEqual((va_a2 >> 47) & 3, 0) + self.assertEqual((va_b >> 47) & 3, 1) + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into(" pte_invalid + report = check(_spec(root, mem, [ + {"label": "qmd", "va": f"{va:#x}", "access": "write"}, + ])) + row = report["contract"][0] + self.assertEqual(row["status"], "fault") + self.assertEqual(row["fault"], "pte_invalid") + self.assertIsNone(row["first_missing"]) + self.assertEqual(report["needed_pages"], []) + self.assertTrue(report["complete"]) + + def test_check_64k_dual_missing_big_table(self): + va = 0x204600000 + root, pd2, pd1, pd0 = 0x60000000, 0x50001000, 0x50002000, 0x50003000 + big_pt, small_pt = 0x80000000, 0x80001000 + i0, i1, i2 = (va >> 47) & 3, (va >> 38) & 0x1FF, (va >> 29) & 0x1FF + i3 = (va >> 21) & 0xFF + mem = MemoryView() + page = bytearray(0x1000) + struct.pack_into("> 16) & 0x1FF) + + def test_check_is_deterministic(self): + va = 0x204608000 + root = 0x60000000 + full = _tree_for_va(root, va, 0x12345000) + partial = _drop(full, {0x50002000}) + spec = _spec(root, partial, [ + {"label": "qmd", "va": f"{va:#x}", "access": "write"}, + ]) + self.assertEqual(json.dumps(check(spec), sort_keys=True), + json.dumps(check(spec), sort_keys=True)) + + def test_check_spec_errors(self): + with self.assertRaises(AuditError): + check({"format": SPEC_FORMAT}) + with self.assertRaises(AuditError): + check({"format": "nope", "pdb_pa": "0x1000", "contract": []}) + with self.assertRaises(AuditError): + check(_spec(0x60000000, MemoryView(), [], gmmu="ver9")) + with self.assertRaises(AuditError): + check(_spec(0x60000000, MemoryView(), [{"label": "x"}])) + + def test_cli_check_exit_codes_and_artifact(self): + va = 0x204608000 + root = 0x60000000 + full = _tree_for_va(root, va, 0x12345000) + partial = _drop(full, {0x50002000}) + spec = _spec(root, partial, [ + {"label": "qmd", "va": f"{va:#x}", "access": "write"}, + ]) + with tempfile.TemporaryDirectory() as tmp: + spec_path = os.path.join(tmp, "spec.json") + with open(spec_path, "w", encoding="utf-8") as fh: + json.dump(spec, fh) + rc, out, _ = _run_main(["check", "--spec", spec_path, "--out", tmp]) + self.assertEqual(rc, 3) + self.assertIn("need 0x50002000", out) + self.assertIn("dump next:", out) + report_path = os.path.join(tmp, "launch-audit-check.json") + self.assertTrue(os.path.isfile(report_path)) + with open(report_path, encoding="utf-8") as fh: + report = json.load(fh) + self.assertEqual(report["format"], CHECK_FORMAT) + self.assertEqual(report["needed_pages"][0]["pa"], "0x50002000") + # a complete view -> exit 0 + spec["memory"] = [{"pa": f"{pa:#x}", "hex": b.hex()} + for pa, b in _blobs(full)] + with open(spec_path, "w", encoding="utf-8") as fh: + json.dump(spec, fh) + rc, out, _ = _run_main(["check", "--spec", spec_path]) + self.assertEqual(rc, 0) + self.assertIn("complete", out) + + +class DemoAndCliTests(unittest.TestCase): + def test_demo_is_green_and_deterministic(self): + first = run_demo() + second = run_demo() + self.assertEqual(first["format"], "greeneye-launch-audit-demo/1") + self.assertTrue(first["ok"]) + self.assertEqual(json.dumps(first, sort_keys=True), + json.dumps(second, sort_keys=True)) + + def test_audit_is_deterministic(self): + va, shadow = 0x204608000, 0x15E8FA000 + root = 0x60000000 + spec = _spec(root, _tree_for_va(root, va, shadow), [ + {"label": "qmd", "va": f"{va:#x}", "access": "write"}, + ]) + self.assertEqual(json.dumps(audit(spec), sort_keys=True), + json.dumps(audit(spec), sort_keys=True)) + + def test_cli_demo_and_hints(self): + rc, out, _ = _run_main(["demo"]) + self.assertEqual(rc, 0) + self.assertIn("captured channel contract: ok=True", out) + self.assertIn("engine-write-shadow-host-checks-guest", out) + rc, out, _ = _run_main(["capture-hints"]) + self.assertEqual(rc, 0) + self.assertIn("CAPTURE HINTS", out) + + def test_engine_write_shadow_demo_verdicts(self): + demo = run_demo() + scenarios = {s["name"]: s for s in demo["engine_write_shadow"]} + mismatch = scenarios["engine-write-shadow-host-checks-guest"] + aligned = scenarios["engine-write-shadow-bookkeeping-aligned"] + self.assertEqual(mismatch["expected_verdict"], VERDICT_PA_MISMATCH) + self.assertEqual( + mismatch["report"]["contract"][0]["verdict"], VERDICT_PA_MISMATCH) + self.assertEqual(aligned["expected_verdict"], VERDICT_OK) + self.assertEqual( + aligned["report"]["contract"][0]["verdict"], VERDICT_OK) + self.assertTrue(demo["ok"]) + # the original demo sections are still present and unchanged in shape + self.assertTrue(demo["captured_channel"]["ok"]) + self.assertEqual(len(demo["qmd_hypotheses"]), 3) + + def test_cli_run_writes_report_and_exit_codes(self): + va, shadow = 0x204608000, 0x15E8FA000 + root = 0x60000000 + spec = _spec(root, _tree_for_va(root, va, shadow), [ + {"label": "qmd", "va": f"{va:#x}", "access": "write"}, + ]) + with tempfile.TemporaryDirectory() as tmp: + spec_path = os.path.join(tmp, "spec.json") + with open(spec_path, "w", encoding="utf-8") as fh: + json.dump(spec, fh) + rc, out, _ = _run_main(["run", "--spec", spec_path, "--out", tmp]) + self.assertEqual(rc, 0) + self.assertTrue(os.path.isfile(os.path.join(tmp, "launch-audit.json"))) + # mismatch -> exit 3 + spec["contract"][0]["expect_pa"] = f"{va:#x}" + with open(spec_path, "w", encoding="utf-8") as fh: + json.dump(spec, fh) + rc, _, _ = _run_main(["run", "--spec", spec_path]) + self.assertEqual(rc, 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py new file mode 100644 index 0000000..0f01c01 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py @@ -0,0 +1,291 @@ +"""libos port receive-path model tests (sim/libos_port.py, docs/51 section 7). + +Hermetic: a fake machine (register file only) and a fake base syscall model; +no image, no boot, no machine. The tests pin the port-syscall ABI observed on +the real 570.144 image (send args a0-a3, receive args a4-a7, wait id t1, +timeout t2, return a0/a4/a5/a6), the async receive post, the boot handshake +send, the host doorbell wake's external-interrupt handoff, and the strict +"never complete an uncompleted receive" rule. + +Offline only: no hardware, kernel module, mdev, guest, network or QEMU. + +Run:: + + cd open-iov/debug/risc-v-gsp-rm-vm + python3 -m unittest tests.test_libos_port -v +""" +from __future__ import annotations + +import json +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.libos import INTERRUPT_ARM_EXTERNAL, TASK_PORT_SEND_RECV_WAIT # noqa: E402 +from sim.libos_port import ( # noqa: E402 + EXTINTR_WAKE_CONTRACT, + HANDSHAKE_CONTRACT, + REPORT_FORMAT, + SHUTTLE_ANY, + TIMEOUT_INFINITE, + LibosPortModel, + role_for, +) + + +class FakeMachine: + """Register file only: the port model never touches memory.""" + + def __init__(self): + self.x = [0] * 32 + self.insns = 0 + self.pc = 0 + + def set(self, index: int, value: int) -> None: + self.x[index] = value & 0xFFFFFFFFFFFFFFFF + + +class FakeBase: + """Minimal base syscall model: handles arm, records and stops the rest.""" + + name = "fake-libos" + + def __init__(self): + self.calls: list[int] = [] + self.armed_mask = 0 + self.stopped: dict | None = None + + def handle(self, machine, pc: int = 0): + number = int(machine.x[5]) + if number == INTERRUPT_ARM_EXTERNAL: + self.armed_mask |= int(machine.x[10]) + return True, 0 + self.calls.append(number) + self.stopped = {"num": number, "name": f"sys_{number}", "pc": int(pc)} + return False, None + + +def _port_call(machine, *, send=None, recv=None, wait=None, + timeout=TIMEOUT_INFINITE) -> None: + """Write one ``TASK_PORT_SEND_RECV_WAIT`` argument set (explicit fields).""" + machine.set(5, TASK_PORT_SEND_RECV_WAIT) # t0 + for index, value in enumerate(send or (0, 0, 0, 0), start=10): + machine.set(index, value) + for index, value in enumerate(recv or (0, 0, 0, 0), start=14): + machine.set(index, value) + machine.set(6, 0 if wait is None else wait) # t1 + machine.set(7, timeout) # t2 + + +class PortSurfaceTests(unittest.TestCase): + """Role table and argument contract.""" + + def test_observed_roles(self): + self.assertEqual(role_for(6), "extintr") + self.assertEqual(role_for(7), "notification") + self.assertEqual(role_for(8), "request") + self.assertEqual(role_for(12), "handshake") + self.assertEqual(role_for(13), "event") + self.assertEqual(role_for(99), "data") + + def test_async_post_keeps_receive_args_and_returns_ok(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + _port_call(machine, recv=(10, 6, 0, 0)) + handled, result = model.handle(machine, pc=0x4018038) + self.assertTrue(handled) + self.assertEqual(result, 0) + self.assertEqual(len(model.posts), 1) + self.assertEqual(model.posts[0]["shuttle"], 10) + self.assertEqual(model.posts[0]["role"], "extintr") + # The kernel returns LIBOS_OK through a0; the wrapper masks a0 only. + self.assertEqual(machine.x[14], 10) + self.assertEqual(machine.x[15], 6) + + def test_send_completes_with_the_transcribe_contract(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + _port_call(machine, send=(1, 12, 0, 0, ), wait=1) + handled, result = model.handle(machine, pc=0x4019CD4) + self.assertTrue(handled) + self.assertEqual(result, 0) + self.assertEqual(len(model.sends), 1) + self.assertEqual(model.sends[0]["contract"], HANDSHAKE_CONTRACT) + self.assertEqual(model.sends[0]["role"], "handshake") + self.assertEqual(machine.x[14], 0) # completed size + self.assertEqual(machine.x[15], 0) # originating task + self.assertEqual(machine.x[16], 1) # completed shuttle + self.assertEqual(machine.x[17], 0) + + def test_blocking_wait_stops_without_completion(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + _port_call(machine, recv=(9, 8, 0x4015070, 0x808)) + model.handle(machine, pc=0x4018038) + _port_call(machine, wait=SHUTTLE_ANY) + handled, result = model.handle(machine, pc=0x4019DF4) + self.assertFalse(handled) + self.assertIsNone(result) + self.assertIsNotNone(model.stopped) + self.assertEqual(model.stopped["reason"], "port-wait-blocks") + self.assertEqual(model.stopped["num"], TASK_PORT_SEND_RECV_WAIT) + self.assertEqual(model.waits[-1]["reason"], "port-wait-blocks") + + def test_unmodelled_syscall_is_recorded_and_stopped(self): + base = FakeBase() + model = LibosPortModel(base=base) + machine = FakeMachine() + machine.set(5, 6) # task_state_query + machine.set(10, 0) + handled, result = model.handle(machine, pc=0x401A46C) + self.assertFalse(handled) + self.assertIsNone(result) + self.assertEqual(model.stopped["num"], 6) + self.assertEqual(model.stopped["reason"], "unmodelled-syscall") + self.assertEqual(base.calls, [6]) + + +class DoorbellWakeTests(unittest.TestCase): + """The host wake's external-interrupt handoff (RTOS report contract).""" + + def test_kick_before_post_and_arm_is_delivered_at_post(self): + base = FakeBase() + model = LibosPortModel(base=base) + machine = FakeMachine() + event = model.host_kick(queue=0, value=0) + self.assertIsNone(event["delivered"]) + self.assertTrue(event["pending"]) + # Arm first, then post the extintr receive: the pending wake fills it. + machine.set(5, INTERRUPT_ARM_EXTERNAL) + machine.set(10, 0xFF) + model.handle(machine, pc=0x4018054) + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018038) + self.assertEqual(model.kicks_pending, 0) + self.assertEqual([entry["shuttle"] for entry in model.completed], [10]) + self.assertEqual(model.completed[0]["payload_size"], 0) + self.assertEqual(model.completed[0]["origin_task"], 0) + self.assertEqual(model.completed[0]["contract"], EXTINTR_WAKE_CONTRACT) + + def test_wake_requires_an_arm(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018038) + event = model.host_kick() + self.assertIsNone(event["delivered"]) + self.assertEqual(model.kicks_pending, 1) + self.assertEqual(model.completed, []) + # Without interrupts armed the wait still blocks (no fabricated wake). + _port_call(machine, wait=SHUTTLE_ANY) + handled, _ = model.handle(machine, pc=0x4019DF4) + self.assertFalse(handled) + + def test_arm_seeds_from_the_attached_base_model(self): + base = FakeBase() + base.armed_mask = 0xFF + model = LibosPortModel(base=base) + self.assertEqual(model.armed_mask, 0xFF) + machine = FakeMachine() + model.host_kick() + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018038) + self.assertEqual(model.kicks_pending, 0) + self.assertEqual([entry["shuttle"] for entry in model.completed], [10]) + + def test_wait_returns_the_wake_with_the_argument_contract(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + model.host_kick() + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018038) + machine.set(5, INTERRUPT_ARM_EXTERNAL) + machine.set(10, 0xFF) + model.handle(machine, pc=0x4018054) + _port_call(machine, wait=SHUTTLE_ANY) + handled, result = model.handle(machine, pc=0x4019DF4) + self.assertTrue(handled) + self.assertEqual(result, 0) + self.assertEqual(machine.x[14], 0) # completed size + self.assertEqual(machine.x[15], 0) # originating task + self.assertEqual(machine.x[16], 10) # completed shuttle + self.assertEqual(model.waits[-1]["completed"], 10) + self.assertEqual(model.waits[-1]["role"], "extintr") + + def test_one_wake_completes_one_receive_only(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + model.host_kick() + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018038) + machine.set(5, INTERRUPT_ARM_EXTERNAL) + machine.set(10, 0xFF) + model.handle(machine, pc=0x4018054) + self.assertEqual(len(model.completed), 1) + # Consume the wake, re-post and wait again: nothing completes. + _port_call(machine, wait=SHUTTLE_ANY) + handled, _ = model.handle(machine, pc=0x4019DF4) + self.assertTrue(handled) + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018C8) + _port_call(machine, wait=SHUTTLE_ANY) + handled, _ = model.handle(machine, pc=0x4019DF4) + self.assertFalse(handled) + self.assertEqual(model.completed, []) + + +class StateTests(unittest.TestCase): + """Deterministic, serialisable report state.""" + + def _run(self) -> dict: + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + model.host_kick(queue=0, value=0) + _port_call(machine, recv=(10, 6, 0, 0)) + model.handle(machine, pc=0x4018038) + machine.set(5, INTERRUPT_ARM_EXTERNAL) + machine.set(10, 0xFF) + model.handle(machine, pc=0x4018054) + _port_call(machine, recv=(9, 8, 0x4015070, 0x808)) + model.handle(machine, pc=0x4018038) + _port_call(machine, wait=SHUTTLE_ANY) + model.handle(machine, pc=0x4019DF4) + return model.state() + + def test_state_is_byte_identical_across_runs(self): + first = self._run() + second = self._run() + self.assertEqual(first["format"], REPORT_FORMAT) + self.assertEqual(json.dumps(first, sort_keys=True), + json.dumps(second, sort_keys=True)) + self.assertEqual(first["armed_mask"], 0xFF) + self.assertEqual(first["host_kicks"], 1) + self.assertEqual(first["kicks_pending"], 0) + self.assertEqual([post["role"] for post in first["posts"]], + ["extintr", "request"]) + self.assertEqual(first["completed"], []) + self.assertEqual(first["waits"][-1]["completed"], 10) + + def test_state_has_no_host_timestamps(self): + state = self._run() + + def keys(value): + if isinstance(value, dict): + for key, item in value.items(): + yield key + yield from keys(item) + elif isinstance(value, list): + for item in value: + yield from keys(item) + + for key in keys(state): + self.assertNotIn(str(key), ("t", "ts", "timestamp", "time")) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py b/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py new file mode 100644 index 0000000..b43dc00 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py @@ -0,0 +1,335 @@ +"""Tests for the offline GSP-RM log decoder (harness/log_decode.py). + +Crafted inputs cover both real dump forms: the raw ring hexdump +(``LOGINIT-raw:`` + rows) and the dmesg/evidence record lines +(``rec seq=.. recType=.. errorCode=.. source=..`` plus ``assertCallStack`` +frames, ``entry[k]:`` aggregates and the liblogdecode ``rec NNN ts=..`` +text form). A real-file test runs only when the campaign dump is present. + +Run:: + + cd TOOL_ROOT && python3 -m unittest tests.test_log_decode -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, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.log_decode import ( # noqa: E402 + SCHEMA, + LogDecodeError, + apply_filters, + decode_path, + decode_text, + load_symbols, + main, + run_demo, +) + +REAL_LOGDUMP = ( + "/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/" + "evidence-20261002-gr/21-logdump-grctx4.txt" +) + +RAW_TEXT = """\ +vgsp-boot: libos task log buffers (read-only) +LOGINIT-raw: 64 bytes (allocated 64) VA=0xffff0000 dma=0x1000 +00000000: 00000005 00000000 0000000a deadbeef +00000010: 01020304 05060708 090a0b0c 0d0e0f10 +""" + +RECORDS_TEXT = """\ +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine="" +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000,0x0] (offsets from load base 0x4000000) +[Tue Oct 6 12:40:53 2026] vgsp-nocat: entry[0]: count=1 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=1 source="ASSERT" +an unmatched filler line +rec 7 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x401000,0x402000 +rec 9 ts=0x18dbb29118ec7280 meta=0x0000000005bb9bd0 args=[0x10, 0x20] +""" + +SYMBOLS = { + "symbols": [ + {"name": "gr_init", "addr": "0x401000"}, + {"name": "gr_done", "addr": 0x403000}, + ] +} + + +def _write(tmp: str, name: str, text: str) -> str: + path = os.path.join(tmp, name) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def _records(report: dict, kind: str = "record") -> list[dict]: + return [r for r in report["records"] if r["kind"] == kind] + + +class RawHexTests(unittest.TestCase): + def test_header_and_rows_decode_to_words(self): + report = decode_text(RAW_TEXT, source="raw.txt") + self.assertEqual(report["schema"], SCHEMA) + self.assertEqual(report["source"], "raw.txt") + self.assertEqual(report["count"], 1) + self.assertEqual(report["unparsed"], 1) # the intro line is not a record + raw = report["records"][0] + self.assertEqual(raw["kind"], "raw-hex") + self.assertEqual(raw["label"], "LOGINIT") + self.assertEqual(raw["source_name"], "LOGINIT") + self.assertEqual(len(raw["frames"]), 8) + self.assertEqual(raw["frames"][0], "0x5") + self.assertEqual(raw["frames"][3], "0xdeadbeef") + self.assertEqual(raw["frames"][7], "0xd0e0f10") + # raw is the decoded little-endian byte buffer of the rows + self.assertEqual(len(raw["raw"]), 2 * 8 * 8) + self.assertEqual(bytes.fromhex(raw["raw"])[:8].hex(), "0500000000000000") + self.assertEqual(raw["raw"].endswith("100f0e0d00000000"), True) + + def test_two_sections_stay_separate(self): + text = RAW_TEXT + ( + "LOGRM-raw: 16 bytes (allocated 16) VA=0x1 dma=0x2\n" + "00000000: 11111111 22222222 33333333 44444444\n" + ) + report = decode_text(text) + raw = _records(report, "raw-hex") + self.assertEqual(len(raw), 2) + self.assertEqual(raw[1]["label"], "LOGRM") + self.assertEqual(raw[1]["frames"], ["0x11111111", "0x22222222", + "0x33333333", "0x44444444"]) + + def test_blank_and_comment_lines_are_structure(self): + report = decode_text("\n# LOGINIT\n" + RAW_TEXT) + self.assertEqual(report["unparsed"], 1) # only the intro line + + +class RecordLineTests(unittest.TestCase): + def setUp(self): + self.report = decode_text(RECORDS_TEXT) + self.rec = self.report["records"][0] + + def test_rec_line_fields(self): + self.assertEqual(self.rec["kind"], "record") + self.assertEqual(self.rec["label"], "vgsp-nocat") + self.assertEqual(self.rec["seq"], 1) + self.assertEqual(self.rec["rec_type"], 5) + self.assertEqual(self.rec["error_code"], "0x57f1344") + self.assertEqual(self.rec["source_name"], "ASSERT") + + def test_assert_callstack_merges_frames(self): + self.assertEqual(self.rec["frames"], ["0x401000", "0x402000", "0x0"]) + # no duplicate standalone continuation record was emitted + self.assertEqual(len(_records(self.report)), 4) + + def test_entry_aggregate_line(self): + entry = self.report["records"][1] + self.assertEqual(entry["rec_type"], 5) + self.assertEqual(entry["error_code"], "0x57f1344") + self.assertEqual(entry["source_name"], "ASSERT") + self.assertIn("entry[0]:", entry["raw"]) + self.assertIn("vgsp-nocat:", entry["raw"]) + + def test_decoded_rec_shapes_with_and_without_brackets(self): + records = _records(self.report) + plain, bracketed = records[2], records[3] + self.assertEqual(plain["seq"], 7) + self.assertEqual(plain["frames"], ["0x401000", "0x402000"]) + self.assertEqual(bracketed["seq"], 9) + self.assertEqual(bracketed["frames"], ["0x10", "0x20"]) + + def test_unparsed_accounting(self): + self.assertEqual(self.report["unparsed"], 1) + self.assertEqual(len(self.report["records"]), 4) + + def test_histogram(self): + hist = self.report["histogram"] + self.assertEqual(hist["rec_type"], {"5": 2}) + self.assertEqual(hist["error_code"], {"0x57f1344": 2}) + self.assertEqual(hist["source"], {"ASSERT": 2}) + + def test_standalone_continuation_becomes_record(self): + text = ("vgsp-nocat: rec seq=4 assertCallStack count=1 " + "[0x1000,0x2000] (offsets from load base 0x4000000)\n") + report = decode_text(text) + self.assertEqual(report["count"], 1) + self.assertEqual(report["records"][0]["seq"], 4) + self.assertEqual(report["records"][0]["frames"], ["0x1000", "0x2000"]) + + +class SymbolTests(unittest.TestCase): + def test_nearest_below_with_offset(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "symbols.json", json.dumps(SYMBOLS)) + table = load_symbols(path) + self.assertEqual(table.resolve(0x401000), ("gr_init", 0x0)) + self.assertEqual(table.resolve(0x402fff), ("gr_init", 0x1FFF)) + self.assertEqual(table.resolve(0x403000), ("gr_done", 0x0)) + self.assertIsNone(table.resolve(0x400fff)) # below the first symbol + self.assertIsNone(table.resolve(0x0)) + + def test_top_level_list_shape_accepted(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "symbols.json", + json.dumps([{"name": "a", "addr": "0x10"}])) + table = load_symbols(path) + self.assertEqual(table.resolve(0x1f), ("a", 0xF)) + + def test_resolved_entries_align_with_frames(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "symbols.json", json.dumps(SYMBOLS)) + report = decode_text(RECORDS_TEXT, symbols=load_symbols(path)) + resolved = report["records"][0]["resolved"] + self.assertEqual(resolved[0], + {"frame": "0x401000", "symbol": "gr_init", + "offset": "0x0"}) + self.assertEqual(resolved[1], + {"frame": "0x402000", "symbol": "gr_init", + "offset": "0x1000"}) + self.assertEqual(resolved[2], + {"frame": "0x0", "symbol": None, "offset": None}) + + def test_malformed_symbols_fail_cleanly(self): + cases = [ + "{not json", + json.dumps([{"name": "no-addr"}]), + json.dumps({"symbols": "nope"}), + json.dumps({"nope": []}), + json.dumps([42]), + json.dumps([{"name": "x", "addr": "zzz"}]), + ] + with tempfile.TemporaryDirectory() as tmp: + for index, payload in enumerate(cases): + path = _write(tmp, f"bad{index}.json", payload) + with self.assertRaises(LogDecodeError): + load_symbols(path) + self.assertRaises(LogDecodeError, load_symbols, + os.path.join(tmp, "missing.json")) + # the CLI exits non-zero with a clear message, not a traceback + good = _write(tmp, "good.txt", RECORDS_TEXT) + bad = _write(tmp, "bad.json", cases[0]) + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + code = main(["decode", good, "--symbols", bad]) + self.assertEqual(code, 2) + self.assertIn("symbols JSON", stderr.getvalue()) + + +class FilterTests(unittest.TestCase): + def setUp(self): + self.report = decode_text(RECORDS_TEXT) + self.raw_report = decode_text(RAW_TEXT) + + def test_kind_filter(self): + self.assertEqual(apply_filters(self.report, kind="record")["count"], 4) + self.assertEqual(apply_filters(self.report, kind="raw-hex")["count"], 0) + self.assertEqual(apply_filters(self.raw_report, kind="raw-hex")["count"], 1) + + def test_rec_type_filter_accepts_hex_and_decimal(self): + for value in ("5", "0x5"): + filtered = apply_filters(self.report, rec_type=value) + self.assertEqual(filtered["count"], 2) + self.assertTrue(all(r["rec_type"] == 5 for r in filtered["records"])) + + def test_error_code_filter(self): + filtered = apply_filters(self.report, error_code="0x57f1344") + self.assertEqual(filtered["count"], 2) + zeros = apply_filters(self.report, error_code="0x1") + self.assertEqual(zeros["count"], 0) + + def test_source_filter_is_case_insensitive_substring(self): + self.assertEqual(apply_filters(self.report, source="assert")["count"], 2) + self.assertEqual(apply_filters(self.report, source="Display")["count"], 0) + + def test_limit_and_combined_filters(self): + self.assertEqual(apply_filters(self.report, limit=1)["count"], 1) + combined = apply_filters(self.report, kind="record", source="ASSERT", + limit=1) + self.assertEqual(combined["count"], 1) + self.assertEqual(combined["records"][0]["seq"], 1) + self.assertEqual(apply_filters(self.report, limit=0)["count"], 0) + + def test_bad_filter_value_raises(self): + with self.assertRaises(LogDecodeError): + apply_filters(self.report, rec_type="zzz") + with self.assertRaises(LogDecodeError): + apply_filters(self.report, kind="bogus") + with self.assertRaises(LogDecodeError): + apply_filters(self.report, limit=-1) + + def test_histogram_follows_filter(self): + filtered = apply_filters(self.report, kind="record", limit=1) + self.assertEqual(filtered["histogram"]["rec_type"], {"5": 1}) + self.assertEqual(filtered["unparsed"], self.report["unparsed"]) + + +class CliTests(unittest.TestCase): + def test_decode_writes_json_report(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "records.txt", RECORDS_TEXT) + out = os.path.join(tmp, "out", "report.json") + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + code = main(["decode", path, "--json", out, + "--source", "ASSERT", "--limit", "1"]) + self.assertEqual(code, 0) + with open(out, "r", encoding="utf-8") as fh: + report = json.load(fh) + self.assertEqual(report["schema"], SCHEMA) + self.assertEqual(report["source"], "records.txt") + self.assertEqual(report["count"], 1) + self.assertIn("report_json", stdout.getvalue()) + + def test_missing_input_fails_cleanly(self): + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + code = main(["decode", "/nonexistent/log.txt"]) + self.assertEqual(code, 2) + self.assertIn("input not found", stderr.getvalue()) + + def test_demo_runs_deterministic(self): + first = run_demo() + second = run_demo() + self.assertTrue(first["ok"], first) + self.assertEqual(first["raw"]["count"], second["raw"]["count"]) + self.assertEqual(first["records"]["records"][0]["frames"], + second["records"]["records"][0]["frames"]) + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + code = main(["demo"]) + self.assertEqual(code, 0) + self.assertIn('"ok": true', stdout.getvalue()) + + +@unittest.skipUnless(os.path.isfile(REAL_LOGDUMP), + f"real logdump not present: {REAL_LOGDUMP}") +class RealLogdumpTests(unittest.TestCase): + """Decode the campaign dump when it is present; skipped otherwise.""" + + def test_real_logdump_decodes(self): + report = decode_path(REAL_LOGDUMP) + self.assertEqual(report["schema"], SCHEMA) + self.assertEqual(report["source"], os.path.basename(REAL_LOGDUMP)) + self.assertGreater(report["count"], 0) + self.assertTrue(any(report["histogram"].values()), + report["histogram"]) + raw_labels = {r["label"] for r in _records(report, "raw-hex")} + self.assertEqual(raw_labels, {"LOGINIT", "LOGRM"}) + framed = [r for r in _records(report) + if r.get("seq") is not None and r.get("frames")] + self.assertTrue(framed) + self.assertEqual(report["count"], + len(report["records"])) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py b/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py new file mode 100644 index 0000000..749157e --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py @@ -0,0 +1,342 @@ +"""Message-queue attach tests (sim/msgq_attach.py, docs/51 section 7). + +Hermetic tests drive the pure classification/report functions with crafted +inputs -- no image, no boot, no machine. One image test runs the real 570.144 +firmware with a small bounded budget when the image is present and asserts +honest, deterministic outcomes only: it never requires the message loop to be +reached, because the tool must report a first divergence rather than fake one. + +Offline only: no hardware, kernel module, mdev, guest, network or QEMU. + +Run:: + + cd open-iov/debug/risc-v-gsp-rm-vm + python3 -m unittest tests.test_msgq_attach -v +""" +from __future__ import annotations + +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.host_traffic import ( # noqa: E402 + DOORBELL_BASE, + DOORBELL_SIZE, + GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, + GSP_MSG_QUEUE_ELEMENT_SIZE_MIN, + MSGQ_ENTRY_OFF, + MSGQ_RX_HDR_OFF, + MSGQ_TX_HEADER_BYTES, + SHARED_MEM_BASE, + HostTraffic, +) +from sim.msgq_attach import ( # noqa: E402 + DEFAULT_IMAGE, + MAX_RUN_CYCLES, + REPORT_FORMAT, + MsgqAttachError, + RingGeometry, + assemble_report, + classify_ring_access, + missing_semantic_after_handoff, + ring_geometry_from_host, + run_attach, +) + +#: Small total budget: boot-to-B3 needs ~474k instructions; the rest is a +#: bounded continuation tail. Two runs stay well under the test time cap. +SMALL_BUDGET = 600_000 + + +def _geometry() -> RingGeometry: + return RingGeometry( + shared_base=SHARED_MEM_BASE, + cmd_base=SHARED_MEM_BASE, + stat_base=SHARED_MEM_BASE + 0x40000, + cmd_size=0x40000, + stat_size=0x40000, + msg_size=GSP_MSG_QUEUE_ELEMENT_SIZE_MIN, + entry_off=MSGQ_ENTRY_OFF, + rx_hdr_off=MSGQ_RX_HDR_OFF, + doorbell_base=DOORBELL_BASE, + doorbell_size=DOORBELL_SIZE, + ) + + +def _image() -> dict: + return {"path": "/nonexistent/image.bin", "sha256": "0" * 64, "size": 1} + + +def _profile() -> dict: + return {"name": "crafted", "path": "/nonexistent/profile.json"} + + +def _boot(reached: bool) -> dict: + return { + "ran": True, + "skipped": False, + "rung": "B3" if reached else "B1", + "message_loop_wait_reached": reached, + "instructions_executed": 474043 if reached else 1000, + "stop": {"reason": "ecall" if reached else "budget", + "pc": 0x4018038 if reached else 0x4009000}, + } + + +def _assemble(reached: bool, polls, divergence) -> dict: + return assemble_report( + image=_image(), + profile=_profile(), + boot=_boot(reached), + attach={"cmd_base": SHARED_MEM_BASE}, + doorbell={"requested": True}, + polls=polls, + divergence=divergence, + missing_mmio={"count": 0, "first": None}, + determinism={"instructions": 474043 if reached else 1000}, + ) + + +class ClassifierTests(unittest.TestCase): + """Pure ring-access classification (no machine, no boot).""" + + def test_geometry_from_host_matches_the_attached_layout(self): + host = HostTraffic() + try: + geometry = ring_geometry_from_host(host) + finally: + host.close() + self.assertEqual(geometry.cmd_base, SHARED_MEM_BASE) + self.assertEqual(geometry.stat_base, SHARED_MEM_BASE + 0x40000) + self.assertEqual(geometry.msg_size, GSP_MSG_QUEUE_ELEMENT_SIZE_MIN) + self.assertEqual(geometry.entry_off, MSGQ_ENTRY_OFF) + self.assertEqual(geometry.rx_hdr_off, MSGQ_RX_HDR_OFF) + self.assertEqual(geometry.doorbell_base, DOORBELL_BASE) + self.assertEqual(geometry.doorbell_size, DOORBELL_SIZE) + + def test_queue_head_classification(self): + geometry = _geometry() + poll = classify_ring_access(geometry.doorbell_base + 8, 4, 0x11, geometry) + self.assertTrue(poll["in_ring"]) + self.assertEqual(poll["kind"], "queue-head") + self.assertEqual(poll["queue"], 1) + self.assertEqual(poll["register"], "NV_PGSP_QUEUE_HEAD(1)") + self.assertEqual(poll["window"], "PRIV+0x110c00") + self.assertEqual(poll["value"], 0x11) + + def test_tx_header_classification(self): + geometry = _geometry() + poll = classify_ring_access(geometry.cmd_base + 16, 4, 0, geometry) + self.assertTrue(poll["in_ring"]) + self.assertEqual(poll["kind"], "tx-header") + self.assertEqual(poll["buffer"], "command") + self.assertEqual(poll["field"], "writePtr") + self.assertEqual(poll["field_offset"], 16) + self.assertEqual(MSGQ_TX_HEADER_BYTES, 32) + + def test_rx_header_classification_carries_the_cross_link(self): + geometry = _geometry() + poll = classify_ring_access(geometry.stat_base + MSGQ_RX_HDR_OFF, 4, 0, + geometry) + self.assertEqual(poll["kind"], "rx-header") + self.assertEqual(poll["buffer"], "status") + self.assertEqual(poll["cross_linked_owner"], "command") + self.assertEqual(poll["semantics"], "readPtr") + + def test_entry_classification_and_slot(self): + geometry = _geometry() + addr = (geometry.cmd_base + MSGQ_ENTRY_OFF + + 3 * GSP_MSG_QUEUE_ELEMENT_SIZE_MIN + 0x40) + poll = classify_ring_access(addr, 4, 0xAB, geometry) + self.assertEqual(poll["kind"], "entry") + self.assertEqual(poll["buffer"], "command") + self.assertEqual(poll["entry_index"], 3) + self.assertEqual(poll["slot_offset"], 0x40) + self.assertEqual(poll["region"], "entry-payload") + header = classify_ring_access( + geometry.cmd_base + MSGQ_ENTRY_OFF + 0x28, 4, 0, geometry) + self.assertEqual(header["region"], "entry-header") + self.assertEqual(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, 0x30) + + def test_non_ring_access_is_other(self): + geometry = _geometry() + poll = classify_ring_access(0xDEAD, 4, 0, geometry) + self.assertFalse(poll["in_ring"]) + self.assertEqual(poll["kind"], "other") + + +class ReportTests(unittest.TestCase): + """Pure report assembly: schema, outcomes and honest divergence.""" + + def test_not_reached_keeps_divergence_and_no_poll(self): + geometry = _geometry() + divergence = {"kind": "budget", + "reason": "message-loop port wait not reached within budget"} + report = _assemble(False, [], divergence) + self.assertEqual(report["format"], REPORT_FORMAT) + self.assertEqual(report["outcome"], "not-reached") + self.assertFalse(report["ok"]) + self.assertIsNone(report["first_poll"]) + self.assertEqual(report["subsequent_polls"], []) + self.assertEqual(report["poll_count"], 0) + self.assertTrue(report["divergence"]["reason"]) + self.assertEqual(report["determinism"]["report_sha256"], + _assemble(False, [], divergence)["determinism"]["report_sha256"]) + self.assertEqual(geometry.doorbell_base, DOORBELL_BASE) + + def test_reached_with_poll_is_classified(self): + geometry = _geometry() + poll = classify_ring_access(geometry.doorbell_base, 4, 0, geometry) + poll["pc"] = 0x4019000 + report = _assemble(True, [poll], None) + self.assertEqual(report["outcome"], "wait-reached-poll-classified") + self.assertTrue(report["ok"]) + self.assertEqual(report["first_poll"]["kind"], "queue-head") + self.assertEqual(report["first_poll"]["pc"], 0x4019000) + self.assertEqual(report["subsequent_polls"], []) + self.assertIsNone(report["divergence"]) + + def test_reached_without_poll_is_honest(self): + report = _assemble(True, [], None) + self.assertEqual(report["outcome"], "wait-reached-no-poll") + self.assertTrue(report["ok"]) + self.assertIsNone(report["first_poll"]) + + def test_subsequent_polls_are_bounded_to_sixteen(self): + geometry = _geometry() + polls = [] + for index in range(20): + poll = classify_ring_access(geometry.cmd_base + 8, 4, index, geometry) + poll["pc"] = 0x4018000 + index + polls.append(poll) + report = _assemble(True, polls, None) + self.assertEqual(report["poll_count"], 20) + self.assertEqual(report["first_poll"]["value"], 0) + self.assertEqual(len(report["subsequent_polls"]), 16) + self.assertEqual(report["subsequent_polls"][-1]["value"], 16) + + def test_absent_image_is_a_clean_skip(self): + report = run_attach(image="/nonexistent/gsp-image.bin", + max_cycles=SMALL_BUDGET) + self.assertEqual(report["format"], REPORT_FORMAT) + self.assertEqual(report["outcome"], "skipped") + self.assertFalse(report["ok"]) + self.assertTrue(report["boot"]["reason"]) + + def test_bad_budget_is_refused(self): + with self.assertRaises(MsgqAttachError): + run_attach(image="/nonexistent/gsp-image.bin", max_cycles=0) + with self.assertRaises(MsgqAttachError): + run_attach(image="/nonexistent/gsp-image.bin", + max_cycles=MAX_RUN_CYCLES + 1) + with self.assertRaises(MsgqAttachError): + run_attach(image="/nonexistent/gsp-image.bin", max_cycles=True) + + +class PortModelReportTests(unittest.TestCase): + """The additive port_model / missing_semantic report fields.""" + + def test_defaults_are_additive_and_honest(self): + report = _assemble(True, [], None) + self.assertEqual(report["port_model"], {"enabled": False}) + self.assertIsNone(report["missing_semantic"]) + self.assertEqual(report["format"], REPORT_FORMAT) + + def test_port_model_state_is_kept_verbatim(self): + state = {"enabled": True, "format": "greeneye-msgq-port/1", + "armed_mask": 0xFF, "host_kicks": 1} + report = assemble_report( + image=_image(), + profile=_profile(), + boot=_boot(True), + attach={}, + doorbell={"requested": True}, + polls=[], + divergence=None, + missing_mmio={"count": 0, "first": None}, + determinism={"instructions": 1}, + port_model=state, + missing_semantic={"stage": "after-extintr-handoff"}, + ) + self.assertTrue(report["port_model"]["enabled"]) + self.assertEqual(report["port_model"]["armed_mask"], 0xFF) + self.assertEqual(report["missing_semantic"]["stage"], + "after-extintr-handoff") + + def test_missing_semantic_after_handoff_is_precise(self): + stopped = {"num": 6, "name": "task_state_query", "pc": 0x401A46C, + "reason": "unmodelled-syscall"} + missing = missing_semantic_after_handoff(stopped=stopped, polls=[]) + self.assertEqual(missing["stage"], "after-extintr-handoff") + self.assertEqual(missing["stop"]["num"], 6) + self.assertEqual(missing["polls"], 0) + self.assertGreaterEqual(len(missing["missing"]), 3) + self.assertTrue(missing["next_step"]) + self.assertTrue(all(isinstance(item, str) for item in missing["observed"])) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent") +class ImageTests(unittest.TestCase): + """Bounded real-firmware run: deterministic, honest outcome only. + + Limitation: the budget is deliberately small (600k instructions; boot-to-B3 + needs ~474k). If the firmware needs more, the report is expected to take + the honest ``not-reached`` path with a non-empty divergence reason -- this + test accepts either outcome and asserts only determinism, never success. + With the default port model the run advances past the host wake to the + external-interrupt dispatch and reports the exact unmodelled semantic + (``task-state-query``); a fabricated poll is never accepted. + """ + + def test_bounded_run_is_deterministic_and_honest(self): + first = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET, + doorbell=True) + second = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET, + doorbell=True) + allowed = ("not-reached", "wait-reached-no-poll", + "wait-reached-poll-classified") + for report in (first, second): + self.assertEqual(report["format"], REPORT_FORMAT) + self.assertIn(report["outcome"], allowed) + self.assertGreater(report["image"]["size"], 0) + self.assertEqual(len(report["image"]["sha256"]), 64) + self.assertTrue(report["port_model"]["enabled"]) + if not report["boot"]["message_loop_wait_reached"]: + self.assertIsNotNone(report["divergence"]) + self.assertTrue(report["divergence"]["reason"]) + if report["first_poll"] is not None: + self.assertTrue(report["first_poll"]["in_ring"]) + self.assertTrue(report["first_poll"]["kind"] != "other") + divergence = report["divergence"] + if divergence and divergence.get("kind") == "task-state-query": + # The port model delivered the doorbell wake (extintr shuttle + # 10) and stopped at the libos trap/report path; the missing + # semantic is named, and no poll is claimed. + self.assertIsNone(report["first_poll"]) + self.assertIsNotNone(report["missing_semantic"]) + self.assertEqual(report["missing_semantic"]["stage"], + "after-extintr-handoff") + completed = [wait.get("completed") + for wait in report["port_model"].get("waits", [])] + self.assertIn(10, completed) + self.assertEqual(first["boot"]["message_loop_wait_reached"], + second["boot"]["message_loop_wait_reached"]) + self.assertEqual(first["boot"]["instructions_executed"], + second["boot"]["instructions_executed"]) + self.assertEqual(first["outcome"], second["outcome"]) + self.assertEqual(first["first_poll"], second["first_poll"]) + self.assertEqual(first["divergence"], second["divergence"]) + self.assertEqual(first["continuation"], second["continuation"]) + self.assertEqual(first["port_model"], second["port_model"]) + self.assertEqual(first["missing_semantic"], second["missing_semantic"]) + self.assertEqual(first["determinism"]["report_sha256"], + second["determinism"]["report_sha256"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rehost.py b/debug/risc-v-gsp-rm-vm/tests/test_rehost.py new file mode 100644 index 0000000..3c88355 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_rehost.py @@ -0,0 +1,135 @@ +"""Tests for the surgical-rehost harness (harness/rehost.py, docs/46). + +Fixtures: the demo firmware (built by ``make demo``) is rehosted offline. The +real-image probes are skipped when the firmware is absent, and are bounded to +one short call each. No hardware, no module, no guest, no network. + +Run:: + + python3 -m unittest tests.test_rehost -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.rehost import ( # noqa: E402 + DEFAULT_IMAGE, + DEMO_ELF, + DEMO_FORMAT, + REHOST_FORMAT, + RehostError, + main, + parse_hex_bytes, + parse_mem_spec, + parse_snap_spec, + rehost, + run_demo, +) + + +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() + + +class SpecTests(unittest.TestCase): + def test_parse_hex_bytes_and_mem_spec(self): + self.assertEqual(parse_hex_bytes("aa bb cc"), b"\xaa\xbb\xcc") + addr, data = parse_mem_spec("0x1000=0102") + self.assertEqual((addr, data), (0x1000, b"\x01\x02")) + + def test_bad_specs_are_clean_errors(self): + with self.assertRaises(RehostError): + parse_hex_bytes("abc") + with self.assertRaises(RehostError): + parse_mem_spec("0x1000") + with self.assertRaises(RehostError): + parse_snap_spec("0x1000") + + def test_missing_image_is_clean_error(self): + with self.assertRaises(RehostError): + rehost(image="/nonexistent/fw.bin", entry=0x1000) + + +@unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware not built") +class DemoRehostTests(unittest.TestCase): + def test_demo_rehost_is_deterministic_and_correct(self): + payload = run_demo() + self.assertEqual(payload["format"], DEMO_FORMAT) + self.assertTrue(payload["ok"]) + self.assertEqual(payload["first"]["status"], "return") + self.assertEqual(payload["first"]["retval"], payload["expected"]) + self.assertEqual(payload["first"]["retval"], payload["second"]["retval"]) + + def test_rehost_writes_report_and_session(self): + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + report = rehost( + image=DEMO_ELF, + entry="rpc_sum", + args=(0x80012000, 4), + mem=((0x80012000, b"\x01\x02\x03\x04"),), + snap=((0x80012000, 4),), + budget=50_000, + timeout=20.0, + out=tmp, + record=True, + name="rehost-test", + ) + self.assertEqual(report["format"], REHOST_FORMAT) + self.assertEqual(report["status"], "return") + self.assertEqual(report["retval"], 10) + self.assertTrue(os.path.isfile(report["report_json"])) + self.assertEqual(len(report["snapshots"]), 1) + self.assertEqual(report["snapshots"][0]["changed_bytes"], 0) + self.assertTrue(report["snapshots"][0]["before_sha256"]) + self.assertIn("fidelity", " ".join(report["limits"]).lower()) + + def test_cli_demo_returns_zero(self): + rc, out, _ = _run_main(["demo"]) + self.assertEqual(rc, 0) + self.assertTrue(json.loads(out)["ok"]) + + def test_cli_missing_image_returns_two(self): + rc, _, err = _run_main(["run", "--image", "/nonexistent/fw.bin", "--entry", "0x1000"]) + self.assertEqual(rc, 2) + self.assertIn("error:", err) + + +class RealImageProbeTests(unittest.TestCase): + """Bounded probes on the shipped 570.144 image; skipped without it.""" + + @unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") + def test_gate_probe_reproduces_invalid_state_without_state(self): + report = rehost( + image=DEFAULT_IMAGE, + profile_path=os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json"), + entry=0x5B01210, # per-buffer commit gate (from live analysis; numeric only) + args=(0,), + budget=50_000, + timeout=20.0, + record=False, + name="probe-gate-test", + ) + self.assertEqual(report["format"], REHOST_FORMAT) + self.assertIn(report["status"], ("return", "fault", "budget", "ecall")) + # With no engine-context state the gate family returns NV_ERR_INVALID_STATE. + self.assertEqual(report["retval"], 0x40) + self.assertIn("Not a boot", report["limits"][0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py b/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py new file mode 100644 index 0000000..cb62bea --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py @@ -0,0 +1,409 @@ +"""Tests for the GSP RPC capture ingestion harness (harness/rm_capture.py, docs/45). + +All fixtures are synthetic text written in-process or to temporary +directories: no capture, no hardware, no driver, no simulator run, no QEMU, +no network. The decode/compare/stimulus loop is offline file-in/file-out. +Tests that would need a real capture assert the clean error path instead. + +Run:: + + python3 -m unittest tests.test_rm_capture -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, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.rm_capture import ( # noqa: E402 + CAPTURE_FORMAT, + COMPARE_FORMAT, + DEMO_FORMAT, + GSP_RM_ALLOC, + GSP_RM_CONTROL, + MANUAL_DECODE_BASELINE_SECONDS, + RmCaptureError, + STIMULUS_FORMAT, + WORKFLOW_THRESHOLD_FRACTION, + canonical_json, + compare, + decode_file, + decode_text, + load_capture, + main, + print_capture_hints, + run_compare, + run_decode, + run_demo, + run_stimulus, + stimulus_from_capture, +) + + +def _hex_lines(tag: str, payload: bytes) -> list[str]: + return [f"GE-{tag} 000:" + "".join(f" {b:02x}" for b in payload)] + + +def _sample_log() -> str: + """A small log exercising every tag; params are contiguous single chunks.""" + return "\n".join( + [ + "[ 100.000000] NVRM: GE-GOLDEN-TRIGGER gspclient=1 mig=0", + "[ 100.000010] NVRM: GE-GOLDEN-BEGIN", + "[ 100.000020] NVRM: GE-ALLOC cls=0x90f1 hClient=0xbaba0001 hParent=0x0 " + "hObject=0xbaba1001 size=0x10 paramp=0xffffffffc0a00000 ret=0xffffffffc0b00000", + "[ 100.000020] NVRM: " + _hex_lines("ALLOC", bytes(range(16)))[0], + "[ 100.000030] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 " + "hObject=0xabcd0080 size=0x8 paramp=0xffffffffc0a10000 ret=0xffffffffc0b00000", + "[ 100.000030] NVRM: " + _hex_lines("CTRL", bytes([1, 2, 3, 4, 5, 6, 7, 8]))[0], + "[ 100.000040] NVRM: GE-SEND fn=0x0000004c len=0x00000028 ret=0xffffffffc0b00000", + "[ 100.000040] NVRM: " + + _hex_lines("SEND", (0xC2000006).to_bytes(4, "little") + + (0xABCD0080).to_bytes(4, "little") + + (0x2080012B).to_bytes(4, "little") + + bytes(4))[0], + "[ 100.000050] NVRM: GE-CTRL-DONE cmd=0x2080012b rpc_status=0x0 params_status=0x0", + "[ 100.000060] NVRM: GE-GOLDEN-END status=0x0", + ] + ) + "\n" + + +def _follow(text: str, header_fragment: str) -> str: + """Return the hex block emitted immediately after a header line.""" + lines = text.splitlines() + for index, line in enumerate(lines): + if header_fragment in line: + return lines[index + 1] + raise AssertionError(f"fragment not found: {header_fragment}") + + +def _chunked_control_log(size: int) -> str: + payload = bytes((i * 5) & 0xFF for i in range(size)) + lines = ["[ 1.0] NVRM: GE-GOLDEN-BEGIN"] + lines.append( + f"[ 1.1] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 " + f"hObject=0xabcd0080 size={size:#x} paramp=0x0 ret=0x0" + ) + for start in range(0, len(payload), 192): + piece = payload[start : start + 192] + lines.append( + f"[ 1.1] NVRM: GE-CTRL {start:03x}:" + + "".join(f" {b:02x}" for b in piece) + ) + lines.append("[ 1.2] NVRM: GE-GOLDEN-END status=0x0") + return "\n".join(lines) + "\n" + + +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() + + +class DecodeTests(unittest.TestCase): + def test_decodes_every_record_kind_in_order(self): + capture = decode_text(_sample_log(), source="sample") + kinds = [record.kind for record in capture.records] + self.assertEqual( + kinds, + ["golden", "golden", "alloc", "control", "send", "ctrl-done", "golden"], + ) + trigger, begin = capture.records[0], capture.records[1] + self.assertEqual(trigger.marker, "trigger") + self.assertEqual(trigger.gspclient, 1) + self.assertEqual(trigger.mig, 0) + self.assertEqual(begin.marker, "begin") + alloc = capture.records[2] + self.assertEqual(alloc.hclass, 0x90F1) + self.assertEqual(alloc.hclient, 0xBABA0001) + self.assertEqual(alloc.hobject, 0xBABA1001) + self.assertEqual(alloc.size, 0x10) + self.assertEqual(alloc.caller, 0xFFFFFFFFC0B00000) + control = capture.records[3] + self.assertEqual(control.cmd, 0x2080012B) + self.assertEqual(control.params, bytes([1, 2, 3, 4, 5, 6, 7, 8])) + send = capture.records[4] + self.assertEqual(send.fn, 0x4C) + self.assertEqual(send.length, 0x28) + self.assertEqual(send.expected_params_size, 0x28 - 32) + self.assertEqual(capture.records[5].rpc_status, 0) + self.assertEqual(capture.records[6].status, 0) + self.assertEqual(capture.golden_status(), 0) + self.assertFalse(capture.warnings) + + def test_dmesg_timestamp_and_prefix_are_stripped(self): + capture = decode_text(_sample_log(), source="sample") + self.assertEqual(capture.records[0].ts, " 100.000000") + + def test_multichunk_payload_is_assembled(self): + capture = decode_text(_chunked_control_log(560), source="chunked") + control = capture.records[1] + self.assertEqual(len(control.params), 560) + self.assertFalse(control.truncated) + self.assertEqual(control.chunks[0].__len__(), 192) + self.assertIn(0xC0, control.chunks) + self.assertIn(0x180, control.chunks) + self.assertEqual(control.params[192], (192 * 5) & 0xFF) + + def test_truncated_payload_is_flagged(self): + # The runbook helper caps at 192 bytes: declared 560, captured 192. + capture = decode_text(_chunked_control_log(192), source="truncated") + control = capture.records[1] + control.size = 560 + self.assertTrue(control.truncated) + self.assertEqual(control.expected_params_size, 560) + self.assertEqual(len(control.params), 192) + + def test_orphan_hex_warns_and_is_not_attached(self): + capture = decode_text("[ 1.0] NVRM: GE-CTRL 000: 01 02 03 04\n", source="orphan") + self.assertEqual(capture.records, []) + self.assertEqual(len(capture.warnings), 1) + self.assertIn("orphan CTRL hex", capture.warnings[0]) + + def test_wire_id_derived_from_send_payload(self): + control_send = decode_text(_sample_log(), source="s").records[4] + self.assertEqual(control_send.wire_id, 0x2080012B) + alloc_payload = ( + (0xBABA0001).to_bytes(4, "little") + + (0x0).to_bytes(4, "little") + + (0xBABA1002).to_bytes(4, "little") + + (0x0040).to_bytes(4, "little") + ) + log = ( + "[ 1.0] NVRM: GE-SEND fn=0x00000067 len=0x30 ret=0x0\n" + "[ 1.0] NVRM: GE-SEND 000:" + "".join(f" {b:02x}" for b in alloc_payload) + "\n" + ) + record = decode_text(log, source="a").records[0] + self.assertEqual(record.fn, GSP_RM_ALLOC) + self.assertEqual(record.wire_id, 0x0040) + self.assertNotEqual(GSP_RM_CONTROL, GSP_RM_ALLOC) + + def test_overrun_is_an_anomaly(self): + log = ( + "[ 1.0] NVRM: GE-CTRL cmd=0x1 hClient=0x0 hObject=0x0 size=0x4 " + "paramp=0x0 ret=0x0\n" + "[ 1.0] NVRM: GE-CTRL 000: 01 02 03 04 05 06 07 08\n" + ) + record = decode_text(log, source="overrun").records[0] + self.assertTrue(record.overrun) + self.assertTrue(any("> declared" in item for item in record.anomalies)) + + def test_golden_status_none_without_end_record(self): + log = "[ 1.0] NVRM: GE-GOLDEN-BEGIN\n" + self.assertIsNone(decode_text(log, source="x").golden_status()) + + +class CaptureIOTests(unittest.TestCase): + def test_decode_write_load_roundtrip(self): + with tempfile.TemporaryDirectory() as tmp: + log_path = os.path.join(tmp, "capture.log") + with open(log_path, "w", encoding="utf-8") as fh: + fh.write(_sample_log()) + out_dir = os.path.join(tmp, "out") + payload = run_decode(log_path, out_dir) + self.assertEqual(payload["format"], CAPTURE_FORMAT) + self.assertTrue(os.path.isfile(payload["artifacts"]["capture_json"])) + self.assertTrue(os.path.isfile(payload["artifacts"]["capture_md"])) + loaded = load_capture(payload["artifacts"]["capture_json"]) + self.assertEqual(len(loaded.records), len(decode_file(log_path).records)) + self.assertEqual(loaded.records[3].cmd, 0x2080012B) + + def test_load_capture_autodetects_raw_log(self): + with tempfile.TemporaryDirectory() as tmp: + log_path = os.path.join(tmp, "raw.log") + with open(log_path, "w", encoding="utf-8") as fh: + fh.write(_sample_log()) + loaded = load_capture(log_path) + self.assertEqual(loaded.records[0].kind, "golden") + + def test_load_capture_rejects_unknown_format(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "bad.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump({"format": "something-else", "records": []}, fh) + with self.assertRaises(RmCaptureError): + load_capture(path) + + def test_missing_file_is_clean_error(self): + with self.assertRaises(RmCaptureError): + decode_file("/nonexistent/capture.log") + + +class CompareTests(unittest.TestCase): + def _captures(self): + capture_a = decode_text(_sample_log(), source="a") + capture_b = decode_text(_sample_log(), source="b") + return capture_a, capture_b + + def test_identical_captures_fully_agree(self): + capture_a, capture_b = self._captures() + report = compare(capture_a, capture_b) + self.assertEqual(report["format"], COMPARE_FORMAT) + summary = report["summary"] + self.assertEqual(summary["mismatch"], 0) + self.assertEqual(summary["a_only"], 0) + self.assertEqual(summary["b_only"], 0) + self.assertEqual(summary["agree"], len(capture_a.records)) + self.assertEqual(summary["agreement_pct"], 100.0) + self.assertEqual(summary["coverage_pct"], 100.0) + + def test_dropped_record_is_a_only(self): + capture_a, capture_b = self._captures() + capture_b.records = [ + record for record in capture_b.records if record.kind != "alloc" + ] + report = compare(capture_a, capture_b) + self.assertEqual(report["summary"]["a_only"], 1) + self.assertEqual(report["a_only"][0]["label"].startswith("ALLOC"), True) + + def test_param_byte_flip_is_shape_agree(self): + capture_a, capture_b = self._captures() + control = next(r for r in capture_b.records if r.kind == "control") + control.chunks[0] = bytes([9]) + control.chunks[0][1:] + report = compare(capture_a, capture_b) + self.assertEqual(report["summary"]["shape_agree"], 1) + row = next(r for r in report["rows"] if r["status"] == "shape-agree") + self.assertEqual(row["detail"]["first_diff_offset"], 0) + self.assertLess(row["detail"]["byte_agreement_pct"], 100.0) + + def test_truncated_pair_is_partial(self): + capture_a = decode_text(_chunked_control_log(192), source="a") + capture_b = decode_text(_chunked_control_log(192), source="b") + for capture in (capture_a, capture_b): + capture.records[1].size = 560 + report = compare(capture_a, capture_b) + self.assertEqual(report["summary"]["partial"], 1) + self.assertTrue(report["rows"][1]["a_truncated"]) + + def test_compare_is_deterministic_and_self_describing(self): + capture_a, capture_b = self._captures() + report_one = compare(capture_a, capture_b) + report_two = compare(capture_a, capture_b) + self.assertEqual(canonical_json(report_one), canonical_json(report_two)) + self.assertEqual(report_one["format"], COMPARE_FORMAT) + self.assertIn("L2", report_one["fidelity"]["levels"]) + self.assertTrue(report_one["limits"]) + + def test_run_compare_writes_json_and_markdown(self): + with tempfile.TemporaryDirectory() as tmp: + log_path = os.path.join(tmp, "capture.log") + with open(log_path, "w", encoding="utf-8") as fh: + fh.write(_sample_log()) + report = run_compare(log_path, log_path, os.path.join(tmp, "out")) + self.assertTrue(os.path.isfile(report["artifacts"]["report_json"])) + self.assertTrue(os.path.isfile(report["artifacts"]["report_md"])) + with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh: + self.assertIn("# GSP RPC capture comparison", fh.read()) + + +class StimulusTests(unittest.TestCase): + def test_stimulus_lists_rpcs_in_order_without_handles(self): + capture = decode_text(_sample_log(), source="s") + stimulus = stimulus_from_capture(capture) + self.assertEqual(stimulus["format"], STIMULUS_FORMAT) + kinds = [message["kind"] for message in stimulus["messages"]] + self.assertEqual(kinds, ["alloc", "control", "send"]) + self.assertEqual(stimulus["messages"][1]["cmd"], 0x2080012B) + self.assertEqual(stimulus["messages"][2]["fn"], GSP_RM_CONTROL) + self.assertEqual(stimulus["messages"][2]["wire_id"], 0x2080012B) + self.assertNotIn("hclient", stimulus["messages"][1]) + self.assertEqual(len(stimulus["markers"]), 3) + self.assertTrue(stimulus["limits"]) + + def test_stimulus_roundtrips_params_and_flags_truncation(self): + capture = decode_text(_chunked_control_log(192), source="s") + capture.records[1].size = 560 + stimulus = stimulus_from_capture(capture) + message = stimulus["messages"][0] + self.assertTrue(message["truncated"]) + self.assertEqual(message["params_size_expected"], 560) + self.assertEqual(bytes.fromhex(message["params_hex"]), capture.records[1].params) + + def test_run_stimulus_writes_file(self): + with tempfile.TemporaryDirectory() as tmp: + out_path = os.path.join(tmp, "nested", "stimulus.json") + log_path = os.path.join(tmp, "capture.log") + with open(log_path, "w", encoding="utf-8") as fh: + fh.write(_sample_log()) + payload = run_stimulus(log_path, out_path) + self.assertTrue(os.path.isfile(out_path)) + self.assertEqual(payload["format"], STIMULUS_FORMAT) + with open(out_path, "r", encoding="utf-8") as fh: + written = json.load(fh) + self.assertEqual(written["format"], STIMULUS_FORMAT) + + +class DemoAndCliTests(unittest.TestCase): + def test_demo_is_ok_and_within_threshold(self): + payload = run_demo() + self.assertEqual(payload["format"], DEMO_FORMAT) + self.assertTrue(payload["ok"]) + self.assertLessEqual( + payload["wall_seconds"], + MANUAL_DECODE_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION, + ) + self.assertGreaterEqual(payload["summary"]["agree"], 1) + self.assertEqual(payload["summary"]["shape_agree"], 1) + self.assertEqual(payload["summary"]["a_only"], 1) + + def test_demo_is_deterministic_apart_from_timing(self): + first = run_demo() + second = run_demo() + for payload in (first, second): + payload.pop("wall_seconds", None) + payload.pop("artifacts", None) + self.assertEqual(canonical_json(first), canonical_json(second)) + + def test_demo_writes_artifacts(self): + with tempfile.TemporaryDirectory() as tmp: + payload = run_demo(tmp) + for key in ("capture_a", "replay_b", "compare", "stimulus"): + self.assertTrue(os.path.isfile(payload["artifacts"][key])) + + def test_cli_decode_compare_stimulus_return_zero(self): + with tempfile.TemporaryDirectory() as tmp: + log_path = os.path.join(tmp, "capture.log") + with open(log_path, "w", encoding="utf-8") as fh: + fh.write(_sample_log()) + rc, out, _ = _run_main(["decode", "--log", log_path, "--out", tmp]) + self.assertEqual(rc, 0) + self.assertEqual(json.loads(out)["format"], CAPTURE_FORMAT) + rc, out, _ = _run_main( + ["compare", "--a", log_path, "--b", log_path, "--out", tmp] + ) + self.assertEqual(rc, 0) + self.assertEqual(json.loads(out)["format"], COMPARE_FORMAT) + rc, out, _ = _run_main( + ["stimulus", "--capture", log_path, "--out", os.path.join(tmp, "s.json")] + ) + self.assertEqual(rc, 0) + self.assertEqual(json.loads(out)["format"], STIMULUS_FORMAT) + + def test_cli_missing_input_returns_two(self): + rc, _, err = _run_main(["decode", "--log", "/nonexistent/capture.log"]) + self.assertEqual(rc, 2) + self.assertIn("error:", err) + + def test_hints_execute_nothing_and_name_the_rules(self): + out = io.StringIO() + with contextlib.redirect_stdout(out): + print_capture_hints() + text = out.getvalue() + self.assertIn("executes", text) + self.assertIn("NOTHING", text) + self.assertIn("GE-SEND", text) + self.assertIn("192 bytes", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py b/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py new file mode 100644 index 0000000..b82d09c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py @@ -0,0 +1,364 @@ +"""Tests for the server-state snapshot collector (harness/server_snapshot.py). + +Crafted evidence covers every supported shape (``LABEL[offset]=0xVALUE``, +labelled hexdump blocks, ``LOGINIT-raw`` headers, record lines, duplicate +labels and unparseable dump-looking lines). Optional tests run against the +real 2026-10-02 campaign backups when they are present; they are skipped +otherwise. All tests are offline, hermetic and touch no hardware. + +Run:: + + python3 -m unittest tests.test_server_snapshot -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +from unittest import mock + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.server_snapshot import ( # noqa: E402 + DIFF_FORMAT, + SNAPSHOT_FORMAT, + SnapshotError, + collect, + diff_snapshots, + main, +) + +EVIDENCE_VICTORY = ("/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/" + "evidence-20261002-victory") +LADDER_M12 = os.path.join(EVIDENCE_VICTORY, "06-ladder-m12.txt") +PHYS_STATE_M26 = os.path.join(EVIDENCE_VICTORY, "20-gsp-phys-state-m26.txt") +EVIDENCE_GR = ("/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/" + "evidence-20261002-gr") +LOGDUMP_GRCTX4 = os.path.join(EVIDENCE_GR, "21-logdump-grctx4.txt") + +FIXED_CREATED = "2026-01-01T00:00:00Z" + + +def _run_main(argv): + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main(argv) + return rc, out.getvalue(), err.getvalue() + + +class ServerSnapshotTests(unittest.TestCase): + """Collector normalization, collision, diff, CLI and determinism.""" + + def setUp(self): + holder = tempfile.TemporaryDirectory(prefix="server-snapshot-test-") + self.addCleanup(holder.cleanup) + self.tmp = holder.name + + def _write(self, name, text): + path = os.path.join(self.tmp, name) + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + return path + + def _collect_text(self, name, text, **kwargs): + path = self._write(name, text) + kwargs.setdefault("source", f"test:{name}") + kwargs.setdefault("created", FIXED_CREATED) + snapshot, stats = collect(files=[path], **kwargs) + return snapshot, stats, path + + @staticmethod + def _chunks(snapshot): + return {chunk["label"]: chunk for chunk in snapshot["chunks"]} + + # -- crafted shapes ---------------------------------------------------- + + def test_labeled_value_lines(self): + snapshot, stats, _ = self._collect_text( + "values.txt", + "alpha[0]=0x00000011\n" + "alpha[4]=0x00000022\n" + "beta[8]=0x0000000112345678\n", + ) + chunks = self._chunks(snapshot) + self.assertEqual(chunks["alpha"]["base"], "0x0") + self.assertEqual(chunks["alpha"]["words"], + ["0x00000011", "0x00000022"]) + self.assertEqual(chunks["beta"]["base"], "0x8") + self.assertEqual(chunks["beta"]["words"], + ["0x12345678", "0x00000001"]) + self.assertEqual(stats["unparsed"], 0) + self.assertEqual(stats["conflicts"], 0) + + def test_labeled_hexdump_block(self): + snapshot, stats, _ = self._collect_text( + "block.txt", + "preamble text that is not a header\n" + "SOME_BUF: 64 bytes of server state\n" + "00000000: 00000001 00000002 00000003 00000004\n" + "00000010: 00000005 00000006 00000007 00000008\n" + "RAMFC: a second block with a nonzero base\n" + "00000020: 0000000a 0000000b\n", + ) + chunks = self._chunks(snapshot) + self.assertEqual(chunks["SOME_BUF"]["base"], "0x0") + self.assertEqual( + chunks["SOME_BUF"]["words"], + ["0x00000001", "0x00000002", "0x00000003", "0x00000004", + "0x00000005", "0x00000006", "0x00000007", "0x00000008"]) + self.assertEqual(chunks["RAMFC"]["base"], "0x20") + self.assertEqual(chunks["RAMFC"]["words"], + ["0x0000000a", "0x0000000b"]) + self.assertEqual(stats["unparsed"], 0) + self.assertEqual(stats["chunks"], 2) + + def test_logininit_raw_block(self): + snapshot, stats, _ = self._collect_text( + "loginit.txt", + "vgsp-boot: libos task log buffers (read-only)\n" + "LOGINIT-raw: 65536 bytes (allocated 65536) " + "VA=0xffff88a673af0000 dma=0x4f3af0000\n" + "00000000: 000002d2 00000000 05bb9e10 00000000\n" + "00000010: f7660780 18dbb28f fe2c0000 61800001\n", + ) + chunk = self._chunks(snapshot)["LOGINIT-raw"] + self.assertEqual(chunk["base"], "0x0") + self.assertEqual( + chunk["words"], + ["0x000002d2", "0x00000000", "0x05bb9e10", "0x00000000", + "0xf7660780", "0x18dbb28f", "0xfe2c0000", "0x61800001"]) + self.assertEqual(stats["unparsed"], 0) + + def test_inline_labeled_rows_group_into_one_chunk(self): + snapshot, _, _ = self._collect_text( + "inline.txt", + "[dmesg] vgsp-boot: rm-call: INSTANCE[0000]: " + "00000000 00000000 0a545003 00000001\n" + "[dmesg] vgsp-boot: rm-call: INSTANCE[0020]: " + "0000face 00000000 00000000 00000000\n", + ) + chunk = self._chunks(snapshot)["INSTANCE"] + self.assertEqual(chunk["base"], "0x0") + self.assertEqual(chunk["words"], + ["0x00000000", "0x00000000", "0x0a545003", + "0x00000001", "0x0000face", "0x00000000", + "0x00000000", "0x00000000"]) + + def test_record_lines(self): + snapshot, stats, _ = self._collect_text( + "records.txt", + 'vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 ' + 'subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" ' + 'faultingEngine=""\n', + ) + self.assertEqual(len(snapshot["records"]), 1) + record = snapshot["records"][0] + self.assertEqual(record["kind"], "record") + self.assertEqual(record["rec_type"], 5) + self.assertEqual(record["error_code"], 0x57F1344) + self.assertEqual(record["source_name"], "ASSERT") + self.assertEqual(stats["records"], 1) + self.assertEqual(snapshot["chunks"], []) + + # -- duplicates, collisions, unparsed --------------------------------- + + def test_identical_duplicate_merged(self): + snapshot, stats, _ = self._collect_text( + "merged.txt", + "INSTANCE[0000]: 00000001 00000002\n" + "plain prose closes the first block\n" + "INSTANCE[0000]: 00000001 00000002\n", + ) + self.assertEqual(list(self._chunks(snapshot)), ["INSTANCE"]) + self.assertEqual(stats["merged"], 1) + self.assertEqual(stats["collisions"], 0) + + def test_conflicting_duplicate_renamed(self): + snapshot, stats, _ = self._collect_text( + "collision.txt", + "INSTANCE[0000]: 00000001 00000002\n" + "plain prose closes the first block\n" + "INSTANCE[0000]: 00000003 00000004\n", + ) + self.assertEqual(list(self._chunks(snapshot)), + ["INSTANCE", "INSTANCE#2"]) + self.assertEqual(stats["collisions"], 1) + self.assertEqual(stats["collision_list"], + [{"label": "INSTANCE", "renamed": "INSTANCE#2", + "base": "0x0", "words": 2}]) + + def test_unparsed_candidates_counted(self): + snapshot, stats, _ = self._collect_text( + "unparsed.txt", + "BROKEN[0x10]: zzzz\n" + "00000000: zz00\n" + "ordinary prose is not a candidate\n", + ) + self.assertEqual(snapshot["chunks"], []) + self.assertEqual(stats["unparsed"], 2) + + # -- diff -------------------------------------------------------------- + + def test_diff_changed_added_removed(self): + snap_a, _, _ = self._collect_text( + "run-a.txt", + "MOCK[0000]: 00000001 00000002\n" + "shared[0]=0x00000007\n" + "only_a[0]=0x00000009\n", + ) + snap_b, _, _ = self._collect_text( + "run-b.txt", + "MOCK[0000]: 00000001 00000003 00000004\n" + "shared[0]=0x00000007\n" + "only_b[0]=0x0000000a\n", + ) + report = diff_snapshots(snap_a, snap_b) + self.assertEqual(report["schema"], DIFF_FORMAT) + rows = {row["label"]: row for row in report["chunks"]} + self.assertEqual(rows["MOCK"]["status"], "changed") + self.assertEqual(rows["MOCK"]["changed"], + [{"index": 1, "a": "0x00000002", "b": "0x00000003"}]) + self.assertEqual(rows["MOCK"]["added"], + [{"index": 2, "value": "0x00000004"}]) + self.assertEqual(rows["MOCK"]["removed"], []) + self.assertEqual(rows["shared"]["status"], "unchanged") + self.assertEqual(rows["only_a"]["status"], "removed") + self.assertEqual(rows["only_a"]["removed"], + [{"index": 0, "value": "0x00000009"}]) + self.assertEqual(rows["only_b"]["status"], "added") + self.assertEqual(rows["only_b"]["added"], + [{"index": 0, "value": "0x0000000a"}]) + summary = report["summary"] + self.assertFalse(summary["identical"]) + self.assertEqual(summary["chunks_added"], 1) + self.assertEqual(summary["chunks_removed"], 1) + self.assertEqual(summary["chunks_changed"], 1) + self.assertEqual(summary["chunks_unchanged"], 1) + self.assertEqual(summary["words_added"], 2) + self.assertEqual(summary["words_removed"], 1) + self.assertEqual(summary["words_changed"], 1) + + def test_diff_identical(self): + snapshot, _, _ = self._collect_text( + "same.txt", "INSTANCE[0000]: 00000001 00000002\n") + report = diff_snapshots(snapshot, snapshot) + self.assertTrue(report["summary"]["identical"]) + self.assertEqual(report["summary"]["words_changed"], 0) + self.assertEqual([row["status"] for row in report["chunks"]], + ["unchanged"]) + + def test_diff_rejects_wrong_schema(self): + with self.assertRaises(SnapshotError): + diff_snapshots({"schema": "other"}, {"schema": SNAPSHOT_FORMAT}) + + # -- determinism and CLI ---------------------------------------------- + + def test_determinism_with_source_date_epoch(self): + path = self._write( + "det.txt", + "INSTANCE[0000]: 00000001 00000002\n" + 'vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 ' + 'source="ASSERT"\n', + ) + with mock.patch.dict(os.environ, {"SOURCE_DATE_EPOCH": "1700000000"}): + first, _ = collect(files=[path], source="det") + second, _ = collect(files=[path], source="det") + self.assertEqual(first["created"], "2023-11-14T22:13:20Z") + self.assertEqual(first, second) + self.assertEqual(json.dumps(first, sort_keys=True), + json.dumps(second, sort_keys=True)) + + def test_cli_collect_stdout_and_json(self): + path = self._write("cli.txt", "INSTANCE[0000]: 00000001 00000002\n") + rc, out, err = _run_main( + ["collect", "--file", path, "--source", "cli", + "--created", FIXED_CREATED]) + self.assertEqual(rc, 0) + snapshot = json.loads(out) + self.assertEqual(set(snapshot), + {"schema", "source", "created", "chunks", "records"}) + self.assertEqual(snapshot["schema"], SNAPSHOT_FORMAT) + self.assertEqual(snapshot["source"], "cli") + self.assertIn("1 chunk(s)", err) + + json_path = os.path.join(self.tmp, "snapshot.json") + rc, out, _ = _run_main( + ["collect", "--file", path, "--json", json_path, + "--created", FIXED_CREATED]) + self.assertEqual(rc, 0) + with open(json_path, "r", encoding="utf-8") as handle: + written = json.load(handle) + self.assertEqual(written["schema"], SNAPSHOT_FORMAT) + self.assertEqual(json.loads(out)["chunks"], len(written["chunks"])) + + def test_cli_errors(self): + rc, _, err = _run_main(["collect"]) + self.assertEqual(rc, 2) + self.assertIn("error:", err) + missing = os.path.join(self.tmp, "missing.txt") + rc, _, _ = _run_main(["collect", "--file", missing]) + self.assertEqual(rc, 2) + + def test_discover_include_globs(self): + self._write("keep.txt", "AAA[0]=0x00000001\n") + self._write("skip.bin", "BBB[0]=0x00000002\n") + snapshot, _ = collect(dirs=[self.tmp], source="glob", + created=FIXED_CREATED) + self.assertEqual(list(self._chunks(snapshot)), ["AAA"]) + snapshot, _ = collect(dirs=[self.tmp], includes=["*.bin"], + source="glob", created=FIXED_CREATED) + self.assertEqual(list(self._chunks(snapshot)), ["AAA", "BBB"]) + + def test_demo_cli(self): + rc, out, _ = _run_main(["demo"]) + self.assertEqual(rc, 0) + self.assertIn("MOCK", out) + summary = json.loads(out[out.index("{"):]) + self.assertFalse(summary["identical"]) + self.assertEqual(summary["words_changed"], 2) + + # -- real campaign evidence (optional) -------------------------------- + + @unittest.skipUnless(os.path.isdir(EVIDENCE_VICTORY), + "victory evidence backup not present") + def test_real_victory_evidence(self): + files = [path for path in (LADDER_M12, PHYS_STATE_M26) + if os.path.isfile(path)] + if not files: + self.skipTest("expected victory evidence files missing") + snapshot, stats = collect(files=files, source="evidence-victory", + created=FIXED_CREATED) + self.assertTrue(snapshot["chunks"]) + chunks = self._chunks(snapshot) + self.assertIn("INSTANCE", chunks) + self.assertEqual(chunks["INSTANCE"]["base"], "0x0") + self.assertEqual(len(chunks["INSTANCE"]["words"]), 1024) + self.assertIn("pt", chunks) + self.assertEqual(len(chunks["pt"]["words"]), 5) + self.assertGreaterEqual(stats["records"], 1) + self.assertEqual(stats["files"], len(files)) + + @unittest.skipUnless(os.path.isfile(LOGDUMP_GRCTX4), + "gr logdump evidence not present") + def test_real_loginit_raw_block(self): + snapshot, stats = collect(files=[LOGDUMP_GRCTX4], + source="evidence-gr", + created=FIXED_CREATED) + chunks = self._chunks(snapshot) + self.assertIn("LOGINIT-raw", chunks) + self.assertIn("LOGRM-raw", chunks) + self.assertEqual(chunks["LOGINIT-raw"]["base"], "0x0") + self.assertEqual(len(chunks["LOGINIT-raw"]["words"]), 65536 // 4) + self.assertEqual(len(chunks["LOGRM-raw"]["words"]), 65536 // 4) + self.assertEqual(stats["unparsed"], 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py b/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py new file mode 100644 index 0000000..5c05ccd --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py @@ -0,0 +1,518 @@ +"""Tests for the engine-visible state decoder (harness/state_decode.py). + +Offline only: text/hexdump/binary/snapshot-JSON parsing, the bundled +INSTANCE/RAMFC field table, map override/extension, decode/diff reports and +the CLI. The real-evidence test parses a frozen live ladder trace and skips +when the backup is absent. No hardware, no module, no guest, no network. + +Run:: + + python3 -m unittest tests.test_state_decode -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import struct +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.state_decode import ( # noqa: E402 + DECODE_FORMAT, + DEMO_FORMAT, + DIFF_FORMAT, + MAP_FORMAT, + SNAPSHOT_FORMAT, + StateDecodeError, + decode_chunk, + decode_specs, + diff_reports, + load_input, + load_map, + main, + run_demo, +) + +EVIDENCE = os.environ.get("GREENEYE_STATE_DECODE_EVIDENCE") or os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "backups", + "evidence-20261002-victory", "06-ladder-m12.txt")) + + +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 _write(path: str, text: str) -> str: + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + return path + + +def _write_snapshot(tmp: str, name: str, chunks: list[dict]) -> str: + path = os.path.join(tmp, name) + with open(path, "w", encoding="utf-8") as handle: + json.dump({"schema": SNAPSHOT_FORMAT, "source": name, + "chunks": chunks}, handle) + return path + + +def _write_map(tmp: str, name: str, doc) -> str: + path = os.path.join(tmp, name) + with open(path, "w", encoding="utf-8") as handle: + json.dump(doc, handle) + return path + + +def _report_from_chunks(chunks: list[dict], map_path: str | None = None) -> dict: + entries, source = load_map(map_path) + return { + "format": DECODE_FORMAT, + "map_source": source, + "sources": [], + "chunks": [decode_chunk(chunk, entries) for chunk in chunks], + "limits": [], + } + + +def _chunk_field(report: dict, label: str, name: str) -> dict: + for chunk in report["chunks"]: + if chunk["label"] != label: + continue + for field in chunk["fields"]: + if field["name"] == name: + return field + raise AssertionError(f"field {name} not found in chunk {label}") + + +def _spec(path: str, label: str | None = None, base=None, + binary: bool = False) -> dict: + return {"path": path, "label": label, "base": base, "binary": binary} + + +class ParsingTests(unittest.TestCase): + def test_text_lines_parse_labels_offsets_and_values(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "lines.txt"), "\n".join([ + "banner line that must be ignored", + "INSTANCE[0x10]=0xface", + "mid-text noise", + "Instance[00ac]=0x00030000", + "INSTANCE[72]=0x1", # decimal offset 72 == 0x48 + "INSTANCE[0x4c]=64206", # decimal value 64206 == 0xface + "", + ])) + document = load_input(_spec(path)) + self.assertEqual(document["format"], "text") + self.assertEqual(len(document["chunks"]), 1) + chunk = document["chunks"][0] + self.assertEqual(chunk["label"], "INSTANCE") # first spelling wins + self.assertEqual(chunk["words"][0x10], 0xFACE) + self.assertEqual(chunk["words"][0xAC], 0x00030000) + self.assertEqual(chunk["words"][0x48], 0x1) + self.assertEqual(chunk["words"][0x4C], 64206) + + def test_hexdump_rows_parse_with_label_and_base(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "dump.txt"), "\n".join([ + "00000000: 000002d2 00000000 00000000 00000000", + "00000010: 0000face 00000000 00000000 00000000", + ])) + document = load_input(_spec(path, label="INSTANCE", + base="0x1f4000000")) + chunk = document["chunks"][0] + self.assertEqual(chunk["label"], "INSTANCE") + self.assertEqual(chunk["base"], 0x1F4000000) + self.assertEqual(chunk["words"][0x0], 0x2D2) + self.assertEqual(chunk["words"][0x4], 0x0) + self.assertEqual(chunk["words"][0x10], 0xFACE) + + def test_raw_binary_parses_little_endian_words(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "ramfc.bin") + with open(path, "wb") as handle: + handle.write(struct.pack("<4I", 0x2D2, 0, 0x30000, 0)) + document = load_input(_spec(path, label="RAMFC", binary=True)) + self.assertEqual(document["format"], "binary") + chunk = document["chunks"][0] + self.assertEqual(chunk["label"], "RAMFC") + self.assertEqual(chunk["words"][0x0], 0x2D2) + self.assertEqual(chunk["words"][0x8], 0x30000) + + def test_raw_binary_is_auto_detected(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "auto.bin") + with open(path, "wb") as handle: + handle.write(struct.pack("<8I", 1, 2, 3, 4, 5, 6, 7, 8)) + document = load_input(_spec(path, label="WINDOW")) + self.assertEqual(document["format"], "binary") + chunk = document["chunks"][0] + self.assertEqual(chunk["words"][0x1C], 8) + + def test_snapshot_json_parses_exactly(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_snapshot(tmp, "snap.json", [{ + "label": "RAMFC", "base": "0x1000", + "words": ["0x2d2", 0, "0x30000", "0x0"], + }]) + document = load_input(_spec(path)) + self.assertEqual(document["format"], "snapshot-json") + chunk = document["chunks"][0] + self.assertEqual(chunk["label"], "RAMFC") + self.assertEqual(chunk["base"], 0x1000) + self.assertEqual(chunk["words"][0x0], 0x2D2) + self.assertEqual(chunk["words"][0x8], 0x30000) + + def test_snapshot_schema_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "other.json") + with open(path, "w", encoding="utf-8") as handle: + json.dump({"schema": "something-else/1", "chunks": []}, handle) + with self.assertRaises(StateDecodeError): + load_input(_spec(path)) + + def test_unparseable_text_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "nothing.txt"), + "just prose\nwith no fields at all\n") + with self.assertRaises(StateDecodeError): + load_input(_spec(path)) + + def test_bad_binary_size_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "odd.bin") + with open(path, "wb") as handle: + handle.write(b"\x01\x02\x03") + with self.assertRaises(StateDecodeError): + load_input(_spec(path, binary=True)) + + +class DecodeTests(unittest.TestCase): + def _ramfc_target(self, word48: int) -> dict: + report = _report_from_chunks([ + {"label": "RAMFC", "base": 0, "words": {0x48: word48}}, + ]) + return _chunk_field(report, "RAMFC", "RAMFC_TARGET") + + def test_ramfc_target_both_bits_set(self): + field = self._ramfc_target(0x00030000) + self.assertEqual(field["offset"], "0x48") + self.assertEqual(field["raw"], "0x00030000") + self.assertEqual(field["kind"], "bits") + bits = {bit["name"]: bit for bit in field["bitfields"]} + self.assertEqual(bits["ENG_CTX_VALID"]["value"], 1) + self.assertEqual(bits["ENG_CTX_VALID"]["state"], "set") + self.assertEqual(bits["CE_CTX_VALID"]["value"], 1) + self.assertEqual(bits["CE_CTX_VALID"]["state"], "set") + + def test_ramfc_target_only_eng_bit16(self): + field = self._ramfc_target(0x00010000) + bits = {bit["name"]: bit for bit in field["bitfields"]} + self.assertEqual(bits["ENG_CTX_VALID"]["state"], "set") + self.assertEqual(bits["CE_CTX_VALID"]["state"], "clear") + + def test_ramfc_target_only_ce_bit17(self): + field = self._ramfc_target(0x00020000) + bits = {bit["name"]: bit for bit in field["bitfields"]} + self.assertEqual(bits["ENG_CTX_VALID"]["state"], "clear") + self.assertEqual(bits["CE_CTX_VALID"]["state"], "set") + + def test_instance_fields_and_unmapped_words(self): + report = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, + "words": {0x0: 0, 0x10: 0xFACE, 0x48: 0x100000, 0x4C: 0x30000}, + }]) + chunk = report["chunks"][0] + names = {field["name"] for field in chunk["fields"]} + self.assertEqual(names, {"MAGIC_0FACE", "GPFIFO_VA"}) + self.assertEqual(chunk["word_count"], 4) + self.assertEqual(chunk["unmapped_count"], 2) + self.assertEqual({word["offset"] for word in chunk["unmapped"]}, + {"0x0", "0x4c"}) + self.assertEqual(chunk["words"]["0x48"], "0x00100000") + + def test_decode_report_format_and_determinism(self): + report = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x48: 0x100000}, + }]) + self.assertEqual(report["format"], DECODE_FORMAT) + again = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x48: 0x100000}, + }]) + self.assertEqual(json.dumps(report, sort_keys=True), + json.dumps(again, sort_keys=True)) + + +class MapTests(unittest.TestCase): + def test_override_replaces_bundled_and_extends(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_map(tmp, "map.json", { + "format": MAP_FORMAT, + "entries": [ + {"chunk": "INSTANCE", "label": "GPFIFO_VA_ALT", + "offset": "0x48", "width": 4, "note": "override"}, + {"chunk": "INSTANCE", "label": "CTX_MARK", + "offset": "0x4c", "width": 4, + "bits": [ + {"name": "ENG", "lo": 16, "hi": 16}, + {"name": "CE", "lo": 17, "hi": 17}, + {"name": "MODE", "lo": 0, "hi": 7, + "values": {"0": "zero", "1": "one"}}, + ]}, + ], + }) + report = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, + "words": {0x10: 0xFACE, 0x48: 0x100000, 0x4C: 0x00030001}, + }], map_path=path) + chunk = report["chunks"][0] + names = [field["name"] for field in chunk["fields"]] + self.assertIn("GPFIFO_VA_ALT", names) + self.assertNotIn("GPFIFO_VA", names) + self.assertIn("MAGIC_0FACE", names) # bundled table still extends + mark = _chunk_field(report, "INSTANCE", "CTX_MARK") + bits = {bit["name"]: bit for bit in mark["bitfields"]} + self.assertEqual(bits["MODE"]["state"], "one") + self.assertEqual(bits["ENG"]["state"], "set") + self.assertEqual(bits["CE"]["state"], "set") + self.assertEqual(report["map_source"], os.path.abspath(path)) + + def test_replace_mode_drops_bundled_table(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_map(tmp, "replace.json", { + "format": MAP_FORMAT, + "mode": "replace", + "entries": [ + {"chunk": "RAMFC", "label": "TARGET_ALT", + "offset": "0x48", "width": 4, + "bits": [{"name": "VALID", "lo": 16, "hi": 16}]}, + ], + }) + report = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE}, + }, { + "label": "RAMFC", "base": 0, "words": {0x48: 0x10000}, + }], map_path=path) + self.assertEqual(report["chunks"][0]["fields"], []) + field = _chunk_field(report, "RAMFC", "TARGET_ALT") + self.assertEqual(field["bitfields"][0]["state"], "set") + + def test_bad_map_rejected(self): + cases = [ + {"entries": [{"chunk": "X", "label": "Y", "offset": "0x48", + "width": 3}]}, + {"entries": [{"chunk": "X", "label": "Y", "offset": "0x48", + "width": 4, + "bits": [{"name": "B", "lo": 0, "hi": 99}]}]}, + {"entries": [{"chunk": "X", "label": "Y", "offset": "0x0", + "width": 4}], "mode": "sideways"}, + ] + for doc in cases: + with self.subTest(doc=doc): + with tempfile.TemporaryDirectory() as tmp: + path = _write_map(tmp, "bad.json", doc) + with self.assertRaises(StateDecodeError): + load_map(path) + + +class DiffTests(unittest.TestCase): + def test_diff_detects_changed_word_and_named_field(self): + left = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, + "words": {0x10: 0xFACE, 0x48: 0x100000}, + }]) + right = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, + "words": {0x10: 0xFACE, 0x48: 0x200000}, + }]) + report = diff_reports(left, right) + self.assertEqual(report["format"], DIFF_FORMAT) + self.assertFalse(report["identical"]) + chunk = report["chunks"][0] + self.assertEqual(chunk["status"], "changed") + self.assertEqual(chunk["changed"], + [{"offset": "0x48", "old": "0x00100000", + "new": "0x00200000"}]) + self.assertEqual(len(chunk["changed_fields"]), 1) + field = chunk["changed_fields"][0] + self.assertEqual(field["name"], "GPFIFO_VA") + self.assertEqual(field["old"], "0x00100000") + self.assertEqual(field["new"], "0x00200000") + self.assertEqual(report["summary"]["changed_words"], 1) + self.assertEqual(report["summary"]["changed_fields"], 1) + + def test_diff_reports_bitfield_state_change(self): + left = _report_from_chunks([ + {"label": "RAMFC", "base": 0, "words": {0x48: 0x00030000}}]) + right = _report_from_chunks([ + {"label": "RAMFC", "base": 0, "words": {0x48: 0x00010000}}]) + report = diff_reports(left, right) + field = report["chunks"][0]["changed_fields"][0] + self.assertEqual(field["name"], "RAMFC_TARGET") + bits = {bit["name"]: bit for bit in field["bits"]} + self.assertEqual(bits["ENG_CTX_VALID"]["old"], "set") + self.assertEqual(bits["ENG_CTX_VALID"]["new"], "set") + self.assertEqual(bits["CE_CTX_VALID"]["old"], "set") + self.assertEqual(bits["CE_CTX_VALID"]["new"], "clear") + + def test_diff_identical_is_empty_and_deterministic(self): + left = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE}, + }]) + right = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x10: 0xFACE}, + }]) + report = diff_reports(left, right) + self.assertTrue(report["identical"]) + self.assertEqual(report["summary"]["changed_words"], 0) + again = diff_reports(left, right) + self.assertEqual(json.dumps(report, sort_keys=True), + json.dumps(again, sort_keys=True)) + + def test_diff_added_and_removed_words(self): + left = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x0: 1, 0x4: 2}, + }]) + right = _report_from_chunks([{ + "label": "INSTANCE", "base": 0, "words": {0x0: 1, 0x8: 3}, + }]) + report = diff_reports(left, right) + chunk = report["chunks"][0] + self.assertEqual([row["offset"] for row in chunk["removed"]], ["0x4"]) + self.assertEqual([row["offset"] for row in chunk["added"]], ["0x8"]) + self.assertEqual(report["summary"]["added"], 1) + self.assertEqual(report["summary"]["removed"], 1) + + +class CliAndDemoTests(unittest.TestCase): + def test_cli_decode_prints_table_and_writes_json(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "dump.txt"), "\n".join([ + "00000000: 00000000 00000000 00000000 00000000", + "00000010: 0000face 00000000 00000000 00000000", + "00000020: 00000000 00000000 00000000 00000000", + "00000030: 00000000 00000000 00000000 00000000", + "00000040: 00000000 00000000 00100000 00030000", + ])) + out_path = os.path.join(tmp, "decoded.json") + rc, out, err = _run_main(["decode", "--input", path, + "--label", "INSTANCE", + "--json", out_path]) + self.assertEqual(rc, 0, err) + self.assertIn("GPFIFO_VA", out) + self.assertIn("0x00100000", out) + with open(out_path, encoding="utf-8") as handle: + report = json.load(handle) + self.assertEqual(report["format"], DECODE_FORMAT) + self.assertEqual(len(report["chunks"]), 1) + self.assertEqual(_chunk_field(report, "INSTANCE", + "GPFIFO_VA")["raw"], "0x00100000") + + def test_cli_multiple_inputs_with_per_input_labels(self): + with tempfile.TemporaryDirectory() as tmp: + left = _write(os.path.join(tmp, "left.txt"), + "00000000: 00000001 00000000 00000000 00000000\n") + right = _write(os.path.join(tmp, "right.txt"), + "00000000: 00000002 00000000 00000000 00000000\n") + rc, out, err = _run_main([ + "decode", "--input", left, "--label", "LEFT", + "--input", right, "--label", "RIGHT"]) + self.assertEqual(rc, 0, err) + self.assertIn("chunk LEFT", out) + self.assertIn("chunk RIGHT", out) + + def test_cli_decode_text_lines_reports_canonical_words(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "lines.txt"), + "INSTANCE[0x48]=0x100000\n") + out_path = os.path.join(tmp, "decoded.json") + rc, _, err = _run_main(["decode", "--input", path, + "--json", out_path]) + self.assertEqual(rc, 0, err) + with open(out_path, encoding="utf-8") as handle: + report = json.load(handle) + chunk = report["chunks"][0] + self.assertEqual(chunk["label"], "INSTANCE") + self.assertEqual(chunk["words"]["0x48"], "0x00100000") + self.assertEqual(chunk["unmapped_count"], 0) + + def test_cli_diff_exit_codes(self): + with tempfile.TemporaryDirectory() as tmp: + left = _write_snapshot(tmp, "left.json", [{ + "label": "RAMFC", "base": "0x0", + "words": ["0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", + "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", + "0x0", "0x0", "0x0", "0x0", "0x30000", "0x0"], + }]) + right = _write_snapshot(tmp, "right.json", [{ + "label": "RAMFC", "base": "0x0", + "words": ["0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", + "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", "0x0", + "0x0", "0x0", "0x0", "0x0", "0x10000", "0x0"], + }]) + rc, out, err = _run_main(["diff", left, right]) + self.assertEqual(rc, 3, err) + self.assertIn("CE_CTX_VALID: set -> clear", out) + rc, out, err = _run_main(["diff", left, left]) + self.assertEqual(rc, 0, err) + self.assertIn("identical=True", out) + + def test_cli_diff_accepts_decoded_reports(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_snapshot(tmp, "snap.json", [{ + "label": "INSTANCE", "base": "0x0", + "words": ["0xface", "0x0", "0x0", "0x0"], + }]) + report_path = os.path.join(tmp, "decoded.json") + rc, _, err = _run_main(["decode", "--input", path, + "--json", report_path]) + self.assertEqual(rc, 0, err) + rc, out, err = _run_main(["diff", report_path, report_path]) + self.assertEqual(rc, 0, err) + self.assertIn("identical=True", out) + + def test_cli_demo_green(self): + rc, out, err = _run_main(["demo"]) + self.assertEqual(rc, 0, err) + self.assertIn("ok=True", out) + self.assertIn("RAMFC_TARGET", out) + self.assertIn("CE_CTX_VALID: set -> clear", out) + + def test_demo_api_is_self_contained_and_deterministic(self): + first = run_demo() + second = run_demo() + self.assertEqual(first["format"], DEMO_FORMAT) + self.assertTrue(first["ok"]) + self.assertEqual(first["diff"]["summary"]["changed_words"], 2) + self.assertEqual(first["diff"]["summary"]["changed_fields"], 2) + self.assertEqual(json.dumps(first["diff"], sort_keys=True), + json.dumps(second["diff"], sort_keys=True)) + + +@unittest.skipUnless(os.path.isfile(EVIDENCE), "ladder evidence absent") +class EvidenceTests(unittest.TestCase): + def test_ladder_trace_decodes_instance_fields(self): + entries, source = load_map(None) + report = decode_specs([_spec(EVIDENCE)], entries, source) + self.assertGreaterEqual(len(report["chunks"]), 1) + chunks = {chunk["label"]: chunk for chunk in report["chunks"]} + self.assertIn("INSTANCE", chunks) + instance = chunks["INSTANCE"] + self.assertGreater(instance["word_count"], 0) + fields = {field["name"]: field for field in instance["fields"]} + self.assertEqual(fields["GPFIFO_VA"]["raw"], "0x00100000") + self.assertEqual(fields["MAGIC_0FACE"]["raw"], "0x0000face") + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py b/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py new file mode 100644 index 0000000..439f36d --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py @@ -0,0 +1,311 @@ +"""Tests for the RM-object state-seeding harness (harness/state_seed.py, docs/52). + +Loader tests run against the native machine directly (no firmware image); the +FECS mechanism proof and the construct-gate attempt run on the shipped 570.144 +image and skip when it is absent. No hardware, no module, no guest, no +network. + +Run:: + + python3 -m unittest tests.test_state_seed -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, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.machine import Machine # noqa: E402 + +from harness.rehost import DEFAULT_IMAGE # noqa: E402 +from harness.state_seed import ( # noqa: E402 + DEMO_FORMAT, + REPORT_FORMAT, + SEED_FORMAT, + StateSeedError, + apply_seed, + fecs_seed, + load_seed, + main, + run_seed, + target_seed, +) +from harness.valprobe import encode_return_stub # noqa: E402 + +PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") + + +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 _machine() -> Machine: + return Machine(memory_mb=64, ram_base=0, unknown_mmio="lenient") + + +def _two_object_seed() -> dict: + return { + "format": SEED_FORMAT, + "name": "loader-test", + "entry": "0x5290dd4", + "allocator": {"base": "0x100000", "align": 0x100}, + "objects": [ + {"name": "client", "type": "client", "size": "0x2000", "fields": { + "0x1be0": "@ctx", + "0x1dd8": {"value": 0, "size": 8}, + "0x1aa8": "unknown", + }}, + {"name": "ctx", "type": "graphics_context", "size": "0x1000", "fields": { + "0x190": "@table", + "0x161": {"value": 0, "size": 1}, + }}, + {"name": "table", "type": "rm_object", "size": "0x100", "fields": { + "0x30": {"value": 0xCAFE, "size": 8}, + }}, + ], + "registers": {"a0": "@client", "a1": "@ctx", "t1": 0xC}, + "stack": "0x300000", + "memory": [{"pa": "0x200000", "u32": 7}], + "stubs": [{"name": "getter", "addr": "0x3f0000", "return": "0x280000"}], + "provenance": {"author": "test"}, + } + + +class SchemaTests(unittest.TestCase): + def test_layout_allocates_objects_in_order(self): + plan = load_seed(_two_object_seed()) + by_name = {item["name"]: item for item in plan["layout"]} + self.assertEqual(by_name["client"]["pa"], 0x100000) + self.assertEqual(by_name["ctx"]["pa"], 0x102000) + self.assertEqual(by_name["table"]["pa"], 0x103000) + + def test_pointer_and_value_fields_are_encoded(self): + plan = load_seed(_two_object_seed()) + writes = {item["label"]: item for item in plan["writes"]} + self.assertEqual(bytes.fromhex(writes["client+0x1be0"]["hex"]), + (0x102000).to_bytes(8, "little")) + self.assertEqual(bytes.fromhex(writes["ctx+0x190"]["hex"]), + (0x103000).to_bytes(8, "little")) + self.assertEqual(bytes.fromhex(writes["table+0x30"]["hex"]), + (0xCAFE).to_bytes(8, "little")) + self.assertEqual(bytes.fromhex(writes["mem+0x200000"]["hex"]), + (7).to_bytes(4, "little")) + + def test_unknown_marker_is_reported_and_not_written(self): + plan = load_seed(_two_object_seed()) + unknown = {(item["object"], item["off"]) for item in plan["unknown_fields"]} + self.assertIn(("client", "0x1aa8"), unknown) + labels = {item["label"] for item in plan["writes"]} + self.assertNotIn("client+0x1aa8", labels) + + def test_pc_register_becomes_entry(self): + seed = _two_object_seed() + del seed["entry"] + seed["registers"]["pc"] = "0x5290dd4" + plan = load_seed(seed) + self.assertEqual(plan["entry"], 0x5290DD4) + self.assertNotIn("pc", plan["registers"]) + + def test_plan_is_deterministic(self): + first = load_seed(_two_object_seed()) + second = load_seed(_two_object_seed()) + self.assertEqual(first["writes"], second["writes"]) + self.assertEqual(first["layout"], second["layout"]) + + def test_bad_format_rejected(self): + seed = _two_object_seed() + seed["format"] = "greeneye-state-seed/2" + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_missing_entry_rejected(self): + seed = _two_object_seed() + del seed["entry"] + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_unknown_object_type_rejected(self): + seed = _two_object_seed() + seed["objects"].append({"name": "x", "type": "gpu_of_doom"}) + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_duplicate_symbol_rejected(self): + seed = _two_object_seed() + seed["objects"].append({"name": "client", "type": "client"}) + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_unknown_pointer_rejected(self): + seed = _two_object_seed() + seed["objects"][0]["fields"]["0x1de8"] = "@does-not-exist" + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_bad_memory_item_rejected(self): + seed = _two_object_seed() + seed["memory"] = [{"pa": "0x200000"}] + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_overlapping_objects_rejected(self): + seed = _two_object_seed() + seed["objects"][1]["pa"] = "0x100800" + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_bad_register_rejected(self): + seed = _two_object_seed() + seed["registers"]["r99"] = 1 + with self.assertRaises(StateSeedError): + load_seed(seed) + + +class LoaderTests(unittest.TestCase): + def test_apply_seed_places_bytes_and_registers(self): + plan = load_seed(_two_object_seed()) + machine = _machine() + apply_seed(machine, plan) + client = 0x100000 + self.assertEqual(machine.read_mem(client + 0x1be0, 8), + (0x102000).to_bytes(8, "little")) + self.assertEqual(machine.read_mem(client + 0x1dd8, 8), b"\x00" * 8) + self.assertEqual(machine.read_mem(client + 0x1aa8, 8), b"\x00" * 8) + self.assertEqual(machine.read_mem(0x200000, 4), (7).to_bytes(4, "little")) + self.assertEqual(machine.x[10], client) + self.assertEqual(machine.x[11], 0x102000) + self.assertEqual(machine.x[6], 0xC) + self.assertEqual(machine.read_mem(0x3F0000, 12), encode_return_stub(0x280000)) + + def test_run_seed_missing_image_is_clean_error(self): + with self.assertRaises(StateSeedError): + run_seed(_two_object_seed(), image="/nonexistent/fw.bin") + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class FecsMechanismTests(unittest.TestCase): + """Prove the seeder on a path that needs no RM objects (docs/47 predicate).""" + + def test_seeded_slot_value_drives_the_verdict(self): + nonnull = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False, + name="state-seed-test") + null = run_seed(fecs_seed(slot=0), profile_path=PROFILE, record=False, + name="state-seed-test") + self.assertEqual(nonnull["format"], REPORT_FORMAT) + self.assertEqual(nonnull["status"], "outcome") + self.assertEqual(nonnull["outcome"], "refuse-fecs-slot-exists-0x1f") + self.assertEqual(null["status"], "outcome") + self.assertEqual(null["outcome"], "accept-commit-path") + + def test_fecs_run_is_deterministic(self): + first = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False, + name="state-seed-test") + second = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False, + name="state-seed-test") + self.assertEqual(first["outcome"], second["outcome"]) + self.assertEqual(first["instructions"], second["instructions"]) + self.assertEqual(first["state_reads"], second["state_reads"]) + + def test_report_records_session_and_json(self): + with tempfile.TemporaryDirectory() as tmp: + report = run_seed(fecs_seed(slot=1), profile_path=PROFILE, out=tmp, + record=True, name="state-seed-test") + self.assertTrue(os.path.isfile(report["report_json"])) + with open(report["report_json"], "r", encoding="utf-8") as fh: + doc = json.load(fh) + self.assertEqual(doc["format"], REPORT_FORMAT) + self.assertTrue(report["session"]) + self.assertTrue(os.path.isdir(report["session"])) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class TargetGateTests(unittest.TestCase): + """Attempt FUN_ram_052a70ec with a crafted client/channel/context seed.""" + + def test_minimal_seed_reports_first_gate(self): + report = run_seed(target_seed(advance=False), profile_path=PROFILE, + record=False, name="state-seed-test") + self.assertEqual(report["status"], "fault") + self.assertIsNotNone(report["gate"]) + self.assertEqual(report["gate"]["kind"], "null_indirect_call") + self.assertEqual(report["gate"]["label"], "engine_node+0x600") + self.assertEqual(report["gate"]["value"], 0) + labels = [item["label"] for item in report["state_reads"]] + self.assertIn("client+0x1de8", labels) + self.assertLess(labels.index("client+0x1de8"), + labels.index("engine_node+0x600")) + self.assertTrue(report["unknown_fields"]) + + def test_stub_advances_to_the_next_gate(self): + report = run_seed(target_seed(advance=True), profile_path=PROFILE, + record=False, name="state-seed-test") + self.assertEqual(report["status"], "fault") + self.assertEqual(report["gate"]["kind"], "null_indirect_call") + self.assertEqual(report["gate"]["label"], "channel+0x898") + + def test_target_seed_is_deterministic(self): + first = run_seed(target_seed(advance=False), profile_path=PROFILE, + record=False, name="state-seed-test") + second = run_seed(target_seed(advance=False), profile_path=PROFILE, + record=False, name="state-seed-test") + self.assertEqual(first["gate"], second["gate"]) + self.assertEqual(first["state_reads"], second["state_reads"]) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class CliTests(unittest.TestCase): + def test_emit_and_validate_round_trip(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "fecs.json") + rc, out, _ = _run_main(["emit", "--kind", "fecs", "--out", path]) + self.assertEqual(rc, 0) + self.assertTrue(os.path.isfile(path)) + rc, out, _ = _run_main(["validate", "--seed", path]) + self.assertEqual(rc, 0) + doc = json.loads(out) + self.assertEqual(doc["format"], SEED_FORMAT) + self.assertEqual(doc["entry"], "0x5290dd4") + + def test_run_emitted_seed_with_out(self): + with tempfile.TemporaryDirectory() as tmp: + seed_path = os.path.join(tmp, "fecs.json") + rc, _, _ = _run_main(["emit", "--kind", "fecs", "--out", seed_path]) + self.assertEqual(rc, 0) + out_dir = os.path.join(tmp, "out") + rc, out, _ = _run_main(["run", "--seed", seed_path, "--out", out_dir, + "--no-record"]) + self.assertEqual(rc, 0) + doc = json.loads(out) + self.assertEqual(doc["status"], "outcome") + self.assertEqual(doc["outcome"], "refuse-fecs-slot-exists-0x1f") + self.assertTrue(os.path.isfile( + os.path.join(out_dir, "state-seed-report.json"))) + + def test_run_with_bad_seed_returns_two(self): + rc, _, err = _run_main(["run", "--seed", "/nonexistent/seed.json"]) + self.assertEqual(rc, 2) + self.assertIn("error:", err) + + def test_demo_returns_zero(self): + rc, out, _ = _run_main(["demo"]) + self.assertEqual(rc, 0) + doc = json.loads(out) + self.assertEqual(doc["format"], DEMO_FORMAT) + self.assertTrue(doc["ok"]) + self.assertIsNotNone(doc["target"]["gate"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_valprobe.py b/debug/risc-v-gsp-rm-vm/tests/test_valprobe.py new file mode 100644 index 0000000..cbf530a --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_valprobe.py @@ -0,0 +1,200 @@ +"""Tests for the validation-predicate probe (harness/valprobe.py, docs/47). + +The bundled FECS-promote matrix runs offline on the shipped 570.144 image +(skipped when the image is absent). No hardware, no module, no guest, no +network. + +Run:: + + python3 -m unittest tests.test_valprobe -v +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import struct +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.decode import K_JALR, K_LUI, K_OPI, decode32 # noqa: E402 + +from harness.rehost import DEFAULT_IMAGE, _parse_u64 # noqa: E402 +from harness.valprobe import ( # noqa: E402 + MATRIX_FORMAT, + PROBE_FORMAT, + PROMOTE_VALIDATION, + SPEC_FORMAT, + ValprobeError, + decode_memory_spec, + encode_return_stub, + fecs_spec, + main, + reg_index, + run_spec, +) + +PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") + + +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() + + +class EncodingTests(unittest.TestCase): + def test_stub_decodes_as_lui_addi_ret(self): + code = encode_return_stub(0x2FFFF9800) + self.assertEqual(len(code), 12) + words = struct.unpack("> 12 + lo = struct.unpack("> 20 + if lo >= 0x800: + lo -= 0x1000 + self.assertEqual((hi << 12) + lo, value) + + def test_memory_spec_widths(self): + self.assertEqual(decode_memory_spec({"addr": "0x10", "hex": "aabb"}), + (0x10, b"\xaa\xbb")) + self.assertEqual(decode_memory_spec({"addr": "0x10", "u64": 1}), + (0x10, b"\x01" + b"\x00" * 7)) + self.assertEqual(decode_memory_spec({"addr": "0x10", "u32": 2}), + (0x10, b"\x02\x00\x00\x00")) + self.assertEqual(decode_memory_spec({"addr": "0x10", "u16": 3}), + (0x10, b"\x03\x00")) + self.assertEqual(decode_memory_spec({"addr": "0x10", "u8": 4}), + (0x10, b"\x04")) + with self.assertRaises(ValprobeError): + decode_memory_spec({"addr": "0x10"}) + + def test_register_aliases(self): + for name, index in (("s2", 18), ("x18", 18), ("a3", 13), ("t1", 6), ("fp", 8)): + self.assertEqual(reg_index(name), index) + with self.assertRaises(ValprobeError): + reg_index("r99") + + +class SpecTests(unittest.TestCase): + def test_fecs_spec_shape(self): + spec = fecs_spec() + self.assertEqual(spec["format"], SPEC_FORMAT) + self.assertEqual(_parse_u64(spec["target"]), PROMOTE_VALIDATION) + self.assertGreaterEqual(len(spec["variants"]), 15) + for variant in spec["variants"]: + self.assertTrue(variant["label"]) + self.assertIn(variant["expect"], set(spec["outcomes"].values())) + self.assertTrue(variant["stubs"]) + + def test_fecs_spec_layout_stays_inside_ram(self): + spec = fecs_spec() + profile_ram_mb = 12288 + ram_hi = profile_ram_mb * 1024 * 1024 + for variant in spec["variants"]: + for item in variant["memory"]: + self.assertLess(_parse_u64(item["addr"]) + 8, ram_hi) + + def test_missing_spec_is_clean_error(self): + rc, _, err = _run_main(["run", "--spec", "/nonexistent/spec.json"]) + self.assertEqual(rc, 2) + self.assertIn("error:", err) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class RealImageMatrixTests(unittest.TestCase): + def _reduced_spec(self) -> dict: + spec = fecs_spec() + keep = { + "FECS init vidmem, slot non-NULL", + "FECS init vidmem, slot NULL", + "FECS init sysmem attr 0x6, slot non-NULL", + "FECS no-init mapped, slot non-NULL", + "PM init", + } + spec["variants"] = [v for v in spec["variants"] if v["label"] in keep] + return spec + + def test_matrix_reproduces_live_verdicts_deterministically(self): + spec = self._reduced_spec() + first = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test") + second = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test") + self.assertEqual(first["format"], MATRIX_FORMAT) + self.assertEqual(first["evaluated"], len(spec["variants"])) + self.assertEqual(first["failed"], 0) + verdicts_a = [(r["label"], r["outcome"]) for r in first["variants"]] + verdicts_b = [(r["label"], r["outcome"]) for r in second["variants"]] + self.assertEqual(verdicts_a, verdicts_b) + by_label = dict(verdicts_a) + self.assertEqual(by_label["FECS init vidmem, slot non-NULL"], + "refuse-fecs-slot-exists-0x1f") + self.assertEqual(by_label["FECS init vidmem, slot NULL"], "accept-commit-path") + self.assertEqual(by_label["FECS no-init mapped, slot non-NULL"], "accept-skip-path") + + def test_probe_reports_firmware_slot_index_and_stub(self): + spec = self._reduced_spec() + spec["variants"] = [v for v in spec["variants"] + if v["label"] == "FECS init vidmem, slot non-NULL"] + matrix = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test") + row = matrix["variants"][0] + self.assertEqual(row["format"], PROBE_FORMAT) + self.assertEqual(row["status"], "outcome") + self.assertEqual(row["instructions"], row["instructions"]) + self.assertTrue(row["stubs"]) + slot_index = [r for r in row["reads"] if r.get("label") == "firmware-mapped-slot-index"] + self.assertEqual(slot_index[0]["u32"], 6) # firmware maps FECS -> slot 6 + + def test_boot_to_b3_then_probe(self): + spec = self._reduced_spec() + spec["variants"] = [v for v in spec["variants"] + if v["label"] == "FECS init vidmem, slot non-NULL"] + spec["boot_to"] = "B3" + matrix = run_spec(spec=spec, profile_path=PROFILE, record=False, name="valprobe-test") + row = matrix["variants"][0] + self.assertIsNotNone(row["boot"]) + self.assertEqual(row["boot"]["rung"], "B3") + self.assertTrue(row["boot"]["reached"]) + self.assertEqual(row["outcome"], "refuse-fecs-slot-exists-0x1f") + + def test_matrix_writes_report_json(self): + spec = self._reduced_spec() + spec["variants"] = spec["variants"][:2] + with tempfile.TemporaryDirectory() as tmp: + matrix = run_spec(spec=spec, profile_path=PROFILE, out=tmp, + record=False, name="valprobe-test") + self.assertTrue(os.path.isfile(os.path.join(tmp, "valprobe-matrix.json"))) + with open(matrix["report_json"], "r", encoding="utf-8") as fh: + doc = json.load(fh) + self.assertEqual(doc["format"], MATRIX_FORMAT) + self.assertEqual(len(doc["variants"]), 2) + + def test_cli_sweep_returns_zero(self): + rc, out, _ = _run_main(["sweep", "--no-record"]) + self.assertEqual(rc, 0) + self.assertIn("PASS", out) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3