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

"""Tests for the GSP RPC capture ingestion harness (harness/rm_capture.py, docs/45).

All fixtures are synthetic text written in-process or to temporary
directories: no capture, no hardware, no driver, no simulator run, no QEMU,
no network.  The decode/compare/stimulus loop is offline file-in/file-out.
Tests that would need a real capture assert the clean error path instead.

Run::

    python3 -m unittest tests.test_rm_capture -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.rm_capture import (  # noqa: E402
    CAPTURE_FORMAT,
    COMPARE_FORMAT,
    DEMO_FORMAT,
    GSP_RM_ALLOC,
    GSP_RM_CONTROL,
    MANUAL_DECODE_BASELINE_SECONDS,
    RmCaptureError,
    STIMULUS_FORMAT,
    WORKFLOW_THRESHOLD_FRACTION,
    canonical_json,
    compare,
    decode_file,
    decode_text,
    load_capture,
    main,
    print_capture_hints,
    run_compare,
    run_decode,
    run_demo,
    run_stimulus,
    stimulus_from_capture,
)


def _hex_lines(tag: str, payload: bytes) -> list[str]:
    return [f"GE-{tag} 000:" + "".join(f" {b:02x}" for b in payload)]


def _sample_log() -> str:
    """A small log exercising every tag; params are contiguous single chunks."""
    return "\n".join(
        [
            "[ 100.000000] NVRM: GE-GOLDEN-TRIGGER gspclient=1 mig=0",
            "[ 100.000010] NVRM: GE-GOLDEN-BEGIN",
            "[ 100.000020] NVRM: GE-ALLOC cls=0x90f1 hClient=0xbaba0001 hParent=0x0 "
            "hObject=0xbaba1001 size=0x10 paramp=0xffffffffc0a00000 ret=0xffffffffc0b00000",
            "[ 100.000020] NVRM: " + _hex_lines("ALLOC", bytes(range(16)))[0],
            "[ 100.000030] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 "
            "hObject=0xabcd0080 size=0x8 paramp=0xffffffffc0a10000 ret=0xffffffffc0b00000",
            "[ 100.000030] NVRM: " + _hex_lines("CTRL", bytes([1, 2, 3, 4, 5, 6, 7, 8]))[0],
            "[ 100.000040] NVRM: GE-SEND fn=0x0000004c len=0x00000028 ret=0xffffffffc0b00000",
            "[ 100.000040] NVRM: "
            + _hex_lines("SEND", (0xC2000006).to_bytes(4, "little")
                         + (0xABCD0080).to_bytes(4, "little")
                         + (0x2080012B).to_bytes(4, "little")
                         + bytes(4))[0],
            "[ 100.000050] NVRM: GE-CTRL-DONE cmd=0x2080012b rpc_status=0x0 params_status=0x0",
            "[ 100.000060] NVRM: GE-GOLDEN-END status=0x0",
        ]
    ) + "\n"


def _follow(text: str, header_fragment: str) -> str:
    """Return the hex block emitted immediately after a header line."""
    lines = text.splitlines()
    for index, line in enumerate(lines):
        if header_fragment in line:
            return lines[index + 1]
    raise AssertionError(f"fragment not found: {header_fragment}")


def _chunked_control_log(size: int) -> str:
    payload = bytes((i * 5) & 0xFF for i in range(size))
    lines = ["[ 1.0] NVRM: GE-GOLDEN-BEGIN"]
    lines.append(
        f"[ 1.1] NVRM: GE-CTRL cmd=0x2080012b hClient=0xc2000006 "
        f"hObject=0xabcd0080 size={size:#x} paramp=0x0 ret=0x0"
    )
    for start in range(0, len(payload), 192):
        piece = payload[start : start + 192]
        lines.append(
            f"[ 1.1] NVRM: GE-CTRL {start:03x}:"
            + "".join(f" {b:02x}" for b in piece)
        )
    lines.append("[ 1.2] NVRM: GE-GOLDEN-END status=0x0")
    return "\n".join(lines) + "\n"


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 DecodeTests(unittest.TestCase):
    def test_decodes_every_record_kind_in_order(self):
        capture = decode_text(_sample_log(), source="sample")
        kinds = [record.kind for record in capture.records]
        self.assertEqual(
            kinds,
            ["golden", "golden", "alloc", "control", "send", "ctrl-done", "golden"],
        )
        trigger, begin = capture.records[0], capture.records[1]
        self.assertEqual(trigger.marker, "trigger")
        self.assertEqual(trigger.gspclient, 1)
        self.assertEqual(trigger.mig, 0)
        self.assertEqual(begin.marker, "begin")
        alloc = capture.records[2]
        self.assertEqual(alloc.hclass, 0x90F1)
        self.assertEqual(alloc.hclient, 0xBABA0001)
        self.assertEqual(alloc.hobject, 0xBABA1001)
        self.assertEqual(alloc.size, 0x10)
        self.assertEqual(alloc.caller, 0xFFFFFFFFC0B00000)
        control = capture.records[3]
        self.assertEqual(control.cmd, 0x2080012B)
        self.assertEqual(control.params, bytes([1, 2, 3, 4, 5, 6, 7, 8]))
        send = capture.records[4]
        self.assertEqual(send.fn, 0x4C)
        self.assertEqual(send.length, 0x28)
        self.assertEqual(send.expected_params_size, 0x28 - 32)
        self.assertEqual(capture.records[5].rpc_status, 0)
        self.assertEqual(capture.records[6].status, 0)
        self.assertEqual(capture.golden_status(), 0)
        self.assertFalse(capture.warnings)

    def test_dmesg_timestamp_and_prefix_are_stripped(self):
        capture = decode_text(_sample_log(), source="sample")
        self.assertEqual(capture.records[0].ts, " 100.000000")

    def test_multichunk_payload_is_assembled(self):
        capture = decode_text(_chunked_control_log(560), source="chunked")
        control = capture.records[1]
        self.assertEqual(len(control.params), 560)
        self.assertFalse(control.truncated)
        self.assertEqual(control.chunks[0].__len__(), 192)
        self.assertIn(0xC0, control.chunks)
        self.assertIn(0x180, control.chunks)
        self.assertEqual(control.params[192], (192 * 5) & 0xFF)

    def test_truncated_payload_is_flagged(self):
        # The runbook helper caps at 192 bytes: declared 560, captured 192.
        capture = decode_text(_chunked_control_log(192), source="truncated")
        control = capture.records[1]
        control.size = 560
        self.assertTrue(control.truncated)
        self.assertEqual(control.expected_params_size, 560)
        self.assertEqual(len(control.params), 192)

    def test_orphan_hex_warns_and_is_not_attached(self):
        capture = decode_text("[ 1.0] NVRM: GE-CTRL 000: 01 02 03 04\n", source="orphan")
        self.assertEqual(capture.records, [])
        self.assertEqual(len(capture.warnings), 1)
        self.assertIn("orphan CTRL hex", capture.warnings[0])

    def test_wire_id_derived_from_send_payload(self):
        control_send = decode_text(_sample_log(), source="s").records[4]
        self.assertEqual(control_send.wire_id, 0x2080012B)
        alloc_payload = (
            (0xBABA0001).to_bytes(4, "little")
            + (0x0).to_bytes(4, "little")
            + (0xBABA1002).to_bytes(4, "little")
            + (0x0040).to_bytes(4, "little")
        )
        log = (
            "[ 1.0] NVRM: GE-SEND fn=0x00000067 len=0x30 ret=0x0\n"
            "[ 1.0] NVRM: GE-SEND 000:" + "".join(f" {b:02x}" for b in alloc_payload) + "\n"
        )
        record = decode_text(log, source="a").records[0]
        self.assertEqual(record.fn, GSP_RM_ALLOC)
        self.assertEqual(record.wire_id, 0x0040)
        self.assertNotEqual(GSP_RM_CONTROL, GSP_RM_ALLOC)

    def test_overrun_is_an_anomaly(self):
        log = (
            "[ 1.0] NVRM: GE-CTRL cmd=0x1 hClient=0x0 hObject=0x0 size=0x4 "
            "paramp=0x0 ret=0x0\n"
            "[ 1.0] NVRM: GE-CTRL 000: 01 02 03 04 05 06 07 08\n"
        )
        record = decode_text(log, source="overrun").records[0]
        self.assertTrue(record.overrun)
        self.assertTrue(any("> declared" in item for item in record.anomalies))

    def test_golden_status_none_without_end_record(self):
        log = "[ 1.0] NVRM: GE-GOLDEN-BEGIN\n"
        self.assertIsNone(decode_text(log, source="x").golden_status())


class CaptureIOTests(unittest.TestCase):
    def test_decode_write_load_roundtrip(self):
        with tempfile.TemporaryDirectory() as tmp:
            log_path = os.path.join(tmp, "capture.log")
            with open(log_path, "w", encoding="utf-8") as fh:
                fh.write(_sample_log())
            out_dir = os.path.join(tmp, "out")
            payload = run_decode(log_path, out_dir)
            self.assertEqual(payload["format"], CAPTURE_FORMAT)
            self.assertTrue(os.path.isfile(payload["artifacts"]["capture_json"]))
            self.assertTrue(os.path.isfile(payload["artifacts"]["capture_md"]))
            loaded = load_capture(payload["artifacts"]["capture_json"])
            self.assertEqual(len(loaded.records), len(decode_file(log_path).records))
            self.assertEqual(loaded.records[3].cmd, 0x2080012B)

    def test_load_capture_autodetects_raw_log(self):
        with tempfile.TemporaryDirectory() as tmp:
            log_path = os.path.join(tmp, "raw.log")
            with open(log_path, "w", encoding="utf-8") as fh:
                fh.write(_sample_log())
            loaded = load_capture(log_path)
            self.assertEqual(loaded.records[0].kind, "golden")

    def test_load_capture_rejects_unknown_format(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "bad.json")
            with open(path, "w", encoding="utf-8") as fh:
                json.dump({"format": "something-else", "records": []}, fh)
            with self.assertRaises(RmCaptureError):
                load_capture(path)

    def test_missing_file_is_clean_error(self):
        with self.assertRaises(RmCaptureError):
            decode_file("/nonexistent/capture.log")


class CompareTests(unittest.TestCase):
    def _captures(self):
        capture_a = decode_text(_sample_log(), source="a")
        capture_b = decode_text(_sample_log(), source="b")
        return capture_a, capture_b

    def test_identical_captures_fully_agree(self):
        capture_a, capture_b = self._captures()
        report = compare(capture_a, capture_b)
        self.assertEqual(report["format"], COMPARE_FORMAT)
        summary = report["summary"]
        self.assertEqual(summary["mismatch"], 0)
        self.assertEqual(summary["a_only"], 0)
        self.assertEqual(summary["b_only"], 0)
        self.assertEqual(summary["agree"], len(capture_a.records))
        self.assertEqual(summary["agreement_pct"], 100.0)
        self.assertEqual(summary["coverage_pct"], 100.0)

    def test_dropped_record_is_a_only(self):
        capture_a, capture_b = self._captures()
        capture_b.records = [
            record for record in capture_b.records if record.kind != "alloc"
        ]
        report = compare(capture_a, capture_b)
        self.assertEqual(report["summary"]["a_only"], 1)
        self.assertEqual(report["a_only"][0]["label"].startswith("ALLOC"), True)

    def test_param_byte_flip_is_shape_agree(self):
        capture_a, capture_b = self._captures()
        control = next(r for r in capture_b.records if r.kind == "control")
        control.chunks[0] = bytes([9]) + control.chunks[0][1:]
        report = compare(capture_a, capture_b)
        self.assertEqual(report["summary"]["shape_agree"], 1)
        row = next(r for r in report["rows"] if r["status"] == "shape-agree")
        self.assertEqual(row["detail"]["first_diff_offset"], 0)
        self.assertLess(row["detail"]["byte_agreement_pct"], 100.0)

    def test_truncated_pair_is_partial(self):
        capture_a = decode_text(_chunked_control_log(192), source="a")
        capture_b = decode_text(_chunked_control_log(192), source="b")
        for capture in (capture_a, capture_b):
            capture.records[1].size = 560
        report = compare(capture_a, capture_b)
        self.assertEqual(report["summary"]["partial"], 1)
        self.assertTrue(report["rows"][1]["a_truncated"])

    def test_compare_is_deterministic_and_self_describing(self):
        capture_a, capture_b = self._captures()
        report_one = compare(capture_a, capture_b)
        report_two = compare(capture_a, capture_b)
        self.assertEqual(canonical_json(report_one), canonical_json(report_two))
        self.assertEqual(report_one["format"], COMPARE_FORMAT)
        self.assertIn("L2", report_one["fidelity"]["levels"])
        self.assertTrue(report_one["limits"])

    def test_run_compare_writes_json_and_markdown(self):
        with tempfile.TemporaryDirectory() as tmp:
            log_path = os.path.join(tmp, "capture.log")
            with open(log_path, "w", encoding="utf-8") as fh:
                fh.write(_sample_log())
            report = run_compare(log_path, log_path, os.path.join(tmp, "out"))
            self.assertTrue(os.path.isfile(report["artifacts"]["report_json"]))
            self.assertTrue(os.path.isfile(report["artifacts"]["report_md"]))
            with open(report["artifacts"]["report_md"], "r", encoding="utf-8") as fh:
                self.assertIn("# GSP RPC capture comparison", fh.read())


class StimulusTests(unittest.TestCase):
    def test_stimulus_lists_rpcs_in_order_without_handles(self):
        capture = decode_text(_sample_log(), source="s")
        stimulus = stimulus_from_capture(capture)
        self.assertEqual(stimulus["format"], STIMULUS_FORMAT)
        kinds = [message["kind"] for message in stimulus["messages"]]
        self.assertEqual(kinds, ["alloc", "control", "send"])
        self.assertEqual(stimulus["messages"][1]["cmd"], 0x2080012B)
        self.assertEqual(stimulus["messages"][2]["fn"], GSP_RM_CONTROL)
        self.assertEqual(stimulus["messages"][2]["wire_id"], 0x2080012B)
        self.assertNotIn("hclient", stimulus["messages"][1])
        self.assertEqual(len(stimulus["markers"]), 3)
        self.assertTrue(stimulus["limits"])

    def test_stimulus_roundtrips_params_and_flags_truncation(self):
        capture = decode_text(_chunked_control_log(192), source="s")
        capture.records[1].size = 560
        stimulus = stimulus_from_capture(capture)
        message = stimulus["messages"][0]
        self.assertTrue(message["truncated"])
        self.assertEqual(message["params_size_expected"], 560)
        self.assertEqual(bytes.fromhex(message["params_hex"]), capture.records[1].params)

    def test_run_stimulus_writes_file(self):
        with tempfile.TemporaryDirectory() as tmp:
            out_path = os.path.join(tmp, "nested", "stimulus.json")
            log_path = os.path.join(tmp, "capture.log")
            with open(log_path, "w", encoding="utf-8") as fh:
                fh.write(_sample_log())
            payload = run_stimulus(log_path, out_path)
            self.assertTrue(os.path.isfile(out_path))
            self.assertEqual(payload["format"], STIMULUS_FORMAT)
            with open(out_path, "r", encoding="utf-8") as fh:
                written = json.load(fh)
            self.assertEqual(written["format"], STIMULUS_FORMAT)


class DemoAndCliTests(unittest.TestCase):
    def test_demo_is_ok_and_within_threshold(self):
        payload = run_demo()
        self.assertEqual(payload["format"], DEMO_FORMAT)
        self.assertTrue(payload["ok"])
        self.assertLessEqual(
            payload["wall_seconds"],
            MANUAL_DECODE_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION,
        )
        self.assertGreaterEqual(payload["summary"]["agree"], 1)
        self.assertEqual(payload["summary"]["shape_agree"], 1)
        self.assertEqual(payload["summary"]["a_only"], 1)

    def test_demo_is_deterministic_apart_from_timing(self):
        first = run_demo()
        second = run_demo()
        for payload in (first, second):
            payload.pop("wall_seconds", None)
            payload.pop("artifacts", None)
        self.assertEqual(canonical_json(first), canonical_json(second))

    def test_demo_writes_artifacts(self):
        with tempfile.TemporaryDirectory() as tmp:
            payload = run_demo(tmp)
            for key in ("capture_a", "replay_b", "compare", "stimulus"):
                self.assertTrue(os.path.isfile(payload["artifacts"][key]))

    def test_cli_decode_compare_stimulus_return_zero(self):
        with tempfile.TemporaryDirectory() as tmp:
            log_path = os.path.join(tmp, "capture.log")
            with open(log_path, "w", encoding="utf-8") as fh:
                fh.write(_sample_log())
            rc, out, _ = _run_main(["decode", "--log", log_path, "--out", tmp])
            self.assertEqual(rc, 0)
            self.assertEqual(json.loads(out)["format"], CAPTURE_FORMAT)
            rc, out, _ = _run_main(
                ["compare", "--a", log_path, "--b", log_path, "--out", tmp]
            )
            self.assertEqual(rc, 0)
            self.assertEqual(json.loads(out)["format"], COMPARE_FORMAT)
            rc, out, _ = _run_main(
                ["stimulus", "--capture", log_path, "--out", os.path.join(tmp, "s.json")]
            )
            self.assertEqual(rc, 0)
            self.assertEqual(json.loads(out)["format"], STIMULUS_FORMAT)

    def test_cli_missing_input_returns_two(self):
        rc, _, err = _run_main(["decode", "--log", "/nonexistent/capture.log"])
        self.assertEqual(rc, 2)
        self.assertIn("error:", err)

    def test_hints_execute_nothing_and_name_the_rules(self):
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            print_capture_hints()
        text = out.getvalue()
        self.assertIn("executes", text)
        self.assertIn("NOTHING", text)
        self.assertIn("GE-SEND", text)
        self.assertIn("192 bytes", text)


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