"""Tests for the engine-visible state decoder (harness/state_decode.py). 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:: 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 AGGREGATE_FORMAT, DECODE_FORMAT, DEMO_FORMAT, DIFF_FORMAT, MAP_FORMAT, SNAPSHOT_FORMAT, StateDecodeError, decode_chunk, decode_specs, diff_reports, load_input, load_map, main, render_diff, 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")) # 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() 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 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([ {"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 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: 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") @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()