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

"""Tests for the server-state snapshot collector (harness/server_snapshot.py).

Crafted evidence covers every supported shape (``LABEL[offset]=0xVALUE``,
labelled hexdump blocks, ``LOGINIT-raw`` headers, record lines, duplicate
labels and unparseable dump-looking lines).  Optional tests run against the
real 2026-10-02 campaign backups when they are present; they are skipped
otherwise.  All tests are offline, hermetic and touch no hardware.

Run::

    python3 -m unittest tests.test_server_snapshot -v
"""
from __future__ import annotations

import contextlib
import io
import json
import os
import sys
import tempfile
import unittest
from unittest import mock

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.server_snapshot import (  # noqa: E402
    DIFF_FORMAT,
    SNAPSHOT_FORMAT,
    SnapshotError,
    collect,
    diff_snapshots,
    main,
)

EVIDENCE_VICTORY = ("/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/"
                    "evidence-20261002-victory")
LADDER_M12 = os.path.join(EVIDENCE_VICTORY, "06-ladder-m12.txt")
PHYS_STATE_M26 = os.path.join(EVIDENCE_VICTORY, "20-gsp-phys-state-m26.txt")
EVIDENCE_GR = ("/home/rmcore/Git/integdev_gpu_drv/open-iov/backups/"
               "evidence-20261002-gr")
LOGDUMP_GRCTX4 = os.path.join(EVIDENCE_GR, "21-logdump-grctx4.txt")

FIXED_CREATED = "2026-01-01T00:00:00Z"


