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

"""Tests for the GMMU/FBIF static<->dynamic join (harness/gmmu_debug.py, docs/44).

All fixtures are synthetic: no firmware, no simulator run, no QEMU, no
hardware.  The end-to-end demo test builds its own fixture in a temporary
directory and checks the frozen threshold result.

Run::

    python3 -m unittest tests.test_gmmu_debug -v
"""
from __future__ import annotations

import contextlib
import io
import json
import os
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.gmmu_debug import (  # noqa: E402
    DEBUG_REPORT_FORMAT,
    HITS_HEADER,
    MANUAL_STATIC_BASELINE_SECONDS,
    TRACE_FORMAT,
    WORKFLOW_THRESHOLD_FRACTION,
    GmmuInputError,
    build_debug_report,
    load_sim_trace,
    load_symbols,
    main,
    run_demo,
    run_report,
    write_hits_tsv,
)


def _write_json(path: str, payload) -> str:
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh)
    return path


def _symbols_json(path: str) -> str:
    return _write_json(
        path,
        {
            "format": "greeneye-symbols/1",
            "symbols": [
                {"addr": 0x4001000, "name": "gmmu_map_alloc", "size": 0x120},
                {"addr": 0x4002000, "name": "fbif_setup", "size": 0x80},
            ],
        },
    )


def _events() -> list[dict]:
    return [
        {"seq": 0, "pc": "0x4002040", "kind": "write", "window": "falcon",
         "offset": "0x624", "name": "FBIF_CTL", "value": "0x1"},
        {"seq": 1, "pc": "0x4002050", "kind": "read", "window": "falcon",
         "offset": "0x8", "name": "IRQSTAT", "value": "0x0"},
        {"seq": 2, "pc": "0x4001010", "kind": "write", "window": "gmmu",
         "offset": "0x0", "name": "GMMU_PDB_LO", "value": "0x400000"},
        {"seq": 3, "pc": "0x4001020", "kind": "write", "window": "gmmu",
         "offset": "0x0", "name": "GMMU_PDB_LO", "value": "0x800000"},
        {"seq": 4, "pc": "0x4009000", "kind": "write", "window": "gmmu",
         "offset": "0x10", "name": "GMMU_MODE", "value": "0x7"},
    ]


def _g1_document() -> dict:
    """The nested shape emitted by sim/devices/observe.py:build_trace."""
    return {
        "schema": TRACE_FORMAT,
        "image_sha256": "ab" * 32,
        "profile": "tu104-570.144",
        "profile_name": "fixture",
        "symbols": "symbols.json",
        "summary": {"windows": 1, "accesses": 2, "retained_accesses": 2,
                    "dropped_accesses": 0},
        "windows": [
            {
                "label": "falcon-observe",
                "base": "0x300110000",
                "size": "0x1000",
                "mode": "observe-only",
                "evidence": "header",
                "counts": {"r": 1, "w": 1, "served": 0, "declined": 2},
                "registers": [
                    {
                        "offset": "0x624",
                        "name": "FBIF_CTL",
                        "access": "rw",
                        "counts": {"r": 0, "w": 1, "served": 0},
                        "accesses": [
                            {"addr": "0x300110624", "offset": "0x624", "size": 4,
                             "rw": "w", "value": "0x1", "pc": "0x4002040", "vt": 123,
                             "window": "falcon-observe", "register": "FBIF_CTL",
                             "served": False, "pc_symbol": "fbif_setup"},
                        ],
                    }
                ],
                "unclassified": {
                    "counts": {"r": 1, "w": 0, "served": 0},
                    "accesses": [
                        {"addr": "0x300110008", "offset": "0x8", "size": 4,
                         "rw": "r", "value": None, "pc": "0x4002050", "vt": 123,
                         "window": "falcon-observe", "register": None,
                         "served": False, "pc_symbol": "fbif_setup"},
                    ],
                },
            }
        ],
    }


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


