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