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

"""Device-model tests: registry, per-block semantics, profile and boot ladder.

Every modelled register assertion here is tied to the evidence documented in
``sim/devices/*.py`` and ``docs/36-GSP-DEVICE-MODEL.md``.  The real-firmware
tests are bounded (<= 20k instructions) and assert *observed* rungs only.
"""
import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from sim.devices import (  # noqa: E402
    CsbWindow,
    DmaTranslator,
    FalconBlock,
    IrqRouter,
    PmcBlock,
    PmuStub,
    PriscvBlock,
    PtimerBlock,
    Wpr2Block,
    load_profile,
    registry_from_profile,
)
from sim.devices.base import DeviceRegistry  # noqa: E402
from sim.devices import falcon as falcon_mod  # noqa: E402
from sim.machine import Machine  # noqa: E402

TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_IMAGE = os.path.normpath(
    os.path.join(TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144",
                 "gsp-570.144.fwimage.bin"))
DEFAULT_PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json")


class RegistryTests(unittest.TestCase):
    def test_containment_and_unhandled_log(self):
        reg = DeviceRegistry()
        dev = PmcBlock(base=0x1000, size=0x1000)
        reg.add(dev)
        ok, value = reg.read(0x1A00, 4)
        self.assertTrue(ok)
        self.assertEqual(value, dev.boot_42)
        # unmodelled offset inside the window: unhandled but window-matched
        ok, _ = reg.read(0x1A04, 4)
        self.assertFalse(ok)
        self.assertEqual(reg.unhandled, 1)
        # outside every window
        ok, _ = reg.read(0x2000, 4)
        self.assertFalse(ok)
        self.assertEqual(reg.unhandled, 2)
        # Only window-matched accesses are recorded; outside-window accesses
        # are counted but left to the machine's unknown log.
        self.assertEqual(len(reg.accesses), 2)
        self.assertFalse(reg.accesses[1]["handled"])

    def test_overlapping_windows_fall_back_on_refusal(self):
        # A later window shadows an earlier one, but an offset the overlay does
        # not model (here the PMU stub refuses everything) falls back to the
        # wider device instead of being logged as unknown.
        reg = DeviceRegistry()
        fal = FalconBlock(base=0x1000, size=0x1000)
        reg.add(fal)
        reg.add(PmuStub(base=0x1000, size=0x1000))
        self.assertTrue(reg.write(0x1040, 4, 0xCA))  # MAILBOX0 via fallback
        self.assertEqual(fal.regs[falcon_mod.MAILBOX0], 0xCA)
        ok, value = reg.read(0x1040, 4)
        self.assertTrue(ok)
        self.assertEqual(value, 0xCA)
        self.assertEqual(reg.unhandled, 0)
        self.assertTrue(all(a["handled"] for a in reg.accesses))

    def test_overlapping_windows_all_refuse_records_once(self):
        reg = DeviceRegistry()
        reg.add(PmuStub(base=0x1000, size=0x1000))
        reg.add(PmuStub(base=0x1000, size=0x1000))
        ok, _ = reg.read(0x1000, 4)
        self.assertFalse(ok)
        self.assertEqual(reg.unhandled, 1)
        self.assertEqual(len(reg.accesses), 1)
        self.assertFalse(reg.accesses[0]["handled"])
        self.assertFalse(reg.write(0x1000, 4, 1))
        self.assertEqual(reg.unhandled, 2)
        self.assertEqual(len(reg.accesses), 2)

    def test_straddling_access_is_not_served(self):
        reg = DeviceRegistry()
        reg.add(PmcBlock(base=0x1000, size=0x10))
        ok, _ = reg.read(0x1008, 8)  # ends at 0x1010, outside the window
        self.assertFalse(ok)

    def test_state_round_trip(self):
        reg = DeviceRegistry()
        fal = FalconBlock(base=0x1000, size=0x1000)
        reg.add(fal)
        reg.write(0x1040, 4, 0xCA)
        snap = reg.state()
        reg2 = DeviceRegistry()
        fal2 = FalconBlock(base=0x1000, size=0x1000)
        reg2.add(fal2)
        reg2.load_state(snap)
        self.assertEqual(fal2.regs[falcon_mod.MAILBOX0], 0xCA)
        self.assertEqual(fal2.state(), fal.state())


