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