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

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