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
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
|
"""Native (M1) backend tests: demo parity with QEMU and engine determinism.
The native simulator is pure Python and always available; every test still
skips cleanly when the demo ELF cannot be loaded. The differential class
additionally requires qemu-system-riscv64 and compares the scripted
hook/watch/leave profile of both backends event for event.
"""
from __future__ import annotations
import os
import shutil
import tempfile
import unittest
import greeneye as ge
from greeneye.errors import RspTimeout
HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.join(HERE, "..")
DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
RPC_HANDLERS = ["rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state"]
def native_available() -> bool:
try:
import greeneye.sim_backend # noqa: F401
except Exception:
return False
return True
def qemu_available() -> bool:
return bool(os.environ.get("GREENEYE_QEMU") or shutil.which("qemu-system-riscv64"))
def demo_available() -> bool:
return os.path.exists(DEMO_ELF) and native_available()
def launch(backend: str, record=None, name: str = "sim"):
return ge.launch(kernel=DEMO_ELF, backend=backend, record=record, name=name)
@unittest.skipUnless(native_available(), "native simulator unavailable")
class NativeIsaTests(unittest.TestCase):
"""Exercise RV64M/A/C paths the demo firmware does not reach."""
# Assembled with llvm-mc -triple=riscv64 -mattr=+m,+a,+c (encodings are
# data, so this runs without a toolchain):
# li a0,6; li a1,7; mul a2,a0,a1; div a5,a1,a0; rem a6,a1,a0;
# srai a4,a1,1; subw a3,a0,a1; lui s0,0x12345; addi sp,sp,-16;
# sd a0,8(sp); ld s1,8(sp); lh s2,8(sp); lbu s3,8(sp);
# amoadd.d s4,a0,(sp); lr.d s5,(sp); sc.d s6,a0,(sp);
# li t0,1; c.addi t0,3; c.mv a2,t0; c.andi a2,15; ebreak
PROGRAM = bytes.fromhex(
"".join(
"""
19 45 9d 45 33 06 b5 02 b3 c7 a5 02 33 e8 a5 02 13 d7 15 40
bb 06 b5 40 37 54 34 12 41 11 2a e4 a2 64 03 19 81 00 83 49 81 00
2f 3a a1 00 af 3a 01 10 2f 3b a1 18 85 42 8d 02 16 86 3d 8a 02 90
""".split()
)
)
BASE = 0x80000000
STACK = 0x80004000
def test_m_a_c_execution(self):
from sim.machine import Machine
machine = Machine()
machine.load_segments([(self.BASE, self.PROGRAM)], self.BASE)
machine.set_reg(2, self.STACK) # sp
reason, pc, _, _ = machine.run(1000)
self.assertEqual(reason, 3) # R_EBREAK: stops before executing it
self.assertEqual(pc, self.BASE + len(self.PROGRAM) - 2)
expected = {
5: 4, # t0: c.addi
8: 0x12345000, # s0: lui
9: 6, # s1: c.sdsp/c.ldsp round-trip
10: 6, # a0
11: 7, # a1
12: 4, # a2: mul then c.mv/c.andi
13: 0xFFFFFFFFFFFFFFFF, # a3: subw sign extension
14: 3, # a4: srai
15: 1, # a5: div
16: 1, # a6: rem
18: 6, # s2: lh
19: 6, # s3: lbu
20: 0, # s4: amoadd old value
21: 6, # s5: lr
22: 0, # s6: sc success
2: self.STACK - 16, # c.addi16sp
}
got = {index: machine.x[index] for index in expected}
self.assertEqual(got, expected)
self.assertEqual(machine.read_mem(self.STACK - 16, 8), (6).to_bytes(8, "little"))
@unittest.skipUnless(demo_available(), "demo firmware or native simulator unavailable")
class NativeSmokeTests(unittest.TestCase):
def test_start_sequence_sets_stack_and_enters_main(self):
session = launch("native", name="unittest-sim-start")
try:
session.step(4) # auipc sp / addi sp / auipc ra / jalr main
stack_top = next(s.value for s in session.image.symbols if s.name == "_stack_top")
self.assertEqual(session.pc, session.symbols.lookup("main"))
self.assertEqual(session.read_regs()["sp"], stack_top)
finally:
session.close()
def test_boot_hook_watch_snapshot_analyze(self):
root = tempfile.mkdtemp(prefix="greeneye-sim-")
session = launch("native", record=root, name="unittest-sim-smoke")
hits = []
try:
ctrl = session.symbols.lookup("gsp_ctrl")
session.hook("rpc_sum", on_enter=lambda hit: hits.append(hit))
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.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)
self.assertEqual(watch_hits[0].watch.kind, "w")
after = session.snapshot(ctrl, 0x188, "ctrl")
self.assertEqual(int.from_bytes(after[0:4], "little"), 0x47535031)
self.assertGreaterEqual(int.from_bytes(after[16:24], "little"), 7)
finally:
directory = getattr(session.recorder, "dir", None)
session.close()
from greeneye.analyze import render_summary
report = render_summary(directory)
self.assertIn("rpc_sum", report)
self.assertIn("watch", report)
def test_hook_args_and_leave_value(self):
session = launch("native", name="unittest-sim-hookargs")
events = []
try:
ctrl = session.symbols.lookup("gsp_ctrl")
session.hook(
"rpc_sum",
on_enter=lambda hit: events.append(("enter", hit.arg(0), hit.arg(1))),
on_leave=lambda hit: events.append(("leave", hit.retval)),
)
session.run(duration=0.2)
# Self-test: rpc_dispatch(4, ctl->buf, 16, 0) after fill(0x5a, 32).
self.assertIn(("enter", ctrl + 0x48, 16), events)
self.assertIn(("leave", 16 * 0x5A), events)
finally:
session.close()
def test_call_and_run_until(self):
session = launch("native", name="unittest-sim-call")
try:
before = session.pc
self.assertEqual(session.call("rpc_ping", (0x1111,)), 0x1111 ^ 0xBEEF)
self.assertEqual(session.pc, before)
hit = session.run_until("rpc_sum", timeout=2.0)
self.assertEqual(hit.kind, "breakpoint")
self.assertEqual(hit.addr, session.symbols.lookup("rpc_sum"))
finally:
session.close()
def test_run_until_timeout(self):
session = launch("native", name="unittest-sim-timeout")
try:
with self.assertRaises(RspTimeout):
session.run_until("rpc_fault", timeout=0.05) # self-test never calls it
finally:
session.close()
@unittest.skipUnless(demo_available(), "demo firmware or native simulator unavailable")
class NativeDeterminismTests(unittest.TestCase):
def _scripted_profile(self):
"""Drive fixed instruction budgets: no host clock in the result."""
session = ge.launch(kernel=DEMO_ELF, backend="native", name="unittest-sim-determinism")
events = []
stops = []
try:
session.hook(
"rpc_sum",
on_enter=lambda hit: events.append(("enter", hit.arg(0), hit.arg(1))),
on_leave=lambda hit: events.append(("leave", hit.retval)),
)
ctrl = session.symbols.lookup("gsp_ctrl")
session.watch(ctrl + 16, size=8, kind="w")
for _ in range(20):
try:
stop = session.client.continue_(timeout=0.02) # exactly 20k insns
except RspTimeout:
stops.append(("budget", session.pc))
continue
session._dispatch_stop(stop)
stops.append(("stop", session.pc))
ctrl_image = session.read_regs()
snapshot = session.client.read_mem(ctrl, 0x188)
finally:
session.close()
return events, stops, ctrl_image["pc"], snapshot
def test_two_runs_identical(self):
first = self._scripted_profile()
second = self._scripted_profile()
self.assertEqual(first, second)
@unittest.skipUnless(demo_available() and qemu_available(), "native simulator and QEMU required")
class DifferentialTests(unittest.TestCase):
"""Run one scripted scenario on both backends and compare the profiles."""
def _profile(self, backend: str):
session = launch(backend, name=f"unittest-diff-{backend}")
events = []
watch_hits = []
try:
for name in RPC_HANDLERS:
session.hook(
name,
on_enter=lambda hit, n=name: events.append(("enter", n, hit.arg(0), hit.arg(1))),
on_leave=lambda hit, n=name: events.append(("leave", n, hit.retval)),
)
ctrl = session.symbols.lookup("gsp_ctrl")
session.watch(ctrl + 16, size=8, kind="w", callback=lambda hit: watch_hits.append(hit.addr))
session.run(duration=0.3)
uart = session.target.read_uart()
bp_counts = [handle.hits for handle in session._bps.values()]
finally:
session.close()
return events, len(watch_hits), bp_counts, uart.splitlines()[0] if uart else ""
def test_hook_watch_profile_parity(self):
qemu = self._profile("qemu")
native = self._profile("native")
self.assertEqual(qemu, native)
self.assertEqual(native[1], 8) # 7 self-test dispatches + ctrl init
self.assertEqual(native[2], [1, 1, 1, 1, 3])
if __name__ == "__main__":
unittest.main()
|
