1 files changed, 81 insertions, 0 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
new file mode 100644
index 0000000..472bc49
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/tests/test_bridge.py
@@ -0,0 +1,81 @@
+"""Integration tests for the Ghidra JSON-lines control bridge."""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import socket
+import unittest
+
+import greeneye as ge
+from greeneye.gdbridge import GhidraBridge
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+DEMO_ELF = os.path.join(HERE, "..", "firmware", "uproc_demo", "build", "uproc_demo.elf")
+
+
+def qemu_available() -> bool:
+    return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
+
+
+@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed")
+@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built")
+class BridgeTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.session = ge.launch(kernel=DEMO_ELF, record=True, name="unittest-bridge")
+        cls.bridge = GhidraBridge(cls.session, port=0)
+        cls.port = cls.bridge.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.bridge.stop()
+        cls.session.close()
+
+    def call(self, cmd, args=None):
+        with socket.create_connection(("127.0.0.1", self.port), timeout=10) as sock:
+            stream = sock.makefile("rwb")
+            stream.write((json.dumps({"id": 1, "cmd": cmd, "args": args or {}}) + "\n").encode())
+            stream.flush()
+            response = json.loads(stream.readline())
+        self.assertTrue(response["ok"], response)
+        return response["result"]
+
+    def test_info_and_resolve(self):
+        info = self.call("info")
+        self.assertEqual(info["name"], "unittest-bridge")
+        addr = self.call("resolve", {"name": "rpc_sum"})
+        self.assertEqual(addr, self.session.symbols.lookup("rpc_sum"))
+
+    def test_run_until_and_hits(self):
+        result = self.call("run_until", {"target": "rpc_sum", "timeout": 5.0})
+        self.assertEqual(result["addr"], self.session.symbols.lookup("rpc_sum"))
+        hits = self.call("hits")
+        self.assertTrue(any(entry["addr"] == result["addr"] for entry in hits), hits)
+
+    def test_memory_and_registers(self):
+        ctrl = self.session.symbols.lookup("gsp_ctrl")
+        hexdata = self.call("read_mem", {"addr": ctrl, "size": 8})
+        self.assertEqual(len(bytes.fromhex(hexdata)), 8)
+        self.call("write_reg", {"name": "t3", "value": 0x1234})
+        self.assertEqual(self.call("read_reg", {"name": "t3"}), 0x1234)
+
+    def test_breakpoints_and_call(self):
+        addr = self.call("bp_set", {"target": "rpc_fill"})
+        self.assertEqual(addr, self.session.symbols.lookup("rpc_fill"))
+        self.call("bp_clear", {"addr": addr})
+        result = self.call("call", {"target": "rpc_ping", "args": [0x1111]})
+        self.assertEqual(result, 0x1111 ^ 0xBEEF)
+
+    def test_unknown_command(self):
+        with socket.create_connection(("127.0.0.1", self.port), timeout=10) as sock:
+            stream = sock.makefile("rwb")
+            stream.write((json.dumps({"id": 2, "cmd": "nope", "args": {}}) + "\n").encode())
+            stream.flush()
+            response = json.loads(stream.readline())
+        self.assertFalse(response["ok"])
+        self.assertIn("unknown bridge command", response["error"])
+
+
+if __name__ == "__main__":
+    unittest.main()