diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py | 364 |
1 files changed, 364 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py b/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py new file mode 100644 index 0000000..b82d09c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_server_snapshot.py @@ -0,0 +1,364 @@ +"""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() |