class PmcTests(unittest.TestCase):
    def test_tu104_value_fields(self):
        dev = PmcBlock(chip_id=0x164)
        value = dev.read(0xA00, 4)
        self.assertEqual((value >> 20) & 0x1FF, 0x164)
        self.assertEqual((value >> 24) & 0x1F, 0x16)  # ARCHITECTURE TU100
        self.assertEqual((value >> 20) & 0xF, 0x4)    # IMPLEMENTATION
        self.assertEqual(dev.read(0xA04, 4), None)
        self.assertFalse(dev.write(0xA00, 4, 0))

    def test_profile_profile_value_override(self):
        dev = PmcBlock(boot_42=0x00123456)
        self.assertEqual(dev.read(0xA00, 4), 0x00123456)

    def test_chip_id_constants_are_distinct(self):
        from sim.devices import pmc as pmc_mod
        ids = [pmc_mod.TU102, pmc_mod.TU104, pmc_mod.TU105, pmc_mod.TU106,
               pmc_mod.TU116, pmc_mod.TU117]
        self.assertEqual(len(ids), len(set(ids)))
        self.assertEqual(pmc_mod.TU117, 0x167)
        self.assertFalse(hasattr(pmc_mod, "TU107"))


class FalconTests(unittest.TestCase):
    def test_irq_set_clear_status(self):
        dev = FalconBlock()
        dev.write(falcon_mod.IRQSSET, 4, falcon_mod.IRQ_HALT)
        self.assertEqual(dev.read(falcon_mod.IRQSTAT, 4), falcon_mod.IRQ_HALT)
        dev.write(falcon_mod.IRQSCLR, 4, falcon_mod.IRQ_HALT)
        self.assertEqual(dev.read(falcon_mod.IRQSTAT, 4), 0)
        # write-only set/clear registers have no read-back
        self.assertIsNone(dev.read(falcon_mod.IRQSSET, 4))

    def test_mailbox_rw(self):
        dev = FalconBlock()
        dev.write(falcon_mod.MAILBOX0, 4, 0xCA)
        dev.write(falcon_mod.MAILBOX1, 4, 0x11DEAD)
        self.assertEqual(dev.read(falcon_mod.MAILBOX0, 4), 0xCA)
        self.assertEqual(dev.read(falcon_mod.MAILBOX1, 4), 0x11DEAD)

    def test_debuginfo_storage_without_shim(self):
        dev = FalconBlock(debuginfo_host_consume=False)
        dev.write(falcon_mod.DEBUGINFO, 4, 0x11DEAD)
        self.assertEqual(dev.read(falcon_mod.DEBUGINFO, 4), 0x11DEAD)

    def test_debuginfo_host_consume_shim(self):
        dev = FalconBlock(debuginfo_host_consume=True)
        dev.write(falcon_mod.DEBUGINFO, 4, 0x11DEAD)
        value = dev.read(falcon_mod.DEBUGINFO, 4)
        self.assertEqual(value, 0x11DEAD)          # value preserved for the reader
        self.assertEqual(dev.read(falcon_mod.DEBUGINFO, 4), 0x1DEAD)  # stage cleared
        self.assertEqual(dev.debuginfo_consumed, 1)
        # debugger peek never consumes
        dev.write(falcon_mod.DEBUGINFO, 4, 0x21DEAD)
        self.assertEqual(dev.peek(falcon_mod.DEBUGINFO, 4), 0x21DEAD)
        self.assertEqual(dev.debuginfo_consumed, 1)

    def test_dma_transfer_through_translator(self):
        ram = bytearray(0x10000)
        ram[0x100:0x108] = b"DEADBEEF"
        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram),
                            windows=[{"pa": 0x0, "sim": 0, "size": 0x10000}])
        dev = FalconBlock(dma=dma)
        # Program: destination = 0x200, source = IMEM + 0x100.  The register
        # encodes ((base1 << 32) | base) << 8; the profile's DMA windows decide
        # which aperture that address resolves to (aperture selection itself is
        # not proven yet and is documented as an unknown).
        dev.write(falcon_mod.DMATRFBASE1, 4, 0)
        dev.write(falcon_mod.DMATRFBASE, 4, 0)
        dev.write(falcon_mod.DMATRFFBOFFS, 4, 0x200)
        dev.write(falcon_mod.DMATRFMOFFS, 4, 0x100)
        cmd = (falcon_mod.CMD_IMEM | falcon_mod.CMD_WRITE
               | (2 << falcon_mod.CMD_SIZE_SHIFT))  # IMEM -> FB, 16 bytes
        dev.write(falcon_mod.DMATRFCMD, 4, cmd)
        self.assertEqual(ram[0x200:0x210], b"DEADBEEF" + bytes(8))
        self.assertTrue(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_IDLE)
        self.assertEqual(dev.dma_transfers, 1)

    def test_dma_refused_when_unmapped(self):
        ram = bytearray(0x1000)
        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram), windows=[])
        dev = FalconBlock(dma=dma)
        dev.write(falcon_mod.DMATRFBASE, 4, 0)
        dev.write(falcon_mod.DMATRFCMD, 4, falcon_mod.CMD_IMEM)
        self.assertFalse(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_IDLE)
        self.assertTrue(dev.regs[falcon_mod.DMATRFCMD] & falcon_mod.CMD_ERROR)
        self.assertEqual(dev.dma_refused, 1)


