From 6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf Mon Sep 17 00:00:00 2001 From: Arthur Hanson Rasmusson Date: Tue, 6 Oct 2026 19:28:39 -0400 Subject: risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests rm_capture (chunked decode), launch_audit, state_seed/state_decode, valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk, msgq_attach, libos_port, log_decode, server_snapshot; doc index updates in the tool overview, operations pages and the repo README. --- debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py | 409 ++++++++++++++++++++++++ 1 file changed, 409 insertions(+) create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py (limited to 'debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py') diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py b/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py new file mode 100644 index 0000000..cb62bea --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_rm_capture.py @@ -0,0 +1,409 @@ +"""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() -- cgit v1.2.3