1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
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()
|