class PriscvTests(unittest.TestCase):
    def test_irq_mask_semantics(self):
        dev = PriscvBlock()
        dev.write(0x2AC, 4, 0x5)
        dev.write(0x2B0, 4, 0x1)
        self.assertEqual(dev.read(0x2B4, 4), 0x4)
        self.assertFalse(dev.write(0x2B4, 4, 0xFFFF))  # read-only

    def test_trace_and_error_registers(self):
        dev = PriscvBlock()
        dev.write(0x344, 4, 0x1234)
        self.assertEqual(dev.read(0x344, 4), 0x1234)
        # observed boot panic reads (R-I4R): read-back is zero, writes refused
        self.assertEqual(dev.read(0x364, 4), 0)
        self.assertEqual(dev.read(0x368, 4), 0)
        self.assertFalse(dev.write(0x368, 4, 0x1))


class TimerTests(unittest.TestCase):
    def test_modeled_time_reads(self):
        clock = {"insns": 0}
        dev = PtimerBlock(ns_per_insn=2.0, clock=lambda: clock["insns"])
        self.assertEqual(dev.read(0x400, 4), 0)
        self.assertEqual(dev.read(0x410, 4), 0)
        clock["insns"] = 16
        # 32 ns -> TIME_0 NSEC field (bits 31:5) = 1
        self.assertEqual(dev.read(0x400, 4), 1)
        self.assertEqual(dev.read(0x410, 4), 32)

    def test_profile_window_reaches_time_registers(self):
        # Regression: the profile window is PRIV+0x9000, so TIME_0/TIME_1 must
        # be block-relative (0x400/0x410); aperture-absolute 0x9400/0x9410 fall
        # outside the declared window and serve nothing.
        profile = {
            "schema": "greeneye-sim-profile/1",
            "devices": {
                "timer": {
                    "ns_per_insn": 1.0,
                    "windows": [{"base": "0x9000", "size": "0x1000"}],
                }
            },
        }
        machine = Machine(memory_mb=1, ram_base=0)
        reg = registry_from_profile(profile, machine=machine)
        machine.insns = 32
        ok, value = reg.read(0x9000 + 0x400, 4, pc=0)
        self.assertTrue(ok)
        self.assertEqual(value, 1)  # (32 ns >> 5)
        ok, value = reg.read(0x9000 + 0x410, 4, pc=0)
        self.assertTrue(ok)
        self.assertEqual(value, 32)

    def test_timer_registration_is_opt_in(self):
        machine = Machine(memory_mb=1, ram_base=0)
        no_section = {
            "schema": "greeneye-sim-profile/1",
            "devices": {"pmc": {"windows": [{"base": "0x1000", "size": "0x1000"}]}},
        }
        reg = registry_from_profile(no_section, machine=machine)
        self.assertIsNone(reg.device_by_name("timer"))
        disabled = {
            "schema": "greeneye-sim-profile/1",
            "devices": {"timer": {"enabled": False,
                                  "windows": [{"base": "0x9000", "size": "0x1000"}]}},
        }
        reg = registry_from_profile(disabled, machine=machine)
        self.assertIsNone(reg.device_by_name("timer"))
        enabled = {
            "schema": "greeneye-sim-profile/1",
            "devices": {"timer": {"ns_per_insn": 1.0,
                                  "windows": [{"base": "0x9000", "size": "0x1000"}]}},
        }
        reg = registry_from_profile(enabled, machine=machine)
        self.assertIsNotNone(reg.device_by_name("timer"))


class CsbTests(unittest.TestCase):
    def test_undeclared_window_serves_nothing(self):
        dev = CsbWindow(base=0x1000, size=0x1000)
        self.assertIsNone(dev.read(0x0, 4))
        self.assertFalse(dev.write(0x0, 4, 1))

    def test_declared_semantics(self):
        dev = CsbWindow(base=0x1000, size=0x1000, registers={
            0x00: {"width": 4, "value": 0xF, "semantics": "w1c"},
            0x04: {"width": 4, "value": 0, "semantics": "rw"},
            0x08: {"width": 4, "value": 0, "semantics": "ro"},
        })
        dev.write(0x00, 4, 0x3)
        self.assertEqual(dev.read(0x00, 4), 0xC)
        dev.write(0x04, 4, 0x77)
        self.assertEqual(dev.read(0x04, 4), 0x77)
        self.assertFalse(dev.write(0x08, 4, 1))


