gabriel / muse public

test_remote_supercharge.py file-level

at sha256:2 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 πŸ’₯ blast risk
sha256:4 Merge branch 'dev' into main · gabriel · Jun 17, 2026
1 """Supercharge tests for muse remote β€” agent-first JSON envelope.
2
3 All seven subcommands (list, add, remove, rename, get-url, set-url, status)
4 must emit ``duration_ms`` and ``exit_code`` in every JSON response β€” success
5 and error alike. Agents poll these to measure latency and confirm outcomes
6 without parsing exit codes separately.
7
8 Coverage tiers
9 --------------
10 I Unit β€” TypedDict field presence
11 II Integration β€” every subcommand JSON success path carries the envelope
12 III Integration β€” every subcommand JSON error path carries the envelope
13 IV End-to-end β€” exit_code in JSON matches the process exit code
14 V Data integrity β€” duration_ms is a non-negative integer; exit_code is int
15 VI Security β€” envelope present even on validation-rejected inputs
16 VII Performance β€” local subcommands complete within 200 ms
17 """
18
19 from __future__ import annotations
20
21 import json
22 import time
23 import pathlib
24 import threading
25
26 import pytest
27
28 from tests.cli_test_helper import CliRunner
29 from muse.core.paths import muse_dir
30
31 runner = CliRunner()
32
33 # ---------------------------------------------------------------------------
34 # Fixtures
35 # ---------------------------------------------------------------------------
36
37 @pytest.fixture()
38 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
39 """Minimal Muse repo with .muse/config.toml wired up."""
40 dot_muse = muse_dir(tmp_path)
41 dot_muse.mkdir()
42 (dot_muse / "config.toml").write_text('[repo]\nname = "test"\n')
43 monkeypatch.chdir(tmp_path)
44 return tmp_path
45
46
47 @pytest.fixture()
48 def repo_with_origin(repo: pathlib.Path) -> pathlib.Path:
49 """Repo pre-loaded with a single remote named 'origin'."""
50 runner.invoke(None, ["remote", "add", "origin", "https://musehub.ai/gabriel/test"])
51 return repo
52
53
54 # ---------------------------------------------------------------------------
55 # I Unit β€” TypedDict field presence
56 # ---------------------------------------------------------------------------
57
58 class TestTypedDictFields:
59 def test_I1_remote_list_json_has_duration_ms(self) -> None:
60 """_RemoteListJson TypedDict must declare duration_ms."""
61 from muse.cli.commands.remote import _RemoteListJson
62 import typing
63 hints = typing.get_type_hints(_RemoteListJson)
64 assert "duration_ms" in hints, "_RemoteListJson missing duration_ms field"
65
66 def test_I2_remote_list_json_has_exit_code(self) -> None:
67 """_RemoteListJson TypedDict must declare exit_code."""
68 from muse.cli.commands.remote import _RemoteListJson
69 import typing
70 hints = typing.get_type_hints(_RemoteListJson)
71 assert "exit_code" in hints, "_RemoteListJson missing exit_code field"
72
73 def test_I3_mutation_json_has_duration_ms(self) -> None:
74 """_RemoteMutationJson TypedDict must declare duration_ms."""
75 from muse.cli.commands.remote import _RemoteMutationJson
76 import typing
77 hints = typing.get_type_hints(_RemoteMutationJson)
78 assert "duration_ms" in hints, "_RemoteMutationJson missing duration_ms field"
79
80 def test_I4_mutation_json_has_exit_code(self) -> None:
81 """_RemoteMutationJson TypedDict must declare exit_code."""
82 from muse.cli.commands.remote import _RemoteMutationJson
83 import typing
84 hints = typing.get_type_hints(_RemoteMutationJson)
85 assert "exit_code" in hints, "_RemoteMutationJson missing exit_code field"
86
87 def test_I5_get_url_json_has_duration_ms(self) -> None:
88 """_RemoteGetUrlJson TypedDict must declare duration_ms."""
89 from muse.cli.commands.remote import _RemoteGetUrlJson
90 import typing
91 hints = typing.get_type_hints(_RemoteGetUrlJson)
92 assert "duration_ms" in hints, "_RemoteGetUrlJson missing duration_ms field"
93
94 def test_I6_get_url_json_has_exit_code(self) -> None:
95 """_RemoteGetUrlJson TypedDict must declare exit_code."""
96 from muse.cli.commands.remote import _RemoteGetUrlJson
97 import typing
98 hints = typing.get_type_hints(_RemoteGetUrlJson)
99 assert "exit_code" in hints, "_RemoteGetUrlJson missing exit_code field"
100
101 def test_I7_status_json_has_duration_ms(self) -> None:
102 """_RemoteStatusJson TypedDict must declare duration_ms."""
103 from muse.cli.commands.remote import _RemoteStatusJson
104 import typing
105 hints = typing.get_type_hints(_RemoteStatusJson)
106 assert "duration_ms" in hints, "_RemoteStatusJson missing duration_ms field"
107
108 def test_I8_status_json_has_exit_code(self) -> None:
109 """_RemoteStatusJson TypedDict must declare exit_code."""
110 from muse.cli.commands.remote import _RemoteStatusJson
111 import typing
112 hints = typing.get_type_hints(_RemoteStatusJson)
113 assert "exit_code" in hints, "_RemoteStatusJson missing exit_code field"
114
115
116 # ---------------------------------------------------------------------------
117 # II Integration β€” success JSON carries envelope
118 # ---------------------------------------------------------------------------
119
120 class TestSuccessEnvelope:
121 def test_II1_list_json_has_envelope(self, repo: pathlib.Path) -> None:
122 """muse remote --json must include duration_ms and exit_code."""
123 r = runner.invoke(None, ["remote", "--json"])
124 data = json.loads(r.output)
125 assert "duration_ms" in data, "list JSON missing duration_ms"
126 assert "exit_code" in data, "list JSON missing exit_code"
127
128 def test_II2_add_json_has_envelope(self, repo: pathlib.Path) -> None:
129 """muse remote add --json must include duration_ms and exit_code."""
130 r = runner.invoke(None, ["remote", "add", "origin",
131 "https://musehub.ai/gabriel/test", "--json"])
132 data = json.loads(r.output)
133 assert "duration_ms" in data, "add JSON missing duration_ms"
134 assert "exit_code" in data, "add JSON missing exit_code"
135
136 def test_II3_remove_json_has_envelope(self, repo_with_origin: pathlib.Path) -> None:
137 """muse remote remove --json must include duration_ms and exit_code."""
138 r = runner.invoke(None, ["remote", "remove", "origin", "--json"])
139 data = json.loads(r.output)
140 assert "duration_ms" in data, "remove JSON missing duration_ms"
141 assert "exit_code" in data, "remove JSON missing exit_code"
142
143 def test_II4_rename_json_has_envelope(self, repo_with_origin: pathlib.Path) -> None:
144 """muse remote rename --json must include duration_ms and exit_code."""
145 r = runner.invoke(None, ["remote", "rename", "origin", "upstream", "--json"])
146 data = json.loads(r.output)
147 assert "duration_ms" in data, "rename JSON missing duration_ms"
148 assert "exit_code" in data, "rename JSON missing exit_code"
149
150 def test_II5_get_url_json_has_envelope(self, repo_with_origin: pathlib.Path) -> None:
151 """muse remote get-url --json must include duration_ms and exit_code."""
152 r = runner.invoke(None, ["remote", "get-url", "origin", "--json"])
153 data = json.loads(r.output)
154 assert "duration_ms" in data, "get-url JSON missing duration_ms"
155 assert "exit_code" in data, "get-url JSON missing exit_code"
156
157 def test_II6_set_url_json_has_envelope(self, repo_with_origin: pathlib.Path) -> None:
158 """muse remote set-url --json must include duration_ms and exit_code."""
159 r = runner.invoke(None, ["remote", "set-url", "origin",
160 "https://musehub.ai/gabriel/new", "--json"])
161 data = json.loads(r.output)
162 assert "duration_ms" in data, "set-url JSON missing duration_ms"
163 assert "exit_code" in data, "set-url JSON missing exit_code"
164
165
166 # ---------------------------------------------------------------------------
167 # III Integration β€” error JSON carries envelope
168 # ---------------------------------------------------------------------------
169
170 class TestErrorEnvelope:
171 def test_III1_add_duplicate_error_has_envelope(self, repo_with_origin: pathlib.Path) -> None:
172 """muse remote add duplicate --json error must include duration_ms and exit_code."""
173 r = runner.invoke(None, ["remote", "add", "origin",
174 "https://musehub.ai/x/y", "--json"])
175 data = json.loads(r.output)
176 assert "duration_ms" in data, "add-duplicate error JSON missing duration_ms"
177 assert "exit_code" in data, "add-duplicate error JSON missing exit_code"
178
179 def test_III2_add_invalid_name_error_has_envelope(self, repo: pathlib.Path) -> None:
180 """muse remote add invalid-name --json error must include envelope."""
181 r = runner.invoke(None, ["remote", "add", "bad name!",
182 "https://musehub.ai/x/y", "--json"])
183 data = json.loads(r.output)
184 assert "duration_ms" in data
185 assert "exit_code" in data
186
187 def test_III3_add_bad_scheme_error_has_envelope(self, repo: pathlib.Path) -> None:
188 """muse remote add ftp:// --json error must include envelope."""
189 r = runner.invoke(None, ["remote", "add", "origin",
190 "ftp://example.com/repo", "--json"])
191 data = json.loads(r.output)
192 assert "duration_ms" in data
193 assert "exit_code" in data
194
195 def test_III4_remove_not_found_error_has_envelope(self, repo: pathlib.Path) -> None:
196 """muse remote remove missing --json error must include envelope."""
197 r = runner.invoke(None, ["remote", "remove", "ghost", "--json"])
198 data = json.loads(r.output)
199 assert "duration_ms" in data
200 assert "exit_code" in data
201
202 def test_III5_rename_not_found_error_has_envelope(self, repo: pathlib.Path) -> None:
203 """muse remote rename missing --json error must include envelope."""
204 r = runner.invoke(None, ["remote", "rename", "ghost", "phantom", "--json"])
205 data = json.loads(r.output)
206 assert "duration_ms" in data
207 assert "exit_code" in data
208
209 def test_III6_get_url_not_found_error_has_envelope(self, repo: pathlib.Path) -> None:
210 """muse remote get-url missing --json error must include envelope."""
211 r = runner.invoke(None, ["remote", "get-url", "ghost", "--json"])
212 data = json.loads(r.output)
213 assert "duration_ms" in data
214 assert "exit_code" in data
215
216 def test_III7_set_url_not_found_error_has_envelope(self, repo: pathlib.Path) -> None:
217 """muse remote set-url missing --json error must include envelope."""
218 r = runner.invoke(None, ["remote", "set-url", "ghost",
219 "https://musehub.ai/x/y", "--json"])
220 data = json.loads(r.output)
221 assert "duration_ms" in data
222 assert "exit_code" in data
223
224
225 # ---------------------------------------------------------------------------
226 # IV End-to-end β€” exit_code in JSON matches process exit code
227 # ---------------------------------------------------------------------------
228
229 class TestExitCodeAccuracy:
230 def test_IV1_add_success_exit_code_is_0(self, repo: pathlib.Path) -> None:
231 """exit_code: 0 in JSON on successful add."""
232 r = runner.invoke(None, ["remote", "add", "origin",
233 "https://musehub.ai/gabriel/test", "--json"])
234 assert r.exit_code == 0
235 assert json.loads(r.output)["exit_code"] == 0
236
237 def test_IV2_add_error_exit_code_is_1(self, repo_with_origin: pathlib.Path) -> None:
238 """exit_code: 1 in JSON when add fails (duplicate)."""
239 r = runner.invoke(None, ["remote", "add", "origin",
240 "https://musehub.ai/x/y", "--json"])
241 assert r.exit_code == 1
242 assert json.loads(r.output)["exit_code"] == 1
243
244 def test_IV3_remove_success_exit_code_is_0(self, repo_with_origin: pathlib.Path) -> None:
245 """exit_code: 0 in JSON on successful remove."""
246 r = runner.invoke(None, ["remote", "remove", "origin", "--json"])
247 assert r.exit_code == 0
248 assert json.loads(r.output)["exit_code"] == 0
249
250 def test_IV4_remove_error_exit_code_is_1(self, repo: pathlib.Path) -> None:
251 """exit_code: 1 in JSON when remove fails (not found)."""
252 r = runner.invoke(None, ["remote", "remove", "ghost", "--json"])
253 assert r.exit_code == 1
254 assert json.loads(r.output)["exit_code"] == 1
255
256 def test_IV5_rename_success_exit_code_is_0(self, repo_with_origin: pathlib.Path) -> None:
257 """exit_code: 0 in JSON on successful rename."""
258 r = runner.invoke(None, ["remote", "rename", "origin", "upstream", "--json"])
259 assert r.exit_code == 0
260 assert json.loads(r.output)["exit_code"] == 0
261
262 def test_IV6_get_url_success_exit_code_is_0(self, repo_with_origin: pathlib.Path) -> None:
263 """exit_code: 0 in JSON on successful get-url."""
264 r = runner.invoke(None, ["remote", "get-url", "origin", "--json"])
265 assert r.exit_code == 0
266 assert json.loads(r.output)["exit_code"] == 0
267
268 def test_IV7_set_url_success_exit_code_is_0(self, repo_with_origin: pathlib.Path) -> None:
269 """exit_code: 0 in JSON on successful set-url."""
270 r = runner.invoke(None, ["remote", "set-url", "origin",
271 "https://musehub.ai/gabriel/new", "--json"])
272 assert r.exit_code == 0
273 assert json.loads(r.output)["exit_code"] == 0
274
275 def test_IV8_list_success_exit_code_is_0(self, repo: pathlib.Path) -> None:
276 """exit_code: 0 in JSON on successful list (even empty)."""
277 r = runner.invoke(None, ["remote", "--json"])
278 assert r.exit_code == 0
279 assert json.loads(r.output)["exit_code"] == 0
280
281
282 # ---------------------------------------------------------------------------
283 # V Data integrity β€” field types and values
284 # ---------------------------------------------------------------------------
285
286 class TestEnvelopeTypes:
287 def test_V1_duration_ms_is_non_negative_int_on_add(self, repo: pathlib.Path) -> None:
288 """duration_ms must be a non-negative integer."""
289 r = runner.invoke(None, ["remote", "add", "origin",
290 "https://musehub.ai/gabriel/test", "--json"])
291 data = json.loads(r.output)
292 assert isinstance(data["duration_ms"], (int, float)), "duration_ms must be numeric"
293 assert data["duration_ms"] >= 0, "duration_ms must be non-negative"
294
295 def test_V2_duration_ms_is_non_negative_int_on_list(self, repo: pathlib.Path) -> None:
296 """duration_ms on list is a non-negative numeric."""
297 r = runner.invoke(None, ["remote", "--json"])
298 data = json.loads(r.output)
299 assert isinstance(data["duration_ms"], (int, float))
300 assert data["duration_ms"] >= 0
301
302 def test_V3_duration_ms_is_non_negative_int_on_error(
303 self, repo: pathlib.Path
304 ) -> None:
305 """duration_ms on error path is a non-negative int."""
306 r = runner.invoke(None, ["remote", "remove", "ghost", "--json"])
307 data = json.loads(r.output)
308 assert isinstance(data["duration_ms"], (int, float))
309 assert data["duration_ms"] >= 0
310
311 def test_V4_exit_code_is_int(self, repo: pathlib.Path) -> None:
312 """exit_code must be a plain int in JSON."""
313 r = runner.invoke(None, ["remote", "--json"])
314 data = json.loads(r.output)
315 assert isinstance(data["exit_code"], int)
316
317 def test_V5_invalid_name_exit_code_matches(self, repo: pathlib.Path) -> None:
318 """exit_code in JSON matches actual exit code for invalid-name errors."""
319 r = runner.invoke(None, ["remote", "add", "bad/name",
320 "https://musehub.ai/x/y", "--json"])
321 data = json.loads(r.output)
322 assert data["exit_code"] == r.exit_code
323
324 def test_V6_all_success_fields_present_add(self, repo: pathlib.Path) -> None:
325 """add success JSON has all documented fields including envelope."""
326 r = runner.invoke(None, ["remote", "add", "origin",
327 "https://musehub.ai/gabriel/test", "--json"])
328 data = json.loads(r.output)
329 for field in ("status", "name", "url", "old_url", "old_name",
330 "new_name", "duration_ms", "exit_code"):
331 assert field in data, f"add JSON missing field: {field}"
332
333 def test_V7_all_success_fields_present_get_url(
334 self, repo_with_origin: pathlib.Path
335 ) -> None:
336 """get-url success JSON has all documented fields including envelope."""
337 r = runner.invoke(None, ["remote", "get-url", "origin", "--json"])
338 data = json.loads(r.output)
339 for field in ("name", "url", "duration_ms", "exit_code"):
340 assert field in data, f"get-url JSON missing field: {field}"
341
342 def test_V8_all_success_fields_present_list(self, repo: pathlib.Path) -> None:
343 """list success JSON has all documented fields including envelope."""
344 r = runner.invoke(None, ["remote", "--json"])
345 data = json.loads(r.output)
346 for field in ("remotes", "duration_ms", "exit_code"):
347 assert field in data, f"list JSON missing field: {field}"
348
349
350 # ---------------------------------------------------------------------------
351 # VI Security β€” envelope present even on adversarial inputs
352 # ---------------------------------------------------------------------------
353
354 class TestSecurityEnvelope:
355 def test_VI1_ansi_in_name_error_has_envelope(self, repo: pathlib.Path) -> None:
356 """ANSI-injected remote name error JSON has envelope."""
357 r = runner.invoke(None, ["remote", "add", "\x1b[31mmalicious",
358 "https://musehub.ai/x/y", "--json"])
359 data = json.loads(r.output)
360 assert "duration_ms" in data
361 assert "exit_code" in data
362 assert data["exit_code"] != 0
363
364 def test_VI2_file_scheme_error_has_envelope(self, repo: pathlib.Path) -> None:
365 """file:// URL scheme rejection carries envelope."""
366 r = runner.invoke(None, ["remote", "add", "malicious",
367 "file:///etc/passwd", "--json"])
368 data = json.loads(r.output)
369 assert "duration_ms" in data
370 assert "exit_code" in data
371 assert data["exit_code"] != 0
372
373 def test_VI3_oversized_name_error_has_envelope(self, repo: pathlib.Path) -> None:
374 """Oversized remote name error JSON has envelope."""
375 long_name = "a" * 101
376 r = runner.invoke(None, ["remote", "add", long_name,
377 "https://musehub.ai/x/y", "--json"])
378 data = json.loads(r.output)
379 assert "duration_ms" in data
380 assert "exit_code" in data
381
382 def test_VI4_oversized_url_error_has_envelope(self, repo: pathlib.Path) -> None:
383 """Oversized URL error JSON has envelope."""
384 long_url = f"https://musehub.ai/{'a' * 2048}"
385 r = runner.invoke(None, ["remote", "add", "origin", long_url, "--json"])
386 data = json.loads(r.output)
387 assert "duration_ms" in data
388 assert "exit_code" in data
389
390 def test_VI5_rename_new_name_invalid_error_has_envelope(
391 self, repo_with_origin: pathlib.Path
392 ) -> None:
393 """Invalid new name in rename error JSON has envelope."""
394 r = runner.invoke(None, ["remote", "rename", "origin",
395 "bad name!", "--json"])
396 data = json.loads(r.output)
397 assert "duration_ms" in data
398 assert "exit_code" in data
399
400
401 # ---------------------------------------------------------------------------
402 # VII Performance β€” local subcommands complete quickly
403 # ---------------------------------------------------------------------------
404
405 class TestPerformance:
406 _LIMIT_MS = 200
407
408 def test_VII1_add_completes_within_limit(self, repo: pathlib.Path) -> None:
409 """muse remote add --json completes within 200 ms."""
410 r = runner.invoke(None, ["remote", "add", "origin",
411 "https://musehub.ai/gabriel/test", "--json"])
412 data = json.loads(r.output)
413 assert data["duration_ms"] <= self._LIMIT_MS, (
414 f"add took {data['duration_ms']} ms β€” exceeds {self._LIMIT_MS} ms limit"
415 )
416
417 def test_VII2_list_completes_within_limit(self, repo: pathlib.Path) -> None:
418 """muse remote --json completes within 200 ms."""
419 r = runner.invoke(None, ["remote", "--json"])
420 data = json.loads(r.output)
421 assert data["duration_ms"] <= self._LIMIT_MS, (
422 f"list took {data['duration_ms']} ms β€” exceeds {self._LIMIT_MS} ms limit"
423 )
424
425 def test_VII3_remove_completes_within_limit(
426 self, repo_with_origin: pathlib.Path
427 ) -> None:
428 """muse remote remove --json completes within 200 ms."""
429 r = runner.invoke(None, ["remote", "remove", "origin", "--json"])
430 data = json.loads(r.output)
431 assert data["duration_ms"] <= self._LIMIT_MS
432
433 def test_VII4_get_url_completes_within_limit(
434 self, repo_with_origin: pathlib.Path
435 ) -> None:
436 """muse remote get-url --json completes within 200 ms."""
437 r = runner.invoke(None, ["remote", "get-url", "origin", "--json"])
438 data = json.loads(r.output)
439 assert data["duration_ms"] <= self._LIMIT_MS
440
441 def test_VII5_set_url_completes_within_limit(
442 self, repo_with_origin: pathlib.Path
443 ) -> None:
444 """muse remote set-url --json completes within 200 ms."""
445 r = runner.invoke(None, ["remote", "set-url", "origin",
446 "https://musehub.ai/gabriel/new", "--json"])
447 data = json.loads(r.output)
448 assert data["duration_ms"] <= self._LIMIT_MS
449
450 def test_VII6_rename_completes_within_limit(
451 self, repo_with_origin: pathlib.Path
452 ) -> None:
453 """muse remote rename --json completes within 200 ms."""
454 r = runner.invoke(None, ["remote", "rename", "origin", "upstream", "--json"])
455 data = json.loads(r.output)
456 assert data["duration_ms"] <= self._LIMIT_MS
457
458
459 class TestRegisterFlags:
460 def test_default_json_out_is_false(self) -> None:
461 import argparse
462 from muse.cli.commands.remote import register
463 p = argparse.ArgumentParser()
464 subs = p.add_subparsers()
465 register(subs)
466 args = p.parse_args(["remote"])
467 assert args.json_out is False
468
469 def test_json_flag_sets_json_out(self) -> None:
470 import argparse
471 from muse.cli.commands.remote import register
472 p = argparse.ArgumentParser()
473 subs = p.add_subparsers()
474 register(subs)
475 args = p.parse_args(["remote", "--json"])
476 assert args.json_out is True
477
478 def test_j_shorthand_sets_json_out(self) -> None:
479 import argparse
480 from muse.cli.commands.remote import register
481 p = argparse.ArgumentParser()
482 subs = p.add_subparsers()
483 register(subs)
484 args = p.parse_args(["remote", "-j"])
485 assert args.json_out is True