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

"""Tests for the GMMU/FBIF oracle comparison (harness/gmmu_oracle.py, docs/44).

All fixtures are synthetic and written to temporary directories: no oracle
capture, no hardware, no simulator run, no QEMU, no network.  The comparison
is offline file-in/file-out.  Tests that would need real inputs are written
to assert the clean error path instead.

Run::

    python3 -m unittest tests.test_gmmu_oracle -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 GmmuInputError, TRACE_FORMAT  # noqa: E402
from harness.gmmu_oracle import (  # noqa: E402
    ORACLE_FORMAT,
    REPORT_FORMAT,
    compare,
    load_oracle,
    load_sim_trace,
    main,
    print_capture_hints,
    run_compare,
)


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


def _oracle_items() -> list[dict]:
    return [
        {
            "name": "FBIF_CTL",
            "window": "falcon",
            "offset": "0x624",
            "value": "0x1",
            "source": "rm-readback",
            "captured_at": "2026-10-03",
        },
        {
            "name": "GMMU_PDB_LO",
            "window": "gmmu",
            "offset": "0x0",
            "value": "0x400000",
            "source": "rm-readback",
            "captured_at": "2026-10-03",
        },
    ]


def _sim_events() -> list[dict]:
    return [
        {
            "seq": 0,
            "pc": "0x4002040",
            "kind": "write",
            "window": "falcon",
            "offset": "0x624",
            "name": "FBIF_CTL",
            "size": 4,
            "value": "0x1",
        },
        {
            "seq": 1,
            "pc": "0x4001010",
            "kind": "write",
            "window": "gmmu",
            "offset": "0x0",
            "name": "GMMU_PDB_LO",
            "size": 4,
            "value": "0x400000",
        },
    ]


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 IngestTests(unittest.TestCase):
    def test_oracle_object_and_bare_list(self):
        with tempfile.TemporaryDirectory() as tmp:
            tagged = _write_json(
                os.path.join(tmp, "o1.json"),
                {"format": ORACLE_FORMAT, "captures": _oracle_items()},
            )
            bare = _write_json(os.path.join(tmp, "o2.json"), _oracle_items())
            self.assertEqual(load_oracle(tagged).format, ORACLE_FORMAT)
            self.assertEqual(load_oracle(bare).format, "unlabeled-list")
            self.assertEqual(len(load_oracle(tagged).items), 2)
            self.assertEqual(load_oracle(tagged).items[0].value, 1)
            self.assertEqual(load_oracle(tagged).items[1].offset, 0)

    def test_oracle_wrong_format_is_clean_error(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = _write_json(
                os.path.join(tmp, "o.json"), {"format": "something-else", "captures": []}
            )
            with self.assertRaises(GmmuInputError):
                load_oracle(path)

    def test_oracle_missing_file_is_clean_error(self):
        with self.assertRaises(GmmuInputError):
            load_oracle("/does/not/exist.json")

    def test_trace_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",
                            "value": 1,
                        }
                    ],
                },
            )
            trace = load_sim_trace(path)
            obs = trace.observations[0]
            self.assertEqual(obs.kind, "write")
            self.assertEqual(obs.window, "falcon")
            self.assertEqual(obs.offset, 0x624)
            self.assertEqual(obs.name, "FBIF_CTL")
            self.assertEqual(obs.value, 1)

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


class ComparisonTests(unittest.TestCase):
    def _compare(self, items, events):
        with tempfile.TemporaryDirectory() as tmp:
            oracle_path = _write_json(
                os.path.join(tmp, "oracle.json"),
                {"format": ORACLE_FORMAT, "captures": items},
            )
            trace_path = _write_json(
                os.path.join(tmp, "trace.json"), {"format": TRACE_FORMAT, "events": events}
            )
            return compare(load_oracle(oracle_path), load_sim_trace(trace_path))

    def test_full_agreement(self):
        report = self._compare(_oracle_items(), _sim_events())
        summary = report["summary"]
        self.assertEqual(summary["oracle_items"], 2)
        self.assertEqual(summary["compared"], 2)
        self.assertEqual(summary["agreed"], 2)
        self.assertEqual(summary["mismatched"], 0)
        self.assertEqual(summary["oracle_only"], 0)
        self.assertEqual(summary["simulator_only"], 0)
        self.assertEqual(summary["agreement_pct"], 100.0)
        self.assertEqual(summary["coverage_pct"], 100.0)
        self.assertTrue(all(row["status"] == "agree" for row in report["rows"]))
        self.assertTrue(all(row["fidelity"] == "observed vs modelled" for row in report["rows"]))

    def test_mismatch_is_reported(self):
        events = _sim_events()
        events[0]["value"] = "0x2"
        report = self._compare(_oracle_items(), events)
        summary = report["summary"]
        self.assertEqual(summary["mismatched"], 1)
        self.assertEqual(summary["agreed"], 1)
        self.assertEqual(summary["agreement_pct"], 50.0)
        mismatch = next(row for row in report["rows"] if row["status"] == "mismatch")
        self.assertEqual(mismatch["oracle_value"], 1)
        self.assertEqual(mismatch["sim_value"], 2)

    def test_missing_simulator_observation(self):
        report = self._compare(_oracle_items(), _sim_events()[:1])
        summary = report["summary"]
        self.assertEqual(summary["oracle_only"], 1)
        self.assertEqual(summary["coverage_pct"], 50.0)
        self.assertEqual(len(report["unmatched_oracle"]), 1)
        self.assertEqual(report["unmatched_oracle"][0]["register"], "GMMU_PDB_LO")
        oracle_only = next(row for row in report["rows"] if row["status"] == "oracle-only")
        self.assertIn("observed", oracle_only["fidelity"])

    def test_missing_oracle_counterpart(self):
        extra = {
            "seq": 9,
            "pc": "0x4003010",
            "kind": "write",
            "window": "gmmu",
            "offset": "0x200",
            "name": "PTE_DATA",
            "value": "0x1",
        }
        report = self._compare(_oracle_items(), _sim_events() + [extra])
        self.assertEqual(report["summary"]["simulator_only"], 1)
        self.assertEqual(len(report["unmatched_simulator"]), 1)
        self.assertEqual(report["unmatched_simulator"][0]["register"], "PTE_DATA")
        self.assertEqual(report["summary"]["simulator_match_pct"], 66.67)

    def test_name_fallback_when_oracle_has_no_offset(self):
        items = [{"name": "IRQSTAT", "window": "falcon", "value": "0x0", "source": "rm"}]
        events = [
            {
                "seq": 0,
                "pc": 0x4002050,
                "kind": "read",
                "window": "falcon",
                "offset": "0x8",
                "name": "IRQSTAT",
                "value": "0x0",
            }
        ]
        report = self._compare(items, events)
        self.assertEqual(report["summary"]["agreed"], 1)
        self.assertEqual(report["rows"][0]["matched_by"], "window-name")

    def test_no_value_is_not_agreement(self):
        items = [{"name": "FBIF_CTL", "window": "falcon", "offset": "0x624"}]
        report = self._compare(items, _sim_events()[:1])
        self.assertEqual(report["summary"]["no_value"], 1)
        self.assertEqual(report["summary"]["agreement_pct"], None)

    def test_g1_nested_document_compares(self):
        document = {
            "schema": TRACE_FORMAT,
            "image_sha256": "ab" * 32,
            "profile": "tu104-570.144",
            "windows": [
                {
                    "label": "falcon-observe",
                    "registers": [
                        {
                            "offset": "0x624",
                            "name": "FBIF_CTL",
                            "access": "rw",
                            "accesses": [
                                {"addr": "0x300110624", "offset": "0x624",
                                 "size": 4, "rw": "w", "value": "0x1",
                                 "pc": "0x4002040", "window": "falcon-observe",
                                 "register": "FBIF_CTL", "served": False,
                                 "pc_symbol": None},
                            ],
                        }
                    ],
                }
            ],
        }
        with tempfile.TemporaryDirectory() as tmp:
            oracle_path = _write_json(
                os.path.join(tmp, "oracle.json"),
                {
                    "format": ORACLE_FORMAT,
                    "captures": [
                        {"name": "FBIF_CTL", "window": "falcon-observe",
                         "offset": "0x624", "value": "0x1", "source": "rm"}
                    ],
                },
            )
            trace_path = _write_json(os.path.join(tmp, "trace.json"), document)
            report = compare(load_oracle(oracle_path), load_sim_trace(trace_path))
        self.assertEqual(report["summary"]["oracle_items"], 1)
        self.assertEqual(report["summary"]["agreed"], 1)
        self.assertEqual(report["inputs"]["simulator"]["format"], TRACE_FORMAT)

    def test_empty_inputs(self):
        report = self._compare([], [])
        summary = report["summary"]
        self.assertEqual(summary["oracle_items"], 0)
        self.assertEqual(summary["compared"], 0)
        self.assertEqual(summary["agreement_pct"], None)
        self.assertEqual(summary["coverage_pct"], None)
        self.assertEqual(report["rows"], [])
        self.assertEqual(report["unmatched_simulator"], [])


class SessionIngestTests(unittest.TestCase):
    def test_session_directory_with_blobs_trace(self):
        with tempfile.TemporaryDirectory() as tmp:
            session = os.path.join(tmp, "session")
            blobs = os.path.join(session, "blobs")
            os.makedirs(blobs)
            _write_json(
                os.path.join(session, "session.json"),
                {"format": "greeneye-session/1", "meta": {}},
            )
            _write_json(
                os.path.join(blobs, "gmmu-fbif-trace.json"),
                {"format": TRACE_FORMAT, "events": _sim_events()},
            )
            trace = load_sim_trace(session)
            self.assertEqual(len(trace.observations), 2)
            self.assertEqual(trace.format, TRACE_FORMAT)

    def test_session_without_trace_is_clean_error(self):
        with tempfile.TemporaryDirectory() as tmp:
            with self.assertRaises(GmmuInputError):
                load_sim_trace(tmp)


class ReportArtifactTests(unittest.TestCase):
    def test_run_compare_writes_json_and_markdown(self):
        with tempfile.TemporaryDirectory() as tmp:
            oracle_path = _write_json(
                os.path.join(tmp, "oracle.json"),
                {"format": ORACLE_FORMAT, "captures": _oracle_items()},
            )
            trace_path = _write_json(
                os.path.join(tmp, "trace.json"),
                {"format": TRACE_FORMAT, "events": _sim_events()},
            )
            out = os.path.join(tmp, "out")
            report = run_compare(oracle_path, trace_path, out)
            self.assertEqual(report["format"], REPORT_FORMAT)
            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
            self.assertTrue(os.path.isfile(report["artifacts"]["report_md"]))
            with open(report["artifacts"]["report_json"], "r", encoding="utf-8") as fh:
                loaded = json.load(fh)
            self.assertEqual(loaded["summary"]["agreed"], 2)
            with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh:
                md = fh.read()
            self.assertIn("oracle vs simulator report", md)
            self.assertIn("No cycle/timing claim", md)
            self.assertIn("Oracle items with no simulator observation", md)


class CliTests(unittest.TestCase):
    def test_print_capture_hints_executes_nothing(self):
        out, err = io.StringIO(), io.StringIO()
        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
            print_capture_hints(out)
        text = out.getvalue()
        self.assertIn("SUPERVISED, READ-ONLY, HUMAN-RUN ONLY", text)
        self.assertIn("executes NOTHING", text)
        self.assertIn("RUNBOOK P1/P7", text)
        self.assertEqual(err.getvalue(), "")

    def test_print_capture_hints_cli(self):
        rc, stdout, _stderr = _run_main(["--print-capture-hints"])
        self.assertEqual(rc, 0)
        self.assertIn("rmharness/harness.sh", stdout)
        self.assertIn("gmmu_oracle.py compare", stdout)

    def test_compare_cli_writes_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            oracle_path = _write_json(
                os.path.join(tmp, "oracle.json"),
                {"format": ORACLE_FORMAT, "captures": _oracle_items()},
            )
            trace_path = _write_json(
                os.path.join(tmp, "trace.json"),
                {"format": TRACE_FORMAT, "events": _sim_events()},
            )
            out = os.path.join(tmp, "out")
            rc, stdout, _stderr = _run_main(
                ["compare", "--oracle", oracle_path, "--sim", trace_path, "--out", out]
            )
            self.assertEqual(rc, 0)
            payload = json.loads(stdout[stdout.index("{"):stdout.rindex("}") + 1])
            self.assertEqual(payload["summary"]["agreed"], 2)
            self.assertTrue(os.path.isfile(os.path.join(out, "gmmu-oracle-report.json")))

    def test_compare_missing_inputs_exits_2(self):
        with tempfile.TemporaryDirectory() as tmp:
            rc, _stdout, stderr = _run_main(
                ["compare", "--oracle", "/nope.json", "--sim", "/nope.json", "--out", tmp]
            )
            self.assertEqual(rc, 2)
            self.assertIn("oracle not found", stderr)

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