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

"""Device-map (H2.1/H2.2 start) tests: harvest, merge, query, lenient MMIO.

Unit tests use synthetic QEMU logs and tiny hand-assembled RV64 programs, so
they need neither QEMU nor a firmware build.  The real-image test is bounded
(200k instructions, no QEMU) and skips when the TU104 570.144 fwimage is
absent.
"""
from __future__ import annotations

import json
import os
import tempfile
import unittest

from greeneye.device_map import (
    SCHEMA,
    DeviceMap,
    harvest_from_text,
    harvest_probe_session,
)
from greeneye.errors import GreenEyeError
from sim.machine import R_BUDGET, R_EBREAK, R_FAULT, Machine

HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.normpath(os.path.join(HERE, ".."))
DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
REAL_IMAGE = os.path.normpath(
    os.path.join(
        HERE, "..", "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin"
    )
)

BASE = 0x80000000
TOUCH_A = 0x2000000000000A00  # QEMU probe first-touch address (570.144)
TOUCH_B = 0x2000000000000100
BIAS = 0x80000000

# lui t0,0x20000; slli t0,t0,32; addi t0,t0,0x600; addi t0,t0,0x400;
# lw a0,0(t0); sw a0,0(t0); ebreak
PROGRAM_LOAD_STORE = bytes.fromhex(
    "b702002093920202938202609382024003a5020023a0a20073001000"
)
# ... lw a0,0(t0); lui t1,0x20000; slli t1,t1,32; addi t1,t1,0x100;
# sd a0,0(t1); ebreak  (two distinct unknown addresses)
PROGRAM_TWO_TOUCHES = bytes.fromhex(
    "b702002093920202938202609382024003a502003703002013130302130303102330a30073001000"
)

SAMPLE_LOG = """\
Invalid read at addr 0x2000000000000A00, size 4, region '(null)', reason: rejected
riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000005, epc:0x0000000084009014, tval:0x2000000000000a00, desc=fault_load
Invalid read at addr 0x2000000000000A00, size 4, region '(null)', reason: rejected
riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000005, epc:0x0000000084009014, tval:0x2000000000000a00, desc=fault_load
Invalid write at addr 0x2000000000001000, size 8, region '(null)', reason: rejected
riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000007, epc:0x0000000084009020, tval:0x2000000000001000, desc=fault_store
"""


def _run_program(program: bytes, **kwargs):
    machine = Machine(**kwargs)
    machine.load_segments([(BASE, program)], BASE)
    return machine, machine.run(100)


class HarvestTests(unittest.TestCase):
    def test_harvest_from_text_fields(self):
        dm = harvest_from_text(SAMPLE_LOG, bias=BIAS)
        entries = {(e.address, e.width, e.kind): e for e in dm.entries()}
        self.assertEqual(set(entries), {(TOUCH_A, 4, "r"), (0x2000000000001000, 8, "w")})
        read = entries[(TOUCH_A, 4, "r")]
        self.assertEqual(read.hits, 2)
        self.assertEqual(read.first_epc, 0x84009014)
        self.assertEqual(read.first_epc_fw, 0x4009014)
        self.assertEqual(read.source, "probe")
        self.assertIn("fault_load", read.notes)
        write = entries[(0x2000000000001000, 8, "w")]
        self.assertEqual(write.hits, 1)
        self.assertEqual(write.first_epc_fw, 0x4009020)
        self.assertEqual(dm.meta["bias"], BIAS)

    def test_harvest_skips_fetch_faults(self):
        log = (
            "riscv_cpu_do_interrupt: hart:0, async:0, cause:0000000000000001, "
            "epc:0x0000000000000000, tval:0x0, desc=fault_fetch\n"
        )
        self.assertEqual(len(harvest_from_text(log, bias=BIAS)), 0)

    def test_harvest_probe_session_blob(self):
        with tempfile.TemporaryDirectory() as root:
            os.makedirs(os.path.join(root, "blobs"))
            with open(os.path.join(root, "blobs", "qemu.log"), "w", encoding="utf-8") as fh:
                fh.write(SAMPLE_LOG)
            session = {
                "format": "greeneye-session/1",
                "bias": BIAS,
                "entry": 0x4009000,
                "firmware": "/nonexistent/gsp-570.144.fwimage.bin",
                "qemu_log_blob": "blobs/qemu.log",
                "qemu_fault_count": 3,
            }
            with open(os.path.join(root, "session.json"), "w", encoding="utf-8") as fh:
                json.dump(session, fh)
            dm = harvest_probe_session(root)
        self.assertEqual(len(dm), 2)
        self.assertEqual(dm.meta["bias"], BIAS)
        self.assertEqual(dm.meta["qemu_fault_count"], 3)
        self.assertNotIn("firmware_sha256", dm.meta)  # firmware absent: honest to omit

    def test_harvest_probe_session_first_fault_fallback(self):
        with tempfile.TemporaryDirectory() as root:
            session = {
                "bias": BIAS,
                "qemu_first_fault": {
                    "epc": 0x84009014,
                    "cause": 5,
                    "desc": "fault_load",
                    "access": {"rw": "read", "access_addr": TOUCH_A, "size": 4,
                               "region": "(null)", "reason": "rejected"},
                },
            }
            with open(os.path.join(root, "session.json"), "w", encoding="utf-8") as fh:
                json.dump(session, fh)
            dm = harvest_probe_session(root)
        self.assertEqual(len(dm), 1)
        self.assertEqual(dm.lookup(TOUCH_A, 4), "zero")  # observed, no value recorded

    def test_harvest_bias_override(self):
        with tempfile.TemporaryDirectory() as root:
            session = {
                "qemu_first_fault": {
                    "epc": 0x84009014,
                    "cause": 5,
                    "desc": "fault_load",
                    "access": {"rw": "read", "access_addr": TOUCH_A, "size": 4,
                               "region": "(null)", "reason": "rejected"},
                },
            }
            with open(os.path.join(root, "session.json"), "w", encoding="utf-8") as fh:
                json.dump(session, fh)
            dm = harvest_probe_session(root, bias=BIAS)
        entry = dm.entries()[0]
        self.assertEqual(entry.first_epc, 0x84009014)
        self.assertEqual(entry.first_epc_fw, 0x4009014)
        self.assertEqual(dm.meta["bias"], BIAS)

    def test_harvest_missing_session_raises(self):
        with tempfile.TemporaryDirectory() as root:
            with self.assertRaises(GreenEyeError):
                harvest_probe_session(os.path.join(root, "nope"))


