{{ message }}
Repository navigation
Expand file tree
/
Copy pathtest_client_e2e.py
More file actions
352 lines (282 loc) · 12.1 KB
/
Copy pathtest_client_e2e.py
File metadata and controls
352 lines (282 loc) · 12.1 KB
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
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
"""E2E Client Tests"""
import pytest
from copilot import CopilotClient, RuntimeConnection
from copilot.client import (
ModelCapabilities,
ModelInfo,
ModelLimits,
ModelSupports,
StopError,
)
from copilot.session import PermissionHandler
from .testharness import CLI_PATH, DEFAULT_GITHUB_TOKEN, E2ETestContext
class TestClient:
@pytest.mark.asyncio
async def test_should_start_and_connect_to_server_using_stdio(self):
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.start()
pong = await client.ping("test message")
assert pong.message == "pong: test message"
assert pong.timestamp is not None
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_start_and_connect_to_server_using_tcp(self):
client = CopilotClient(connection=RuntimeConnection.for_tcp(path=CLI_PATH))
try:
await client.start()
pong = await client.ping("test message")
assert pong.message == "pong: test message"
assert pong.timestamp is not None
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_raise_exception_group_on_failed_cleanup(self):
import asyncio
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.create_session(on_permission_request=PermissionHandler.approve_all)
# Kill the server process to force cleanup to fail
process = client._process
assert process is not None
process.kill()
await asyncio.to_thread(process.wait, 10.0)
try:
await client.stop()
except ExceptionGroup as exc:
assert len(exc.exceptions) > 0
assert isinstance(exc.exceptions[0], StopError)
assert "Failed to disconnect session" in exc.exceptions[0].message
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_force_stop_without_cleanup(self):
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
await client.create_session(on_permission_request=PermissionHandler.approve_all)
await client.force_stop()
@pytest.mark.asyncio
async def test_should_get_status_with_version_and_protocol_info(self):
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.start()
status = await client.get_status()
assert hasattr(status, "version")
assert isinstance(status.version, str)
assert hasattr(status, "protocol_version")
assert isinstance(status.protocol_version, int)
assert status.protocol_version >= 1
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_get_auth_status(self):
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.start()
auth_status = await client.get_auth_status()
assert hasattr(auth_status, "isAuthenticated")
assert isinstance(auth_status.isAuthenticated, bool)
if auth_status.isAuthenticated:
assert hasattr(auth_status, "authType")
assert hasattr(auth_status, "statusMessage")
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_list_models_when_authenticated(self, ctx: E2ETestContext):
client = CopilotClient(
connection=RuntimeConnection.for_stdio(path=CLI_PATH),
env=ctx.get_env(),
working_directory=ctx.work_dir,
github_token=DEFAULT_GITHUB_TOKEN,
)
try:
await client.start()
auth_status = await client.get_auth_status()
assert auth_status.isAuthenticated
models = await client.list_models()
assert isinstance(models, list)
assert models
model = models[0]
assert hasattr(model, "id")
assert hasattr(model, "name")
assert hasattr(model, "capabilities")
assert hasattr(model.capabilities, "supports")
assert hasattr(model.capabilities, "limits")
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_cache_models_list(self, ctx: E2ETestContext):
"""Test that list_models caches results to avoid rate limiting"""
client = CopilotClient(
connection=RuntimeConnection.for_stdio(path=CLI_PATH),
env=ctx.get_env(),
working_directory=ctx.work_dir,
github_token=DEFAULT_GITHUB_TOKEN,
)
try:
await client.start()
auth_status = await client.get_auth_status()
assert auth_status.isAuthenticated
# First call should fetch from backend
models1 = await client.list_models()
assert isinstance(models1, list)
assert models1
# Second call should return from cache (different list object but same content)
models2 = await client.list_models()
assert models2 is not models1, "Should return a copy, not the same object"
assert len(models2) == len(models1), "Cached results should have same content"
assert models1[0].id == models2[0].id, "Cached models should match"
# After stopping, cache should be cleared
await client.stop()
# Restart and verify cache is empty
await client.start()
# Check authentication again after restart
auth_status = await client.get_auth_status()
assert auth_status.isAuthenticated
models3 = await client.list_models()
assert models3 is not models1, "Cache should be cleared after disconnect"
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_report_error_with_stderr_when_cli_fails_to_start(self):
"""Test that CLI startup errors include stderr output in the error message."""
client = CopilotClient(
connection=RuntimeConnection.for_stdio(
path=CLI_PATH, args=["--nonexistent-flag-for-testing"]
)
)
try:
with pytest.raises(RuntimeError) as exc_info:
await client.start()
error_message = str(exc_info.value)
# Verify we get the stderr output in the error message
assert "stderr" in error_message, (
f"Expected error to contain 'stderr', got: {error_message}"
)
assert "nonexistent" in error_message, (
f"Expected error to contain 'nonexistent', got: {error_message}"
)
# Verify subsequent calls also fail (don't hang)
with pytest.raises(Exception) as exc_info2:
session = await client.create_session(
on_permission_request=PermissionHandler.approve_all
)
await session.send("test")
# Retirement rejects new session admission; before it wins, process exit
# preserves stderr or the broken transport reports EINVAL/EPIPE.
error_msg = str(exc_info2.value).lower()
if error_msg == "client not connected":
assert isinstance(exc_info2.value, RuntimeError)
elif "cli process exited with code" in error_msg:
assert (
error_msg.partition("stderr:")[2]
== error_message.lower().partition("stderr:")[2]
)
else:
assert "invalid" in error_msg or "pipe" in error_msg or "closed" in error_msg
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_not_throw_when_disposing_session_after_stopping_client(self):
"""Disconnecting a session after the client is stopped must not raise."""
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.start()
session = await client.create_session(
on_permission_request=PermissionHandler.approve_all
)
# Stop the client first; subsequent session disconnect should be harmless.
await client.stop()
# Should not raise.
await session.disconnect()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_create_session_without_permission_handler(self):
"""`create_session` allows omitting an `on_permission_request` handler."""
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.start()
session = await client.create_session()
assert session.session_id
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_should_resume_session_without_permission_handler(self):
"""`resume_session` allows omitting an `on_permission_request` handler."""
client = CopilotClient(connection=RuntimeConnection.for_stdio(path=CLI_PATH))
try:
await client.start()
session = await client.create_session()
resumed = await client.resume_session(session.session_id)
assert resumed.session_id == session.session_id
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_list_models_with_custom_handler_calls_handler(self):
"""A custom `on_list_models` handler is invoked instead of the CLI RPC."""
custom_models = [
ModelInfo(
id="my-custom-model",
name="My Custom Model",
capabilities=ModelCapabilities(
supports=ModelSupports(vision=False, reasoning_effort=False),
limits=ModelLimits(max_context_window_tokens=128000),
),
)
]
call_count = 0
def on_list_models():
nonlocal call_count
call_count += 1
return custom_models
client = CopilotClient(
connection=RuntimeConnection.for_stdio(path=CLI_PATH),
on_list_models=on_list_models,
)
try:
await client.start()
models = await client.list_models()
assert call_count == 1
assert len(models) == 1
assert models[0].id == "my-custom-model"
await client.stop()
finally:
await client.force_stop()
@pytest.mark.asyncio
async def test_list_models_with_custom_handler_works_without_start(self):
"""The custom `on_list_models` handler is callable even before `start()`."""
custom_models = [
ModelInfo(
id="no-start-model",
name="No Start Model",
capabilities=ModelCapabilities(
supports=ModelSupports(vision=False, reasoning_effort=False),
limits=ModelLimits(max_context_window_tokens=128000),
),
)
]
call_count = 0
def on_list_models():
nonlocal call_count
call_count += 1
return custom_models
client = CopilotClient(
connection=RuntimeConnection.for_stdio(path=CLI_PATH),
on_list_models=on_list_models,
)
try:
models = await client.list_models()
assert call_count == 1
assert len(models) == 1
assert models[0].id == "no-start-model"
finally:
await client.force_stop()
You can’t perform that action at this time.