Mercurial

comparison mrjunejune/inference/mock_sidecar_test.py @ 261:b401627fc49e

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add JRPG mock flows and interactive previews Add scripted mock SSE commands, custom event forwarding, animated chat turns, full-height message navigation, and a cyberpunk resume dossier. Co-authored-by: Copilot <[email protected]>
author MrJuneJune <mrjunejune@users.noreply.github.com>
date Wed, 05 Aug 2026 20:38:32 -0700
parents
children 056790c4fb0d
comparison
equal deleted inserted replaced
260:1f9877b637e9 261:b401627fc49e
1 import asyncio
2 import json
3 import pathlib
4 import tempfile
5 import unittest
6
7 from mrjunejune.inference.mock_sidecar import (
8 MockConfig,
9 MockSidecar,
10 load_mock_config,
11 )
12
13
14 class MockSidecarTest(unittest.IsolatedAsyncioTestCase):
15 async def asyncSetUp(self):
16 self.events = []
17
18 async def emit(payload):
19 self.events.append(payload)
20
21 packaged = load_mock_config(
22 pathlib.Path(__file__).with_name("mock_responses.json")
23 )
24 self.config = MockConfig(
25 delay_ms=0,
26 fallback=packaged.fallback,
27 commands=packaged.commands,
28 )
29 self.sidecar = MockSidecar(emit, self.config)
30
31 def test_packaged_commands_are_selectable(self):
32 self.assertEqual(self.config.select("!hello").name, "!hello")
33 self.assertEqual(self.config.select("!response please").name, "!response")
34 self.assertEqual(self.config.select("ordinary prompt").name, "!response")
35 self.assertEqual(self.config.select("!tool").name, "!tool")
36 self.assertEqual(self.config.select("!plan").name, "!plan")
37
38 def test_response_file_rejects_invalid_scripts(self):
39 invalid_configs = (
40 {
41 "fallback": "!hello",
42 "commands": {
43 "!hello": {
44 "events": [
45 {
46 "type": "assistant.completed",
47 "content": "Hello",
48 }
49 ],
50 "unknown": True,
51 }
52 },
53 },
54 {
55 "fallback": "hello",
56 "commands": {
57 "hello": {
58 "events": [
59 {
60 "type": "assistant.completed",
61 "content": "Hello",
62 }
63 ]
64 }
65 },
66 },
67 {
68 "fallback": "!hello",
69 "commands": {
70 "!hello": {
71 "events": [
72 {
73 "type": "assistant.delta",
74 "delta": "Different",
75 },
76 {
77 "type": "assistant.completed",
78 "content": "Hello",
79 },
80 ]
81 }
82 },
83 },
84 {
85 "delay_ms": 60001,
86 "fallback": "!hello",
87 "commands": {
88 "!hello": {
89 "events": [
90 {
91 "type": "assistant.completed",
92 "content": "Hello",
93 }
94 ]
95 }
96 },
97 },
98 {
99 "fallback": "!hello",
100 "commands": {
101 "!hello": {
102 "events": [
103 {
104 "type": "assistant.completed",
105 "content": "Hello",
106 },
107 {
108 "type": "assistant.delta",
109 "delta": "Hello",
110 },
111 ]
112 }
113 },
114 },
115 )
116 for payload in invalid_configs:
117 with self.subTest(payload=payload):
118 with tempfile.TemporaryDirectory() as root:
119 path = pathlib.Path(root) / "responses.json"
120 path.write_text(json.dumps(payload), encoding="utf-8")
121 with self.assertRaises(ValueError):
122 load_mock_config(path)
123
124 async def test_hello_command_streams_scripted_events(self):
125 await self.sidecar.dispatch(
126 {
127 "command": "turn.start",
128 "request_id": "request-1",
129 "conversation_id": "conversation-1",
130 "prompt": "!hello",
131 }
132 )
133 await self.sidecar.wait_for_idle()
134
135 self.assertEqual(
136 [event["type"] for event in self.events],
137 [
138 "turn.accepted",
139 "assistant.delta",
140 "assistant.delta",
141 "assistant.completed",
142 "turn.done",
143 ],
144 )
145 deltas = "".join(
146 event["delta"]
147 for event in self.events
148 if event["type"] == "assistant.delta"
149 )
150 completed = next(
151 event["content"]
152 for event in self.events
153 if event["type"] == "assistant.completed"
154 )
155 self.assertEqual(deltas, completed)
156 self.assertEqual(self.events[-1]["mock_command"], "!hello")
157
158 async def test_tool_command_preserves_custom_event_payloads(self):
159 await self.sidecar.dispatch(
160 {
161 "command": "turn.start",
162 "request_id": "request-2",
163 "conversation_id": "conversation-2",
164 "prompt": "!tool",
165 }
166 )
167 await self.sidecar.wait_for_idle()
168
169 started = next(
170 event for event in self.events if event["type"] == "tool.started"
171 )
172 completed = next(
173 event for event in self.events if event["type"] == "tool.completed"
174 )
175 self.assertEqual(started["tool"]["name"], "site_search")
176 self.assertEqual(completed["tool"]["result"]["matches"], 3)
177
178 async def test_error_command_emits_failed_turn(self):
179 await self.sidecar.dispatch(
180 {
181 "command": "turn.start",
182 "request_id": "request-3",
183 "conversation_id": "conversation-3",
184 "prompt": "!error",
185 }
186 )
187 await self.sidecar.wait_for_idle()
188
189 self.assertEqual(
190 [event["type"] for event in self.events],
191 ["turn.accepted", "turn.error", "turn.done"],
192 )
193 self.assertTrue(self.events[-1]["failed"])
194
195 async def test_abort_stops_active_command(self):
196 self.sidecar._config = MockConfig(
197 delay_ms=50,
198 fallback=self.config.fallback,
199 commands=self.config.commands,
200 )
201 await self.sidecar.dispatch(
202 {
203 "command": "turn.start",
204 "request_id": "request-4",
205 "conversation_id": "conversation-4",
206 "prompt": "!response",
207 }
208 )
209 await asyncio.sleep(0.01)
210 await self.sidecar.dispatch(
211 {
212 "command": "turn.abort",
213 "request_id": "abort-4",
214 "conversation_id": "conversation-4",
215 }
216 )
217
218 target_done = next(
219 event
220 for event in self.events
221 if event["type"] == "turn.done"
222 and event["request_id"] == "request-4"
223 )
224 self.assertTrue(target_done["aborted"])
225
226
227 if __name__ == "__main__":
228 unittest.main()