diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_session_call.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_session_call.py | 195 |
1 files changed, 195 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_session_call.py b/debug/risc-v-gsp-rm-vm/tests/test_session_call.py new file mode 100644 index 0000000..aab173c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_session_call.py @@ -0,0 +1,195 @@ +"""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() |