"""Session.call and Session.close invariant tests with a fake RSP client. These lock in two contracts that hardware behavior alone did not cover: * ``Session.call`` restores every ABI register plus ``pc`` after both a normal return and a timeout. QEMU names x8 ``fp`` (not ``s0``), so both spellings must be handled. * ``Session.close`` clears every breakpoint and watchpoint it registered and closes the transport. """ from __future__ import annotations import os import shutil import unittest import greeneye as ge from greeneye.errors import RspTimeout from greeneye.rsp import StopReply from greeneye.session import RESTORE_REGS, Session HERE = os.path.dirname(os.path.abspath(__file__)) TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") #: QEMU's target.xml spelling: x8 is ``fp`` and ``s0`` is not present. QEMU_REG_NAMES = ( ["zero", "ra", "sp", "gp", "tp", "t0", "t1", "t2", "fp", "s1"] + [f"s{i}" for i in range(2, 12)] + [f"a{i}" for i in range(8)] + [f"t{i}" for i in range(3, 7)] + ["pc"] ) class FakeClient: """Minimal RSP client double with breakpoint/watchpoint accounting.""" def __init__(self, *, timeout_on_continue: bool = False, retval: int = 0x1234): self.regs = {name: 0x100 + index for index, name in enumerate(QEMU_REG_NAMES)} self.regs["pc"] = 0x80000000 self.writes: list[tuple[str, int]] = [] self.sw_breaks: set[int] = set() self.hw_breaks: set[int] = set() self.watches: set[tuple[int, int, str]] = set() self.closed = False self.timeout_on_continue = timeout_on_continue self.retval = retval self.continue_count = 0 def read_registers(self) -> dict[str, int]: return dict(self.regs) def read_reg(self, name: str) -> int: return self.regs["fp" if name == "s0" and "s0" not in self.regs else name] def write_reg(self, name: str, value: int) -> None: key = "fp" if name == "s0" and "s0" not in self.regs else name self.regs[key] = value self.writes.append((name, value)) @property def pc(self) -> int: return self.regs["pc"] def set_sw_break(self, addr: int, kind: int = 4) -> None: self.sw_breaks.add(addr) def clear_sw_break(self, addr: int, kind: int = 4) -> None: self.sw_breaks.discard(addr) def set_hw_break(self, addr: int, kind: int = 4) -> None: self.hw_breaks.add(addr) def clear_hw_break(self, addr: int, kind: int = 4) -> None: self.hw_breaks.discard(addr) def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: self.watches.add((addr, size, kind)) def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None: self.watches.discard((addr, size, kind)) def continue_(self, timeout: float | None = None) -> StopReply: self.continue_count += 1 if self.timeout_on_continue: raise RspTimeout("target never returned") self.regs["pc"] = self.regs["ra"] # returns to the scratch trampoline self.regs["a0"] = self.retval return StopReply(raw="T05") def interrupt(self, timeout: float | None = None) -> StopReply: return StopReply(raw="T02") def close(self) -> None: self.closed = True class CallRestoreTests(unittest.TestCase): def _restored(self, client: FakeClient, saved: dict[str, int]) -> list[str]: missing = [] for name, value in saved.items(): if name not in RESTORE_REGS: continue if client.regs.get(name) != value: missing.append(name) if saved.get("fp") != client.regs.get("fp"): missing.append("fp") return missing def test_call_success_restores_abi_regs_and_pc(self): client = FakeClient(retval=0xCAFE) session = Session(client=client, name="fake-success") saved = session.read_regs() result = session.call(0x80001000, (0x41, 0x42), timeout=1.0, scratch=0x801FF000) self.assertEqual(result, 0xCAFE) self.assertEqual(self._restored(client, saved), []) self.assertEqual(client.sw_breaks, set(), "temporary call breakpoint must be gone") session.close() def test_call_timeout_restores_abi_regs_and_pc(self): client = FakeClient(timeout_on_continue=True) session = Session(client=client, name="fake-timeout") saved = session.read_regs() with self.assertRaises(RspTimeout): session.call(0x80001000, (), timeout=0.05, scratch=0x801FF000) self.assertEqual(self._restored(client, saved), []) self.assertIn("fp", [name for name, _ in client.writes], "x8/fp must be restored under QEMU's name") self.assertEqual(client.sw_breaks, set(), "temporary call breakpoint must be gone") session.close() @unittest.skipUnless(shutil.which("qemu-system-riscv64"), "qemu-system-riscv64 not installed") @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built") def test_real_qemu_register_names_include_fp_not_s0(self): session = ge.launch(kernel=DEMO_ELF, name="unittest-call-names") try: regs = session.read_regs() self.assertIn("fp", regs) self.assertNotIn("s0", regs) self.assertLessEqual(set(RESTORE_REGS) - {"s0"}, set(regs)) finally: session.close() @unittest.skipUnless(shutil.which("qemu-system-riscv64"), "qemu-system-riscv64 not installed") @unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built") def test_real_qemu_loop_call_timeout_restores_registers(self): session = ge.launch(kernel=DEMO_ELF, name="unittest-call-loop") loop_addr = 0x80200000 # free RAM past the firmware segments try: session.client.write_mem(loop_addr, b"\x6f\x00\x00\x00") # jal x0, 0 saved = session.read_regs() with self.assertRaises(RspTimeout): session.call(loop_addr, (), timeout=0.3, scratch=0x801FF000) after = session.read_regs() for name in RESTORE_REGS: if name in saved and name != "s0": self.assertEqual(after.get(name), saved[name], f"{name} not restored after timeout") self.assertEqual(after.get("fp"), saved.get("fp"), "fp (x8/s0) not restored after timeout") finally: session.close() class CloseCleanupTests(unittest.TestCase): def test_close_clears_breakpoints_watchpoints_and_transport(self): client = FakeClient() session = Session(client=client, name="fake-close") session.hook(0x80001000) session.hook(0x80001000) # two on one address: only drop after the last session.breakpoint(0x80002000, kind="hw") session.watch(0x80003000, size=8, kind="w") session.watch(0x80003040, size=4, kind="r") self.assertTrue(client.sw_breaks) self.assertTrue(client.hw_breaks) self.assertEqual(len(client.watches), 2) session.close() self.assertEqual(client.sw_breaks, set()) self.assertEqual(client.hw_breaks, set()) self.assertEqual(client.watches, set()) self.assertTrue(client.closed) def test_close_removes_only_last_breakpoint_at_address(self): client = FakeClient() session = Session(client=client, name="fake-last-bp") first = session.hook(0x80001000) session.hook(0x80001000) first.detach() self.assertEqual(client.sw_breaks, {0x80001000}) session.close() self.assertEqual(client.sw_breaks, set()) if __name__ == "__main__": unittest.main()