path: root/debug/risc-v-gsp-rm-vm/tests/test_state_decode.py

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