diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_libos_port.py | 186 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py | 114 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_state_decode.py | 331 |
3 files changed, 616 insertions, 15 deletions
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 index 0f01c01..befa14f 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_libos_port.py @@ -29,11 +29,15 @@ if TOOL_ROOT not in sys.path: from sim.libos import INTERRUPT_ARM_EXTERNAL, TASK_PORT_SEND_RECV_WAIT # noqa: E402 from sim.libos_port import ( # noqa: E402 EXTINTR_WAKE_CONTRACT, + EXTERNAL_INTERRUPT_XCAUSE, HANDSHAKE_CONTRACT, + NETWORK_REPORT_FORMAT, REPORT_FORMAT, SHUTTLE_ANY, + SPURIOUS_XEPC, TIMEOUT_INFINITE, LibosPortModel, + LibosPortNetwork, role_for, ) @@ -50,6 +54,26 @@ class FakeMachine: self.x[index] = value & 0xFFFFFFFFFFFFFFFF +class FakeMemoryMachine(FakeMachine): + """Register file plus flat memory for the routing model tests.""" + + def __init__(self, size: int = 0x10000): + super().__init__() + self.ram = bytearray(size) + self.ram_base = 0 + self.ram_size = size + + def read_mem(self, addr: int, size: int) -> bytes: + if addr < 0 or addr + size > self.ram_size: + raise ValueError("out of range") + return bytes(self.ram[addr:addr + size]) + + def write_mem(self, addr: int, data: bytes) -> None: + if addr < 0 or addr + len(data) > self.ram_size: + raise ValueError("out of range") + self.ram[addr:addr + len(data)] = data + + class FakeBase: """Minimal base syscall model: handles arm, records and stops the rest.""" @@ -287,5 +311,167 @@ class StateTests(unittest.TestCase): self.assertNotIn(str(key), ("t", "ts", "timestamp", "time")) +class SyncPortTests(unittest.TestCase): + """Synchronous receive semantics (waitId != 0 must block, not fake OK).""" + + def test_sync_recv_without_completion_blocks(self): + model = LibosPortModel(base=FakeBase()) + machine = FakeMachine() + _port_call(machine, recv=(9, 8, 0x100, 0x10), wait=9) + handled, result = model.handle(machine, pc=0x4019B00) + self.assertFalse(handled) + self.assertIsNone(result) + self.assertEqual(model.stopped["reason"], "sync-recv-blocks") + self.assertEqual(model.waits[-1]["reason"], "sync-recv-blocks") + self.assertEqual(model.waits[-1]["shuttle"], 9) + + def test_sync_recv_returns_a_queued_completion(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, recv=(9, 8, 0x200, 0x10), wait=9) + handled, result = model.handle(machine, pc=0x4019B00) + self.assertTrue(handled) + self.assertEqual(result, 0) + self.assertEqual(machine.x[16], 10) + self.assertEqual(model.waits[-1]["completed"], 10) + + +class NetworkModelTests(unittest.TestCase): + """The two-task routing model (opt-in, additive).""" + + def _network(self): + return LibosPortNetwork(base=FakeBase(), tasks=("init", "rm"), + current="init") + + def test_state_format_is_the_network_schema(self): + model = self._network() + self.assertEqual(model.state()["format"], NETWORK_REPORT_FORMAT) + self.assertEqual(model.state()["tasks"], ["init", "rm"]) + + def test_cross_task_send_copies_the_firmware_payload(self): + model = self._network() + machine = FakeMemoryMachine() + model.switch("init", pc=0) + _port_call(machine, recv=(9, 8, 0x1000, 0x808)) + model.handle(machine, pc=0x4019B00) + machine.write_mem(0x2000, b"\x11\x22\x33\x44") + model.switch("rm", pc=0) + _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), + wait=2) + handled, _ = model.handle(machine, pc=0x5AD507C) + self.assertFalse(handled) # the sync receive blocks + self.assertEqual(model.stopped["reason"], "sync-recv-blocks") + completions = model.task_completions["init"] + self.assertEqual(len(completions), 1) + self.assertEqual(completions[0]["payload_size"], 4) + self.assertEqual(completions[0]["origin_task"], 1) + self.assertEqual(machine.read_mem(0x1000, 4), b"\x11\x22\x33\x44") + self.assertEqual(model.cross_task_payloads, + [{"port": 8, "size": 4, "from": "rm", "to": "init", + "kind": "send"}]) + + def test_reply_completes_the_sender_receive_and_resume_is_idempotent(self): + model = self._network() + machine = FakeMemoryMachine() + model.switch("init", pc=0) + _port_call(machine, recv=(9, 8, 0x1000, 0x808)) + model.handle(machine, pc=0x4019B00) + machine.write_mem(0x2000, b"\xab\xcd\xef\x01") + model.switch("rm", pc=0) + _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), + wait=2) + model.handle(machine, pc=0x5AD507C) + model.stopped = None + model.switch("init", pc=0) + _port_call(machine, wait=SHUTTLE_ANY) + handled, _ = model.handle(machine, pc=0x4019DF4) + self.assertTrue(handled) + self.assertEqual(machine.x[16], 9) + machine.write_mem(0x1000, b"\x7f") + _port_call(machine, send=(9, 0, 0x1000, 1), wait=9) + handled, _ = model.handle(machine, pc=0x401829C) + self.assertTrue(handled) + model.stopped = None + model.switch("rm", pc=0) + _port_call(machine, send=(1, 8, 0x2000, 4), recv=(2, 11, 0x3000, 0x100), + wait=2) + handled, _ = model.handle(machine, pc=0x5AD507C) + self.assertTrue(handled) + self.assertEqual(machine.x[16], 2) + self.assertEqual(machine.read_mem(0x3000, 1), b"\x7f") + # the resumed receive did not re-post a stale entry + self.assertNotIn(2, model.task_recvs["rm"]) + self.assertEqual(model.cross_task_payloads[-1], + {"port": 0, "size": 1, "from": "init", "to": "rm", + "kind": "reply"}) + + def test_task_state_query_serves_the_modelled_trap_state(self): + model = self._network() + machine = FakeMemoryMachine() + model.switch("init", pc=0) + 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) + model.handle(machine, pc=0x4019DF4) + machine.set(5, 6) + machine.set(10, 1) + machine.set(11, 0x4000) + handled, result = model.handle(machine, pc=0x401A46C) + self.assertTrue(handled) + self.assertEqual(result, 0) + state = machine.read_mem(0x4000, 288) + self.assertEqual(int.from_bytes(state[256:264], "little"), SPURIOUS_XEPC) + self.assertEqual(int.from_bytes(state[264:272], "little"), + EXTERNAL_INTERRUPT_XCAUSE) + self.assertEqual(model.queries[-1]["task_index"], 1) + + def test_memory_query_and_shuttle_reset_are_services(self): + model = self._network() + machine = FakeMemoryMachine() + machine.set(5, 5) + machine.set(10, 0xFF) + machine.set(11, 0x1000) + handled, result = model.handle(machine, pc=0x5AC44FC) + self.assertTrue(handled) + self.assertEqual(result, 0) + self.assertEqual(machine.x[12], 0x1000) + self.assertEqual(machine.x[14], 1) + machine.set(5, 7) + machine.set(10, 1) + handled, result = model.handle(machine, pc=0x5AB7D8C) + self.assertTrue(handled) + self.assertEqual(result, 0) + self.assertEqual([s["service"] for s in model.services], + ["memory_query", "shuttle_reset"]) + + def test_network_state_is_byte_identical_across_runs(self): + def scenario(): + model = self._network() + machine = FakeMemoryMachine() + model.switch("init", pc=0) + _port_call(machine, recv=(9, 8, 0x1000, 0x808)) + model.handle(machine, pc=0x4019B00) + model.switch("rm", pc=0) + _port_call(machine, send=(1, 8, 0x2000, 4), + recv=(2, 11, 0x3000, 0x100), wait=2) + model.handle(machine, pc=0x5AD507C) + return model.state() + + first = scenario() + second = scenario() + self.assertEqual(json.dumps(first, sort_keys=True), + json.dumps(second, sort_keys=True)) + + 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 index 749157e..90643e9 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_msgq_attach.py @@ -43,7 +43,9 @@ from sim.msgq_attach import ( # noqa: E402 RingGeometry, assemble_report, classify_ring_access, + image_tasks, missing_semantic_after_handoff, + missing_semantic_after_second_task, ring_geometry_from_host, run_attach, ) @@ -236,6 +238,69 @@ class ReportTests(unittest.TestCase): max_cycles=MAX_RUN_CYCLES + 1) with self.assertRaises(MsgqAttachError): run_attach(image="/nonexistent/gsp-image.bin", max_cycles=True) + with self.assertRaises(MsgqAttachError): + run_attach(image="/nonexistent/gsp-image.bin", + second_task_budget=-1) + with self.assertRaises(MsgqAttachError): + run_attach(image="/nonexistent/gsp-image.bin", + second_task_budget=MAX_RUN_CYCLES + 1) + + +class TaskTableTests(unittest.TestCase): + """Data-driven task discovery from the real image section table.""" + + @unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent") + def test_two_tasks_are_discovered_with_entries_and_stacks(self): + doc = image_tasks(DEFAULT_IMAGE) + self.assertEqual(doc["format"], "greeneye-image-tasks/1") + names = [task["name"] for task in doc["tasks"]] + self.assertEqual(names, ["init", "rm"]) + for task in doc["tasks"]: + self.assertGreaterEqual(task["entry"], task["text_base"]) + self.assertLess(task["entry"], + task["text_base"] + task["text_size"]) + self.assertGreater(task["stack_top"], task["stack_base"]) + self.assertGreater(task["text_size"], 0) + rm = doc["tasks"][1] + self.assertGreater(rm["text_size"], doc["tasks"][0]["text_size"]) + + def test_absent_image_yields_an_empty_table(self): + doc = image_tasks("/nonexistent/image.bin") + self.assertEqual(doc, {"format": "greeneye-image-tasks/1", "tasks": []}) + + def test_second_task_missing_semantic_is_precise(self): + doc = image_tasks(DEFAULT_IMAGE) if os.path.isfile(DEFAULT_IMAGE) \ + else {"format": "greeneye-image-tasks/1", + "tasks": [{"name": "init"}, {"name": "rm"}]} + missing = missing_semantic_after_second_task( + tasks_doc=doc, + schedule=[{"task": "init", "reason": "port-wait-blocks"}, + {"task": "rm", "reason": "scheduled"}], + frontier={"task": "rm", "reason": "second-task-slice-budget"}, + polls=[]) + self.assertEqual(missing["stage"], "second-task-frontier") + self.assertEqual(missing["polls"], 0) + self.assertEqual(missing["second_task"], "rm") + self.assertGreaterEqual(len(missing["missing"]), 3) + self.assertTrue(missing["next_step"]) + + def test_report_carries_tasks_and_schedule_additively(self): + 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}, + tasks={"format": "greeneye-image-tasks/1", + "tasks": [{"name": "init"}, {"name": "rm"}]}, + schedule=[{"task": "rm", "reason": "scheduled"}], + ) + self.assertEqual([task["name"] for task in report["tasks"]["tasks"]], + ["init", "rm"]) + self.assertEqual(report["schedule"][0]["task"], "rm") + # the schema tag and the original keys are untouched + self.assertEqual(report["format"], REPORT_FORMAT) + self.assertIn("first_poll", report) + self.assertIn("missing_semantic", report) class PortModelReportTests(unittest.TestCase): @@ -285,12 +350,12 @@ 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. + needs ~474k). The scheduled second task (the image-discovered RM task) + completes its boot port handshake inside the bounded slice and then runs + its own start-up past the cap, so the report is expected to take the + honest ``wait-reached-no-poll`` frontier path -- this test accepts any of + the allowed outcomes and asserts determinism, never success. A fabricated + poll is never accepted. """ def test_bounded_run_is_deterministic_and_honest(self): @@ -306,6 +371,12 @@ class ImageTests(unittest.TestCase): self.assertGreater(report["image"]["size"], 0) self.assertEqual(len(report["image"]["sha256"]), 64) self.assertTrue(report["port_model"]["enabled"]) + self.assertEqual(report["port_model"]["format"], + "greeneye-msgq-port-network/1") + self.assertEqual([task["name"] for task in report["tasks"]["tasks"]], + ["init", "rm"]) + self.assertTrue(report["schedule"]) + self.assertIn("rm", [step.get("task") for step in report["schedule"]]) if not report["boot"]["message_loop_wait_reached"]: self.assertIsNotNone(report["divergence"]) self.assertTrue(report["divergence"]["reason"]) @@ -313,17 +384,22 @@ class ImageTests(unittest.TestCase): 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. + if divergence and divergence.get("kind") == "second-task-frontier": + # The doorbell wake was consumed by the init task and the + # discovered RM task completed the boot handshake before its + # start-up ran past the bounded slice; the exact frontier 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") + "second-task-frontier") completed = [wait.get("completed") for wait in report["port_model"].get("waits", [])] self.assertIn(10, completed) + payloads = report["port_model"].get("cross_task_payloads", []) + self.assertTrue(any(item.get("from") == "rm" + and item.get("to") == "init" + for item in payloads)) self.assertEqual(first["boot"]["message_loop_wait_reached"], second["boot"]["message_loop_wait_reached"]) self.assertEqual(first["boot"]["instructions_executed"], @@ -334,9 +410,25 @@ class ImageTests(unittest.TestCase): 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["tasks"], second["tasks"]) + self.assertEqual(first["schedule"], second["schedule"]) self.assertEqual(first["determinism"]["report_sha256"], second["determinism"]["report_sha256"]) + def test_single_task_mode_keeps_the_handoff_classification(self): + report = run_attach(image=DEFAULT_IMAGE, max_cycles=SMALL_BUDGET, + doorbell=True, second_task=False) + self.assertIn(report["outcome"], + ("not-reached", "wait-reached-no-poll", + "wait-reached-poll-classified")) + self.assertEqual(report["port_model"]["format"], "greeneye-msgq-port/1") + self.assertFalse(report["schedule"]) + divergence = report["divergence"] + if divergence and divergence.get("kind") == "task-state-query": + self.assertIsNone(report["first_poll"]) + self.assertEqual(report["missing_semantic"]["stage"], + "after-extintr-handoff") + 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 index 5c05ccd..3485ced 100644 --- a/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py +++ b/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py @@ -1,9 +1,10 @@ """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. +Offline only: text/hexdump/binary/snapshot-JSON/aggregate-group parsing, the +bundled INSTANCE/RAMFC field table, map override/extension, decode/diff +reports and the CLI. The real-evidence tests parse frozen live ladder traces +and an aggregate GSP physical-state dump, and skip when the backups are +absent. No hardware, no module, no guest, no network. Run:: @@ -26,6 +27,7 @@ if TOOL_ROOT not in sys.path: sys.path.insert(0, TOOL_ROOT) from harness.state_decode import ( # noqa: E402 + AGGREGATE_FORMAT, DECODE_FORMAT, DEMO_FORMAT, DIFF_FORMAT, @@ -38,6 +40,7 @@ from harness.state_decode import ( # noqa: E402 load_input, load_map, main, + render_diff, run_demo, ) @@ -45,6 +48,17 @@ 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")) +# Our own evidence artifact directory (not a vendor path). The aggregate +# dump is one group per line: ``stage: primed=1 kicked=1 tx_seq_num=106``. +AGGREGATE_EVIDENCE = os.environ.get( + "GREENEYE_STATE_DECODE_EVIDENCE_AGG") or os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "backups", + "evidence-20261002-victory", "20-gsp-phys-state-m26.txt")) +AGGREGATE_EVIDENCE_PEER = os.environ.get( + "GREENEYE_STATE_DECODE_EVIDENCE_AGG_PEER") or os.path.normpath( + os.path.join(TOOL_ROOT, "..", "..", "backups", + "evidence-20261002-victory", "20-gsp-phys-state-m29.txt")) + def _run_main(argv: list[str]) -> tuple[int, str, str]: out, err = io.StringIO(), io.StringIO() @@ -197,6 +211,147 @@ class ParsingTests(unittest.TestCase): load_input(_spec(path, binary=True)) +class AggregateParsingTests(unittest.TestCase): + def _decode(self, tmp: str, name: str, text: str) -> dict: + entries, source = load_map(None) + path = _write(os.path.join(tmp, name), text) + return decode_specs([_spec(path)], entries, source) + + def test_aggregate_line_decodes_types_and_preserves_raw(self): + text = "\n".join([ + "stage: primed=1 kicked=1 tx_seq_num=106 variant=V6", + "ctx: pa=0x1f4000000 size=0x133000 last_ret=-5 " + "userd=sysmem (host-writable)", + ]) + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "agg.txt"), text) + document = load_input(_spec(path)) + self.assertEqual(document["format"], AGGREGATE_FORMAT) + self.assertEqual([chunk["label"] for chunk in document["chunks"]], + ["PHYS_stage", "PHYS_ctx"]) + stage = document["chunks"][0] + self.assertEqual(stage["format"], AGGREGATE_FORMAT) + self.assertIn("stage: primed=1", stage["raw"]) + entries, source = load_map(None) + report = decode_specs([_spec(path)], entries, source) + chunks = {chunk["label"]: chunk for chunk in report["chunks"]} + stage = chunks["PHYS_stage"] + self.assertEqual(stage["input_format"], AGGREGATE_FORMAT) + self.assertEqual(stage["word_count"], 0) + self.assertEqual(stage["kv_count"], 4) + self.assertEqual(stage["raw"], + "stage: primed=1 kicked=1 tx_seq_num=106 " + "variant=V6") + fields = {field["name"]: field for field in stage["fields"]} + self.assertEqual(fields["primed"]["value"], 1) + self.assertTrue(fields["primed"]["numeric"]) + self.assertEqual(fields["tx_seq_num"]["value"], 106) + self.assertEqual(fields["variant"]["value"], "V6") + self.assertFalse(fields["variant"]["numeric"]) + self.assertEqual(fields["variant"]["raw"], "V6") + self.assertEqual(fields["variant"]["kind"], "kv") + self.assertIsNone(fields["variant"]["offset"]) + ctx = {field["name"]: field for field in chunks["PHYS_ctx"]["fields"]} + self.assertEqual(ctx["pa"]["value"], 0x1F4000000) + self.assertEqual(ctx["size"]["value"], 0x133000) + self.assertEqual(ctx["last_ret"]["value"], -5) + self.assertEqual(ctx["userd"]["value"], "sysmem") + self.assertNotIn("host-writable", ctx) + + def test_aggregate_missing_value_and_unparsed_lines(self): + with tempfile.TemporaryDirectory() as tmp: + report = self._decode(tmp, "agg.txt", "\n".join([ + "banner line that must be ignored", + "status_ring: writePtr slot=29; last produced elements " + "(max 8):", + "peer: host status readPtr @cmd+0x20 = 8", + "empty: a= b=1", + ])) + self.assertEqual([chunk["label"] for chunk in report["chunks"]], + ["PHYS_empty"]) + fields = {field["name"]: field for field in + report["chunks"][0]["fields"]} + self.assertEqual(fields["a"]["value"], "") + self.assertEqual(fields["a"]["raw"], "") + self.assertFalse(fields["a"]["numeric"]) + self.assertEqual(fields["b"]["value"], 1) + + def test_aggregate_single_kv_group_is_captured(self): + with tempfile.TemporaryDirectory() as tmp: + report = self._decode(tmp, "agg.txt", "guest_ram: active=0\n") + self.assertEqual([chunk["label"] for chunk in report["chunks"]], + ["PHYS_guest_ram"]) + field = report["chunks"][0]["fields"][0] + self.assertEqual(field["name"], "active") + self.assertEqual(field["value"], 0) + + def test_aggregate_pipe_groups_and_duplicate_keys(self): + with tempfile.TemporaryDirectory() as tmp: + report = self._decode(tmp, "agg.txt", "\n".join([ + "rm_call_userd: userd=sysmem (host-writable) | " + "rm_call_pdb: vaBase=0x100000 vaGpfifo=0x200000", + "queue_dma: shared=0x1a86b6000 ptes=129 | " + "radix3=0x105370000 fmc=0x142d25000", + "guest_ram: active=1 gpa=0x0 gpa=0x100000000 " + "size=0x80000000 size=0x80000000", + ])) + self.assertEqual([chunk["label"] for chunk in report["chunks"]], [ + "PHYS_guest_ram", "PHYS_queue_dma", "PHYS_rm_call_pdb", + "PHYS_rm_call_userd", + ]) + chunks = {chunk["label"]: chunk for chunk in report["chunks"]} + userd = {field["name"]: field for field in + chunks["PHYS_rm_call_userd"]["fields"]} + self.assertEqual(userd["userd"]["value"], "sysmem") + pdb = [field["name"] for field in chunks["PHYS_rm_call_pdb"]["fields"]] + self.assertEqual(pdb, ["vaBase", "vaGpfifo"]) + queue = [field["name"] for field in chunks["PHYS_queue_dma"]["fields"]] + self.assertEqual(queue, ["shared", "ptes", "radix3", "fmc"]) + gpa = [field for field in chunks["PHYS_guest_ram"]["fields"] + if field["name"] == "gpa"] + self.assertEqual([field["index"] for field in gpa], [0, 1]) + self.assertEqual([field["value"] for field in gpa], + [0, 0x100000000]) + + def test_mixed_word_and_aggregate_lines(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "mixed.txt"), "\n".join([ + "INSTANCE[0x10]=0xface", + "stage: a=1 b=2", + ])) + document = load_input(_spec(path)) + self.assertEqual(document["format"], "text") + self.assertEqual([chunk["label"] for chunk in document["chunks"]], + ["INSTANCE", "PHYS_stage"]) + + def test_unstructured_text_still_rejected(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "none.txt"), "\n".join([ + "status_ring: writePtr slot=29; last produced elements:", + "peer: host status readPtr @cmd+0x20 = 8", + ])) + with self.assertRaises(StateDecodeError): + load_input(_spec(path)) + + def test_aggregate_decode_is_deterministic(self): + with tempfile.TemporaryDirectory() as tmp: + text = "stage: a=1 b=0x10 c=up\n" + first = self._decode(tmp, "one.txt", text) + second = self._decode(tmp, "two.txt", text) + stripped = [ + {key: value for key, value in chunk.items() + if key != "source"} + for chunk in first["chunks"] + ] + other = [ + {key: value for key, value in chunk.items() + if key != "source"} + for chunk in second["chunks"] + ] + self.assertEqual(json.dumps(stripped, sort_keys=True), + json.dumps(other, sort_keys=True)) + + class DecodeTests(unittest.TestCase): def _ramfc_target(self, word48: int) -> dict: report = _report_from_chunks([ @@ -394,6 +549,132 @@ class DiffTests(unittest.TestCase): self.assertEqual(report["summary"]["removed"], 1) +class AggregateDiffTests(unittest.TestCase): + def _pair(self, tmp: str, left_text: str, right_text: str) -> tuple: + entries, source = load_map(None) + left = _write(os.path.join(tmp, "left.txt"), left_text) + right = _write(os.path.join(tmp, "right.txt"), right_text) + return ( + decode_specs([_spec(left)], entries, source), + decode_specs([_spec(right)], entries, source), + ) + + def test_diff_classifies_added_removed_changed_kvs(self): + with tempfile.TemporaryDirectory() as tmp: + left, right = self._pair( + tmp, + "stage: a=1 b=2 gone=1 variant=V6\n", + "stage: a=1 b=3 variant=V7 extra=4\n") + report = diff_reports(left, right) + self.assertEqual(report["format"], DIFF_FORMAT) + self.assertFalse(report["identical"]) + chunk = report["chunks"][0] + self.assertEqual(chunk["label"], "PHYS_stage") + self.assertEqual(chunk["status"], "changed") + self.assertEqual(chunk["added"], []) + self.assertEqual(chunk["removed"], []) + self.assertEqual(chunk["changed"], []) + rows = {row["name"]: row for row in chunk["changed_fields"]} + self.assertEqual(rows["b"]["status"], "changed") + self.assertEqual((rows["b"]["old"], rows["b"]["new"]), (2, 3)) + self.assertEqual(rows["variant"]["status"], "changed") + self.assertEqual((rows["variant"]["old"], rows["variant"]["new"]), + ("V6", "V7")) + self.assertEqual(rows["extra"]["status"], "added") + self.assertIsNone(rows["extra"]["old"]) + self.assertEqual(rows["extra"]["new"], 4) + self.assertEqual(rows["gone"]["status"], "removed") + self.assertEqual(rows["gone"]["old"], 1) + self.assertIsNone(rows["gone"]["new"]) + self.assertEqual([row["name"] for row in chunk["fields_added"]], + ["extra"]) + self.assertEqual([row["name"] for row in chunk["fields_removed"]], + ["gone"]) + self.assertEqual([row["name"] for row in chunk["fields_changed"]], + ["b", "variant"]) + self.assertEqual(report["summary"]["fields_added"], 1) + self.assertEqual(report["summary"]["fields_removed"], 1) + self.assertEqual(report["summary"]["fields_changed"], 2) + self.assertEqual(report["summary"]["changed_fields"], 4) + again = diff_reports(left, right) + self.assertEqual(json.dumps(report, sort_keys=True), + json.dumps(again, sort_keys=True)) + + def test_diff_aggregate_identical_and_one_sided(self): + with tempfile.TemporaryDirectory() as tmp: + left, right = self._pair(tmp, "stage: a=1 b=up\n", + "stage: a=1 b=up\n") + report = diff_reports(left, right) + self.assertTrue(report["identical"]) + self.assertEqual(report["summary"]["fields_added"], 0) + self.assertEqual(report["summary"]["fields_removed"], 0) + self.assertEqual(report["summary"]["fields_changed"], 0) + only_left, only_right = self._pair( + tmp, "stage: a=1 b=2\n", "ctx: x=1\n") + report = diff_reports(only_left, only_right) + labels = {chunk["label"]: chunk for chunk in report["chunks"]} + self.assertEqual(labels["PHYS_stage"]["status"], "left_only") + self.assertEqual( + [row["name"] for row in labels["PHYS_stage"]["fields_removed"]], + ["a", "b"]) + self.assertEqual(labels["PHYS_ctx"]["status"], "right_only") + self.assertEqual( + [row["name"] for row in labels["PHYS_ctx"]["fields_added"]], + ["x"]) + + def test_diff_aggregate_duplicate_key_occurrence(self): + with tempfile.TemporaryDirectory() as tmp: + left, right = self._pair( + tmp, "mem: size=0x10 size=0x20\n", + "mem: size=0x10 size=0x30\n") + chunk = diff_reports(left, right)["chunks"][0] + self.assertEqual(len(chunk["fields_changed"]), 1) + row = chunk["fields_changed"][0] + self.assertEqual(row["name"], "size") + self.assertEqual(row["index"], 1) + self.assertEqual((row["old"], row["new"]), (0x20, 0x30)) + + def test_cli_diff_aggregate_exit_codes_and_kv_rendering(self): + with tempfile.TemporaryDirectory() as tmp: + left = _write(os.path.join(tmp, "left.txt"), + "stage: a=1 b=2 state=up\n") + right = _write(os.path.join(tmp, "right.txt"), + "stage: a=1 b=3 state=up\n") + rc, out, err = _run_main(["diff", left, right]) + self.assertEqual(rc, 3, err) + self.assertIn("PHYS_stage", out) + self.assertIn("kv b changed: 2 -> 3", out) + rc, out, err = _run_main(["diff", left, left]) + self.assertEqual(rc, 0, err) + self.assertIn("identical=True", out) + + def test_render_diff_aggregate_handles_added_and_removed(self): + with tempfile.TemporaryDirectory() as tmp: + left, right = self._pair(tmp, "stage: gone=1\n", + "stage: extra=2\n") + text = render_diff(diff_reports(left, right)) + self.assertIn("kv extra added: - -> 2", text) + self.assertIn("kv gone removed: 1 -> -", text) + + def test_cli_decode_aggregate_writes_report(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(os.path.join(tmp, "agg.txt"), + "stage: primed=1 variant=V6\n") + out_path = os.path.join(tmp, "agg.json") + rc, out, err = _run_main(["decode", "--input", path, + "--json", out_path]) + self.assertEqual(rc, 0, err) + self.assertIn("PHYS_stage", out) + self.assertIn("(aggregate)", out) + with open(out_path, encoding="utf-8") as handle: + report = json.load(handle) + self.assertEqual(report["format"], DECODE_FORMAT) + self.assertEqual(report["sources"][0]["format"], AGGREGATE_FORMAT) + self.assertEqual(report["chunks"][0]["input_format"], + AGGREGATE_FORMAT) + self.assertEqual(report["chunks"][0]["fields"][1]["value"], "V6") + + class CliAndDemoTests(unittest.TestCase): def test_cli_decode_prints_table_and_writes_json(self): with tempfile.TemporaryDirectory() as tmp: @@ -514,5 +795,47 @@ class EvidenceTests(unittest.TestCase): self.assertEqual(fields["MAGIC_0FACE"]["raw"], "0x0000face") +@unittest.skipUnless(os.path.isfile(AGGREGATE_EVIDENCE), + "aggregate GSP physical-state evidence absent") +class AggregateEvidenceTests(unittest.TestCase): + """Frozen GSP physical-state dumps from our own evidence directory. + + Override with GREENEYE_STATE_DECODE_EVIDENCE_AGG (and ..._PEER); skipped + when the files are absent. No hardware, no module, no guest, no network. + """ + + def test_aggregate_evidence_decodes_groups(self): + entries, source = load_map(None) + report = decode_specs([_spec(AGGREGATE_EVIDENCE)], entries, source) + self.assertGreater(len(report["chunks"]), 0) + self.assertEqual(report["sources"][0]["format"], AGGREGATE_FORMAT) + chunks = {chunk["label"]: chunk for chunk in report["chunks"]} + for label in ("PHYS_stage", "PHYS_rm_call", "PHYS_guest_ram", + "PHYS_rm_call_ctx"): + self.assertIn(label, chunks) + stage = {field["name"]: field for field in + chunks["PHYS_stage"]["fields"]} + self.assertEqual(stage["primed"]["value"], 1) + self.assertTrue(stage["tx_seq_num"]["numeric"]) + + def test_aggregate_evidence_diff_peer_shows_changed_kvs(self): + if not os.path.isfile(AGGREGATE_EVIDENCE_PEER): + self.skipTest("aggregate peer evidence absent") + entries, source = load_map(None) + left = decode_specs([_spec(AGGREGATE_EVIDENCE)], entries, source) + right = decode_specs([_spec(AGGREGATE_EVIDENCE_PEER)], entries, source) + report = diff_reports(left, right) + self.assertFalse(report["identical"]) + self.assertGreater(report["summary"]["fields_changed"], 0) + self.assertEqual(report["summary"]["changed_words"], 0) + chunks = {chunk["label"]: chunk for chunk in report["chunks"]} + rows = {row["name"]: row for row in + chunks["PHYS_guest_ram"]["changed_fields"]} + self.assertEqual(rows["active"]["status"], "changed") + self.assertEqual((rows["active"]["old"], rows["active"]["new"]), + (1, 0)) + render_diff(report) + + if __name__ == "__main__": unittest.main() |