diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_log_decode.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_log_decode.py | 335 |
1 files changed, 335 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py b/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py new file mode 100644 index 0000000..b43dc00 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_log_decode.py @@ -0,0 +1,335 @@ +"""Tests for the offline GSP-RM log decoder (harness/log_decode.py). + +Crafted inputs cover both real dump forms: the raw ring hexdump +(``LOGINIT-raw:`` + rows) and the dmesg/evidence record lines +(``rec seq=.. recType=.. errorCode=.. source=..`` plus ``assertCallStack`` +frames, ``entry[k]:`` aggregates and the liblogdecode ``rec NNN ts=..`` +text form). A real-file test runs only when the campaign dump is present. + +Run:: + + cd TOOL_ROOT && python3 -m unittest tests.test_log_decode -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.log_decode import ( # noqa: E402 + SCHEMA, + LogDecodeError, + apply_filters, + decode_path, + decode_text, + load_symbols, + main, + run_demo, +) + +REAL_LOGDUMP = ( + "/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/" + "evidence-20261002-gr/21-logdump-grctx4.txt" +) + +RAW_TEXT = """\ +vgsp-boot: libos task log buffers (read-only) +LOGINIT-raw: 64 bytes (allocated 64) VA=0xffff0000 dma=0x1000 +00000000: 00000005 00000000 0000000a deadbeef +00000010: 01020304 05060708 090a0b0c 0d0e0f10 +""" + +RECORDS_TEXT = """\ +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine="" +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000,0x0] (offsets from load base 0x4000000) +[Tue Oct 6 12:40:53 2026] vgsp-nocat: entry[0]: count=1 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=1 source="ASSERT" +an unmatched filler line +rec 7 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x401000,0x402000 +rec 9 ts=0x18dbb29118ec7280 meta=0x0000000005bb9bd0 args=[0x10, 0x20] +""" + +SYMBOLS = { + "symbols": [ + {"name": "gr_init", "addr": "0x401000"}, + {"name": "gr_done", "addr": 0x403000}, + ] +} + + +def _write(tmp: str, name: str, text: str) -> str: + path = os.path.join(tmp, name) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def _records(report: dict, kind: str = "record") -> list[dict]: + return [r for r in report["records"] if r["kind"] == kind] + + +class RawHexTests(unittest.TestCase): + def test_header_and_rows_decode_to_words(self): + report = decode_text(RAW_TEXT, source="raw.txt") + self.assertEqual(report["schema"], SCHEMA) + self.assertEqual(report["source"], "raw.txt") + self.assertEqual(report["count"], 1) + self.assertEqual(report["unparsed"], 1) # the intro line is not a record + raw = report["records"][0] + self.assertEqual(raw["kind"], "raw-hex") + self.assertEqual(raw["label"], "LOGINIT") + self.assertEqual(raw["source_name"], "LOGINIT") + self.assertEqual(len(raw["frames"]), 8) + self.assertEqual(raw["frames"][0], "0x5") + self.assertEqual(raw["frames"][3], "0xdeadbeef") + self.assertEqual(raw["frames"][7], "0xd0e0f10") + # raw is the decoded little-endian byte buffer of the rows + self.assertEqual(len(raw["raw"]), 2 * 8 * 8) + self.assertEqual(bytes.fromhex(raw["raw"])[:8].hex(), "0500000000000000") + self.assertEqual(raw["raw"].endswith("100f0e0d00000000"), True) + + def test_two_sections_stay_separate(self): + text = RAW_TEXT + ( + "LOGRM-raw: 16 bytes (allocated 16) VA=0x1 dma=0x2\n" + "00000000: 11111111 22222222 33333333 44444444\n" + ) + report = decode_text(text) + raw = _records(report, "raw-hex") + self.assertEqual(len(raw), 2) + self.assertEqual(raw[1]["label"], "LOGRM") + self.assertEqual(raw[1]["frames"], ["0x11111111", "0x22222222", + "0x33333333", "0x44444444"]) + + def test_blank_and_comment_lines_are_structure(self): + report = decode_text("\n# LOGINIT\n" + RAW_TEXT) + self.assertEqual(report["unparsed"], 1) # only the intro line + + +class RecordLineTests(unittest.TestCase): + def setUp(self): + self.report = decode_text(RECORDS_TEXT) + self.rec = self.report["records"][0] + + def test_rec_line_fields(self): + self.assertEqual(self.rec["kind"], "record") + self.assertEqual(self.rec["label"], "vgsp-nocat") + self.assertEqual(self.rec["seq"], 1) + self.assertEqual(self.rec["rec_type"], 5) + self.assertEqual(self.rec["error_code"], "0x57f1344") + self.assertEqual(self.rec["source_name"], "ASSERT") + + def test_assert_callstack_merges_frames(self): + self.assertEqual(self.rec["frames"], ["0x401000", "0x402000", "0x0"]) + # no duplicate standalone continuation record was emitted + self.assertEqual(len(_records(self.report)), 4) + + def test_entry_aggregate_line(self): + entry = self.report["records"][1] + self.assertEqual(entry["rec_type"], 5) + self.assertEqual(entry["error_code"], "0x57f1344") + self.assertEqual(entry["source_name"], "ASSERT") + self.assertIn("entry[0]:", entry["raw"]) + self.assertIn("vgsp-nocat:", entry["raw"]) + + def test_decoded_rec_shapes_with_and_without_brackets(self): + records = _records(self.report) + plain, bracketed = records[2], records[3] + self.assertEqual(plain["seq"], 7) + self.assertEqual(plain["frames"], ["0x401000", "0x402000"]) + self.assertEqual(bracketed["seq"], 9) + self.assertEqual(bracketed["frames"], ["0x10", "0x20"]) + + def test_unparsed_accounting(self): + self.assertEqual(self.report["unparsed"], 1) + self.assertEqual(len(self.report["records"]), 4) + + def test_histogram(self): + hist = self.report["histogram"] + self.assertEqual(hist["rec_type"], {"5": 2}) + self.assertEqual(hist["error_code"], {"0x57f1344": 2}) + self.assertEqual(hist["source"], {"ASSERT": 2}) + + def test_standalone_continuation_becomes_record(self): + text = ("vgsp-nocat: rec seq=4 assertCallStack count=1 " + "[0x1000,0x2000] (offsets from load base 0x4000000)\n") + report = decode_text(text) + self.assertEqual(report["count"], 1) + self.assertEqual(report["records"][0]["seq"], 4) + self.assertEqual(report["records"][0]["frames"], ["0x1000", "0x2000"]) + + +class SymbolTests(unittest.TestCase): + def test_nearest_below_with_offset(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "symbols.json", json.dumps(SYMBOLS)) + table = load_symbols(path) + self.assertEqual(table.resolve(0x401000), ("gr_init", 0x0)) + self.assertEqual(table.resolve(0x402fff), ("gr_init", 0x1FFF)) + self.assertEqual(table.resolve(0x403000), ("gr_done", 0x0)) + self.assertIsNone(table.resolve(0x400fff)) # below the first symbol + self.assertIsNone(table.resolve(0x0)) + + def test_top_level_list_shape_accepted(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "symbols.json", + json.dumps([{"name": "a", "addr": "0x10"}])) + table = load_symbols(path) + self.assertEqual(table.resolve(0x1f), ("a", 0xF)) + + def test_resolved_entries_align_with_frames(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "symbols.json", json.dumps(SYMBOLS)) + report = decode_text(RECORDS_TEXT, symbols=load_symbols(path)) + resolved = report["records"][0]["resolved"] + self.assertEqual(resolved[0], + {"frame": "0x401000", "symbol": "gr_init", + "offset": "0x0"}) + self.assertEqual(resolved[1], + {"frame": "0x402000", "symbol": "gr_init", + "offset": "0x1000"}) + self.assertEqual(resolved[2], + {"frame": "0x0", "symbol": None, "offset": None}) + + def test_malformed_symbols_fail_cleanly(self): + cases = [ + "{not json", + json.dumps([{"name": "no-addr"}]), + json.dumps({"symbols": "nope"}), + json.dumps({"nope": []}), + json.dumps([42]), + json.dumps([{"name": "x", "addr": "zzz"}]), + ] + with tempfile.TemporaryDirectory() as tmp: + for index, payload in enumerate(cases): + path = _write(tmp, f"bad{index}.json", payload) + with self.assertRaises(LogDecodeError): + load_symbols(path) + self.assertRaises(LogDecodeError, load_symbols, + os.path.join(tmp, "missing.json")) + # the CLI exits non-zero with a clear message, not a traceback + good = _write(tmp, "good.txt", RECORDS_TEXT) + bad = _write(tmp, "bad.json", cases[0]) + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + code = main(["decode", good, "--symbols", bad]) + self.assertEqual(code, 2) + self.assertIn("symbols JSON", stderr.getvalue()) + + +class FilterTests(unittest.TestCase): + def setUp(self): + self.report = decode_text(RECORDS_TEXT) + self.raw_report = decode_text(RAW_TEXT) + + def test_kind_filter(self): + self.assertEqual(apply_filters(self.report, kind="record")["count"], 4) + self.assertEqual(apply_filters(self.report, kind="raw-hex")["count"], 0) + self.assertEqual(apply_filters(self.raw_report, kind="raw-hex")["count"], 1) + + def test_rec_type_filter_accepts_hex_and_decimal(self): + for value in ("5", "0x5"): + filtered = apply_filters(self.report, rec_type=value) + self.assertEqual(filtered["count"], 2) + self.assertTrue(all(r["rec_type"] == 5 for r in filtered["records"])) + + def test_error_code_filter(self): + filtered = apply_filters(self.report, error_code="0x57f1344") + self.assertEqual(filtered["count"], 2) + zeros = apply_filters(self.report, error_code="0x1") + self.assertEqual(zeros["count"], 0) + + def test_source_filter_is_case_insensitive_substring(self): + self.assertEqual(apply_filters(self.report, source="assert")["count"], 2) + self.assertEqual(apply_filters(self.report, source="Display")["count"], 0) + + def test_limit_and_combined_filters(self): + self.assertEqual(apply_filters(self.report, limit=1)["count"], 1) + combined = apply_filters(self.report, kind="record", source="ASSERT", + limit=1) + self.assertEqual(combined["count"], 1) + self.assertEqual(combined["records"][0]["seq"], 1) + self.assertEqual(apply_filters(self.report, limit=0)["count"], 0) + + def test_bad_filter_value_raises(self): + with self.assertRaises(LogDecodeError): + apply_filters(self.report, rec_type="zzz") + with self.assertRaises(LogDecodeError): + apply_filters(self.report, kind="bogus") + with self.assertRaises(LogDecodeError): + apply_filters(self.report, limit=-1) + + def test_histogram_follows_filter(self): + filtered = apply_filters(self.report, kind="record", limit=1) + self.assertEqual(filtered["histogram"]["rec_type"], {"5": 1}) + self.assertEqual(filtered["unparsed"], self.report["unparsed"]) + + +class CliTests(unittest.TestCase): + def test_decode_writes_json_report(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write(tmp, "records.txt", RECORDS_TEXT) + out = os.path.join(tmp, "out", "report.json") + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + code = main(["decode", path, "--json", out, + "--source", "ASSERT", "--limit", "1"]) + self.assertEqual(code, 0) + with open(out, "r", encoding="utf-8") as fh: + report = json.load(fh) + self.assertEqual(report["schema"], SCHEMA) + self.assertEqual(report["source"], "records.txt") + self.assertEqual(report["count"], 1) + self.assertIn("report_json", stdout.getvalue()) + + def test_missing_input_fails_cleanly(self): + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + code = main(["decode", "/nonexistent/log.txt"]) + self.assertEqual(code, 2) + self.assertIn("input not found", stderr.getvalue()) + + def test_demo_runs_deterministic(self): + first = run_demo() + second = run_demo() + self.assertTrue(first["ok"], first) + self.assertEqual(first["raw"]["count"], second["raw"]["count"]) + self.assertEqual(first["records"]["records"][0]["frames"], + second["records"]["records"][0]["frames"]) + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + code = main(["demo"]) + self.assertEqual(code, 0) + self.assertIn('"ok": true', stdout.getvalue()) + + +@unittest.skipUnless(os.path.isfile(REAL_LOGDUMP), + f"real logdump not present: {REAL_LOGDUMP}") +class RealLogdumpTests(unittest.TestCase): + """Decode the campaign dump when it is present; skipped otherwise.""" + + def test_real_logdump_decodes(self): + report = decode_path(REAL_LOGDUMP) + self.assertEqual(report["schema"], SCHEMA) + self.assertEqual(report["source"], os.path.basename(REAL_LOGDUMP)) + self.assertGreater(report["count"], 0) + self.assertTrue(any(report["histogram"].values()), + report["histogram"]) + raw_labels = {r["label"] for r in _records(report, "raw-hex")} + self.assertEqual(raw_labels, {"LOGINIT", "LOGRM"}) + framed = [r for r in _records(report) + if r.get("seq") is not None and r.get("frames")] + self.assertTrue(framed) + self.assertEqual(report["count"], + len(report["records"])) + + +if __name__ == "__main__": + unittest.main() |