class MapIoTests(unittest.TestCase):
    def _map(self) -> DeviceMap:
        dm = DeviceMap(meta={"firmware": "gsp.bin", "bias": BIAS})
        dm.record(TOUCH_A, 4, "r", pc=0x84009014, bias=BIAS, source="probe")
        dm.record(0x2000, 8, "w", value=7, pc=0x84009020, bias=BIAS, source="manual")
        return dm

    def test_roundtrip_is_byte_identical(self):
        dm = self._map()
        with tempfile.TemporaryDirectory() as root:
            path = os.path.join(root, "map.json")
            dm.save(path)
            first = open(path, "rb").read()
            again = DeviceMap.load(path)
            again.save(path)
            self.assertEqual(open(path, "rb").read(), first)
        self.assertEqual(again.to_dict(), dm.to_dict())
        self.assertEqual(again.meta["bias"], BIAS)

    def test_entries_are_deterministically_ordered(self):
        dm = DeviceMap()
        for addr in (0x3000, 0x1000, 0x2000):
            dm.record(addr, 4, "r")
        self.assertEqual([e.address for e in dm.entries()], [0x1000, 0x2000, 0x3000])

    def test_merge_sums_hits_and_unions_values(self):
        left = DeviceMap()
        left.record(0x1000, 4, "r", value=1)
        left.record(0x1000, 4, "r", value=1)
        right = DeviceMap(meta={"source": "manual"})
        right.record(0x1000, 4, "r", value=2)
        right.record(0x2000, 8, "w")
        merged = left.merge(right)
        entries = {(e.address, e.kind): e for e in merged.entries()}
        self.assertEqual(entries[(0x1000, "r")].hits, 3)
        self.assertEqual(entries[(0x1000, "r")].values, (1, 2))
        self.assertEqual(entries[(0x2000, "w")].hits, 1)
        self.assertEqual(merged.meta["source"], "manual")

    def test_rejects_unknown_schema(self):
        with self.assertRaises(GreenEyeError):
            DeviceMap.from_json(json.dumps({"schema": "greeneye-device-map/2", "entries": []}))
        self.assertEqual(SCHEMA, "greeneye-device-map/1")


class LookupTests(unittest.TestCase):
    def test_unimplemented_zero_and_value(self):
        dm = DeviceMap()
        self.assertEqual(dm.lookup(0x1234, 4), "unimplemented")
        dm.record(0x2000, 4, "r", value=0xAB)
        self.assertEqual(dm.lookup(0x2000, 4), "value:0xab")
        self.assertEqual(dm.lookup(0x2000, 8), "value:0xab")  # width fallback
        dm.record(0x3000, 4, "r", value=None)
        self.assertEqual(dm.lookup(0x3000, 4), "zero")

    def test_write_and_opposite_direction(self):
        dm = DeviceMap()
        dm.record(0x2000, 4, "r", value=5)
        self.assertEqual(dm.lookup(0x2000, 4, write=True), "zero")  # absorb
        dm.record(0x3000, 4, "w", value=9)
        self.assertEqual(dm.lookup(0x3000, 4, write=True), "zero")
        self.assertEqual(dm.lookup(0x3000, 4), "zero")  # write-only reads back zero


