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