class Wpr2Tests(unittest.TestCase):
    def test_range_encoding(self):
        dev = Wpr2Block()
        dev.set_range(0x2000000, 0x6000000)
        lo = dev.read(0x1FA824, 4)
        hi = dev.read(0x1FA828, 4)
        self.assertEqual((lo << 12), 0x2000000)
        self.assertEqual((hi << 12) + 0x20000, 0x6000000)
        self.assertIsNone(dev.read(0x1FA800, 4))

    def test_state_derives_range_after_writes(self):
        # Regression: writes update regs only; state() must derive start/end
        # from the registers rather than reporting a stale set_range range.
        dev = Wpr2Block()
        dev.write(0x1FA824, 4, 0x2000000 >> 12)
        dev.write(0x1FA828, 4, (0x6000000 - 0x20000) >> 12)
        state = dev.state()
        self.assertEqual(state["start"], 0x2000000)
        self.assertEqual(state["end"], 0x6000000)
        restored = Wpr2Block()
        restored.load_state(state)
        self.assertEqual(restored.start, 0x2000000)
        self.assertEqual(restored.end, 0x6000000)
        # A stale serialised start/end never wins over the registers.
        stale = Wpr2Block()
        stale.load_state({"regs": {str(0x1FA824): 0x2000},
                          "start": 0xDEAD, "end": 0xBEEF})
        self.assertEqual(stale.start, 0x2000000)
        self.assertIsNone(stale.end)


class PmuTests(unittest.TestCase):
    def test_stub_never_fakes(self):
        dev = PmuStub(base=0x1000, size=0x1000)
        self.assertIsNone(dev.read(0x0, 4))
        self.assertFalse(dev.write(0x0, 4, 1))
        self.assertEqual(dev.seen["r:0x0:4"], 1)


class IrqRouterTests(unittest.TestCase):
    def test_aggregation(self):
        fal = FalconBlock()
        pri = PriscvBlock()
        router = IrqRouter(falcon=fal, priscv=pri)
        fal.write(falcon_mod.IRQSSET, 4, falcon_mod.IRQ_HALT)
        pri.write(0x2AC, 4, 0x2)
        state = router.pending()
        self.assertTrue(state["falcon_halt"])
        self.assertEqual(state["riscv_irq_mask"], 0x2)


class DmaTranslatorTests(unittest.TestCase):
    def test_translate_and_refuse(self):
        ram = bytearray(0x100)
        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram),
                            windows=[{"pa": 0x1000, "sim": 0, "size": 0x100}])
        self.assertEqual(dma.translate(0x1000), 0)
        self.assertEqual(dma.translate(0x10FF), 0xFF)
        self.assertIsNone(dma.translate(0x2000))
        self.assertTrue(dma.write(0x1000, b"AB"))
        self.assertEqual(dma.read(0x1000, 2), b"AB")
        self.assertIsNone(dma.read(0x2000, 2))
        self.assertEqual(dma.refused, 1)

    def test_copy_counts_only_completed_transfers_and_exports_events(self):
        ram = bytearray(0x100)
        ram[0:4] = b"ABCD"
        dma = DmaTranslator(ram=ram, ram_base=0, ram_size=len(ram),
                            windows=[{"pa": 0x1000, "sim": 0, "size": 0x100}])
        # Destination outside every window: refused, not a transfer.
        self.assertFalse(dma.copy(0x1000, 0x2000, 4))
        self.assertEqual(dma.transfers, 0)
        self.assertEqual(dma.refused, 1)
        # Both mapped: one completed transfer.
        self.assertTrue(dma.copy(0x1000, 0x1004, 4))
        self.assertEqual(dma.transfers, 1)
        self.assertEqual(ram[0x4:0x8], b"ABCD")
        state = dma.state()
        self.assertIn("events", state)
        self.assertEqual(state["events"][0]["kind"], "w")
        self.assertEqual(len(state["events"]), 1)