class LenientMachineTests(unittest.TestCase):
    def test_strict_default_faults_without_logging(self):
        machine, result = _run_program(PROGRAM_LOAD_STORE)
        self.assertEqual(result[0], R_FAULT)
        self.assertEqual(result[1], BASE + 16)  # PC of the faulting load
        self.assertEqual(result[2], 5)  # load access fault
        self.assertEqual(result[3], TOUCH_A)
        self.assertEqual(machine.unknown_mmio, "fault")
        self.assertEqual(machine.mmio_log, [])
        self.assertEqual(machine.mmio_unknown, 0)

    def test_bad_policy_rejected(self):
        with self.assertRaises(ValueError):
            Machine(unknown_mmio="ignore")

    def test_lenient_returns_zero_and_logs(self):
        machine, result = _run_program(PROGRAM_TWO_TOUCHES, unknown_mmio="lenient")
        self.assertEqual(result[0], R_EBREAK)
        self.assertEqual(machine.x[10], 0)  # load returned 0
        self.assertEqual(machine.mmio_unknown, 2)
        self.assertEqual(
            machine.mmio_log,
            [
                {"pc": BASE + 16, "addr": TOUCH_A, "size": 4, "rw": "r"},
                {"pc": BASE + 32, "addr": TOUCH_B, "size": 8, "rw": "w"},
            ],
        )
        self.assertIsNone(machine.device_map)  # no learning requested

    def test_lenient_learns_observed_values(self):
        machine, result = _run_program(
            PROGRAM_LOAD_STORE, unknown_mmio="lenient", learn_mmio=True
        )
        self.assertEqual(result[0], R_EBREAK)
        self.assertEqual(machine.mmio_unknown, 1)  # the store is served once learned
        entry = machine.device_map.entries()[0]
        self.assertEqual((entry.address, entry.kind, entry.values), (TOUCH_A, "r", (0,)))
        self.assertEqual(entry.first_epc_fw, BASE + 16)  # bias 0: PC unchanged
        self.assertEqual(machine.device_map.lookup(TOUCH_A, 4), "value:0x0")

    def test_map_serves_mapped_values_even_in_strict_mode(self):
        dm = DeviceMap()
        dm.record(TOUCH_A, 4, "r", value=0x12345678)
        machine, result = _run_program(PROGRAM_LOAD_STORE, device_map=dm)
        self.assertEqual(result[0], R_EBREAK)
        self.assertEqual(machine.x[10], 0x12345678)
        self.assertEqual(machine.mmio_unknown, 0)

    def test_ram_and_uart_are_unaffected(self):
        # lui t0,0x10000; addi a0,zero,'A'; sb a0,0(t0); lbu a1,5(t0); ebreak
        program = bytes.fromhex("b7020010130510042380a20083c5520073001000")
        machine, result = _run_program(program, unknown_mmio="lenient")
        self.assertEqual(result[0], R_EBREAK)
        self.assertEqual(machine.uart.text(), "A")
        self.assertEqual(machine.x[11], 0x60)  # LSR: THRE | TEMT
        self.assertEqual(machine.mmio_unknown, 0)


class LaunchNativeTests(unittest.TestCase):
    @unittest.skipUnless(os.path.isfile(DEMO_ELF), "demo firmware absent")
    def test_launch_native_accepts_lenient_options(self):
        from greeneye.sim_backend import launch_native

        with tempfile.TemporaryDirectory() as root:
            map_path = os.path.join(root, "map.json")
            DeviceMap(meta={"firmware": "demo"}).save(map_path)
            session = launch_native(
                kernel=DEMO_ELF,
                device_map=map_path,
                unknown_mmio="lenient",
                name="unittest-device-map",
            )
            try:
                self.assertEqual(session.target.machine.unknown_mmio, "lenient")
                self.assertEqual(session.target.mmio_unknown, 0)
                self.assertEqual(session.target.mmio_log, [])
            finally:
                session.close()


@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "TU104 570.144 fwimage absent")
class RealImageLenientTests(unittest.TestCase):
    def test_first_touch_matches_qemu_probe(self):
        from greeneye.elf import load_elf

        image = load_elf(REAL_IMAGE)
        dm = DeviceMap(meta={"bias": BIAS})
        machine = Machine(
            device_map=dm, unknown_mmio="lenient", learn_mmio=True, bias=BIAS
        )
        machine.load_segments(image.flat_load_plan(bias=BIAS), image.relocated_entry(bias=BIAS))
        self.assertEqual(machine.pc, 0x84009000)
        reason, pc, _, _ = machine.run(200_000)
        self.assertGreaterEqual(machine.mmio_unknown, 1)
        self.assertEqual(
            machine.mmio_log[0],
            {"pc": 0x84009014, "addr": TOUCH_A, "size": 4, "rw": "r"},
        )
        # Lenient execution runs past the QEMU first fault to the budget.
        self.assertEqual(reason, R_BUDGET)
        self.assertEqual(machine.insns, 200_000)
        self.assertEqual(pc, 0x84009070)
        entry = dm.entries()[0]
        self.assertEqual(entry.first_epc_fw, 0x4009014)
        self.assertEqual(entry.values, (0,))
        self.assertEqual(dm.meta["bias"], BIAS)


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