diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/tests/test_rsp.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/tests/test_rsp.py | 175 |
1 files changed, 175 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/tests/test_rsp.py b/debug/risc-v-gsp-rm-vm/tests/test_rsp.py new file mode 100644 index 0000000..70b9f45 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_rsp.py @@ -0,0 +1,175 @@ +"""Unit tests for the RSP client using an in-process fake stub.""" +from __future__ import annotations + +import socket +import threading +import unittest + +from greeneye.rsp import ( + RV64_FALLBACK_REGS, + RspClient, + checksum, + encode_packet, + parse_stop_reply, + _parse_target_xml, +) + +TARGET_XML = """<?xml version="1.0"?> +<target><architecture>riscv:rv64</architecture> +<feature name="org.gnu.gdb.riscv.cpu"> +<reg name="x0" bitsize="64" regnum="0"/> +<reg name="ra" bitsize="64" regnum="1"/> +<reg name="pc" bitsize="64" regnum="32"/> +</feature></target>""" + + +class FakeStub(threading.Thread): + """Minimal RSP server for unit tests.""" + + def __init__(self, sock: socket.socket): + super().__init__(daemon=True) + self.sock = sock + self.no_ack = False + self.stopped = True + self.regs = bytearray(33 * 8) + self.regs[8:16] = (0xAA).to_bytes(8, "little") # ra (x1) + self.regs[16:24] = (0x80000194).to_bytes(8, "little") # pc (3-reg test layout) + self.memory = bytearray(0x10000) + for i in range(len(self.memory)): + self.memory[i] = i & 0xFF + self.bps: set[int] = set() + + def run(self): + buf = b"" + while True: + try: + chunk = self.sock.recv(4096) + except OSError: + return + if not chunk: + return + buf += chunk + while b"#" in buf and b"$" in buf: + start = buf.index(b"$") + end = buf.index(b"#", start) + if len(buf) < end + 3: + break + payload = buf[start + 1 : end] + buf = buf[end + 3 :] + self.handle(payload) + + def send(self, text: str): + data = encode_packet(text.encode()) + self.sock.sendall(data) + + def handle(self, payload: bytes): + cmd = payload.decode("latin1") + if not self.no_ack: + self.sock.sendall(b"+") + if cmd.startswith("qSupported"): + self.send("PacketSize=1000;qXfer:features:read+;multiprocess+") + elif cmd == "QStartNoAckMode": + self.send("") + self.no_ack = True + elif cmd == "?": + self.send("T05thread:p01.01;") + elif cmd in ("Hg0", "Hc-1"): + self.send("OK") + elif cmd.startswith("qXfer:features:read:target.xml"): + self.send("l" + TARGET_XML) + elif cmd.startswith("qXfer"): + self.send("E01") + elif cmd == "g": + self.send(bytes(self.regs).hex()) + elif cmd.startswith("p"): + regno = int(cmd[1:], 16) + if regno == 32: + self.send(self.regs[16:24].hex()) + else: + self.send(self.regs[regno * 8 : regno * 8 + 8].hex()) + elif cmd.startswith("m"): + addr_s, len_s = cmd[1:].split(",") + addr, length = int(addr_s, 16), int(len_s, 16) + self.send(bytes(self.memory[addr : addr + length]).hex()) + elif cmd.startswith("M"): + head, data = cmd[1:].split(":", 1) + addr_s, len_s = head.split(",") + addr, length = int(addr_s, 16), int(len_s, 16) + blob = bytes.fromhex(data) + self.memory[addr : addr + length] = blob + self.send("OK") + elif cmd.startswith("Z0"): + self.bps.add(int(cmd.split(",")[1], 16)) + self.send("OK") + elif cmd.startswith("z0"): + self.bps.discard(int(cmd.split(",")[1], 16)) + self.send("OK") + elif cmd.startswith("Z2") or cmd.startswith("z2"): + self.send("OK") + elif cmd in ("c", "s"): + self.stopped = True + self.send("T05thread:p01.01;") + else: + self.send("") + + +class RspTests(unittest.TestCase): + def setUp(self): + self.client_sock, self.server_sock = socket.socketpair() + self.stub = FakeStub(self.server_sock) + self.stub.start() + self.client = RspClient(timeout=2.0) + self.client.sock = self.client_sock + self.client.sync() + + def tearDown(self): + self.client.close() + self.server_sock.close() + + def test_checksum_and_encode(self): + self.assertEqual(checksum(b"OK"), (ord("O") + ord("K")) & 0xFF) + self.assertEqual(encode_packet(b"OK"), b"$OK#9a") + + def test_parse_stop_reply(self): + stop = parse_stop_reply("T05watch:0000000080001000;thread:p01.01;") + self.assertEqual(stop.signal, 5) + self.assertEqual(stop.watch_addr, 0x80001000) + self.assertEqual(stop.watch_kind, "watch") + self.assertEqual(stop.thread, "p01.01") + self.assertTrue(stop.is_stop) + exit_stop = parse_stop_reply("W00") + self.assertTrue(exit_stop.is_exit) + + def test_target_xml_parse(self): + regs = _parse_target_xml(TARGET_XML) + self.assertEqual([r.name for r in regs], ["x0", "ra", "pc"]) + self.assertEqual([r.number for r in regs], [0, 1, 32]) + by_name = {r.name: r for r in regs} + self.assertEqual(by_name["ra"].offset, 8) + self.assertEqual(by_name["pc"].offset, 16) + self.assertEqual(len(RV64_FALLBACK_REGS), 33) + self.assertEqual(RV64_FALLBACK_REGS[32].name, "pc") + + def test_registers_and_memory(self): + regs = self.client.read_registers() + self.assertEqual(regs["ra"], 0xAA) + self.assertEqual(regs["pc"], 0x80000194) + self.assertEqual(self.client.read_reg("pc"), 0x80000194) + data = self.client.read_mem(0x10, 8) + self.assertEqual(data, bytes(range(0x10, 0x18))) + self.client.write_mem(0x20, b"\xde\xad\xbe\xef") + self.assertEqual(self.client.read_mem(0x20, 4), b"\xde\xad\xbe\xef") + self.assertEqual(self.client.read_u32(0x20), 0xEFBEADDE) + + def test_breakpoints_watchpoints_continue(self): + self.client.set_sw_break(0x80000048) + self.client.clear_sw_break(0x80000048) + self.client.set_watch(0x80000850, size=8, kind="w") + stop = self.client.continue_() + self.assertEqual(stop.signal, 5) + stop = self.client.step() + self.assertEqual(stop.signal, 5) + + +if __name__ == "__main__": + unittest.main() |