gabriel / muse public
test_agent_supercharge.py python
1,061 lines 43.0 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago
1 """Supercharge tests for ``muse agent``, ``muse agent-config``, and ``muse agent-map``.
2
3 Seven-Tier Coverage Matrix
4 --------------------------
5
6 Tier 1 — TestTypedDictAgent
7 _KeygenJson, _RegisterJson, _ListJson must have schema_version, exit_code,
8 duration_ms annotations. _ListJson must exist (does not yet).
9
10 Tier 2 — TestTypedDictAgentConfig
11 _InitJson, _SyncJson, _ReadJson, _StatusJson, _InspectJson, _SetJson must
12 exist and carry schema_version, exit_code, duration_ms.
13
14 Tier 3 — TestTypedDictAgentMap
15 _AgentMapJson must exist and carry schema_version, exit_code, duration_ms,
16 mode.
17
18 Tier 4 — TestUnitFingerprint / TestUnitEmitError
19 _fingerprint correctness — known inputs, empty bytes, determinism.
20 _emit_error json/human paths, error/message keys, always exits 1.
21
22 Tier 5 — TestAliasRegistration
23 -j alias present on agent keygen, agent list, agent register, agent-map.
24
25 Tier 6 — TestDocstrings
26 run_list, run_keygen, run_register mention schema_version; register()
27 docstring for agent mentions -j.
28
29 Tier 7 — TestEndToEnd / TestStress / TestDataIntegrity / TestSecurity / TestPerformance
30 Live CLI invocations, stress runs, type invariants, hostile input
31 survival, timing bounds.
32
33 All tests should be RED until the implementation adds the missing TypedDicts,
34 envelope fields, and -j aliases described in the task.
35 """
36
37 from __future__ import annotations
38
39 import argparse
40 import hashlib
41 import io
42 import json
43 import os
44 import pathlib
45 import sys
46 import threading
47 import time
48 from typing import get_type_hints
49
50 import pytest
51
52 from tests.cli_test_helper import CliRunner, InvokeResult
53
54 runner = CliRunner()
55
56 # ---------------------------------------------------------------------------
57 # Constants
58 # ---------------------------------------------------------------------------
59
60 _TEST_HUB = "http://test.example.com"
61 _TEST_MNEMONIC = (
62 "abandon abandon abandon abandon abandon abandon abandon abandon "
63 "abandon abandon abandon about"
64 )
65
66 # ---------------------------------------------------------------------------
67 # Helpers
68 # ---------------------------------------------------------------------------
69
70
71 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
72 """Invoke runner with CWD set to *repo*."""
73 saved = os.getcwd()
74 try:
75 os.chdir(repo)
76 return runner.invoke(None, args)
77 finally:
78 os.chdir(saved)
79
80
81 # ---------------------------------------------------------------------------
82 # Fixtures
83 # ---------------------------------------------------------------------------
84
85
86 @pytest.fixture()
87 def cfg_repo(tmp_path: pathlib.Path) -> pathlib.Path:
88 """Minimal muse repo suitable for agent-config tests."""
89 saved = os.getcwd()
90 try:
91 os.chdir(tmp_path)
92 r = runner.invoke(None, ["init"])
93 assert r.exit_code == 0, f"init failed: {r.output}"
94 finally:
95 os.chdir(saved)
96 return tmp_path
97
98
99 @pytest.fixture()
100 def identity_repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
101 """Repo with a synthetic identity entry for agent keygen/register tests."""
102 saved = os.getcwd()
103 try:
104 os.chdir(tmp_path)
105 r = runner.invoke(None, ["init"])
106 assert r.exit_code == 0, f"init failed: {r.output}"
107 finally:
108 os.chdir(saved)
109
110 # Inject a fake identity so _require_mnemonic does not bail
111 identity_dir = pathlib.Path.home() / ".muse"
112 identity_dir.mkdir(parents=True, exist_ok=True)
113 identity_file = identity_dir / "identity.toml"
114
115 # Back up existing identity if present
116 backup: bytes | None = None
117 if identity_file.exists():
118 backup = identity_file.read_bytes()
119
120 monkeypatch.setattr(
121 "muse.core.identity.load_identity",
122 lambda url: {"mnemonic": _TEST_MNEMONIC, "handle": "test-agent"},
123 )
124
125 yield tmp_path
126
127 # Restore backup
128 if backup is not None:
129 identity_file.write_bytes(backup)
130
131
132 # ===========================================================================
133 # Tier 1 — TestTypedDictAgent
134 # ===========================================================================
135
136
137 class TestTypedDictAgent:
138 """_KeygenJson/_RegisterJson/_ListJson must carry envelope fields."""
139
140 # ── _KeygenJson ──────────────────────────────────────────────────────────
141
142 def test_keygen_json_has_schema_version(self) -> None:
143 from muse.cli.commands.agent import _KeygenJson
144 assert "schema_version" in get_type_hints(_KeygenJson)
145
146 def test_keygen_json_has_exit_code(self) -> None:
147 from muse.cli.commands.agent import _KeygenJson
148 assert "exit_code" in get_type_hints(_KeygenJson)
149
150 def test_keygen_json_has_duration_ms(self) -> None:
151 from muse.cli.commands.agent import _KeygenJson
152 assert "duration_ms" in get_type_hints(_KeygenJson)
153
154 def test_keygen_json_existing_fields_preserved(self) -> None:
155 from muse.cli.commands.agent import _KeygenJson
156 hints = get_type_hints(_KeygenJson)
157 for field in ("status", "hub", "account", "msign_path", "public_key_b64",
158 "fingerprint", "hd_seed_b64"):
159 assert field in hints, f"_KeygenJson missing existing field: {field}"
160
161 # ── _RegisterJson ────────────────────────────────────────────────────────
162
163 def test_register_json_has_schema_version(self) -> None:
164 from muse.cli.commands.agent import _RegisterJson
165 assert "schema_version" in get_type_hints(_RegisterJson)
166
167 def test_register_json_has_exit_code(self) -> None:
168 from muse.cli.commands.agent import _RegisterJson
169 assert "exit_code" in get_type_hints(_RegisterJson)
170
171 def test_register_json_has_duration_ms(self) -> None:
172 from muse.cli.commands.agent import _RegisterJson
173 assert "duration_ms" in get_type_hints(_RegisterJson)
174
175 def test_register_json_existing_fields_preserved(self) -> None:
176 from muse.cli.commands.agent import _RegisterJson
177 hints = get_type_hints(_RegisterJson)
178 for field in ("status", "name", "account", "hub", "msign_path"):
179 assert field in hints, f"_RegisterJson missing existing field: {field}"
180
181 # ── _ListJson ────────────────────────────────────────────────────────────
182
183 def test_list_json_exists(self) -> None:
184 """_ListJson TypedDict must be importable — does not yet exist."""
185 from muse.cli.commands.agent import _ListJson # noqa: F401
186
187 def test_list_json_has_schema_version(self) -> None:
188 from muse.cli.commands.agent import _ListJson
189 assert "schema_version" in get_type_hints(_ListJson)
190
191 def test_list_json_has_exit_code(self) -> None:
192 from muse.cli.commands.agent import _ListJson
193 assert "exit_code" in get_type_hints(_ListJson)
194
195 def test_list_json_has_duration_ms(self) -> None:
196 from muse.cli.commands.agent import _ListJson
197 assert "duration_ms" in get_type_hints(_ListJson)
198
199 def test_list_json_has_slots_field(self) -> None:
200 from muse.cli.commands.agent import _ListJson
201 assert "slots" in get_type_hints(_ListJson)
202
203 def test_list_json_has_mode_field(self) -> None:
204 from muse.cli.commands.agent import _ListJson
205 assert "mode" in get_type_hints(_ListJson)
206
207
208 # ===========================================================================
209 # Tier 2 — TestTypedDictAgentConfig
210 # ===========================================================================
211
212
213 class TestTypedDictAgentConfig:
214 """_InitJson/_SyncJson/_ReadJson/_StatusJson/_InspectJson/_SetJson must exist."""
215
216 # ── _InitJson ─────────────────────────────────────────────────────────────
217
218 def test_init_json_exists(self) -> None:
219 from muse.cli.commands.agent_config import _InitJson # noqa: F401
220
221 def test_init_json_has_schema_version(self) -> None:
222 from muse.cli.commands.agent_config import _InitJson
223 assert "schema_version" in get_type_hints(_InitJson)
224
225 def test_init_json_has_exit_code(self) -> None:
226 from muse.cli.commands.agent_config import _InitJson
227 assert "exit_code" in get_type_hints(_InitJson)
228
229 def test_init_json_has_duration_ms(self) -> None:
230 from muse.cli.commands.agent_config import _InitJson
231 assert "duration_ms" in get_type_hints(_InitJson)
232
233 def test_init_json_has_path_field(self) -> None:
234 from muse.cli.commands.agent_config import _InitJson
235 assert "path" in get_type_hints(_InitJson)
236
237 # ── _SyncJson ─────────────────────────────────────────────────────────────
238
239 def test_sync_json_exists(self) -> None:
240 from muse.cli.commands.agent_config import _SyncJson # noqa: F401
241
242 def test_sync_json_has_schema_version(self) -> None:
243 from muse.cli.commands.agent_config import _SyncJson
244 assert "schema_version" in get_type_hints(_SyncJson)
245
246 def test_sync_json_has_exit_code(self) -> None:
247 from muse.cli.commands.agent_config import _SyncJson
248 assert "exit_code" in get_type_hints(_SyncJson)
249
250 def test_sync_json_has_duration_ms(self) -> None:
251 from muse.cli.commands.agent_config import _SyncJson
252 assert "duration_ms" in get_type_hints(_SyncJson)
253
254 def test_sync_json_has_adapters_field(self) -> None:
255 from muse.cli.commands.agent_config import _SyncJson
256 assert "adapters" in get_type_hints(_SyncJson)
257
258 # ── _ReadJson ─────────────────────────────────────────────────────────────
259
260 def test_read_json_exists(self) -> None:
261 from muse.cli.commands.agent_config import _ReadJson # noqa: F401
262
263 def test_read_json_has_schema_version(self) -> None:
264 from muse.cli.commands.agent_config import _ReadJson
265 assert "schema_version" in get_type_hints(_ReadJson)
266
267 def test_read_json_has_exit_code(self) -> None:
268 from muse.cli.commands.agent_config import _ReadJson
269 assert "exit_code" in get_type_hints(_ReadJson)
270
271 def test_read_json_has_duration_ms(self) -> None:
272 from muse.cli.commands.agent_config import _ReadJson
273 assert "duration_ms" in get_type_hints(_ReadJson)
274
275 def test_read_json_has_content_field(self) -> None:
276 from muse.cli.commands.agent_config import _ReadJson
277 assert "content" in get_type_hints(_ReadJson)
278
279 # ── _StatusJson ───────────────────────────────────────────────────────────
280
281 def test_status_json_exists(self) -> None:
282 from muse.cli.commands.agent_config import _StatusJson # noqa: F401
283
284 def test_status_json_has_schema_version(self) -> None:
285 from muse.cli.commands.agent_config import _StatusJson
286 assert "schema_version" in get_type_hints(_StatusJson)
287
288 def test_status_json_has_exit_code(self) -> None:
289 from muse.cli.commands.agent_config import _StatusJson
290 assert "exit_code" in get_type_hints(_StatusJson)
291
292 def test_status_json_has_duration_ms(self) -> None:
293 from muse.cli.commands.agent_config import _StatusJson
294 assert "duration_ms" in get_type_hints(_StatusJson)
295
296 def test_status_json_has_ready_field(self) -> None:
297 from muse.cli.commands.agent_config import _StatusJson
298 assert "ready" in get_type_hints(_StatusJson)
299
300 # ── _InspectJson ──────────────────────────────────────────────────────────
301
302 def test_inspect_json_exists(self) -> None:
303 from muse.cli.commands.agent_config import _InspectJson # noqa: F401
304
305 def test_inspect_json_has_schema_version(self) -> None:
306 from muse.cli.commands.agent_config import _InspectJson
307 assert "schema_version" in get_type_hints(_InspectJson)
308
309 def test_inspect_json_has_exit_code(self) -> None:
310 from muse.cli.commands.agent_config import _InspectJson
311 assert "exit_code" in get_type_hints(_InspectJson)
312
313 def test_inspect_json_has_duration_ms(self) -> None:
314 from muse.cli.commands.agent_config import _InspectJson
315 assert "duration_ms" in get_type_hints(_InspectJson)
316
317 def test_inspect_json_has_context_field(self) -> None:
318 from muse.cli.commands.agent_config import _InspectJson
319 assert "context" in get_type_hints(_InspectJson)
320
321 # ── _SetJson ──────────────────────────────────────────────────────────────
322
323 def test_set_json_exists(self) -> None:
324 from muse.cli.commands.agent_config import _SetJson # noqa: F401
325
326 def test_set_json_has_schema_version(self) -> None:
327 from muse.cli.commands.agent_config import _SetJson
328 assert "schema_version" in get_type_hints(_SetJson)
329
330 def test_set_json_has_exit_code(self) -> None:
331 from muse.cli.commands.agent_config import _SetJson
332 assert "exit_code" in get_type_hints(_SetJson)
333
334 def test_set_json_has_duration_ms(self) -> None:
335 from muse.cli.commands.agent_config import _SetJson
336 assert "duration_ms" in get_type_hints(_SetJson)
337
338 def test_set_json_has_adapters_field(self) -> None:
339 from muse.cli.commands.agent_config import _SetJson
340 assert "adapters" in get_type_hints(_SetJson)
341
342
343 # ===========================================================================
344 # Tier 3 — TestTypedDictAgentMap
345 # ===========================================================================
346
347
348 class TestTypedDictAgentMap:
349 """_AgentMapJson must exist and carry envelope fields."""
350
351 def test_agent_map_json_exists(self) -> None:
352 from muse.cli.commands.agent_map import _AgentMapJson # noqa: F401
353
354 def test_agent_map_json_has_schema_version(self) -> None:
355 from muse.cli.commands.agent_map import _AgentMapJson
356 assert "schema_version" in get_type_hints(_AgentMapJson)
357
358 def test_agent_map_json_has_exit_code(self) -> None:
359 from muse.cli.commands.agent_map import _AgentMapJson
360 assert "exit_code" in get_type_hints(_AgentMapJson)
361
362 def test_agent_map_json_has_duration_ms(self) -> None:
363 from muse.cli.commands.agent_map import _AgentMapJson
364 assert "duration_ms" in get_type_hints(_AgentMapJson)
365
366 def test_agent_map_json_has_mode(self) -> None:
367 from muse.cli.commands.agent_map import _AgentMapJson
368 assert "mode" in get_type_hints(_AgentMapJson)
369
370 def test_agent_map_json_has_track(self) -> None:
371 from muse.cli.commands.agent_map import _AgentMapJson
372 assert "track" in get_type_hints(_AgentMapJson)
373
374 def test_agent_map_json_has_attributions(self) -> None:
375 from muse.cli.commands.agent_map import _AgentMapJson
376 assert "attributions" in get_type_hints(_AgentMapJson)
377
378 def test_bar_attribution_existing_fields_preserved(self) -> None:
379 from muse.cli.commands.agent_map import BarAttribution
380 hints = get_type_hints(BarAttribution)
381 for field in ("bar", "author", "commit_id", "message"):
382 assert field in hints, f"BarAttribution missing field: {field}"
383
384
385 # ===========================================================================
386 # Tier 4 — TestUnitFingerprint / TestUnitEmitError
387 # ===========================================================================
388
389
390 class TestUnitFingerprint:
391 """_fingerprint correctness tests."""
392
393 def test_known_input(self) -> None:
394 from muse.cli.commands.agent import _fingerprint
395 data = b"\x00" * 32
396 expected = hashlib.sha256(data).hexdigest()
397 assert _fingerprint(data) == expected
398
399 def test_returns_hex_string(self) -> None:
400 from muse.cli.commands.agent import _fingerprint
401 result = _fingerprint(b"test data")
402 assert isinstance(result, str)
403 assert all(c in "0123456789abcdef" for c in result)
404
405 def test_hex_length_is_64(self) -> None:
406 from muse.cli.commands.agent import _fingerprint
407 assert len(_fingerprint(b"x" * 32)) == 64
408
409 def test_empty_bytes(self) -> None:
410 from muse.cli.commands.agent import _fingerprint
411 expected = hashlib.sha256(b"").hexdigest()
412 assert _fingerprint(b"") == expected
413
414 def test_different_inputs_differ(self) -> None:
415 from muse.cli.commands.agent import _fingerprint
416 assert _fingerprint(b"aaa") != _fingerprint(b"bbb")
417
418 def test_deterministic(self) -> None:
419 from muse.cli.commands.agent import _fingerprint
420 data = b"deterministic input"
421 assert _fingerprint(data) == _fingerprint(data)
422
423 def test_single_byte_inputs(self) -> None:
424 from muse.cli.commands.agent import _fingerprint
425 fp0 = _fingerprint(b"\x00")
426 fp1 = _fingerprint(b"\x01")
427 assert fp0 != fp1
428
429 def test_matches_hashlib_directly(self) -> None:
430 from muse.cli.commands.agent import _fingerprint
431 data = b"public_key_bytes_example"
432 assert _fingerprint(data) == hashlib.sha256(data).hexdigest()
433
434
435 class TestUnitEmitError:
436 """_emit_error json/human paths."""
437
438 def _capture_emit_error(self, error: str, message: str, as_json: bool) -> tuple[str, str, int]:
439 """Run _emit_error, return (stdout_text, stderr_text, exit_code)."""
440 from muse.cli.commands.agent import _emit_error
441 stdout_buf = io.StringIO()
442 stderr_buf = io.StringIO()
443 exit_code = 0
444 orig_stdout, orig_stderr = sys.stdout, sys.stderr
445 sys.stdout = stdout_buf
446 sys.stderr = stderr_buf
447 try:
448 _emit_error(error, message, as_json)
449 except SystemExit as exc:
450 exit_code = int(exc.code) if exc.code is not None else 0
451 finally:
452 sys.stdout = orig_stdout
453 sys.stderr = orig_stderr
454 return stdout_buf.getvalue(), stderr_buf.getvalue(), exit_code
455
456 def test_json_mode_writes_to_stdout(self) -> None:
457 stdout, _, _ = self._capture_emit_error("err_code", "msg text", True)
458 assert stdout.strip() != ""
459
460 def test_json_mode_valid_json(self) -> None:
461 stdout, _, _ = self._capture_emit_error("err_code", "msg text", True)
462 json.loads(stdout) # must not raise
463
464 def test_json_mode_has_error_key(self) -> None:
465 stdout, _, _ = self._capture_emit_error("err_code", "msg text", True)
466 d = json.loads(stdout)
467 assert "error" in d
468
469 def test_json_mode_has_message_key(self) -> None:
470 stdout, _, _ = self._capture_emit_error("err_code", "msg text", True)
471 d = json.loads(stdout)
472 assert "message" in d
473
474 def test_json_mode_error_value_correct(self) -> None:
475 stdout, _, _ = self._capture_emit_error("my_error", "some message", True)
476 d = json.loads(stdout)
477 assert d["error"] == "my_error"
478
479 def test_json_mode_message_value_correct(self) -> None:
480 stdout, _, _ = self._capture_emit_error("err", "expected message", True)
481 d = json.loads(stdout)
482 assert d["message"] == "expected message"
483
484 def test_json_mode_exits_1(self) -> None:
485 _, _, code = self._capture_emit_error("err", "msg", True)
486 assert code == 1
487
488 def test_human_mode_writes_to_stderr(self) -> None:
489 _, stderr, _ = self._capture_emit_error("err", "human message", False)
490 assert stderr.strip() != ""
491
492 def test_human_mode_exits_1(self) -> None:
493 _, _, code = self._capture_emit_error("err", "msg", False)
494 assert code == 1
495
496 def test_human_mode_stdout_empty(self) -> None:
497 stdout, _, _ = self._capture_emit_error("err", "msg", False)
498 assert stdout.strip() == ""
499
500
501 # ===========================================================================
502 # Tier 5 — TestAliasRegistration
503 # ===========================================================================
504
505
506 class TestAliasRegistration:
507 """``-j`` alias must be registered on all three agent subcommands and agent-map."""
508
509 def _build_agent_parser(self) -> argparse.ArgumentParser:
510 from muse.cli.commands.agent import register
511 p = argparse.ArgumentParser()
512 sub = p.add_subparsers()
513 register(sub)
514 return p
515
516 def test_agent_keygen_j_alias_parses(self) -> None:
517 p = self._build_agent_parser()
518 ns = p.parse_args(["agent", "keygen", "--account", "0", "--hub", _TEST_HUB, "-j"])
519 assert getattr(ns, "json", False) is True
520
521 def test_agent_list_j_alias_parses(self) -> None:
522 p = self._build_agent_parser()
523 ns = p.parse_args(["agent", "list", "--hub", _TEST_HUB, "-j"])
524 assert getattr(ns, "json", False) is True
525
526 def test_agent_register_j_alias_parses(self) -> None:
527 p = self._build_agent_parser()
528 ns = p.parse_args([
529 "agent", "register",
530 "--account", "1",
531 "--name", "test-agent",
532 "--hub", _TEST_HUB,
533 "-j",
534 ])
535 assert getattr(ns, "json", False) is True
536
537 def test_agent_map_j_alias_registered(self) -> None:
538 from muse.cli.commands.agent_map import register
539 p = argparse.ArgumentParser()
540 sub = p.add_subparsers()
541 register(sub)
542 ns = p.parse_args(["agent-map", "tracks/test.mid", "-j"])
543 # -j should set json/as_json to True
544 assert getattr(ns, "json", False) is True or getattr(ns, "as_json", False) is True
545
546 def test_agent_keygen_json_flag_still_works(self) -> None:
547 p = self._build_agent_parser()
548 ns = p.parse_args(["agent", "keygen", "--account", "0", "--hub", _TEST_HUB, "--json"])
549 assert getattr(ns, "json", False) is True
550
551 def test_agent_list_json_flag_still_works(self) -> None:
552 p = self._build_agent_parser()
553 ns = p.parse_args(["agent", "list", "--hub", _TEST_HUB, "--json"])
554 assert getattr(ns, "json", False) is True
555
556 def test_agent_register_json_flag_still_works(self) -> None:
557 p = self._build_agent_parser()
558 ns = p.parse_args([
559 "agent", "register",
560 "--account", "1",
561 "--name", "test-agent",
562 "--hub", _TEST_HUB,
563 "--json",
564 ])
565 assert getattr(ns, "json", False) is True
566
567
568 # ===========================================================================
569 # Tier 6 — TestDocstrings
570 # ===========================================================================
571
572
573 class TestDocstrings:
574 """Key functions must document new envelope fields in their docstrings."""
575
576 def test_run_keygen_docstring_mentions_schema_version(self) -> None:
577 from muse.cli.commands.agent import run_keygen
578 assert run_keygen.__doc__ is not None
579 assert "schema_version" in run_keygen.__doc__
580
581 def test_run_list_docstring_mentions_schema_version(self) -> None:
582 from muse.cli.commands.agent import run_list
583 assert run_list.__doc__ is not None
584 assert "schema_version" in run_list.__doc__
585
586 def test_run_register_docstring_mentions_schema_version(self) -> None:
587 from muse.cli.commands.agent import run_register
588 assert run_register.__doc__ is not None
589 assert "schema_version" in run_register.__doc__
590
591 def test_register_docstring_mentions_j_alias(self) -> None:
592 from muse.cli.commands.agent import register
593 assert register.__doc__ is not None
594 assert "-j" in register.__doc__
595
596 def test_run_list_docstring_mentions_duration_ms(self) -> None:
597 from muse.cli.commands.agent import run_list
598 assert run_list.__doc__ is not None
599 assert "duration_ms" in run_list.__doc__
600
601 def test_run_keygen_docstring_mentions_duration_ms(self) -> None:
602 from muse.cli.commands.agent import run_keygen
603 assert run_keygen.__doc__ is not None
604 assert "duration_ms" in run_keygen.__doc__
605
606
607 # ===========================================================================
608 # Tier 7a — TestEndToEnd
609 # ===========================================================================
610
611
612 class TestEndToEnd:
613 """Live CLI invocations against real (tmp) repos."""
614
615 # ── agent list -j ────────────────────────────────────────────────────────
616
617 def test_agent_list_j_exits_zero(self) -> None:
618 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
619 assert r.exit_code == 0, r.output
620
621 def test_agent_list_j_valid_json(self) -> None:
622 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
623 assert r.exit_code == 0, r.output
624 json.loads(r.output)
625
626 def test_agent_list_j_has_schema_version(self) -> None:
627 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
628 assert r.exit_code == 0, r.output
629 d = json.loads(r.output)
630 assert "schema_version" in d
631
632 def test_agent_list_j_has_exit_code(self) -> None:
633 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
634 d = json.loads(r.output)
635 assert "exit_code" in d
636
637 def test_agent_list_j_has_duration_ms(self) -> None:
638 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
639 d = json.loads(r.output)
640 assert "duration_ms" in d
641
642 def test_agent_list_j_has_slots(self) -> None:
643 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
644 d = json.loads(r.output)
645 assert "slots" in d
646
647 def test_agent_list_j_slots_is_list(self) -> None:
648 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
649 d = json.loads(r.output)
650 assert isinstance(d["slots"], list)
651
652 def test_agent_list_j_exit_code_matches_process(self) -> None:
653 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
654 d = json.loads(r.output)
655 assert d["exit_code"] == r.exit_code
656
657 def test_agent_list_json_flag_same_as_j(self) -> None:
658 r1 = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
659 r2 = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"])
660 d1 = json.loads(r1.output)
661 d2 = json.loads(r2.output)
662 d1.pop("duration_ms", None)
663 d2.pop("duration_ms", None)
664 assert set(d1.keys()) == set(d2.keys())
665
666 # ── agent-config init -j ──────────────────────────────────────────────────
667
668 def test_agent_config_init_j_exits_zero(self, cfg_repo: pathlib.Path) -> None:
669 r = _invoke(cfg_repo, ["agent-config", "init", "-j"])
670 assert r.exit_code == 0, r.output
671
672 def test_agent_config_init_j_valid_json(self, cfg_repo: pathlib.Path) -> None:
673 r = _invoke(cfg_repo, ["agent-config", "init", "-j"])
674 assert r.exit_code == 0, r.output
675 json.loads(r.output)
676
677 def test_agent_config_init_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None:
678 r = _invoke(cfg_repo, ["agent-config", "init", "-j"])
679 assert r.exit_code == 0, r.output
680 d = json.loads(r.output)
681 assert "schema_version" in d
682
683 def test_agent_config_init_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None:
684 r = _invoke(cfg_repo, ["agent-config", "init", "-j"])
685 d = json.loads(r.output)
686 assert "exit_code" in d
687
688 def test_agent_config_init_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None:
689 r = _invoke(cfg_repo, ["agent-config", "init", "-j"])
690 d = json.loads(r.output)
691 assert "duration_ms" in d
692
693 # ── agent-config status -j ────────────────────────────────────────────────
694
695 def test_agent_config_status_j_exits_zero(self, cfg_repo: pathlib.Path) -> None:
696 _invoke(cfg_repo, ["agent-config", "init", "-j"])
697 r = _invoke(cfg_repo, ["agent-config", "status", "-j"])
698 assert r.exit_code == 0, r.output
699
700 def test_agent_config_status_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None:
701 _invoke(cfg_repo, ["agent-config", "init", "-j"])
702 r = _invoke(cfg_repo, ["agent-config", "status", "-j"])
703 assert r.exit_code == 0, r.output
704 d = json.loads(r.output)
705 assert "schema_version" in d
706
707 def test_agent_config_status_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None:
708 _invoke(cfg_repo, ["agent-config", "init", "-j"])
709 r = _invoke(cfg_repo, ["agent-config", "status", "-j"])
710 d = json.loads(r.output)
711 assert "exit_code" in d
712
713 def test_agent_config_status_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None:
714 _invoke(cfg_repo, ["agent-config", "init", "-j"])
715 r = _invoke(cfg_repo, ["agent-config", "status", "-j"])
716 d = json.loads(r.output)
717 assert "duration_ms" in d
718
719 # ── agent-config inspect -j ────────────────────────────────────────────────
720
721 def test_agent_config_inspect_j_exits_zero(self, cfg_repo: pathlib.Path) -> None:
722 _invoke(cfg_repo, ["agent-config", "init", "-j"])
723 r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"])
724 assert r.exit_code == 0, r.output
725
726 def test_agent_config_inspect_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None:
727 _invoke(cfg_repo, ["agent-config", "init", "-j"])
728 r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"])
729 assert r.exit_code == 0, r.output
730 d = json.loads(r.output)
731 assert "schema_version" in d
732
733 def test_agent_config_inspect_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None:
734 _invoke(cfg_repo, ["agent-config", "init", "-j"])
735 r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"])
736 d = json.loads(r.output)
737 assert "exit_code" in d
738
739 def test_agent_config_inspect_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None:
740 _invoke(cfg_repo, ["agent-config", "init", "-j"])
741 r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"])
742 d = json.loads(r.output)
743 assert "duration_ms" in d
744
745 # ── agent-config read -j (after init) ────────────────────────────────────
746
747 def test_agent_config_read_j_exits_zero(self, cfg_repo: pathlib.Path) -> None:
748 _invoke(cfg_repo, ["agent-config", "init"])
749 r = _invoke(cfg_repo, ["agent-config", "read", "-j"])
750 assert r.exit_code == 0, r.output
751
752 def test_agent_config_read_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None:
753 _invoke(cfg_repo, ["agent-config", "init"])
754 r = _invoke(cfg_repo, ["agent-config", "read", "-j"])
755 assert r.exit_code == 0, r.output
756 d = json.loads(r.output)
757 assert "schema_version" in d
758
759 def test_agent_config_read_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None:
760 _invoke(cfg_repo, ["agent-config", "init"])
761 r = _invoke(cfg_repo, ["agent-config", "read", "-j"])
762 d = json.loads(r.output)
763 assert "exit_code" in d
764
765 def test_agent_config_read_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None:
766 _invoke(cfg_repo, ["agent-config", "init"])
767 r = _invoke(cfg_repo, ["agent-config", "read", "-j"])
768 d = json.loads(r.output)
769 assert "duration_ms" in d
770
771 # ── agent-config sync -j (after init) ────────────────────────────────────
772
773 def test_agent_config_sync_j_exits_zero(self, cfg_repo: pathlib.Path) -> None:
774 _invoke(cfg_repo, ["agent-config", "init"])
775 r = _invoke(cfg_repo, ["agent-config", "sync", "-j"])
776 assert r.exit_code == 0, r.output
777
778 def test_agent_config_sync_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None:
779 _invoke(cfg_repo, ["agent-config", "init"])
780 r = _invoke(cfg_repo, ["agent-config", "sync", "-j"])
781 assert r.exit_code == 0, r.output
782 d = json.loads(r.output)
783 assert "schema_version" in d
784
785 def test_agent_config_sync_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None:
786 _invoke(cfg_repo, ["agent-config", "init"])
787 r = _invoke(cfg_repo, ["agent-config", "sync", "-j"])
788 d = json.loads(r.output)
789 assert "exit_code" in d
790
791 def test_agent_config_sync_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None:
792 _invoke(cfg_repo, ["agent-config", "init"])
793 r = _invoke(cfg_repo, ["agent-config", "sync", "-j"])
794 d = json.loads(r.output)
795 assert "duration_ms" in d
796
797
798 # ===========================================================================
799 # Tier 7b — TestStress
800 # ===========================================================================
801
802
803 class TestStress:
804 """High-volume calls to verify no crashes, leaks, or race conditions."""
805
806 def test_fingerprint_1000_calls_no_crash(self) -> None:
807 from muse.cli.commands.agent import _fingerprint
808 for i in range(1000):
809 result = _fingerprint(i.to_bytes(4, "big"))
810 assert len(result) == 64
811
812 def test_emit_error_json_500_calls(self) -> None:
813 from muse.cli.commands.agent import _emit_error
814 for i in range(500):
815 stdout_buf = io.StringIO()
816 orig = sys.stdout
817 sys.stdout = stdout_buf
818 try:
819 try:
820 _emit_error(f"err_{i}", f"message_{i}", True)
821 except SystemExit:
822 pass
823 finally:
824 sys.stdout = orig
825 d = json.loads(stdout_buf.getvalue())
826 assert d["error"] == f"err_{i}"
827
828 def test_agent_config_status_50_times(self, cfg_repo: pathlib.Path) -> None:
829 _invoke(cfg_repo, ["agent-config", "init"])
830 for _ in range(50):
831 r = _invoke(cfg_repo, ["agent-config", "status", "--json"])
832 assert r.exit_code == 0, r.output
833 d = json.loads(r.output)
834 assert "schema_version" in d
835
836 def test_fingerprint_thread_safety(self) -> None:
837 from muse.cli.commands.agent import _fingerprint
838 results: list[str] = []
839 errors: list[Exception] = []
840 lock = threading.Lock()
841
842 def worker(n: int) -> None:
843 try:
844 fp = _fingerprint(n.to_bytes(4, "big"))
845 with lock:
846 results.append(fp)
847 except Exception as exc:
848 with lock:
849 errors.append(exc)
850
851 threads = [threading.Thread(target=worker, args=(i,)) for i in range(100)]
852 for t in threads:
853 t.start()
854 for t in threads:
855 t.join()
856
857 assert not errors, f"Thread errors: {errors}"
858 assert len(results) == 100
859
860
861 # ===========================================================================
862 # Tier 7c — TestDataIntegrity
863 # ===========================================================================
864
865
866 class TestDataIntegrity:
867 """Type invariants that the JSON envelope must satisfy."""
868
869 def test_agent_list_schema_version_is_str(self) -> None:
870 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
871 assert r.exit_code == 0, r.output
872 d = json.loads(r.output)
873 assert isinstance(d["schema_version"], str)
874
875 def test_agent_list_exit_code_is_int(self) -> None:
876 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
877 d = json.loads(r.output)
878 assert isinstance(d["exit_code"], int)
879
880 def test_agent_list_duration_ms_is_float(self) -> None:
881 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
882 d = json.loads(r.output)
883 assert isinstance(d["duration_ms"], float)
884
885 def test_agent_list_slots_is_list(self) -> None:
886 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
887 d = json.loads(r.output)
888 assert isinstance(d["slots"], list)
889
890 def test_agent_list_duration_ms_nonnegative(self) -> None:
891 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
892 d = json.loads(r.output)
893 assert d["duration_ms"] >= 0.0
894
895 def test_agent_config_init_schema_version_is_str(self, cfg_repo: pathlib.Path) -> None:
896 r = _invoke(cfg_repo, ["agent-config", "init", "--json"])
897 assert r.exit_code == 0, r.output
898 d = json.loads(r.output)
899 assert isinstance(d["schema_version"], str)
900
901 def test_agent_config_init_exit_code_is_int(self, cfg_repo: pathlib.Path) -> None:
902 r = _invoke(cfg_repo, ["agent-config", "init", "--json"])
903 d = json.loads(r.output)
904 assert isinstance(d["exit_code"], int)
905
906 def test_agent_config_init_duration_ms_is_float(self, cfg_repo: pathlib.Path) -> None:
907 r = _invoke(cfg_repo, ["agent-config", "init", "--json"])
908 d = json.loads(r.output)
909 assert isinstance(d["duration_ms"], float)
910
911 def test_agent_config_status_exit_code_mirrors_process(self, cfg_repo: pathlib.Path) -> None:
912 _invoke(cfg_repo, ["agent-config", "init"])
913 r = _invoke(cfg_repo, ["agent-config", "status", "--json"])
914 d = json.loads(r.output)
915 assert d["exit_code"] == r.exit_code
916
917 def test_agent_list_schema_version_nonempty(self) -> None:
918 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
919 d = json.loads(r.output)
920 assert d["schema_version"] != ""
921
922
923 # ===========================================================================
924 # Tier 7d — TestSecurity
925 # ===========================================================================
926
927
928 class TestSecurity:
929 """Hostile input must not crash or corrupt JSON output."""
930
931 def test_hostile_hub_url_in_emit_error_json(self) -> None:
932 from muse.cli.commands.agent import _emit_error
933 hostile = "'; DROP TABLE users; --"
934 stdout_buf = io.StringIO()
935 orig = sys.stdout
936 sys.stdout = stdout_buf
937 try:
938 try:
939 _emit_error("sql_test", hostile, True)
940 except SystemExit:
941 pass
942 finally:
943 sys.stdout = orig
944 # Must still be valid JSON
945 d = json.loads(stdout_buf.getvalue())
946 assert d["message"] == hostile
947
948 def test_sql_injection_in_slot_name(self) -> None:
949 """A slot name with SQL injection chars must not crash fingerprint or emit_error."""
950 from muse.cli.commands.agent import _fingerprint
951 evil_name = b"'; DROP TABLE slots; --"
952 fp = _fingerprint(evil_name)
953 assert len(fp) == 64
954
955 def test_unicode_in_track_path_emit_error(self) -> None:
956 from muse.cli.commands.agent import _emit_error
957 unicode_path = "tracks/\u4e2d\u6587\u97f3\u4e50.mid"
958 stdout_buf = io.StringIO()
959 orig = sys.stdout
960 sys.stdout = stdout_buf
961 try:
962 try:
963 _emit_error("unicode_test", unicode_path, True)
964 except SystemExit:
965 pass
966 finally:
967 sys.stdout = orig
968 d = json.loads(stdout_buf.getvalue())
969 assert d["message"] == unicode_path
970
971 def test_very_long_account_name_in_emit_error(self) -> None:
972 from muse.cli.commands.agent import _emit_error
973 long_msg = "x" * 10_000
974 stdout_buf = io.StringIO()
975 orig = sys.stdout
976 sys.stdout = stdout_buf
977 try:
978 try:
979 _emit_error("long_test", long_msg, True)
980 except SystemExit:
981 pass
982 finally:
983 sys.stdout = orig
984 d = json.loads(stdout_buf.getvalue())
985 assert d["message"] == long_msg
986
987 def test_null_bytes_in_fingerprint_input(self) -> None:
988 from muse.cli.commands.agent import _fingerprint
989 data = b"\x00" * 64
990 fp = _fingerprint(data)
991 assert len(fp) == 64
992
993 def test_newline_in_error_message_survives_json(self) -> None:
994 from muse.cli.commands.agent import _emit_error
995 newline_msg = "line one\nline two\nline three"
996 stdout_buf = io.StringIO()
997 orig = sys.stdout
998 sys.stdout = stdout_buf
999 try:
1000 try:
1001 _emit_error("newline_test", newline_msg, True)
1002 except SystemExit:
1003 pass
1004 finally:
1005 sys.stdout = orig
1006 d = json.loads(stdout_buf.getvalue())
1007 assert d["message"] == newline_msg
1008
1009 def test_agent_list_with_encoded_url_returns_json(self) -> None:
1010 encoded_url = "http://test.example.com%2Fpath"
1011 r = runner.invoke(None, ["agent", "list", "--hub", encoded_url, "--json"])
1012 # Should either succeed with JSON or emit a JSON error — not a crash
1013 output = r.output.strip()
1014 assert output, "no output produced"
1015 json.loads(output) # must be valid JSON regardless
1016
1017
1018 # ===========================================================================
1019 # Tier 7e — TestPerformance
1020 # ===========================================================================
1021
1022
1023 class TestPerformance:
1024 """Timing bounds for performance-sensitive paths."""
1025
1026 def test_fingerprint_1000_under_500ms(self) -> None:
1027 from muse.cli.commands.agent import _fingerprint
1028 data = b"perf test input" * 4
1029 start = time.perf_counter()
1030 for _ in range(1000):
1031 _fingerprint(data)
1032 elapsed_ms = (time.perf_counter() - start) * 1000
1033 assert elapsed_ms < 500, f"1000 _fingerprint calls took {elapsed_ms:.1f}ms"
1034
1035 def test_agent_config_status_completes_quickly(self, cfg_repo: pathlib.Path) -> None:
1036 _invoke(cfg_repo, ["agent-config", "init"])
1037 start = time.perf_counter()
1038 r = _invoke(cfg_repo, ["agent-config", "status", "--json"])
1039 elapsed_ms = (time.perf_counter() - start) * 1000
1040 assert r.exit_code == 0, r.output
1041 assert elapsed_ms < 2000, f"agent-config status took {elapsed_ms:.1f}ms"
1042
1043 def test_agent_list_completes_quickly(self) -> None:
1044 start = time.perf_counter()
1045 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
1046 elapsed_ms = (time.perf_counter() - start) * 1000
1047 assert r.exit_code == 0, r.output
1048 assert elapsed_ms < 2000, f"agent list took {elapsed_ms:.1f}ms"
1049
1050 def test_duration_ms_in_list_output_is_plausible(self) -> None:
1051 r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"])
1052 assert r.exit_code == 0, r.output
1053 d = json.loads(r.output)
1054 # duration_ms must be >= 0 and < 5000 (very generous upper bound)
1055 assert 0.0 <= d["duration_ms"] < 5000.0
1056
1057 def test_duration_ms_in_agent_config_init_is_plausible(self, cfg_repo: pathlib.Path) -> None:
1058 r = _invoke(cfg_repo, ["agent-config", "init", "--json"])
1059 assert r.exit_code == 0, r.output
1060 d = json.loads(r.output)
1061 assert 0.0 <= d["duration_ms"] < 5000.0
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 140 days ago