1 files changed, 325 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_device_map.py b/debug/risc-v-gsp-rm-vm/tests/test_device_map.py
new file mode 100644
index 0000000..a38085a
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_device_map.py
@@ -0,0 +1,325 @@
+"""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()