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_msgq_attach.py | 342 +++++++++++++++++++++++ 1 file changed, 342 insertions(+) create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py (limited to 'debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py') 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() -- cgit v1.2.3