"""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()