def _run_main(argv):
    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 ServerSnapshotTests(unittest.TestCase):
    """Collector normalization, collision, diff, CLI and determinism."""

    def setUp(self):
        holder = tempfile.TemporaryDirectory(prefix="server-snapshot-test-")
        self.addCleanup(holder.cleanup)
        self.tmp = holder.name

    def _write(self, name, text):
        path = os.path.join(self.tmp, name)
        with open(path, "w", encoding="utf-8") as handle:
            handle.write(text)
        return path

    def _collect_text(self, name, text, **kwargs):
        path = self._write(name, text)
        kwargs.setdefault("source", f"test:{name}")
        kwargs.setdefault("created", FIXED_CREATED)
        snapshot, stats = collect(files=[path], **kwargs)
        return snapshot, stats, path

    @staticmethod
    def _chunks(snapshot):
        return {chunk["label"]: chunk for chunk in snapshot["chunks"]}

    # -- crafted shapes ----------------------------------------------------

    def test_labeled_value_lines(self):
        snapshot, stats, _ = self._collect_text(
            "values.txt",
            "alpha[0]=0x00000011\n"
            "alpha[4]=0x00000022\n"
            "beta[8]=0x0000000112345678\n",
        )
        chunks = self._chunks(snapshot)
        self.assertEqual(chunks["alpha"]["base"], "0x0")
        self.assertEqual(chunks["alpha"]["words"],
                         ["0x00000011", "0x00000022"])
        self.assertEqual(chunks["beta"]["base"], "0x8")
        self.assertEqual(chunks["beta"]["words"],
                         ["0x12345678", "0x00000001"])
        self.assertEqual(stats["unparsed"], 0)
        self.assertEqual(stats["conflicts"], 0)

    def test_labeled_hexdump_block(self):
        snapshot, stats, _ = self._collect_text(
            "block.txt",
            "preamble text that is not a header\n"
            "SOME_BUF: 64 bytes of server state\n"
            "00000000: 00000001 00000002 00000003 00000004\n"
            "00000010: 00000005 00000006 00000007 00000008\n"
            "RAMFC: a second block with a nonzero base\n"
            "00000020: 0000000a 0000000b\n",
        )
        chunks = self._chunks(snapshot)
        self.assertEqual(chunks["SOME_BUF"]["base"], "0x0")
        self.assertEqual(
            chunks["SOME_BUF"]["words"],
            ["0x00000001", "0x00000002", "0x00000003", "0x00000004",
             "0x00000005", "0x00000006", "0x00000007", "0x00000008"])
        self.assertEqual(chunks["RAMFC"]["base"], "0x20")
        self.assertEqual(chunks["RAMFC"]["words"],
                         ["0x0000000a", "0x0000000b"])
        self.assertEqual(stats["unparsed"], 0)
        self.assertEqual(stats["chunks"], 2)

    def test_logininit_raw_block(self):
        snapshot, stats, _ = self._collect_text(
            "loginit.txt",
            "vgsp-boot: libos task log buffers (read-only)\n"
            "LOGINIT-raw: 65536 bytes (allocated 65536) "
            "VA=0xffff88a673af0000 dma=0x4f3af0000\n"
            "00000000: 000002d2 00000000 05bb9e10 00000000\n"
            "00000010: f7660780 18dbb28f fe2c0000 61800001\n",
        )
        chunk = self._chunks(snapshot)["LOGINIT-raw"]
        self.assertEqual(chunk["base"], "0x0")
        self.assertEqual(
            chunk["words"],
            ["0x000002d2", "0x00000000", "0x05bb9e10", "0x00000000",
             "0xf7660780", "0x18dbb28f", "0xfe2c0000", "0x61800001"])
        self.assertEqual(stats["unparsed"], 0)

    def test_inline_labeled_rows_group_into_one_chunk(self):
        snapshot, _, _ = self._collect_text(
            "inline.txt",
            "[dmesg] vgsp-boot: rm-call: INSTANCE[0000]: "
            "00000000 00000000 0a545003 00000001\n"
            "[dmesg] vgsp-boot: rm-call: INSTANCE[0020]: "
            "0000face 00000000 00000000 00000000\n",
        )
        chunk = self._chunks(snapshot)["INSTANCE"]
        self.assertEqual(chunk["base"], "0x0")
        self.assertEqual(chunk["words"],
                         ["0x00000000", "0x00000000", "0x0a545003",
                          "0x00000001", "0x0000face", "0x00000000",
                          "0x00000000", "0x00000000"])

    def test_record_lines(self):
        snapshot, stats, _ = self._collect_text(
            "records.txt",
            'vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 '
            'subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" '
            'faultingEngine=""\n',
        )
        self.assertEqual(len(snapshot["records"]), 1)
        record = snapshot["records"][0]
        self.assertEqual(record["kind"], "record")
        self.assertEqual(record["rec_type"], 5)
        self.assertEqual(record["error_code"], 0x57F1344)
        self.assertEqual(record["source_name"], "ASSERT")
        self.assertEqual(stats["records"], 1)
        self.assertEqual(snapshot["chunks"], [])

    # -- duplicates, collisions, unparsed ---------------------------------

    def test_identical_duplicate_merged(self):
        snapshot, stats, _ = self._collect_text(
            "merged.txt",
            "INSTANCE[0000]: 00000001 00000002\n"
            "plain prose closes the first block\n"
            "INSTANCE[0000]: 00000001 00000002\n",
        )
        self.assertEqual(list(self._chunks(snapshot)), ["INSTANCE"])
        self.assertEqual(stats["merged"], 1)
        self.assertEqual(stats["collisions"], 0)

    def test_conflicting_duplicate_renamed(self):
        snapshot, stats, _ = self._collect_text(
            "collision.txt",
            "INSTANCE[0000]: 00000001 00000002\n"
            "plain prose closes the first block\n"
            "INSTANCE[0000]: 00000003 00000004\n",
        )
        self.assertEqual(list(self._chunks(snapshot)),
                         ["INSTANCE", "INSTANCE#2"])
        self.assertEqual(stats["collisions"], 1)
        self.assertEqual(stats["collision_list"],
                         [{"label": "INSTANCE", "renamed": "INSTANCE#2",
                           "base": "0x0", "words": 2}])

    def test_unparsed_candidates_counted(self):
        snapshot, stats, _ = self._collect_text(
            "unparsed.txt",
            "BROKEN[0x10]: zzzz\n"
            "00000000: zz00\n"
            "ordinary prose is not a candidate\n",
        )
        self.assertEqual(snapshot["chunks"], [])
        self.assertEqual(stats["unparsed"], 2)

    # -- diff --------------------------------------------------------------

    def test_diff_changed_added_removed(self):
        snap_a, _, _ = self._collect_text(
            "run-a.txt",
            "MOCK[0000]: 00000001 00000002\n"
            "shared[0]=0x00000007\n"
            "only_a[0]=0x00000009\n",
        )
        snap_b, _, _ = self._collect_text(
            "run-b.txt",
            "MOCK[0000]: 00000001 00000003 00000004\n"
            "shared[0]=0x00000007\n"
            "only_b[0]=0x0000000a\n",
        )
        report = diff_snapshots(snap_a, snap_b)
        self.assertEqual(report["schema"], DIFF_FORMAT)
        rows = {row["label"]: row for row in report["chunks"]}
        self.assertEqual(rows["MOCK"]["status"], "changed")
        self.assertEqual(rows["MOCK"]["changed"],
                         [{"index": 1, "a": "0x00000002", "b": "0x00000003"}])
        self.assertEqual(rows["MOCK"]["added"],
                         [{"index": 2, "value": "0x00000004"}])
        self.assertEqual(rows["MOCK"]["removed"], [])
        self.assertEqual(rows["shared"]["status"], "unchanged")
        self.assertEqual(rows["only_a"]["status"], "removed")
        self.assertEqual(rows["only_a"]["removed"],
                         [{"index": 0, "value": "0x00000009"}])
        self.assertEqual(rows["only_b"]["status"], "added")
        self.assertEqual(rows["only_b"]["added"],
                         [{"index": 0, "value": "0x0000000a"}])
        summary = report["summary"]
        self.assertFalse(summary["identical"])
        self.assertEqual(summary["chunks_added"], 1)
        self.assertEqual(summary["chunks_removed"], 1)
        self.assertEqual(summary["chunks_changed"], 1)
        self.assertEqual(summary["chunks_unchanged"], 1)
        self.assertEqual(summary["words_added"], 2)
        self.assertEqual(summary["words_removed"], 1)
        self.assertEqual(summary["words_changed"], 1)

    def test_diff_identical(self):
        snapshot, _, _ = self._collect_text(
            "same.txt", "INSTANCE[0000]: 00000001 00000002\n")
        report = diff_snapshots(snapshot, snapshot)
        self.assertTrue(report["summary"]["identical"])
        self.assertEqual(report["summary"]["words_changed"], 0)
        self.assertEqual([row["status"] for row in report["chunks"]],
                         ["unchanged"])

    def test_diff_rejects_wrong_schema(self):
        with self.assertRaises(SnapshotError):
            diff_snapshots({"schema": "other"}, {"schema": SNAPSHOT_FORMAT})

    # -- determinism and CLI ----------------------------------------------

    def test_determinism_with_source_date_epoch(self):
        path = self._write(
            "det.txt",
            "INSTANCE[0000]: 00000001 00000002\n"
            'vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 '
            'source="ASSERT"\n',
        )
        with mock.patch.dict(os.environ, {"SOURCE_DATE_EPOCH": "1700000000"}):
            first, _ = collect(files=[path], source="det")
            second, _ = collect(files=[path], source="det")
        self.assertEqual(first["created"], "2023-11-14T22:13:20Z")
        self.assertEqual(first, second)
        self.assertEqual(json.dumps(first, sort_keys=True),
                         json.dumps(second, sort_keys=True))

    def test_cli_collect_stdout_and_json(self):
        path = self._write("cli.txt", "INSTANCE[0000]: 00000001 00000002\n")
        rc, out, err = _run_main(
            ["collect", "--file", path, "--source", "cli",
             "--created", FIXED_CREATED])
        self.assertEqual(rc, 0)
        snapshot = json.loads(out)
        self.assertEqual(set(snapshot),
                         {"schema", "source", "created", "chunks", "records"})
        self.assertEqual(snapshot["schema"], SNAPSHOT_FORMAT)
        self.assertEqual(snapshot["source"], "cli")
        self.assertIn("1 chunk(s)", err)

        json_path = os.path.join(self.tmp, "snapshot.json")
        rc, out, _ = _run_main(
            ["collect", "--file", path, "--json", json_path,
             "--created", FIXED_CREATED])
        self.assertEqual(rc, 0)
        with open(json_path, "r", encoding="utf-8") as handle:
            written = json.load(handle)
        self.assertEqual(written["schema"], SNAPSHOT_FORMAT)
        self.assertEqual(json.loads(out)["chunks"], len(written["chunks"]))

    def test_cli_errors(self):
        rc, _, err = _run_main(["collect"])
        self.assertEqual(rc, 2)
        self.assertIn("error:", err)
        missing = os.path.join(self.tmp, "missing.txt")
        rc, _, _ = _run_main(["collect", "--file", missing])
        self.assertEqual(rc, 2)

    def test_discover_include_globs(self):
        self._write("keep.txt", "AAA[0]=0x00000001\n")
        self._write("skip.bin", "BBB[0]=0x00000002\n")
        snapshot, _ = collect(dirs=[self.tmp], source="glob",
                              created=FIXED_CREATED)
        self.assertEqual(list(self._chunks(snapshot)), ["AAA"])
        snapshot, _ = collect(dirs=[self.tmp], includes=["*.bin"],
                              source="glob", created=FIXED_CREATED)
        self.assertEqual(list(self._chunks(snapshot)), ["AAA", "BBB"])

    def test_demo_cli(self):
        rc, out, _ = _run_main(["demo"])
        self.assertEqual(rc, 0)
        self.assertIn("MOCK", out)
        summary = json.loads(out[out.index("{"):])
        self.assertFalse(summary["identical"])
        self.assertEqual(summary["words_changed"], 2)

    # -- real campaign evidence (optional) --------------------------------

    @unittest.skipUnless(os.path.isdir(EVIDENCE_VICTORY),
                         "victory evidence backup not present")
    def test_real_victory_evidence(self):
        files = [path for path in (LADDER_M12, PHYS_STATE_M26)
                 if os.path.isfile(path)]
        if not files:
            self.skipTest("expected victory evidence files missing")
        snapshot, stats = collect(files=files, source="evidence-victory",
                                  created=FIXED_CREATED)
        self.assertTrue(snapshot["chunks"])
        chunks = self._chunks(snapshot)
        self.assertIn("INSTANCE", chunks)
        self.assertEqual(chunks["INSTANCE"]["base"], "0x0")
        self.assertEqual(len(chunks["INSTANCE"]["words"]), 1024)
        self.assertIn("pt", chunks)
        self.assertEqual(len(chunks["pt"]["words"]), 5)
        self.assertGreaterEqual(stats["records"], 1)
        self.assertEqual(stats["files"], len(files))

    @unittest.skipUnless(os.path.isfile(LOGDUMP_GRCTX4),
                         "gr logdump evidence not present")
    def test_real_loginit_raw_block(self):
        snapshot, stats = collect(files=[LOGDUMP_GRCTX4],
                                  source="evidence-gr",
                                  created=FIXED_CREATED)
        chunks = self._chunks(snapshot)
        self.assertIn("LOGINIT-raw", chunks)
        self.assertIn("LOGRM-raw", chunks)
        self.assertEqual(chunks["LOGINIT-raw"]["base"], "0x0")
        self.assertEqual(len(chunks["LOGINIT-raw"]["words"]), 65536 // 4)
        self.assertEqual(len(chunks["LOGRM-raw"]["words"]), 65536 // 4)
        self.assertEqual(stats["unparsed"], 0)


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