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