rename (cosmetic): green-eye -> risc-v-gsp-rm-debugger in user-facing names, docs, CLI, launcher and CI workflow; internal greeneye package/module identifiers unchanged (includes in-flight greeneye module edits present in the tree)

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