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