class TraceIngestTests(unittest.TestCase):
    def test_trace_tagged_and_aliases(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(
                os.path.join(tmp, "t.json"),
                {
                    "format": TRACE_FORMAT,
                    "events": [
                        {"pc": 0x4002040, "rw": "w", "block": "falcon",
                         "reg_offset": 0x624, "register": "FBIF_CTL", "data": 1},
                    ],
                },
            )
            trace = load_sim_trace(path)
            self.assertEqual(trace.format, TRACE_FORMAT)
            obs = trace.observations[0]
            self.assertEqual((obs.kind, obs.window, obs.offset, obs.name, obs.value),
                             ("write", "falcon", 0x624, "FBIF_CTL", 1))

    def test_jsonl_fallback(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "t.jsonl")
            with open(path, "w", encoding="utf-8") as fh:
                fh.write(json.dumps({"format": TRACE_FORMAT}) + "\n")
                fh.write(json.dumps({"t": "write", "addr": "0x624", "size": 4, "pc": "0x4002040"}) + "\n")
                fh.write(json.dumps({"t": "summary", "insns": 1}) + "\n")
            trace = load_sim_trace(path)
            self.assertEqual(len(trace.observations), 1)
            self.assertEqual(trace.observations[0].addr, 0x624)

    def test_empty_event_list_is_legal(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(os.path.join(tmp, "t.json"), {"format": TRACE_FORMAT, "events": []})
            trace = load_sim_trace(path)
            self.assertEqual(trace.observations, [])

    def test_untagged_object_without_events_is_clean_error(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(os.path.join(tmp, "t.json"), {"hello": "world"})
            with self.assertRaises(GmmuInputError):
                load_sim_trace(path)

    def test_missing_trace_is_clean_error(self):
        with self.assertRaises(GmmuInputError):
            load_sim_trace("/does/not/exist.json")

    def test_session_directory_ingest(self):
        with tempfile.TemporaryDirectory() as tmp:
            blobs = os.path.join(tmp, "blobs")
            os.makedirs(blobs)
            _write_json(os.path.join(tmp, "session.json"), {"format": "greeneye-session/1"})
            _write_json(
                os.path.join(blobs, "gmmu-fbif-trace.json"),
                {"format": TRACE_FORMAT, "events": _events()},
            )
            trace = load_sim_trace(tmp)
            self.assertEqual(len(trace.observations), len(_events()))

    def test_g1_nested_document_flattens(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(os.path.join(tmp, "t.json"), _g1_document())
            trace = load_sim_trace(path)
            self.assertEqual(trace.format, TRACE_FORMAT)
            self.assertEqual(len(trace.observations), 2)
            self.assertEqual(trace.source["profile"], "tu104-570.144")
            self.assertEqual(trace.source["summary"]["dropped_accesses"], 0)
            self.assertEqual(trace.source["summary"]["retained_accesses"], 2)
            write_obs = trace.observations[0]
            self.assertEqual(write_obs.kind, "write")
            self.assertEqual(write_obs.name, "FBIF_CTL")
            self.assertEqual(write_obs.window, "falcon-observe")
            self.assertEqual(write_obs.offset, 0x624)
            self.assertEqual(write_obs.addr, 0x300110624)
            self.assertEqual(write_obs.pc, 0x4002040)
            self.assertEqual(write_obs.vt, 123)
            self.assertIs(write_obs.handled, False)
            self.assertEqual(write_obs.function, "fbif_setup")
            read_obs = trace.observations[1]
            self.assertEqual(read_obs.kind, "read")
            self.assertEqual(read_obs.offset, 0x8)
            self.assertIsNone(read_obs.value)

    def test_g1_nested_document_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(os.path.join(tmp, "t.json"), _g1_document())
            report = build_debug_report(load_sim_trace(path), None)
            self.assertEqual(report["summary"]["functions"], 1)
            function = report["functions"][0]
            self.assertEqual(function["function"], "fbif_setup")
            registers = {r["register"]: r for r in function["registers"]}
            self.assertEqual(registers["FBIF_CTL"]["write_count"], 1)
            self.assertEqual(registers["0x8"]["read_count"], 1)


class ObserveProducerIntegrationTests(unittest.TestCase):
    """Ingest a document produced by the real sim/devices/observe.py module.

    The producer is in-tree, so the import is unconditional: if it breaks,
    the G1<->G3 interface has regressed and this test must fail, not skip.
    """

    def test_ingest_real_build_trace_document(self):
        from sim.devices.observe import ObservedWindow, build_trace
        window = ObservedWindow(
            base=0x300110000,
            size=0x1000,
            label="integration",
            registers={"0x624": {"name": "FBIF_CTL", "width": 4, "access": "rw"}},
        )
        self.assertFalse(window.write(0x624, 4, 1, pc=0x4002040))
        self.assertIsNone(window.read(0x8, 4, pc=0x4002050))
        registry = type("Registry", (), {"devices": [window]})()
        document = build_trace(
            registry, image_sha256="cd" * 32, profile="tu104-570.144"
        )
        self.assertEqual(document.get("schema"), TRACE_FORMAT)
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(os.path.join(tmp, "t.json"), document)
            trace = load_sim_trace(path)
        self.assertEqual(trace.format, TRACE_FORMAT)
        self.assertEqual(len(trace.observations), 2)
        names = {obs.name for obs in trace.observations}
        self.assertIn("FBIF_CTL", names)
        write_obs = next(obs for obs in trace.observations if obs.kind == "write")
        self.assertEqual(write_obs.offset, 0x624)
        self.assertEqual(write_obs.value, 1)


class SymbolLoadingTests(unittest.TestCase):
    def test_symbols_json_db(self):
        with tempfile.TemporaryDirectory() as tmp:
            db = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
            self.assertEqual(db.resolve(0x4002040).name, "fbif_setup")

    def test_symbols_plain_map_json(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(os.path.join(tmp, "s.json"), {"0x4001000": "gmmu_map_alloc"})
            db = load_symbols(path)
            self.assertEqual(db.get("gmmu_map_alloc").addr, 0x4001000)

    def test_symbols_tsv(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "s.tsv")
            with open(path, "w", encoding="utf-8") as fh:
                fh.write("0x4001000\tgmmu_map_alloc\n0x4002000\tfbif_setup\n")
            db = load_symbols(path)
            self.assertEqual(db.resolve(0x4002000).name, "fbif_setup")

    def test_missing_symbols_is_clean_error(self):
        with self.assertRaises(GmmuInputError):
            load_symbols("/does/not/exist.tsv")


class ReportTests(unittest.TestCase):
    def _report(self, events=None, symbols=None):
        with tempfile.TemporaryDirectory() as tmp:
            trace_path = _write_json(
                os.path.join(tmp, "t.json"),
                {"format": TRACE_FORMAT, "events": events if events is not None else _events()},
            )
            trace = load_sim_trace(trace_path)
            return build_debug_report(trace, symbols)

    def test_report_shape_and_resolution(self):
        with tempfile.TemporaryDirectory() as tmp:
            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
            report = self._report(symbols=symbols)
            self.assertEqual(report["format"], DEBUG_REPORT_FORMAT)
            summary = report["summary"]
            self.assertEqual(summary["functions"], 3)
            self.assertEqual(summary["writes"], 4)
            self.assertEqual(summary["reads"], 1)
            names = [f["function"] for f in report["functions"]]
            self.assertEqual(names, ["gmmu_map_alloc", "fbif_setup", "sub_0x4009000"])
            fbif = report["functions"][1]
            self.assertEqual(fbif["addr"], 0x4002000)
            registers = {r["register"]: r for r in fbif["registers"]}
            self.assertEqual(registers["FBIF_CTL"]["write_count"], 1)
            self.assertEqual(registers["FBIF_CTL"]["first_pc"], 0x4002040)
            self.assertEqual(registers["FBIF_CTL"]["last_pc"], 0x4002040)
            self.assertEqual(registers["IRQSTAT"]["read_count"], 1)
            unknown = report["functions"][2]
            self.assertEqual(unknown["function"], "sub_0x4009000")
            self.assertEqual(unknown["addr"], 0x4009000)

    def test_sequence_order_and_last_pc(self):
        with tempfile.TemporaryDirectory() as tmp:
            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
            report = self._report(symbols=symbols)
            alloc = report["functions"][0]
            pdb = next(r for r in alloc["registers"] if r["register"] == "GMMU_PDB_LO")
            self.assertEqual(pdb["write_count"], 2)
            self.assertEqual(pdb["values"], [0x400000, 0x800000])
            self.assertEqual(pdb["first_pc"], 0x4001010)
            self.assertEqual(pdb["last_pc"], 0x4001020)

    def test_explicit_function_field_wins(self):
        events = [
            {"seq": 0, "pc": "0x4009000", "kind": "write", "window": "gmmu",
             "offset": "0x10", "name": "GMMU_MODE", "value": "0x7",
             "function": "fbif_setup"},
        ]
        with tempfile.TemporaryDirectory() as tmp:
            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
            report = self._report(events=events, symbols=symbols)
            self.assertEqual(report["functions"][0]["function"], "fbif_setup")
            self.assertEqual(report["functions"][0]["addr"], 0x4002000)

    def test_no_symbols_gives_sub_labels(self):
        report = self._report(symbols=None)
        self.assertTrue(all(f["function"].startswith("sub_") for f in report["functions"]))

    def test_events_without_pc_are_counted_not_attributed(self):
        events = [{"seq": 0, "kind": "write", "window": "gmmu", "offset": "0x10",
                   "name": "GMMU_MODE", "value": "0x7"}]
        report = self._report(events=events, symbols=None)
        self.assertEqual(report["summary"]["unknown_pc_events"], 1)
        self.assertEqual(report["summary"]["functions"], 0)

    def test_empty_trace(self):
        report = self._report(events=[], symbols=None)
        self.assertEqual(report["summary"]["functions"], 0)
        self.assertEqual(report["summary"]["registers"], 0)
        self.assertEqual(report["functions"], [])


class HitsTsvTests(unittest.TestCase):
    def test_hits_tsv_is_markgreeneye_compatible(self):
        with tempfile.TemporaryDirectory() as tmp:
            symbols = load_symbols(_symbols_json(os.path.join(tmp, "s.json")))
            trace = load_sim_trace(
                _write_json(
                    os.path.join(tmp, "t.json"),
                    {"format": TRACE_FORMAT, "events": _events()},
                )
            )
            report = build_debug_report(trace, symbols)
            hits_path = write_hits_tsv(report, os.path.join(tmp, "greeneye_hits.tsv"))
            with open(hits_path, "r", encoding="utf-8") as fh:
                lines = [line.rstrip("\n") for line in fh]
            self.assertEqual(lines[0], HITS_HEADER)
            self.assertTrue(lines[0].startswith("addr"))
            rows = [line.split("\t") for line in lines[1:]]
            self.assertEqual(len(rows), 3)
            for cols in rows:
                self.assertGreaterEqual(len(cols), 2)
                int(cols[0].replace("0x", ""), 16)
                self.assertTrue(cols[1].isdigit())
            first = rows[0]
            self.assertEqual(first[0], "0x4001000")
            self.assertEqual(first[2], "gmmu_map_alloc")
            self.assertIn("write", first[3])
            addrs = {int(cols[0], 16) for cols in rows}
            self.assertIn(0x4009000, addrs)

    def test_hits_tsv_empty_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            report = {"format": DEBUG_REPORT_FORMAT, "functions": []}
            hits_path = write_hits_tsv(report, os.path.join(tmp, "greeneye_hits.tsv"))
            with open(hits_path, "r", encoding="utf-8") as fh:
                self.assertEqual(fh.read().strip(), HITS_HEADER)


class ArtifactTests(unittest.TestCase):
    def test_run_report_writes_artifacts(self):
        with tempfile.TemporaryDirectory() as tmp:
            symbols_path = _symbols_json(os.path.join(tmp, "s.json"))
            trace_path = _write_json(
                os.path.join(tmp, "t.json"),
                {"format": TRACE_FORMAT, "events": _events()},
            )
            out = os.path.join(tmp, "out")
            report = run_report(trace_path, symbols_path, out)
            for key in ("report_json", "report_md", "hits_tsv", "ghidra_command"):
                self.assertIn(key, report["artifacts"])
            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
            self.assertTrue(os.path.isfile(report["artifacts"]["report_md"]))
            self.assertTrue(os.path.isfile(report["artifacts"]["hits_tsv"]))
            self.assertIn("MarkGreenEyeHits.java", report["artifacts"]["ghidra_command"])
            with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh:
                loaded = json.load(fh)
            self.assertEqual(loaded["summary"]["functions"], 3)
            with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh:
                md = fh.read()
            self.assertIn("Per function", md)
            self.assertIn("Ghidra annotation", md)


class DemoTests(unittest.TestCase):
    def test_demo_end_to_end_and_frozen_threshold(self):
        with tempfile.TemporaryDirectory() as tmp:
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                result = run_demo(tmp)
            self.assertIn("RESULT_JSON:", out.getvalue())
            self.assertTrue(result["pass"])
            self.assertLessEqual(result["measured_seconds"], result["threshold_seconds"])
            self.assertEqual(
                result["threshold_seconds"],
                round(MANUAL_STATIC_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION, 4),
            )
            self.assertIn("estimate", result["manual_baseline_label"])
            self.assertEqual(result["summary"]["functions"], 4)
            self.assertEqual(result["summary"]["writes"], 11)
            self.assertEqual(result["summary"]["reads"], 2)
            for key in ("report_json", "report_md", "hits_tsv"):
                self.assertTrue(os.path.isfile(result["artifacts"][key]))
            self.assertTrue(os.path.isfile(os.path.join(tmp, "demo-result.json")))


class CliTests(unittest.TestCase):
    def test_report_cli(self):
        with tempfile.TemporaryDirectory() as tmp:
            symbols_path = _symbols_json(os.path.join(tmp, "s.json"))
            trace_path = _write_json(
                os.path.join(tmp, "t.json"),
                {"format": TRACE_FORMAT, "events": _events()},
            )
            out = os.path.join(tmp, "out")
            rc, stdout, _stderr = _run_main(
                ["report", "--trace", trace_path, "--symbols", symbols_path, "--out", out]
            )
            self.assertEqual(rc, 0)
            payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1])
            self.assertEqual(payload["summary"]["functions"], 3)
            self.assertTrue(os.path.isfile(os.path.join(out, "greeneye_hits.tsv")))

    def test_report_cli_missing_trace_exits_2(self):
        with tempfile.TemporaryDirectory() as tmp:
            rc, _stdout, stderr = _run_main(
                ["report", "--trace", "/nope.json", "--out", os.path.join(tmp, "out")]
            )
            self.assertEqual(rc, 2)
            self.assertIn("trace not found", stderr)

    def test_empty_trace_cli_writes_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            trace_path = _write_json(
                os.path.join(tmp, "t.json"), {"format": TRACE_FORMAT, "events": []}
            )
            out = os.path.join(tmp, "out")
            rc, stdout, _stderr = _run_main(["report", "--trace", trace_path, "--out", out])
            self.assertEqual(rc, 0)
            payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1])
            self.assertEqual(payload["summary"]["functions"], 0)
            self.assertTrue(os.path.isfile(os.path.join(out, "gmmu-debug-report.md")))

    def test_no_command_prints_help(self):
        rc, stdout, _stderr = _run_main([])
        self.assertEqual(rc, 2)
        self.assertIn("usage", stdout.lower())


if __name__ == "__main__":
    unittest.main()