diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_state_seed.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_state_seed.py | 311 |
1 files changed, 311 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py b/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py new file mode 100644 index 0000000..439f36d --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_state_seed.py @@ -0,0 +1,311 @@ +"""Tests for the RM-object state-seeding harness (harness/state_seed.py, docs/52). + +Loader tests run against the native machine directly (no firmware image); the +FECS mechanism proof and the construct-gate attempt run on the shipped 570.144 +image and skip when it is absent. No hardware, no module, no guest, no +network. + +Run:: + + python3 -m unittest tests.test_state_seed -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 sim.machine import Machine # noqa: E402 + +from harness.rehost import DEFAULT_IMAGE # noqa: E402 +from harness.state_seed import ( # noqa: E402 + DEMO_FORMAT, + REPORT_FORMAT, + SEED_FORMAT, + StateSeedError, + apply_seed, + fecs_seed, + load_seed, + main, + run_seed, + target_seed, +) +from harness.valprobe import encode_return_stub # noqa: E402 + +PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") + + +def _run_main(argv: list[str]) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main(argv) + return rc, out.getvalue(), err.getvalue() + + +def _machine() -> Machine: + return Machine(memory_mb=64, ram_base=0, unknown_mmio="lenient") + + +def _two_object_seed() -> dict: + return { + "format": SEED_FORMAT, + "name": "loader-test", + "entry": "0x5290dd4", + "allocator": {"base": "0x100000", "align": 0x100}, + "objects": [ + {"name": "client", "type": "client", "size": "0x2000", "fields": { + "0x1be0": "@ctx", + "0x1dd8": {"value": 0, "size": 8}, + "0x1aa8": "unknown", + }}, + {"name": "ctx", "type": "graphics_context", "size": "0x1000", "fields": { + "0x190": "@table", + "0x161": {"value": 0, "size": 1}, + }}, + {"name": "table", "type": "rm_object", "size": "0x100", "fields": { + "0x30": {"value": 0xCAFE, "size": 8}, + }}, + ], + "registers": {"a0": "@client", "a1": "@ctx", "t1": 0xC}, + "stack": "0x300000", + "memory": [{"pa": "0x200000", "u32": 7}], + "stubs": [{"name": "getter", "addr": "0x3f0000", "return": "0x280000"}], + "provenance": {"author": "test"}, + } + + +class SchemaTests(unittest.TestCase): + def test_layout_allocates_objects_in_order(self): + plan = load_seed(_two_object_seed()) + by_name = {item["name"]: item for item in plan["layout"]} + self.assertEqual(by_name["client"]["pa"], 0x100000) + self.assertEqual(by_name["ctx"]["pa"], 0x102000) + self.assertEqual(by_name["table"]["pa"], 0x103000) + + def test_pointer_and_value_fields_are_encoded(self): + plan = load_seed(_two_object_seed()) + writes = {item["label"]: item for item in plan["writes"]} + self.assertEqual(bytes.fromhex(writes["client+0x1be0"]["hex"]), + (0x102000).to_bytes(8, "little")) + self.assertEqual(bytes.fromhex(writes["ctx+0x190"]["hex"]), + (0x103000).to_bytes(8, "little")) + self.assertEqual(bytes.fromhex(writes["table+0x30"]["hex"]), + (0xCAFE).to_bytes(8, "little")) + self.assertEqual(bytes.fromhex(writes["mem+0x200000"]["hex"]), + (7).to_bytes(4, "little")) + + def test_unknown_marker_is_reported_and_not_written(self): + plan = load_seed(_two_object_seed()) + unknown = {(item["object"], item["off"]) for item in plan["unknown_fields"]} + self.assertIn(("client", "0x1aa8"), unknown) + labels = {item["label"] for item in plan["writes"]} + self.assertNotIn("client+0x1aa8", labels) + + def test_pc_register_becomes_entry(self): + seed = _two_object_seed() + del seed["entry"] + seed["registers"]["pc"] = "0x5290dd4" + plan = load_seed(seed) + self.assertEqual(plan["entry"], 0x5290DD4) + self.assertNotIn("pc", plan["registers"]) + + def test_plan_is_deterministic(self): + first = load_seed(_two_object_seed()) + second = load_seed(_two_object_seed()) + self.assertEqual(first["writes"], second["writes"]) + self.assertEqual(first["layout"], second["layout"]) + + def test_bad_format_rejected(self): + seed = _two_object_seed() + seed["format"] = "greeneye-state-seed/2" + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_missing_entry_rejected(self): + seed = _two_object_seed() + del seed["entry"] + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_unknown_object_type_rejected(self): + seed = _two_object_seed() + seed["objects"].append({"name": "x", "type": "gpu_of_doom"}) + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_duplicate_symbol_rejected(self): + seed = _two_object_seed() + seed["objects"].append({"name": "client", "type": "client"}) + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_unknown_pointer_rejected(self): + seed = _two_object_seed() + seed["objects"][0]["fields"]["0x1de8"] = "@does-not-exist" + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_bad_memory_item_rejected(self): + seed = _two_object_seed() + seed["memory"] = [{"pa": "0x200000"}] + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_overlapping_objects_rejected(self): + seed = _two_object_seed() + seed["objects"][1]["pa"] = "0x100800" + with self.assertRaises(StateSeedError): + load_seed(seed) + + def test_bad_register_rejected(self): + seed = _two_object_seed() + seed["registers"]["r99"] = 1 + with self.assertRaises(StateSeedError): + load_seed(seed) + + +class LoaderTests(unittest.TestCase): + def test_apply_seed_places_bytes_and_registers(self): + plan = load_seed(_two_object_seed()) + machine = _machine() + apply_seed(machine, plan) + client = 0x100000 + self.assertEqual(machine.read_mem(client + 0x1be0, 8), + (0x102000).to_bytes(8, "little")) + self.assertEqual(machine.read_mem(client + 0x1dd8, 8), b"\x00" * 8) + self.assertEqual(machine.read_mem(client + 0x1aa8, 8), b"\x00" * 8) + self.assertEqual(machine.read_mem(0x200000, 4), (7).to_bytes(4, "little")) + self.assertEqual(machine.x[10], client) + self.assertEqual(machine.x[11], 0x102000) + self.assertEqual(machine.x[6], 0xC) + self.assertEqual(machine.read_mem(0x3F0000, 12), encode_return_stub(0x280000)) + + def test_run_seed_missing_image_is_clean_error(self): + with self.assertRaises(StateSeedError): + run_seed(_two_object_seed(), image="/nonexistent/fw.bin") + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class FecsMechanismTests(unittest.TestCase): + """Prove the seeder on a path that needs no RM objects (docs/47 predicate).""" + + def test_seeded_slot_value_drives_the_verdict(self): + nonnull = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False, + name="state-seed-test") + null = run_seed(fecs_seed(slot=0), profile_path=PROFILE, record=False, + name="state-seed-test") + self.assertEqual(nonnull["format"], REPORT_FORMAT) + self.assertEqual(nonnull["status"], "outcome") + self.assertEqual(nonnull["outcome"], "refuse-fecs-slot-exists-0x1f") + self.assertEqual(null["status"], "outcome") + self.assertEqual(null["outcome"], "accept-commit-path") + + def test_fecs_run_is_deterministic(self): + first = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False, + name="state-seed-test") + second = run_seed(fecs_seed(slot=1), profile_path=PROFILE, record=False, + name="state-seed-test") + self.assertEqual(first["outcome"], second["outcome"]) + self.assertEqual(first["instructions"], second["instructions"]) + self.assertEqual(first["state_reads"], second["state_reads"]) + + def test_report_records_session_and_json(self): + with tempfile.TemporaryDirectory() as tmp: + report = run_seed(fecs_seed(slot=1), profile_path=PROFILE, out=tmp, + record=True, name="state-seed-test") + self.assertTrue(os.path.isfile(report["report_json"])) + with open(report["report_json"], "r", encoding="utf-8") as fh: + doc = json.load(fh) + self.assertEqual(doc["format"], REPORT_FORMAT) + self.assertTrue(report["session"]) + self.assertTrue(os.path.isdir(report["session"])) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class TargetGateTests(unittest.TestCase): + """Attempt FUN_ram_052a70ec with a crafted client/channel/context seed.""" + + def test_minimal_seed_reports_first_gate(self): + report = run_seed(target_seed(advance=False), profile_path=PROFILE, + record=False, name="state-seed-test") + self.assertEqual(report["status"], "fault") + self.assertIsNotNone(report["gate"]) + self.assertEqual(report["gate"]["kind"], "null_indirect_call") + self.assertEqual(report["gate"]["label"], "engine_node+0x600") + self.assertEqual(report["gate"]["value"], 0) + labels = [item["label"] for item in report["state_reads"]] + self.assertIn("client+0x1de8", labels) + self.assertLess(labels.index("client+0x1de8"), + labels.index("engine_node+0x600")) + self.assertTrue(report["unknown_fields"]) + + def test_stub_advances_to_the_next_gate(self): + report = run_seed(target_seed(advance=True), profile_path=PROFILE, + record=False, name="state-seed-test") + self.assertEqual(report["status"], "fault") + self.assertEqual(report["gate"]["kind"], "null_indirect_call") + self.assertEqual(report["gate"]["label"], "channel+0x898") + + def test_target_seed_is_deterministic(self): + first = run_seed(target_seed(advance=False), profile_path=PROFILE, + record=False, name="state-seed-test") + second = run_seed(target_seed(advance=False), profile_path=PROFILE, + record=False, name="state-seed-test") + self.assertEqual(first["gate"], second["gate"]) + self.assertEqual(first["state_reads"], second["state_reads"]) + + +@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "570.144 firmware absent") +class CliTests(unittest.TestCase): + def test_emit_and_validate_round_trip(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "fecs.json") + rc, out, _ = _run_main(["emit", "--kind", "fecs", "--out", path]) + self.assertEqual(rc, 0) + self.assertTrue(os.path.isfile(path)) + rc, out, _ = _run_main(["validate", "--seed", path]) + self.assertEqual(rc, 0) + doc = json.loads(out) + self.assertEqual(doc["format"], SEED_FORMAT) + self.assertEqual(doc["entry"], "0x5290dd4") + + def test_run_emitted_seed_with_out(self): + with tempfile.TemporaryDirectory() as tmp: + seed_path = os.path.join(tmp, "fecs.json") + rc, _, _ = _run_main(["emit", "--kind", "fecs", "--out", seed_path]) + self.assertEqual(rc, 0) + out_dir = os.path.join(tmp, "out") + rc, out, _ = _run_main(["run", "--seed", seed_path, "--out", out_dir, + "--no-record"]) + self.assertEqual(rc, 0) + doc = json.loads(out) + self.assertEqual(doc["status"], "outcome") + self.assertEqual(doc["outcome"], "refuse-fecs-slot-exists-0x1f") + self.assertTrue(os.path.isfile( + os.path.join(out_dir, "state-seed-report.json"))) + + def test_run_with_bad_seed_returns_two(self): + rc, _, err = _run_main(["run", "--seed", "/nonexistent/seed.json"]) + self.assertEqual(rc, 2) + self.assertIn("error:", err) + + def test_demo_returns_zero(self): + rc, out, _ = _run_main(["demo"]) + self.assertEqual(rc, 0) + doc = json.loads(out) + self.assertEqual(doc["format"], DEMO_FORMAT) + self.assertTrue(doc["ok"]) + self.assertIsNotNone(doc["target"]["gate"]) + + +if __name__ == "__main__": + unittest.main() |