summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/tests
diff options
context:
space:
mode:
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests')
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_bridge.py31
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_harness.py2
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_mcp.py4
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_session_call.py195
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py2
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py2
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_value_demos.py2
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_version_diff.py38
8 files changed, 270 insertions, 6 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_bridge.py b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
index 7eeb46c..966399a 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
@@ -5,6 +5,7 @@ import json
import os
import shutil
import socket
+import time
import unittest
import greeneye as ge
@@ -23,6 +24,36 @@ class GhidraCommandTests(unittest.TestCase):
self.assertIn("MarkGreenEyeHits.java /tmp/hits.tsv", cmd)
+class BridgeLifecycleTests(unittest.TestCase):
+ """No target needed: start/stop/close are transport-level."""
+
+ def test_refuses_non_loopback_bind(self):
+ bridge = GhidraBridge(object(), host="0.0.0.0", port=0)
+ with self.assertRaises(ValueError):
+ bridge.start()
+
+ def test_close_command_takes_down_the_listener(self):
+ bridge = GhidraBridge(object(), port=0)
+ port = bridge.start()
+ with socket.create_connection(("127.0.0.1", port), timeout=10) as sock:
+ stream = sock.makefile("rwb")
+ stream.write((json.dumps({"id": 1, "cmd": "close", "args": {}}) + "\n").encode())
+ stream.flush()
+ response = json.loads(stream.readline())
+ self.assertTrue(response["ok"])
+ self.assertTrue(response["close"])
+ deadline = time.monotonic() + 5.0
+ while True:
+ try:
+ with socket.create_connection(("127.0.0.1", port), timeout=0.5):
+ pass
+ except OSError:
+ break
+ if time.monotonic() > deadline:
+ self.fail("listener still accepting connections after <close>")
+ time.sleep(0.05)
+
+
def qemu_available() -> bool:
return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_harness.py b/debug/risc-v-gsp-rm-vm/tests/test_harness.py
index 55938ae..2e079cd 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_harness.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_harness.py
@@ -1,4 +1,4 @@
-"""Tests for the green-eye automation harness (roadmap R5/R6, H4.1/H4.3).
+"""Tests for the risc-v-gsp-rm-debugger automation harness (roadmap R5/R6, H4.1/H4.3).
The integration tests drive the two harness CLIs as subprocesses against the
bundled demo firmware and skip cleanly when qemu-system-riscv64 or the demo
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
index 614d397..2df8a4a 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_mcp.py
@@ -1,4 +1,4 @@
-"""Tests for the green-eye MCP server plugin (roadmap R3).
+"""Tests for the risc-v-gsp-rm-debugger MCP server plugin (roadmap R3).
The protocol and tool tests drive an in-memory Session over a stub RSP-shaped
client, so they need no QEMU and no simulator. A standalone subprocess test
@@ -278,7 +278,7 @@ class ReadOnlyToolTests(McpTestCase):
)
payload, is_error = self.call_tool(server, "analyze_current")
self.assertFalse(is_error)
- self.assertIn("# green-eye session summary", payload["report"])
+ self.assertIn("# risc-v-gsp-rm-debugger session summary", payload["report"])
finally:
session.close()
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()
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
index a0635ab..1ccfb04 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_sim_backend.py
@@ -118,7 +118,7 @@ class NativeSmokeTests(unittest.TestCase):
watch_hits = []
session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit))
session.run(duration=0.6)
- self.assertTrue(session.target.wait_for_uart("GREEN-EYE UPROC DEMO", timeout=1.0))
+ self.assertTrue(session.target.wait_for_uart("RISC-V GSP-RM DEBUGGER UPROC DEMO", timeout=1.0))
self.assertGreaterEqual(len(hits), 1, "rpc_sum hook never fired")
self.assertGreaterEqual(len(watch_hits), 7, "rpc_count watchpoint never fired")
self.assertEqual(watch_hits[0].addr, ctrl + 16)
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py b/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
index c0354d2..850942f 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_smoke_qemu.py
@@ -31,7 +31,7 @@ class QemuSmokeTests(unittest.TestCase):
session.hook("rpc_sum", on_enter=lambda hit: hits.append(hit))
session.watch(ctrl + 16, size=8, kind="w")
session.run(duration=0.6)
- self.assertTrue(session.target.wait_for_uart("GREEN-EYE UPROC DEMO", timeout=1.0))
+ self.assertTrue(session.target.wait_for_uart("RISC-V GSP-RM DEBUGGER UPROC DEMO", timeout=1.0))
self.assertGreaterEqual(len(hits), 1, "rpc_sum hook never fired")
after = session.snapshot(ctrl, 0x188, "ctrl")
self.assertEqual(int.from_bytes(after[0:4], "little"), 0x47535031)
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py b/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
index 7640e13..2a8e957 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_value_demos.py
@@ -73,7 +73,7 @@ class ValueDemoTests(unittest.TestCase):
result = _run_demo("s05_snapshot_diff.py", "native")
self.assertTrue(result["success"], result)
self.assertTrue({"result", "rpc_count", "last_rpc", "buf"}.issubset(set(result["changed_field_names"])))
- self.assertEqual(result["after"]["result"], 15)
+ self.assertEqual(result["after"]["result"], len(b"hello-risc-v-gsp-rm-debugger"))
def test_s09_ghidra_join(self):
result = _run_demo("s09_grabbing_a_ghidra_address.py", "qemu")
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py b/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py
index df704df..d20e780 100644
--- a/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py
+++ b/debug/risc-v-gsp-rm-vm/tests/test_version_diff.py
@@ -262,6 +262,44 @@ class RunVersionDiffTests(unittest.TestCase):
self.assertTrue(os.path.exists(os.path.join(out, "version-diff.json")))
+class VersionDiffDryRunTests(unittest.TestCase):
+ def test_dry_run_exits_zero_and_touches_nothing(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ old = os.path.join(tmp, "old.bin")
+ new = os.path.join(tmp, "new.bin")
+ for path in (old, new):
+ with open(path, "wb") as fh:
+ fh.write(NOP * 8)
+ before = {}
+ for name in ("old.bin", "new.bin"):
+ path = os.path.join(tmp, name)
+ with open(path, "rb") as fh:
+ before[name] = (os.path.getmtime(path), fh.read())
+ out = os.path.join(tmp, "out")
+ rc, stdout, _ = _run_main(
+ version_diff_main,
+ ["--old", old, "--new", new, "--out", out, "--dry-run", "--probe-instructions", "4"],
+ )
+ self.assertEqual(rc, 0)
+ plan = json.loads(stdout[: stdout.find("\n[version-diff]")])
+ self.assertEqual(plan["format"], "greeneye-version-diff-plan/1")
+ self.assertEqual(plan["status"], "planned")
+ self.assertFalse(os.path.exists(out), "dry-run must not create --out")
+ for name, (mtime, data) in before.items():
+ path = os.path.join(tmp, name)
+ self.assertEqual(os.path.getmtime(path), mtime)
+ with open(path, "rb") as fh:
+ self.assertEqual(fh.read(), data)
+
+ def test_dry_run_rm610_refused(self):
+ rc, _, err = _run_main(
+ version_diff_main,
+ ["--old", "rm610.elf", "--new", "new.elf", "--dry-run"],
+ )
+ self.assertEqual(rc, 3)
+ self.assertIn("docs/23", err)
+
+
class CiReleaseTests(unittest.TestCase):
def test_dry_run_exits_zero_and_touches_nothing(self):
with tempfile.TemporaryDirectory() as tmp:
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-12 21:28:47 +0000