From 6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf 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_state_decode.py | 518 ++++++++++++++++++++++ 1 file changed, 518 insertions(+) create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_state_decode.py (limited to 'debug/risc-v-gsp-rm-vm/tests/test_state_decode.py') 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() -- cgit v1.2.3