class MachineDeviceIntegrationTests(unittest.TestCase):
    def _machine(self):
        reg = DeviceRegistry()
        reg.add(PmcBlock(base=0x2000000000000000, size=0x1000))
        m = Machine(memory_mb=1, ram_base=0, unknown_mmio="lenient", devices=reg)
        return m, reg

    def test_modeled_device_takes_precedence_over_lenient(self):
        m, reg = self._machine()
        self.assertEqual(m._mmio_load(0x2000000000000A00, 4), PmcBlock().boot_42)
        self.assertEqual(m.mmio_unknown, 0)
        self.assertEqual(len(reg.accesses), 1)

    def test_unknown_inside_window_is_logged(self):
        m, reg = self._machine()
        self.assertEqual(m._mmio_load(0x2000000000000A04, 4), 0)
        self.assertEqual(m.mmio_unknown, 1)
        self.assertFalse(reg.accesses[0]["handled"])

    def test_debugger_peek_and_device_state(self):
        m, reg = self._machine()
        self.assertEqual(m.read_mem(0x2000000000000A00, 4),
                         PmcBlock().boot_42.to_bytes(4, "little"))
        self.assertIsInstance(m.device_state(), dict)
        self.assertEqual(m.device_state()["schema"], "greeneye-sim-devices/1")

    def test_reset_regs_names(self):
        m = Machine(memory_mb=1, ram_base=0, reset_regs={"a0": 0x1234, "sp": 0x8000})
        self.assertEqual(m.x[10], 0x1234)
        self.assertEqual(m.x[2], 0x8000)


@unittest.skipUnless(os.path.isfile(DEFAULT_IMAGE), "real GSP firmware absent")
class BootLadderTests(unittest.TestCase):
    """Bounded real-firmware runs: observed rungs only."""

    def test_profile_loads_and_builds_windows(self):
        profile = load_profile(DEFAULT_PROFILE)
        self.assertEqual(profile["schema"], "greeneye-sim-profile/1")
        m = Machine(memory_mb=16, ram_base=0, unknown_mmio="lenient")
        reg = registry_from_profile(profile, machine=m)
        names = {d.name for d in reg.devices}
        self.assertIn("pmc", names)
        self.assertIn("falcon", names)
        self.assertIn("priscv", names)
        self.assertIsNotNone(reg.dma)

    def test_no_shim_b1_device_model_serves_boot(self):
        from sim.gsp_boot import run_boot
        r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
                     instructions=20_000, record=False, use_shim=False)
        self.assertTrue(r.ran)
        self.assertEqual(r.rung, "B1")
        self.assertIn("bootargs_consumer", r.milestones)
        self.assertEqual(r.mmio_unknown, 0)
        self.assertEqual(r.device_unhandled, 0)

    def test_shim_reaches_message_loop_b3(self):
        from sim.gsp_boot import run_boot
        r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
                     instructions=2_000_000, record=False, use_shim=True)
        self.assertTrue(r.ran)
        self.assertEqual(r.rung, "B3")
        self.assertIn("relocation_copy", r.milestones)
        self.assertEqual(r.mmio_unknown, 0)
        self.assertEqual(r.device_unhandled, 0)
        # The MMU model resolves the resident-section copy, the libos service
        # path runs (DMA/pagetable registration) and the message-loop task
        # blocks in the port wait -- deterministic, not faked.
        self.assertEqual(r.stop_reason, "ecall")
        self.assertEqual(r.stop_pc, 0x4018038)
        stopped = r.libos.get("stopped") or {}
        self.assertEqual(stopped.get("num"), 1)  # task_port_send_recv_wait

    def test_shim_reaches_relocation_plus_mmu(self):
        """The relocation copy reads its real backing through the MPU."""
        from sim.gsp_boot import run_boot
        r = run_boot(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
                     instructions=20_000, record=False, use_shim=True)
        self.assertTrue(r.ran)
        self.assertIn("relocation_copy", r.milestones)
        self.assertEqual(r.mmio_unknown, 0)
        # The run no longer stops at the old B2 fault; with a 20k budget it
        # is mid-init and deterministic.
        self.assertNotEqual(r.stop_pc, 0x400EA0C)

    def test_two_runs_identical(self):
        from sim.gsp_boot import run_boot
        kwargs = dict(image=DEFAULT_IMAGE, profile_path=DEFAULT_PROFILE,
                      instructions=20_000, record=False, use_shim=True)
        a = run_boot(**kwargs)
        b = run_boot(**kwargs)
        self.assertEqual(a.instructions, b.instructions)
        self.assertEqual(a.stop_pc, b.stop_pc)
        self.assertEqual(a.stop_reason, b.stop_reason)
        self.assertEqual(a.device_state_sha256, b.device_state_sha256)
        self.assertEqual(a.registers, b.registers)


if __name__ == "__main__":
    unittest.main()