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

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