gabriel / muse public
test_cmd_domains_hardening.py python
1,916 lines 79.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive hardening tests for ``muse domains``.
2
3 Covers:
4 Unit tests:
5 - _validate_domain_name: valid, path-traversal, reserved, bad chars
6 - _validate_publish_url: http/https OK, file/ftp/data rejected
7 - _active_domain: no root, missing repo.json, explicit domain, missing key
8 - _build_entry: schema present, schema absent, active flag boolean
9 - _post_json: correct wire format, response size cap, non-object JSON
10 - _check_method / _run_validate_plugin: all checks pass, missing method
11
12 Integration tests:
13 - run (dashboard): text output, --json, --new validation + success
14 - run_info: known domain, unknown domain, --json schema
15 - run_use: no repo, unknown domain, known domain switches repo.json, --json
16 - run_validate: all-pass, missing method, --json, multi-domain
17
18 Security tests:
19 - Path traversal in --new rejected
20 - ANSI in domain name sanitized in dashboard
21 - file:// publish hub rejected
22 - Unsanitized server response never echoed raw
23
24 E2E tests (full CLI invocation via CliRunner):
25 - muse domains --json emits boolean 'active' field
26 - muse domains --json includes module_path
27 - muse domains info code --json schema present
28 - muse domains validate --json ok
29 - muse domains use code --json inside repo
30 - muse domains --new myplug creates directory
31
32 Stress tests:
33 - 8 concurrent validate calls on isolated registry snapshots
34 - 8 concurrent _active_domain reads on isolated tmp dirs
35 """
36 from __future__ import annotations
37
38 import http.client
39 import json
40 import pathlib
41 import shutil
42 import threading
43 import urllib.error
44 import urllib.request
45 from contextlib import ExitStack
46 from typing import TYPE_CHECKING
47 from unittest.mock import MagicMock, patch
48
49 import pytest
50
51 from muse.cli.commands.domains import (
52 _DomainEntryJson,
53 _PublishResponse,
54 _ScaffoldJson,
55 _UseJson,
56 _ValidateJson,
57 )
58 from muse.domain import (
59 DriftReport,
60 LiveState,
61 MergeResult,
62 StateSnapshot,
63 StateDelta,
64 SnapshotManifest,
65 )
66 from muse.core.schema import DomainSchema
67 from tests.cli_test_helper import CliRunner, InvokeResult
68
69 if TYPE_CHECKING:
70 from muse.cli.commands.domains import _PublishPayload, _Capabilities, _DimensionDef
71 from muse.core.transport import SigningIdentity
72
73 cli = None # argparse migration — CliRunner ignores this argument
74
75 runner = CliRunner()
76
77 # ---------------------------------------------------------------------------
78 # JSON helpers — each returns a specific TypedDict to satisfy typing_audit
79 # ---------------------------------------------------------------------------
80
81
82 def _first_json_blob(result: InvokeResult) -> str:
83 """Return the first complete JSON blob string from result.output."""
84 output = result.output
85 depth = 0
86 start: int | None = None
87 for i, ch in enumerate(output):
88 if ch in "{[":
89 if start is None:
90 start = i
91 depth += 1
92 elif ch in "}]":
93 depth -= 1
94 if depth == 0 and start is not None:
95 return output[start : i + 1]
96 raise AssertionError(f"No JSON found in output:\n{output!r}")
97
98
99 def _parse_domains_list(result: InvokeResult) -> list[_DomainEntryJson]:
100 """Parse a JSON array of domain entries from result."""
101 parsed: list[_DomainEntryJson] = json.loads(_first_json_blob(result))
102 assert isinstance(parsed, list)
103 return parsed
104
105
106 def _parse_domain_entry(result: InvokeResult) -> _DomainEntryJson:
107 """Parse a single domain entry JSON dict from result."""
108 parsed: _DomainEntryJson = json.loads(_first_json_blob(result))
109 assert isinstance(parsed, dict)
110 return parsed
111
112
113 def _parse_scaffold(result: InvokeResult) -> _ScaffoldJson:
114 parsed: _ScaffoldJson = json.loads(_first_json_blob(result))
115 assert isinstance(parsed, dict)
116 return parsed
117
118
119 def _parse_use(result: InvokeResult) -> _UseJson:
120 parsed: _UseJson = json.loads(_first_json_blob(result))
121 assert isinstance(parsed, dict)
122 return parsed
123
124
125 def _parse_validate(result: InvokeResult) -> _ValidateJson:
126 parsed: _ValidateJson = json.loads(_first_json_blob(result))
127 assert isinstance(parsed, dict)
128 return parsed
129
130
131 def _parse_validate_list(result: InvokeResult) -> list[_ValidateJson]:
132 parsed: list[_ValidateJson] = json.loads(_first_json_blob(result))
133 assert isinstance(parsed, list)
134 return parsed
135
136
137 def _parse_publish(result: InvokeResult) -> _PublishResponse:
138 parsed: _PublishResponse = json.loads(_first_json_blob(result))
139 assert isinstance(parsed, dict)
140 return parsed
141
142
143 # ---------------------------------------------------------------------------
144 # Repository fixture
145 # ---------------------------------------------------------------------------
146
147
148 def _init_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path:
149 """Create a minimal .muse repo structure under tmp_path."""
150 muse = tmp_path / ".muse"
151 muse.mkdir(parents=True)
152 repo_json = {
153 "repo_id": "test-repo",
154 "schema_version": "0.1.5",
155 "created_at": "2026-01-01T00:00:00+00:00",
156 "domain": domain,
157 }
158 (muse / "repo.json").write_text(json.dumps(repo_json), encoding="utf-8")
159 return tmp_path
160
161
162 # ---------------------------------------------------------------------------
163 # Unit — _validate_domain_name
164 # ---------------------------------------------------------------------------
165
166
167 class TestValidateDomainName:
168 def _call(self, name: str) -> int:
169 from muse.cli.commands.domains import _validate_domain_name
170
171 with pytest.raises(SystemExit) as exc_info:
172 _validate_domain_name(name)
173 code = exc_info.value.code
174 assert isinstance(code, int)
175 return code
176
177 def test_valid_lowercase(self) -> None:
178 from muse.cli.commands.domains import _validate_domain_name
179
180 _validate_domain_name("genomics") # must not raise
181
182 def test_valid_with_hyphen(self) -> None:
183 from muse.cli.commands.domains import _validate_domain_name
184
185 _validate_domain_name("spatial-3d")
186
187 def test_valid_with_underscore(self) -> None:
188 from muse.cli.commands.domains import _validate_domain_name
189
190 _validate_domain_name("my_domain")
191
192 def test_path_traversal_dotdot_rejected(self) -> None:
193 assert self._call("../evil") == 1
194
195 def test_path_traversal_nested_rejected(self) -> None:
196 assert self._call("../../etc/passwd") == 1
197
198 def test_uppercase_rejected(self) -> None:
199 assert self._call("Genomics") == 1
200
201 def test_starts_with_digit_rejected(self) -> None:
202 assert self._call("3d-scenes") == 1
203
204 def test_slash_rejected(self) -> None:
205 assert self._call("evil/path") == 1
206
207 def test_null_byte_rejected(self) -> None:
208 assert self._call("evil\x00name") == 1
209
210 def test_space_rejected(self) -> None:
211 assert self._call("my domain") == 1
212
213 def test_reserved_scaffold_rejected(self) -> None:
214 assert self._call("scaffold") == 1
215
216 def test_max_length_64_accepted(self) -> None:
217 from muse.cli.commands.domains import _validate_domain_name
218
219 _validate_domain_name("a" * 64)
220
221 def test_max_length_65_rejected(self) -> None:
222 assert self._call("a" * 65) == 1
223
224 def test_empty_rejected(self) -> None:
225 assert self._call("") == 1
226
227
228 # ---------------------------------------------------------------------------
229 # Unit — _validate_publish_url
230 # ---------------------------------------------------------------------------
231
232
233 class TestValidatePublishUrl:
234 def _call(self, url: str) -> int:
235 from muse.cli.commands.domains import _validate_publish_url
236
237 with pytest.raises(SystemExit) as exc_info:
238 _validate_publish_url(url)
239 code = exc_info.value.code
240 assert isinstance(code, int)
241 return code
242
243 def test_https_ok(self) -> None:
244 from muse.cli.commands.domains import _validate_publish_url
245
246 _validate_publish_url("https://musehub.ai") # must not raise
247
248 def test_http_ok(self) -> None:
249 from muse.cli.commands.domains import _validate_publish_url
250
251 _validate_publish_url("http://localhost:10003")
252
253 def test_file_scheme_rejected(self) -> None:
254 assert self._call("file:///etc/passwd") == 1
255
256 def test_ftp_scheme_rejected(self) -> None:
257 assert self._call("ftp://evil.com") == 1
258
259 def test_data_uri_rejected(self) -> None:
260 assert self._call("data:text/plain,evil") == 1
261
262 def test_empty_scheme_rejected(self) -> None:
263 assert self._call("://evil") == 1
264
265
266 # ---------------------------------------------------------------------------
267 # Unit — _active_domain
268 # ---------------------------------------------------------------------------
269
270
271 class TestActiveDomain:
272 def test_none_root_returns_none(self) -> None:
273 from muse.cli.commands.domains import _active_domain
274
275 assert _active_domain(None) is None
276
277 def test_missing_repo_json_returns_none(self, tmp_path: pathlib.Path) -> None:
278 from muse.cli.commands.domains import _active_domain
279
280 (tmp_path / ".muse").mkdir()
281 assert _active_domain(tmp_path) is None
282
283 def test_explicit_domain_returned(self, tmp_path: pathlib.Path) -> None:
284 from muse.cli.commands.domains import _active_domain
285
286 _init_repo(tmp_path, domain="code")
287 assert _active_domain(tmp_path) == "code"
288
289 def test_missing_domain_key_returns_default(self, tmp_path: pathlib.Path) -> None:
290 from muse.cli.commands.domains import _active_domain, _DEFAULT_DOMAIN
291
292 muse = tmp_path / ".muse"
293 muse.mkdir()
294 (muse / "repo.json").write_text('{"repo_id": "x"}', encoding="utf-8")
295 assert _active_domain(tmp_path) == _DEFAULT_DOMAIN
296
297 def test_corrupt_json_returns_none(self, tmp_path: pathlib.Path) -> None:
298 from muse.cli.commands.domains import _active_domain
299
300 muse = tmp_path / ".muse"
301 muse.mkdir()
302 (muse / "repo.json").write_text("NOT JSON", encoding="utf-8")
303 assert _active_domain(tmp_path) is None
304
305 def test_no_midi_fallback(self, tmp_path: pathlib.Path) -> None:
306 """The old 'midi' fallback is gone — default is _DEFAULT_DOMAIN ('code')."""
307 from muse.cli.commands.domains import _active_domain
308
309 muse = tmp_path / ".muse"
310 muse.mkdir()
311 (muse / "repo.json").write_text('{"domain": ""}', encoding="utf-8")
312 result = _active_domain(tmp_path)
313 assert result != "midi"
314
315
316 # ---------------------------------------------------------------------------
317 # Unit — _build_entry
318 # ---------------------------------------------------------------------------
319
320
321 class TestBuildEntry:
322 def test_schema_present(self) -> None:
323 from muse.cli.commands.domains import _build_entry
324 from muse.plugins.registry import _REGISTRY
325
326 plugin = _REGISTRY["code"]
327 entry = _build_entry("code", plugin, "code")
328 assert entry["domain"] == "code"
329 assert entry["active"] is True
330 assert isinstance(entry["active"], bool)
331 assert "schema" in entry
332 schema = entry["schema"]
333 assert "schema_version" in schema
334 assert "dimensions" in schema
335
336 def test_schema_absent_when_not_implemented(self) -> None:
337 from muse.cli.commands.domains import _build_entry
338
339 mock_plugin = MagicMock()
340 mock_plugin.schema.side_effect = NotImplementedError
341 mock_plugin.__class__ = type("FakeDomain", (), {})
342 entry = _build_entry("fake", mock_plugin, None)
343 assert "schema" not in entry
344 assert entry["active"] is False
345
346 def test_module_path_included(self) -> None:
347 from muse.cli.commands.domains import _build_entry
348 from muse.plugins.registry import _REGISTRY
349
350 plugin = _REGISTRY["scaffold"]
351 entry = _build_entry("scaffold", plugin, None)
352 assert entry["module_path"] == "plugins/scaffold/plugin.py"
353
354 def test_active_false_for_inactive(self) -> None:
355 from muse.cli.commands.domains import _build_entry
356 from muse.plugins.registry import _REGISTRY
357
358 plugin = _REGISTRY["code"]
359 entry = _build_entry("code", plugin, "scaffold")
360 assert entry["active"] is False
361
362 def test_capabilities_list_of_strings(self) -> None:
363 from muse.cli.commands.domains import _build_entry
364 from muse.plugins.registry import _REGISTRY
365
366 plugin = _REGISTRY["code"]
367 entry = _build_entry("code", plugin, None)
368 caps = entry["capabilities"]
369 assert isinstance(caps, list)
370 assert all(isinstance(c, str) for c in caps)
371 assert "Typed Deltas" in caps
372
373
374 # ---------------------------------------------------------------------------
375 # Unit — _post_json
376 # ---------------------------------------------------------------------------
377
378
379 def _make_caps_payload() -> "_PublishPayload":
380 from muse.cli.commands.domains import _PublishPayload, _Capabilities, _DimensionDef
381
382 caps = _Capabilities(
383 dimensions=[_DimensionDef(name="notes", description="Note events")],
384 artifact_types=["mid"],
385 merge_semantics="three_way",
386 supported_commands=["commit"],
387 )
388 return _PublishPayload(
389 author_slug="user",
390 slug="music",
391 display_name="Music",
392 description="Desc",
393 capabilities=caps,
394 viewer_type="midi",
395 version="0.1.0",
396 )
397
398
399 def _make_signing() -> "SigningIdentity":
400 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
401 from muse.core.transport import SigningIdentity
402 return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate())
403
404
405 class TestPostJson:
406 def test_correct_method_and_headers(self) -> None:
407 from muse.cli.commands.domains import _post_json
408
409 captured: list[urllib.request.Request] = []
410
411 def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock:
412 captured.append(req)
413 resp = MagicMock()
414 resp.read = MagicMock(return_value=b'{"domain_id":"1","scoped_id":"@u/s","manifest_hash":"abc"}')
415 resp.__enter__ = lambda s: s
416 resp.__exit__ = MagicMock(return_value=False)
417 return resp
418
419 with patch("urllib.request.urlopen", fake_urlopen):
420 result = _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing())
421
422 assert len(captured) == 1
423 req = captured[0]
424 assert req.get_method() == "POST"
425 assert req.get_header("Content-type") == "application/json"
426 assert req.get_header("Authorization").startswith("MSign ")
427 assert result["scoped_id"] == "@u/s"
428
429 def test_response_size_cap_applied(self) -> None:
430 """resp.read() is called with _MAX_RESPONSE_BYTES, not unlimited."""
431 from muse.cli.commands.domains import _post_json, _MAX_RESPONSE_BYTES
432
433 read_args: list[int] = []
434
435 def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock:
436 resp = MagicMock()
437
438 def _read(n: int) -> bytes:
439 read_args.append(n)
440 return b'{"domain_id":"1","scoped_id":"@u/s","manifest_hash":""}'
441
442 resp.read = _read
443 resp.__enter__ = lambda s: s
444 resp.__exit__ = MagicMock(return_value=False)
445 return resp
446
447 with patch("urllib.request.urlopen", fake_urlopen):
448 _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing())
449
450 assert read_args == [_MAX_RESPONSE_BYTES]
451
452 def test_non_object_json_raises_value_error(self) -> None:
453 from muse.cli.commands.domains import _post_json
454
455 def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock:
456 resp = MagicMock()
457 resp.read = MagicMock(return_value=b"[1,2,3]")
458 resp.__enter__ = lambda s: s
459 resp.__exit__ = MagicMock(return_value=False)
460 return resp
461
462 with patch("urllib.request.urlopen", fake_urlopen):
463 with pytest.raises(ValueError, match="Expected JSON object"):
464 _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing())
465
466 def test_missing_keys_normalised_to_empty_string(self) -> None:
467 from muse.cli.commands.domains import _post_json
468
469 def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock:
470 resp = MagicMock()
471 resp.read = MagicMock(return_value=b"{}")
472 resp.__enter__ = lambda s: s
473 resp.__exit__ = MagicMock(return_value=False)
474 return resp
475
476 with patch("urllib.request.urlopen", fake_urlopen):
477 result = _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing())
478
479 assert result["domain_id"] == ""
480 assert result["scoped_id"] == ""
481 assert result["manifest_hash"] == ""
482
483
484 # ---------------------------------------------------------------------------
485 # Unit — _run_validate_plugin
486 # ---------------------------------------------------------------------------
487
488
489 class _MinimalPlugin:
490 """Stub implementing all required MuseDomainPlugin methods; schema raises NotImplementedError."""
491
492 def snapshot(self, live_state: LiveState) -> StateSnapshot:
493 raise NotImplementedError
494
495 def diff(
496 self,
497 base: StateSnapshot,
498 target: StateSnapshot,
499 *,
500 repo_root: pathlib.Path | None = None,
501 ) -> StateDelta:
502 raise NotImplementedError
503
504 def merge(
505 self,
506 base: StateSnapshot,
507 left: StateSnapshot,
508 right: StateSnapshot,
509 *,
510 repo_root: pathlib.Path | None = None,
511 ) -> MergeResult:
512 raise NotImplementedError
513
514 def drift(self, committed: StateSnapshot, live: LiveState) -> DriftReport:
515 raise NotImplementedError
516
517 def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState:
518 raise NotImplementedError
519
520 def schema(self) -> DomainSchema:
521 raise NotImplementedError
522
523
524 class TestRunValidatePlugin:
525 def test_all_checks_pass_for_code_plugin(self) -> None:
526 from muse.cli.commands.domains import _run_validate_plugin
527 from muse.plugins.registry import _REGISTRY
528
529 result = _run_validate_plugin("code", _REGISTRY["code"], None)
530 assert result["ok"] is True
531 assert result["domain"] == "code"
532 names = [c["name"] for c in result["checks"]]
533 assert "has_method:snapshot" in names
534 assert "schema()" in names
535
536 def test_schema_not_implemented_flagged(self) -> None:
537 """A plugin that has all methods but raises NotImplementedError for schema()
538 must have the schema check fail while all method checks pass."""
539 from muse.cli.commands.domains import _run_validate_plugin
540
541 result = _run_validate_plugin("min", _MinimalPlugin(), None)
542 schema_check = next(c for c in result["checks"] if c["name"] == "schema()")
543 assert schema_check["ok"] is False
544 # All protocol method checks should pass since methods exist and are callable
545 method_checks = [c for c in result["checks"] if c["name"].startswith("has_method:")]
546 assert all(c["ok"] for c in method_checks)
547
548 def test_scaffold_plugin_all_pass(self) -> None:
549 from muse.cli.commands.domains import _run_validate_plugin
550 from muse.plugins.registry import _REGISTRY
551
552 result = _run_validate_plugin("scaffold", _REGISTRY["scaffold"], None)
553 assert result["ok"] is True, [c for c in result["checks"] if not c["ok"]]
554
555
556 # ---------------------------------------------------------------------------
557 # Integration — muse domains (dashboard + --json)
558 # ---------------------------------------------------------------------------
559
560
561 class TestDomainsDashboard:
562 def test_default_text_output_has_registered_domains(self) -> None:
563 result = runner.invoke(cli, ["domains"])
564 assert result.exit_code == 0
565 assert "Registered domains:" in result.output
566
567 def test_json_flag_emits_list(self) -> None:
568 result = runner.invoke(cli, ["domains", "--json"])
569 assert result.exit_code == 0
570 domains = _parse_domains_list(result)
571 assert len(domains) >= 1
572
573 def test_json_active_field_is_boolean(self) -> None:
574 result = runner.invoke(cli, ["domains", "--json"])
575 assert result.exit_code == 0
576 for entry in _parse_domains_list(result):
577 assert isinstance(entry.get("active"), bool), (
578 f"'active' must be bool, got {type(entry.get('active')).__name__}"
579 )
580
581 def test_json_module_path_present(self) -> None:
582 result = runner.invoke(cli, ["domains", "--json"])
583 assert result.exit_code == 0
584 for entry in _parse_domains_list(result):
585 assert "module_path" in entry, f"missing 'module_path' in {entry}"
586
587 def test_json_schema_present_for_code(self) -> None:
588 result = runner.invoke(cli, ["domains", "--json"])
589 assert result.exit_code == 0
590 code_entry = next(e for e in _parse_domains_list(result) if e.get("domain") == "code")
591 assert "schema" in code_entry
592 schema = code_entry["schema"]
593 assert "dimensions" in schema
594
595 def test_json_no_string_true_false(self) -> None:
596 """Agents must never see 'active': 'true' — only 'active': true."""
597 result = runner.invoke(cli, ["domains", "--json"])
598 raw = result.output
599 assert '"active": "true"' not in raw
600 assert '"active": "false"' not in raw
601
602 def test_text_has_muse_domains_new_hint(self) -> None:
603 result = runner.invoke(cli, ["domains"])
604 assert "muse domains --new" in result.output
605
606 def test_text_has_info_hint(self) -> None:
607 result = runner.invoke(cli, ["domains"])
608 assert "muse domains info" in result.output
609
610 def test_text_has_validate_hint(self) -> None:
611 result = runner.invoke(cli, ["domains"])
612 assert "muse domains validate" in result.output
613
614
615 # ---------------------------------------------------------------------------
616 # Integration — muse domains --new
617 # ---------------------------------------------------------------------------
618
619
620 class TestDomainsNew:
621 def test_valid_name_creates_directory(self) -> None:
622 plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins"
623 dest = plugins_dir / "testdomain9"
624 try:
625 result = runner.invoke(cli, ["domains", "--new", "testdomain9"])
626 assert result.exit_code == 0, result.output
627 assert dest.exists()
628 assert (dest / "plugin.py").exists()
629 finally:
630 if dest.exists():
631 shutil.rmtree(str(dest))
632
633 def test_valid_name_json_flag_emits_scaffold_json(self) -> None:
634 plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins"
635 dest = plugins_dir / "testdomain10"
636 try:
637 result = runner.invoke(cli, ["domains", "--new", "testdomain10", "--json"])
638 assert result.exit_code == 0, result.output
639 data = _parse_scaffold(result)
640 assert data["name"] == "testdomain10"
641 assert data["status"] == "ok"
642 assert "class_name" in data
643 assert "path" in data
644 finally:
645 if dest.exists():
646 shutil.rmtree(str(dest))
647
648 def test_path_traversal_rejected(self) -> None:
649 result = runner.invoke(cli, ["domains", "--new", "../evil"])
650 assert result.exit_code != 0
651
652 def test_uppercase_name_rejected(self) -> None:
653 result = runner.invoke(cli, ["domains", "--new", "Genomics"])
654 assert result.exit_code != 0
655
656 def test_scaffold_reserved_rejected(self) -> None:
657 result = runner.invoke(cli, ["domains", "--new", "scaffold"])
658 assert result.exit_code != 0
659
660 def test_duplicate_name_rejected(self) -> None:
661 result = runner.invoke(cli, ["domains", "--new", "code"])
662 assert result.exit_code != 0
663
664 def test_no_pycache_in_created_directory(self) -> None:
665 plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins"
666 dest = plugins_dir / "testdomain11"
667 try:
668 result = runner.invoke(cli, ["domains", "--new", "testdomain11"])
669 assert result.exit_code == 0
670 pycache = dest / "__pycache__"
671 assert not pycache.exists(), "__pycache__ must not be copied"
672 finally:
673 if dest.exists():
674 shutil.rmtree(str(dest))
675
676 def test_class_name_substituted(self) -> None:
677 plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins"
678 dest = plugins_dir / "testdomain12"
679 try:
680 runner.invoke(cli, ["domains", "--new", "testdomain12"])
681 plugin_src = (dest / "plugin.py").read_text(encoding="utf-8")
682 assert "Testdomain12Plugin" in plugin_src
683 assert "ScaffoldPlugin" not in plugin_src
684 finally:
685 if dest.exists():
686 shutil.rmtree(str(dest))
687
688
689 # ---------------------------------------------------------------------------
690 # Integration — muse domains info
691 # ---------------------------------------------------------------------------
692
693
694 class TestDomainsInfo:
695 def test_known_domain_exits_zero(self) -> None:
696 result = runner.invoke(cli, ["domains", "info", "code"])
697 assert result.exit_code == 0
698
699 def test_unknown_domain_exits_nonzero(self) -> None:
700 result = runner.invoke(cli, ["domains", "info", "doesnotexist"])
701 assert result.exit_code != 0
702
703 def test_json_schema_has_required_keys(self) -> None:
704 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
705 assert result.exit_code == 0
706 data = _parse_domain_entry(result)
707 assert data["domain"] == "code"
708 assert isinstance(data["active"], bool)
709 assert "capabilities" in data
710 assert "module_path" in data
711
712 def test_scaffold_has_crdt_capability(self) -> None:
713 result = runner.invoke(cli, ["domains", "info", "scaffold", "--json"])
714 assert result.exit_code == 0
715 data = _parse_domain_entry(result)
716 assert "CRDT" in data["capabilities"]
717
718 def test_text_output_includes_module_path(self) -> None:
719 result = runner.invoke(cli, ["domains", "info", "code"])
720 assert "Module:" in result.output
721 assert "plugins/code/plugin.py" in result.output
722
723 def test_error_on_unknown_domain(self) -> None:
724 result = runner.invoke(cli, ["domains", "info", "ghost"])
725 assert result.exit_code != 0
726 assert "not registered" in result.output
727
728
729 # ---------------------------------------------------------------------------
730 # Integration — muse domains use
731 # ---------------------------------------------------------------------------
732
733
734 class TestDomainsUse:
735 def test_no_repo_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
736 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
737 result = runner.invoke(cli, ["domains", "use", "code"])
738 assert result.exit_code != 0
739
740 def test_unknown_domain_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
741 _init_repo(tmp_path)
742 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
743 result = runner.invoke(cli, ["domains", "use", "doesnotexist"])
744 assert result.exit_code != 0
745
746 def test_known_domain_switches_repo_json(self, tmp_path: pathlib.Path) -> None:
747 _init_repo(tmp_path, domain="scaffold")
748 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
749 result = runner.invoke(cli, ["domains", "use", "code"])
750 assert result.exit_code == 0, result.output
751 data: Manifest = json.loads((tmp_path / ".muse" / "repo.json").read_text(encoding="utf-8"))
752 assert data["domain"] == "code"
753
754 def test_switch_is_idempotent(self, tmp_path: pathlib.Path) -> None:
755 _init_repo(tmp_path, domain="code")
756 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
757 result = runner.invoke(cli, ["domains", "use", "code"])
758 assert result.exit_code == 0
759
760 def test_json_output_has_required_keys(self, tmp_path: pathlib.Path) -> None:
761 _init_repo(tmp_path, domain="scaffold")
762 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
763 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
764 assert result.exit_code == 0, result.output
765 data = _parse_use(result)
766 assert data["domain"] == "code"
767 assert data["status"] == "switched"
768 assert "repo" in data
769
770 def test_existing_repo_fields_preserved(self, tmp_path: pathlib.Path) -> None:
771 _init_repo(tmp_path, domain="scaffold")
772 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
773 runner.invoke(cli, ["domains", "use", "code"])
774 data: Manifest = json.loads((tmp_path / ".muse" / "repo.json").read_text(encoding="utf-8"))
775 assert "repo_id" in data
776 assert "schema_version" in data
777
778
779 # ---------------------------------------------------------------------------
780 # Integration — muse domains validate
781 # ---------------------------------------------------------------------------
782
783
784 class TestDomainsValidate:
785 def test_code_plugin_passes(self) -> None:
786 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
787 result = runner.invoke(cli, ["domains", "validate", "code"])
788 assert result.exit_code == 0
789
790 def test_scaffold_plugin_passes(self) -> None:
791 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
792 result = runner.invoke(cli, ["domains", "validate", "scaffold"])
793 assert result.exit_code == 0
794
795 def test_unknown_domain_exits_nonzero(self) -> None:
796 result = runner.invoke(cli, ["domains", "validate", "ghost"])
797 assert result.exit_code != 0
798
799 def test_json_ok_field_is_boolean(self) -> None:
800 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
801 result = runner.invoke(cli, ["domains", "validate", "code", "--json"])
802 assert result.exit_code == 0
803 data = _parse_validate(result)
804 assert isinstance(data["ok"], bool)
805 assert data["ok"] is True
806
807 def test_json_checks_list_present(self) -> None:
808 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
809 result = runner.invoke(cli, ["domains", "validate", "scaffold", "--json"])
810 data = _parse_validate(result)
811 assert isinstance(data["checks"], list)
812 assert len(data["checks"]) >= 5
813
814 def test_broken_plugin_exits_nonzero(self) -> None:
815 from muse.plugins.registry import _REGISTRY
816
817 mock_plugin = MagicMock(spec=[]) # no methods at all
818 with patch.dict(_REGISTRY, {"broken": mock_plugin}):
819 result = runner.invoke(cli, ["domains", "validate", "broken"])
820 assert result.exit_code != 0
821
822 def test_no_name_validates_active_repo_domain(self, tmp_path: pathlib.Path) -> None:
823 _init_repo(tmp_path, domain="code")
824 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
825 result = runner.invoke(cli, ["domains", "validate"])
826 assert result.exit_code == 0
827 assert "code" in result.output
828
829 def test_no_name_no_repo_validates_all(self) -> None:
830 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
831 result = runner.invoke(cli, ["domains", "validate", "--json"])
832 assert result.exit_code == 0
833 from muse.plugins.registry import _REGISTRY
834 if len(_REGISTRY) > 1:
835 _parse_validate_list(result)
836 else:
837 _parse_validate(result)
838
839
840 # ---------------------------------------------------------------------------
841 # Security
842 # ---------------------------------------------------------------------------
843
844
845 class TestDomainsSecurity:
846 def test_ansi_in_domain_name_sanitized_in_dashboard(self) -> None:
847 """A registry entry with ANSI in its name must not bleed into output."""
848 from muse.plugins.registry import _REGISTRY
849 from muse.plugins.scaffold.plugin import ScaffoldPlugin
850
851 evil_name = "\x1b[31mevil\x1b[0m"
852 mock = ScaffoldPlugin()
853 with patch.dict(_REGISTRY, {evil_name: mock}):
854 result = runner.invoke(cli, ["domains"])
855 assert "\x1b[31m" not in result.output
856
857 def test_file_scheme_publish_hub_rejected(self, tmp_path: pathlib.Path) -> None:
858 """file:// hub URL must be blocked before any network call."""
859 urlopen_called = False
860
861 def _urlopen(req: urllib.request.Request, timeout: float) -> MagicMock:
862 nonlocal urlopen_called
863 urlopen_called = True
864 raise AssertionError("urlopen must not be called with file:// URL")
865
866 _init_repo(tmp_path, domain="code")
867
868 with ExitStack() as stack:
869 stack.enter_context(patch("urllib.request.urlopen", _urlopen))
870 stack.enter_context(
871 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
872 )
873 stack.enter_context(
874 patch("muse.cli.commands.domains.get_signing_identity", return_value="tok")
875 )
876 result = runner.invoke(cli, [
877 "domains", "publish",
878 "--author", "user", "--slug", "music",
879 "--name", "Music", "--description", "Desc",
880 "--viewer-type", "midi",
881 "--hub", "file:///etc/passwd",
882 "--capabilities", '{"dimensions":[],"artifact_types":[],"merge_semantics":"three_way","supported_commands":[]}',
883 ])
884
885 assert result.exit_code != 0
886 assert not urlopen_called
887
888 def test_server_ansi_in_scoped_id_sanitized(self, tmp_path: pathlib.Path) -> None:
889 """Server-returned scoped_id with ANSI (unicode-escaped in JSON) must not appear raw."""
890 _init_repo(tmp_path, domain="code")
891 # JSON uses \u001b for the ESC character — valid JSON but contains ANSI.
892 evil_response = b'{"domain_id":"1","scoped_id":"\\u001b[31mhacked\\u001b[0m","manifest_hash":""}'
893
894 def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock:
895 resp = MagicMock()
896 resp.read = MagicMock(return_value=evil_response)
897 resp.__enter__ = lambda s: s
898 resp.__exit__ = MagicMock(return_value=False)
899 return resp
900
901 with ExitStack() as stack:
902 stack.enter_context(patch("urllib.request.urlopen", fake_urlopen))
903 stack.enter_context(
904 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
905 )
906 stack.enter_context(
907 patch("muse.cli.commands.domains.get_signing_identity", return_value=_make_signing())
908 )
909 result = runner.invoke(cli, [
910 "domains", "publish",
911 "--author", "user", "--slug", "music",
912 "--name", "Music", "--description", "Desc",
913 "--viewer-type", "midi",
914 "--capabilities", '{"dimensions":[],"artifact_types":[],"merge_semantics":"three_way","supported_commands":[]}',
915 ])
916
917 assert result.exit_code == 0
918 assert "\x1b[31m" not in result.output
919
920 def test_path_traversal_new_rejected_without_touching_fs(self) -> None:
921 evil = "../" * 5 + "evil"
922 evil_path = pathlib.Path(__file__).parents[1] / "muse" / "plugins" / evil
923 result = runner.invoke(cli, ["domains", "--new", evil])
924 assert result.exit_code != 0
925 assert not evil_path.exists()
926
927 def test_null_byte_name_rejected(self) -> None:
928 result = runner.invoke(cli, ["domains", "--new", "evil\x00name"])
929 assert result.exit_code != 0
930
931
932 # ---------------------------------------------------------------------------
933 # E2E — verify the muse CLI entry point wires domains correctly
934 # ---------------------------------------------------------------------------
935
936
937 class TestDomainsE2E:
938 def test_domains_help_exits_zero(self) -> None:
939 result = runner.invoke(cli, ["domains", "--help"])
940 assert result.exit_code == 0
941 assert "domains" in result.output.lower()
942
943 def test_domains_info_help(self) -> None:
944 result = runner.invoke(cli, ["domains", "info", "--help"])
945 assert result.exit_code == 0
946
947 def test_domains_use_help(self) -> None:
948 result = runner.invoke(cli, ["domains", "use", "--help"])
949 assert result.exit_code == 0
950
951 def test_domains_validate_help(self) -> None:
952 result = runner.invoke(cli, ["domains", "validate", "--help"])
953 assert result.exit_code == 0
954
955 def test_domains_publish_help(self) -> None:
956 result = runner.invoke(cli, ["domains", "publish", "--help"])
957 assert result.exit_code == 0
958
959 def test_domains_json_is_valid_json(self) -> None:
960 result = runner.invoke(cli, ["domains", "--json"])
961 assert result.exit_code == 0
962 domains = _parse_domains_list(result)
963 assert len(domains) >= 1
964
965 def test_info_json_is_valid_json(self) -> None:
966 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
967 assert result.exit_code == 0
968 _parse_domain_entry(result) # must not raise
969
970 def test_validate_json_is_valid_json(self) -> None:
971 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
972 result = runner.invoke(cli, ["domains", "validate", "code", "--json"])
973 assert result.exit_code == 0
974 _parse_validate(result)
975
976 def test_use_requires_name_arg(self) -> None:
977 result = runner.invoke(cli, ["domains", "use"])
978 assert result.exit_code != 0
979
980
981 # ---------------------------------------------------------------------------
982 # Publish E2E — happy path and error paths
983 # ---------------------------------------------------------------------------
984
985
986 class TestPublishE2E:
987 _CAPS = json.dumps({
988 "dimensions": [{"name": "notes", "description": "Note events"}],
989 "artifact_types": ["mid"],
990 "merge_semantics": "three_way",
991 "supported_commands": ["commit", "diff"],
992 })
993
994 _BASE_ARGS = [
995 "domains", "publish",
996 "--author", "user", "--slug", "music",
997 "--name", "Music", "--description", "Desc",
998 "--viewer-type", "midi",
999 "--capabilities", _CAPS,
1000 ]
1001
1002 def test_successful_publish_text_output(self, tmp_path: pathlib.Path) -> None:
1003 _init_repo(tmp_path, domain="code")
1004 fake_resp = _PublishResponse(domain_id="1", scoped_id="@user/music", manifest_hash="abc")
1005
1006 with ExitStack() as stack:
1007 stack.enter_context(
1008 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
1009 )
1010 stack.enter_context(
1011 patch("muse.cli.commands.domains.get_signing_identity", return_value="tok")
1012 )
1013 stack.enter_context(
1014 patch("muse.cli.commands.domains._post_json", return_value=fake_resp)
1015 )
1016 result = runner.invoke(cli, self._BASE_ARGS)
1017
1018 assert result.exit_code == 0, result.output
1019 assert "@user/music" in result.output
1020
1021 def test_successful_publish_json_output(self, tmp_path: pathlib.Path) -> None:
1022 _init_repo(tmp_path, domain="code")
1023 fake_resp = _PublishResponse(domain_id="1", scoped_id="@user/music", manifest_hash="abc")
1024
1025 with ExitStack() as stack:
1026 stack.enter_context(
1027 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
1028 )
1029 stack.enter_context(
1030 patch("muse.cli.commands.domains.get_signing_identity", return_value="tok")
1031 )
1032 stack.enter_context(
1033 patch("muse.cli.commands.domains._post_json", return_value=fake_resp)
1034 )
1035 result = runner.invoke(cli, [*self._BASE_ARGS, "--json"])
1036
1037 assert result.exit_code == 0
1038 data = _parse_publish(result)
1039 assert data["scoped_id"] == "@user/music"
1040
1041 def test_no_token_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
1042 _init_repo(tmp_path, domain="code")
1043 with ExitStack() as stack:
1044 stack.enter_context(
1045 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
1046 )
1047 stack.enter_context(
1048 patch("muse.cli.commands.domains.get_signing_identity", return_value=None)
1049 )
1050 result = runner.invoke(cli, self._BASE_ARGS)
1051
1052 assert result.exit_code != 0
1053
1054 def test_http_409_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
1055 _init_repo(tmp_path, domain="code")
1056 exc = urllib.error.HTTPError(
1057 url="", code=409, msg="Conflict",
1058 hdrs=http.client.HTTPMessage(), fp=None,
1059 )
1060
1061 with ExitStack() as stack:
1062 stack.enter_context(
1063 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
1064 )
1065 stack.enter_context(
1066 patch("muse.cli.commands.domains.get_signing_identity", return_value="tok")
1067 )
1068 stack.enter_context(
1069 patch("muse.cli.commands.domains._post_json", side_effect=exc)
1070 )
1071 result = runner.invoke(cli, self._BASE_ARGS)
1072
1073 assert result.exit_code != 0
1074
1075 def test_http_401_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
1076 _init_repo(tmp_path, domain="code")
1077 exc = urllib.error.HTTPError(
1078 url="", code=401, msg="Unauthorized",
1079 hdrs=http.client.HTTPMessage(), fp=None,
1080 )
1081
1082 with ExitStack() as stack:
1083 stack.enter_context(
1084 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
1085 )
1086 stack.enter_context(
1087 patch("muse.cli.commands.domains.get_signing_identity", return_value="tok")
1088 )
1089 stack.enter_context(
1090 patch("muse.cli.commands.domains._post_json", side_effect=exc)
1091 )
1092 result = runner.invoke(cli, self._BASE_ARGS)
1093
1094 assert result.exit_code != 0
1095
1096 def test_url_error_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
1097 _init_repo(tmp_path, domain="code")
1098 exc = urllib.error.URLError("connection refused")
1099
1100 with ExitStack() as stack:
1101 stack.enter_context(
1102 patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path)
1103 )
1104 stack.enter_context(
1105 patch("muse.cli.commands.domains.get_signing_identity", return_value="tok")
1106 )
1107 stack.enter_context(
1108 patch("muse.cli.commands.domains._post_json", side_effect=exc)
1109 )
1110 result = runner.invoke(cli, self._BASE_ARGS)
1111
1112 assert result.exit_code != 0
1113
1114
1115 # ---------------------------------------------------------------------------
1116 # Stress tests
1117 # ---------------------------------------------------------------------------
1118
1119
1120 class TestStress:
1121 def test_8_concurrent_validate_calls_isolated(self) -> None:
1122 """Concurrent validate calls on independent plugin instances must not interfere."""
1123 from muse.cli.commands.domains import _run_validate_plugin
1124 from muse.plugins.registry import _REGISTRY
1125
1126 errors: list[str] = []
1127
1128 def _validate(idx: int) -> None:
1129 try:
1130 plugin = _REGISTRY["code"]
1131 result = _run_validate_plugin("code", plugin, None)
1132 assert result["ok"] is True, f"Thread {idx}: validate failed"
1133 except Exception as exc:
1134 errors.append(f"Thread {idx}: {exc}")
1135
1136 threads = [threading.Thread(target=_validate, args=(i,)) for i in range(8)]
1137 for t in threads:
1138 t.start()
1139 for t in threads:
1140 t.join()
1141 assert not errors, f"Concurrent validate failures: {errors}"
1142
1143 def test_8_concurrent_active_domain_reads(self, tmp_path: pathlib.Path) -> None:
1144 """Concurrent _active_domain reads on isolated dirs must all return correctly."""
1145 from muse.cli.commands.domains import _active_domain
1146
1147 roots: list[pathlib.Path] = []
1148 for i in range(8):
1149 rd = tmp_path / f"repo{i}"
1150 _init_repo(rd, domain="code")
1151 roots.append(rd)
1152
1153 errors: list[str] = []
1154
1155 def _read(idx: int) -> None:
1156 try:
1157 result = _active_domain(roots[idx])
1158 assert result == "code", f"Thread {idx}: expected 'code', got {result!r}"
1159 except Exception as exc:
1160 errors.append(f"Thread {idx}: {exc}")
1161
1162 threads = [threading.Thread(target=_read, args=(i,)) for i in range(8)]
1163 for t in threads:
1164 t.start()
1165 for t in threads:
1166 t.join()
1167 assert not errors, f"Concurrent read failures: {errors}"
1168
1169 def test_8_concurrent_build_entry_calls(self) -> None:
1170 """_build_entry must be re-entrant (no shared mutable state)."""
1171 from muse.cli.commands.domains import _build_entry
1172 from muse.plugins.registry import _REGISTRY
1173
1174 errors: list[str] = []
1175
1176 def _build(idx: int) -> None:
1177 try:
1178 plugin = _REGISTRY["scaffold"]
1179 entry = _build_entry("scaffold", plugin, "code" if idx % 2 == 0 else "scaffold")
1180 assert entry["domain"] == "scaffold"
1181 assert isinstance(entry["active"], bool)
1182 except Exception as exc:
1183 errors.append(f"Thread {idx}: {exc}")
1184
1185 threads = [threading.Thread(target=_build, args=(i,)) for i in range(8)]
1186 for t in threads:
1187 t.start()
1188 for t in threads:
1189 t.join()
1190 assert not errors, f"Concurrent build_entry failures: {errors}"
1191
1192
1193 # ---------------------------------------------------------------------------
1194 # Extended — muse domains info (deeper coverage)
1195 # ---------------------------------------------------------------------------
1196
1197
1198 class TestDomainsInfoExtended:
1199 def test_j_alias_works(self) -> None:
1200 result = runner.invoke(cli, ["domains", "info", "code", "-j"])
1201 assert result.exit_code == 0
1202 data = json.loads(result.output.strip())
1203 assert data["domain"] == "code"
1204
1205 def test_help_mentions_json_flag(self) -> None:
1206 result = runner.invoke(cli, ["domains", "info", "--help"])
1207 assert "--json" in result.output or "-j" in result.output
1208
1209 def test_help_shows_exit_codes(self) -> None:
1210 result = runner.invoke(cli, ["domains", "info", "--help"])
1211 assert "Exit code" in result.output or "exit code" in result.output
1212
1213 def test_json_is_compact_single_line(self) -> None:
1214 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1215 assert result.exit_code == 0
1216 lines = [l for l in result.output.splitlines() if l.strip()]
1217 assert len(lines) == 1, f"Expected 1 non-empty line, got {len(lines)}"
1218
1219 def test_json_parses_cleanly(self) -> None:
1220 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1221 data = json.loads(result.output.strip())
1222 assert isinstance(data, dict)
1223
1224 def test_json_domain_field_matches_arg(self) -> None:
1225 result = runner.invoke(cli, ["domains", "info", "scaffold", "--json"])
1226 assert result.exit_code == 0
1227 data = json.loads(result.output.strip())
1228 assert data["domain"] == "scaffold"
1229
1230 def test_json_active_is_bool(self) -> None:
1231 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1232 data = json.loads(result.output.strip())
1233 assert isinstance(data["active"], bool)
1234
1235 def test_json_capabilities_is_list(self) -> None:
1236 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1237 data = json.loads(result.output.strip())
1238 assert isinstance(data["capabilities"], list)
1239 assert len(data["capabilities"]) >= 1
1240
1241 def test_json_module_path_is_string(self) -> None:
1242 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1243 data = json.loads(result.output.strip())
1244 assert isinstance(data["module_path"], str)
1245 assert len(data["module_path"]) > 0
1246
1247 def test_json_code_has_typed_deltas(self) -> None:
1248 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1249 data = json.loads(result.output.strip())
1250 assert "Typed Deltas" in data["capabilities"]
1251
1252 def test_json_code_schema_present(self) -> None:
1253 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1254 data = json.loads(result.output.strip())
1255 assert "schema" in data
1256
1257 def test_json_schema_has_required_keys(self) -> None:
1258 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1259 data = json.loads(result.output.strip())
1260 schema = data["schema"]
1261 for key in ("schema_version", "merge_mode", "description", "dimensions"):
1262 assert key in schema, f"Missing schema key: {key}"
1263
1264 def test_json_schema_dimensions_is_list(self) -> None:
1265 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1266 data = json.loads(result.output.strip())
1267 assert isinstance(data["schema"]["dimensions"], list)
1268 assert len(data["schema"]["dimensions"]) >= 1
1269
1270 def test_json_schema_dimension_has_name_and_description(self) -> None:
1271 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1272 data = json.loads(result.output.strip())
1273 dim = data["schema"]["dimensions"][0]
1274 assert "name" in dim
1275 assert "description" in dim
1276
1277 def test_text_shows_module_path(self) -> None:
1278 result = runner.invoke(cli, ["domains", "info", "code"])
1279 assert "Module:" in result.output
1280
1281 def test_text_shows_capabilities(self) -> None:
1282 result = runner.invoke(cli, ["domains", "info", "code"])
1283 assert "Capabilities:" in result.output
1284 assert "Typed Deltas" in result.output
1285
1286 def test_unknown_domain_error_mentions_name(self) -> None:
1287 result = runner.invoke(cli, ["domains", "info", "nonexistent_domain_xyz"])
1288 assert result.exit_code == 1
1289 assert "nonexistent_domain_xyz" in result.output or "not registered" in result.output
1290
1291 def test_unknown_domain_lists_known_domains(self) -> None:
1292 result = runner.invoke(cli, ["domains", "info", "nope"])
1293 assert result.exit_code == 1
1294 # Should mention at least one known domain (e.g. "code")
1295 assert "code" in result.output
1296
1297
1298 # ---------------------------------------------------------------------------
1299 # Security — muse domains info
1300 # ---------------------------------------------------------------------------
1301
1302
1303 class TestDomainsInfoSecurity:
1304 def test_ansi_in_domain_name_stripped_error(self) -> None:
1305 """ANSI in the unknown domain name is stripped from the error message."""
1306 result = runner.invoke(cli, ["domains", "info", "\x1b[31mevil\x1b[0m"])
1307 assert result.exit_code == 1
1308 assert "\x1b[" not in result.output
1309
1310 def test_ansi_in_module_path_stripped_text(self) -> None:
1311 """module_path with ANSI injected by a plugin is sanitized in text output."""
1312 from unittest.mock import patch as _patch
1313 evil_entry = {
1314 "domain": "code",
1315 "module_path": "plugins/\x1b[31mevil\x1b[0m/plugin.py",
1316 "capabilities": ["Typed Deltas"],
1317 "active": False,
1318 }
1319 with _patch("muse.cli.commands.domains._build_entry", return_value=evil_entry):
1320 result = runner.invoke(cli, ["domains", "info", "code"])
1321 assert result.exit_code == 0
1322 assert "\x1b[" not in result.output
1323
1324 def test_ansi_in_schema_description_stripped_text(self) -> None:
1325 """ANSI in schema description from a plugin is sanitized in text output."""
1326 from unittest.mock import patch as _patch, MagicMock
1327 evil_entry = {
1328 "domain": "code",
1329 "module_path": "plugins/code/plugin.py",
1330 "capabilities": ["Typed Deltas", "Domain Schema"],
1331 "active": False,
1332 "schema": {
1333 "schema_version": "1.0",
1334 "merge_mode": "structured",
1335 "description": "Good domain \x1b[31mevil\x1b[0m",
1336 "dimensions": [{"name": "symbols", "description": "sym"}],
1337 },
1338 }
1339 with _patch("muse.cli.commands.domains._build_entry", return_value=evil_entry):
1340 result = runner.invoke(cli, ["domains", "info", "code"])
1341 assert result.exit_code == 0
1342 assert "\x1b[" not in result.output
1343
1344 def test_ansi_in_dimension_name_stripped_text(self) -> None:
1345 """ANSI in a dimension name from a plugin schema is sanitized in text output."""
1346 from unittest.mock import patch as _patch
1347 evil_entry = {
1348 "domain": "code",
1349 "module_path": "plugins/code/plugin.py",
1350 "capabilities": ["Typed Deltas", "Domain Schema"],
1351 "active": False,
1352 "schema": {
1353 "schema_version": "1.0",
1354 "merge_mode": "structured",
1355 "description": "desc",
1356 "dimensions": [{"name": "\x1b[31mevil\x1b[0m", "description": "bad"}],
1357 },
1358 }
1359 with _patch("muse.cli.commands.domains._build_entry", return_value=evil_entry):
1360 result = runner.invoke(cli, ["domains", "info", "code"])
1361 assert result.exit_code == 0
1362 assert "\x1b[" not in result.output
1363
1364 def test_json_stdout_only_no_stray_text(self) -> None:
1365 """In JSON mode, stdout must be parseable JSON with no surrounding noise."""
1366 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1367 assert result.exit_code == 0
1368 json.loads(result.output.strip())
1369
1370 def test_unknown_domain_no_traceback(self) -> None:
1371 result = runner.invoke(cli, ["domains", "info", "no_such_domain_xyz"])
1372 assert result.exit_code == 1
1373 assert "Traceback" not in result.output
1374
1375
1376 # ---------------------------------------------------------------------------
1377 # Stress — muse domains info
1378 # ---------------------------------------------------------------------------
1379
1380
1381 class TestDomainsInfoStress:
1382 def test_50_sequential_info_calls(self) -> None:
1383 """50 sequential info invocations on 'code' all exit 0."""
1384 for i in range(50):
1385 result = runner.invoke(cli, ["domains", "info", "code", "--json"])
1386 assert result.exit_code == 0, f"Call {i} failed: {result.output}"
1387
1388 def test_all_registered_domains_info_exits_zero(self) -> None:
1389 """Every domain currently in the registry responds to info with exit 0."""
1390 from muse.plugins.registry import _REGISTRY
1391 for name in sorted(_REGISTRY):
1392 result = runner.invoke(cli, ["domains", "info", name, "--json"])
1393 assert result.exit_code == 0, f"domains info {name!r} failed: {result.output}"
1394
1395 def test_concurrent_info_calls_no_shared_state(self) -> None:
1396 """8 threads each call domains info in isolation — no race conditions."""
1397 import threading
1398 errors: list[str] = []
1399 from muse.cli.commands.domains import _build_entry, _active_domain, _find_repo_root
1400 from muse.plugins.registry import _REGISTRY
1401
1402 def worker(idx: int) -> None:
1403 try:
1404 plugin = _REGISTRY.get("code")
1405 if plugin is None:
1406 errors.append(f"Thread {idx}: code plugin not found")
1407 return
1408 active = _active_domain(_find_repo_root())
1409 entry = _build_entry("code", plugin, active)
1410 if entry["domain"] != "code":
1411 errors.append(f"Thread {idx}: wrong domain {entry['domain']!r}")
1412 if not isinstance(entry["active"], bool):
1413 errors.append(f"Thread {idx}: active is not bool")
1414 except Exception as exc:
1415 errors.append(f"Thread {idx}: {exc}")
1416
1417 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1418 for t in threads:
1419 t.start()
1420 for t in threads:
1421 t.join()
1422 assert not errors, f"Concurrent failures: {errors}"
1423
1424
1425 # ---------------------------------------------------------------------------
1426 # Extended — muse domains use (deeper coverage)
1427 # ---------------------------------------------------------------------------
1428
1429
1430 class TestDomainsUseExtended:
1431 def test_j_alias_works(self, tmp_path: pathlib.Path) -> None:
1432 _init_repo(tmp_path, domain="scaffold")
1433 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1434 result = runner.invoke(cli, ["domains", "use", "code", "-j"])
1435 assert result.exit_code == 0, result.output
1436 data = json.loads(result.output.strip())
1437 assert data["domain"] == "code"
1438
1439 def test_help_mentions_json_flag(self) -> None:
1440 result = runner.invoke(cli, ["domains", "use", "--help"])
1441 assert "--json" in result.output or "-j" in result.output
1442
1443 def test_help_shows_exit_codes(self) -> None:
1444 result = runner.invoke(cli, ["domains", "use", "--help"])
1445 assert "Exit code" in result.output or "exit code" in result.output
1446
1447 def test_json_is_compact_single_line(self, tmp_path: pathlib.Path) -> None:
1448 _init_repo(tmp_path, domain="scaffold")
1449 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1450 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
1451 assert result.exit_code == 0, result.output
1452 lines = [l for l in result.output.splitlines() if l.strip()]
1453 assert len(lines) == 1, f"Expected 1 non-empty line, got {len(lines)}"
1454
1455 def test_json_parses_cleanly(self, tmp_path: pathlib.Path) -> None:
1456 _init_repo(tmp_path)
1457 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1458 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
1459 data = json.loads(result.output.strip())
1460 assert isinstance(data, dict)
1461
1462 def test_json_domain_matches_arg(self, tmp_path: pathlib.Path) -> None:
1463 _init_repo(tmp_path, domain="code")
1464 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1465 result = runner.invoke(cli, ["domains", "use", "scaffold", "--json"])
1466 data = json.loads(result.output.strip())
1467 assert data["domain"] == "scaffold"
1468
1469 def test_json_status_is_switched(self, tmp_path: pathlib.Path) -> None:
1470 _init_repo(tmp_path)
1471 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1472 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
1473 data = json.loads(result.output.strip())
1474 assert data["status"] == "switched"
1475
1476 def test_json_repo_ends_with_muse(self, tmp_path: pathlib.Path) -> None:
1477 _init_repo(tmp_path)
1478 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1479 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
1480 data = json.loads(result.output.strip())
1481 assert data["repo"].endswith(".muse")
1482
1483 def test_repo_json_domain_actually_updated(self, tmp_path: pathlib.Path) -> None:
1484 _init_repo(tmp_path, domain="scaffold")
1485 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1486 runner.invoke(cli, ["domains", "use", "code"])
1487 written = json.loads((tmp_path / ".muse" / "repo.json").read_text())
1488 assert written["domain"] == "code"
1489
1490 def test_existing_fields_preserved(self, tmp_path: pathlib.Path) -> None:
1491 _init_repo(tmp_path, domain="scaffold")
1492 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1493 runner.invoke(cli, ["domains", "use", "code"])
1494 written = json.loads((tmp_path / ".muse" / "repo.json").read_text())
1495 assert "repo_id" in written
1496 assert "schema_version" in written
1497
1498 def test_idempotent_same_domain_exits_zero(self, tmp_path: pathlib.Path) -> None:
1499 _init_repo(tmp_path, domain="code")
1500 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1501 result = runner.invoke(cli, ["domains", "use", "code"])
1502 assert result.exit_code == 0
1503
1504 def test_switch_code_to_scaffold(self, tmp_path: pathlib.Path) -> None:
1505 _init_repo(tmp_path, domain="code")
1506 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1507 result = runner.invoke(cli, ["domains", "use", "scaffold"])
1508 assert result.exit_code == 0
1509 written = json.loads((tmp_path / ".muse" / "repo.json").read_text())
1510 assert written["domain"] == "scaffold"
1511
1512 def test_text_success_mentions_domain_name(self, tmp_path: pathlib.Path) -> None:
1513 _init_repo(tmp_path)
1514 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1515 result = runner.invoke(cli, ["domains", "use", "code"])
1516 assert "code" in result.output
1517
1518 def test_text_success_mentions_repo_path(self, tmp_path: pathlib.Path) -> None:
1519 _init_repo(tmp_path)
1520 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1521 result = runner.invoke(cli, ["domains", "use", "code"])
1522 assert "Repo:" in result.output or ".muse" in result.output
1523
1524 def test_unknown_domain_lists_known(self, tmp_path: pathlib.Path) -> None:
1525 _init_repo(tmp_path)
1526 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1527 result = runner.invoke(cli, ["domains", "use", "no_such_domain"])
1528 assert result.exit_code == 1
1529 assert "code" in result.output # known domains listed
1530
1531 def test_no_repo_exits_1(self) -> None:
1532 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1533 result = runner.invoke(cli, ["domains", "use", "code"])
1534 assert result.exit_code == 1
1535
1536 def test_corrupt_repo_json_exits_1(self, tmp_path: pathlib.Path) -> None:
1537 _init_repo(tmp_path)
1538 (tmp_path / ".muse" / "repo.json").write_text("{{bad json}}", encoding="utf-8")
1539 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1540 result = runner.invoke(cli, ["domains", "use", "code"])
1541 assert result.exit_code == 1
1542
1543 def test_extra_repo_json_fields_preserved(self, tmp_path: pathlib.Path) -> None:
1544 """Custom extra fields in repo.json survive a domain switch."""
1545 muse = tmp_path / ".muse"
1546 muse.mkdir(parents=True, exist_ok=True)
1547 (muse / "repo.json").write_text(
1548 json.dumps({
1549 "repo_id": "my-repo",
1550 "schema_version": "1.0",
1551 "domain": "scaffold",
1552 "custom_field": "keep_me",
1553 }),
1554 encoding="utf-8",
1555 )
1556 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1557 result = runner.invoke(cli, ["domains", "use", "code"])
1558 assert result.exit_code == 0
1559 written = json.loads((tmp_path / ".muse" / "repo.json").read_text())
1560 assert written["custom_field"] == "keep_me"
1561 assert written["domain"] == "code"
1562
1563
1564 # ---------------------------------------------------------------------------
1565 # Security — muse domains use
1566 # ---------------------------------------------------------------------------
1567
1568
1569 class TestDomainsUseSecurity:
1570 def test_ansi_in_unknown_domain_name_stripped(self, tmp_path: pathlib.Path) -> None:
1571 _init_repo(tmp_path)
1572 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1573 result = runner.invoke(cli, ["domains", "use", "\x1b[31mevil\x1b[0m"])
1574 assert result.exit_code == 1
1575 assert "\x1b[" not in result.output
1576
1577 def test_unregistered_domain_never_written(self, tmp_path: pathlib.Path) -> None:
1578 """An unregistered domain name is rejected before repo.json is touched."""
1579 _init_repo(tmp_path, domain="code")
1580 original = (tmp_path / ".muse" / "repo.json").read_text()
1581 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1582 result = runner.invoke(cli, ["domains", "use", "malicious_domain"])
1583 assert result.exit_code == 1
1584 # repo.json must be unchanged
1585 assert (tmp_path / ".muse" / "repo.json").read_text() == original
1586
1587 def test_domain_name_sanitized_in_success_text(self, tmp_path: pathlib.Path) -> None:
1588 """Domain name in success text is sanitized (no raw ANSI from env)."""
1589 _init_repo(tmp_path)
1590 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1591 result = runner.invoke(cli, ["domains", "use", "code"])
1592 assert result.exit_code == 0
1593 assert "\x1b[" not in result.output
1594
1595 def test_repo_path_sanitized_in_success_text(self, tmp_path: pathlib.Path) -> None:
1596 """Repo path in success text is sanitized."""
1597 _init_repo(tmp_path)
1598 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1599 result = runner.invoke(cli, ["domains", "use", "code"])
1600 assert result.exit_code == 0
1601 assert "\x1b[" not in result.output
1602
1603 def test_json_stdout_only_no_stray_text(self, tmp_path: pathlib.Path) -> None:
1604 """In JSON mode, stdout must be valid JSON and nothing else."""
1605 _init_repo(tmp_path)
1606 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1607 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
1608 assert result.exit_code == 0
1609 json.loads(result.output.strip())
1610
1611 def test_no_traceback_on_corrupt_repo_json(self, tmp_path: pathlib.Path) -> None:
1612 _init_repo(tmp_path)
1613 (tmp_path / ".muse" / "repo.json").write_text("{bad}", encoding="utf-8")
1614 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1615 result = runner.invoke(cli, ["domains", "use", "code"])
1616 assert result.exit_code == 1
1617 assert "Traceback" not in result.output
1618
1619
1620 # ---------------------------------------------------------------------------
1621 # Stress — muse domains use
1622 # ---------------------------------------------------------------------------
1623
1624
1625 class TestDomainsUseStress:
1626 def test_50_sequential_use_calls_all_succeed(self, tmp_path: pathlib.Path) -> None:
1627 _init_repo(tmp_path, domain="code")
1628 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1629 for i in range(50):
1630 result = runner.invoke(cli, ["domains", "use", "code", "--json"])
1631 assert result.exit_code == 0, f"Call {i} failed: {result.output}"
1632
1633 def test_alternate_domains_100_times(self, tmp_path: pathlib.Path) -> None:
1634 """Switch between code and scaffold 100 times — file integrity preserved."""
1635 _init_repo(tmp_path, domain="code")
1636 domains = ["code", "scaffold"]
1637 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1638 for i in range(100):
1639 target = domains[i % 2]
1640 result = runner.invoke(cli, ["domains", "use", target])
1641 assert result.exit_code == 0, f"Switch {i} to {target!r} failed"
1642 written = json.loads((tmp_path / ".muse" / "repo.json").read_text())
1643 assert written["domain"] in domains
1644
1645 def test_concurrent_use_isolated_repos(self, tmp_path: pathlib.Path) -> None:
1646 """8 threads each switch domain on their own isolated repo."""
1647 import argparse
1648 import threading
1649
1650 from muse.cli.commands.domains import run_use
1651
1652 errors: list[str] = []
1653
1654 def worker(idx: int) -> None:
1655 repo = tmp_path / f"repo_{idx}"
1656 _init_repo(repo, domain="scaffold")
1657 try:
1658 args = argparse.Namespace(use_name="code", as_json=True)
1659 with patch("muse.cli.commands.domains._find_repo_root", return_value=repo):
1660 run_use(args)
1661 written = json.loads((repo / ".muse" / "repo.json").read_text())
1662 if written["domain"] != "code":
1663 errors.append(f"Thread {idx}: domain is {written['domain']!r}")
1664 except SystemExit as exc:
1665 errors.append(f"Thread {idx}: exit {exc.code}")
1666 except Exception as exc:
1667 errors.append(f"Thread {idx}: {exc}")
1668
1669 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1670 for t in threads:
1671 t.start()
1672 for t in threads:
1673 t.join()
1674 assert not errors, f"Concurrent failures: {errors}"
1675
1676
1677 # ---------------------------------------------------------------------------
1678 # Extended — muse domains validate
1679 # ---------------------------------------------------------------------------
1680
1681
1682 class TestDomainsValidateExtended:
1683 def test_j_alias_works(self) -> None:
1684 """-j is equivalent to --json."""
1685 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1686 result = runner.invoke(cli, ["domains", "validate", "code", "-j"])
1687 assert result.exit_code == 0
1688 data = json.loads(result.output.strip())
1689 assert data["domain"] == "code"
1690
1691 def test_help_flag(self) -> None:
1692 result = runner.invoke(cli, ["domains", "validate", "--help"])
1693 assert result.exit_code == 0
1694 assert "validate" in result.output.lower()
1695
1696 def test_json_compact_no_indent(self) -> None:
1697 """JSON output must be a single line (compact, no indent=2)."""
1698 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1699 result = runner.invoke(cli, ["domains", "validate", "code", "-j"])
1700 assert result.exit_code == 0
1701 lines = [l for l in result.output.splitlines() if l.strip()]
1702 assert len(lines) == 1, f"Expected compact JSON, got {len(lines)} lines"
1703
1704 def test_json_fields_present(self) -> None:
1705 """JSON object contains domain, ok, checks."""
1706 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1707 result = runner.invoke(cli, ["domains", "validate", "code", "-j"])
1708 data = json.loads(result.output.strip())
1709 assert set(data.keys()) >= {"domain", "ok", "checks"}
1710
1711 def test_json_domain_matches_arg(self) -> None:
1712 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1713 result = runner.invoke(cli, ["domains", "validate", "scaffold", "-j"])
1714 data = json.loads(result.output.strip())
1715 assert data["domain"] == "scaffold"
1716
1717 def test_json_checks_are_objects(self) -> None:
1718 """Each entry in checks has name, ok, detail."""
1719 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1720 result = runner.invoke(cli, ["domains", "validate", "code", "-j"])
1721 data = json.loads(result.output.strip())
1722 for c in data["checks"]:
1723 assert "name" in c
1724 assert "ok" in c
1725 assert "detail" in c
1726
1727 def test_text_output_shows_checkmarks(self) -> None:
1728 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1729 result = runner.invoke(cli, ["domains", "validate", "code"])
1730 assert "✅" in result.output or "✓" in result.output
1731
1732 def test_text_output_shows_domain_name(self) -> None:
1733 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1734 result = runner.invoke(cli, ["domains", "validate", "scaffold"])
1735 assert "scaffold" in result.output
1736
1737 def test_no_name_in_repo_validates_active_domain(self, tmp_path: pathlib.Path) -> None:
1738 """When inside a repo with domain=code, validate with no name validates code."""
1739 _init_repo(tmp_path, domain="code")
1740 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1741 result = runner.invoke(cli, ["domains", "validate", "-j"])
1742 assert result.exit_code == 0
1743 data = json.loads(result.output.strip())
1744 assert data["domain"] == "code"
1745
1746 def test_no_name_no_repo_validates_all_json(self) -> None:
1747 """With no name and no repo, all registered domains are validated."""
1748 from muse.plugins.registry import _REGISTRY
1749 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1750 result = runner.invoke(cli, ["domains", "validate", "-j"])
1751 assert result.exit_code == 0
1752 if len(_REGISTRY) == 1:
1753 data = json.loads(result.output.strip())
1754 assert "domain" in data
1755 else:
1756 data = json.loads(result.output.strip())
1757 assert isinstance(data, list)
1758 assert len(data) == len(_REGISTRY)
1759
1760 def test_unregistered_domain_exit1(self) -> None:
1761 result = runner.invoke(cli, ["domains", "validate", "does-not-exist"])
1762 assert result.exit_code == 1
1763
1764 def test_broken_plugin_ok_false(self) -> None:
1765 """A plugin with no required methods emits ok=false."""
1766 from muse.plugins.registry import _REGISTRY
1767 from unittest.mock import MagicMock
1768 mock = MagicMock(spec=[])
1769 with patch.dict(_REGISTRY, {"broken2": mock}):
1770 result = runner.invoke(cli, ["domains", "validate", "broken2", "-j"])
1771 assert result.exit_code == 1
1772 data = json.loads(result.output.strip())
1773 assert data["ok"] is False
1774
1775 def test_broken_plugin_checks_have_failed_entries(self) -> None:
1776 from muse.plugins.registry import _REGISTRY
1777 from unittest.mock import MagicMock
1778 mock = MagicMock(spec=[])
1779 with patch.dict(_REGISTRY, {"broken3": mock}):
1780 result = runner.invoke(cli, ["domains", "validate", "broken3", "-j"])
1781 data = json.loads(result.output.strip())
1782 failed = [c for c in data["checks"] if not c["ok"]]
1783 assert len(failed) >= 1
1784
1785 def test_multi_domain_json_is_list(self) -> None:
1786 """When validating all (no name, no repo) with >=2 domains, output is a list."""
1787 from muse.plugins.registry import _REGISTRY
1788 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1789 result = runner.invoke(cli, ["domains", "validate", "-j"])
1790 assert result.exit_code == 0
1791 if len(_REGISTRY) >= 2:
1792 data = json.loads(result.output.strip())
1793 assert isinstance(data, list)
1794
1795 def test_active_domain_not_in_registry_falls_back_to_all(self, tmp_path: pathlib.Path) -> None:
1796 """Active domain removed from registry → falls back to validating all."""
1797 _init_repo(tmp_path, domain="ghost")
1798 with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path):
1799 result = runner.invoke(cli, ["domains", "validate", "-j"])
1800 assert result.exit_code == 0
1801 # should validate registry domains, not error on missing ghost
1802
1803 def test_help_shows_agent_quickstart(self) -> None:
1804 result = runner.invoke(cli, ["domains", "validate", "--help"])
1805 assert result.exit_code == 0
1806 assert "Agent quickstart" in result.output
1807
1808 def test_help_shows_json_schema(self) -> None:
1809 result = runner.invoke(cli, ["domains", "validate", "--help"])
1810 assert "JSON output schema" in result.output
1811
1812 def test_help_shows_exit_codes(self) -> None:
1813 result = runner.invoke(cli, ["domains", "validate", "--help"])
1814 assert "Exit codes" in result.output
1815
1816
1817 # ---------------------------------------------------------------------------
1818 # Security — muse domains validate
1819 # ---------------------------------------------------------------------------
1820
1821
1822 class TestDomainsValidateSecurity:
1823 def test_ansi_in_domain_name_stripped_in_error(self) -> None:
1824 """ANSI in unknown domain name must not bleed into stderr."""
1825 result = runner.invoke(cli, ["domains", "validate", "\x1b[31mevil\x1b[0m"])
1826 assert result.exit_code == 1
1827 assert "\x1b" not in result.output
1828
1829 def test_unregistered_domain_never_checked(self) -> None:
1830 """An unregistered name exits before any check logic runs."""
1831 result = runner.invoke(cli, ["domains", "validate", "notareal_domain_xyz"])
1832 assert result.exit_code == 1
1833 assert "not registered" in result.output or "not registered" in (result.output + "")
1834
1835 def test_json_output_only_on_stdout(self, capsys: pytest.CaptureFixture) -> None:
1836 """JSON is emitted to stdout; errors go to stderr only."""
1837 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1838 result = runner.invoke(cli, ["domains", "validate", "code", "-j"])
1839 assert result.exit_code == 0
1840 json.loads(result.output.strip()) # must parse cleanly
1841
1842 def test_ansi_in_check_detail_sanitized_text_mode(self) -> None:
1843 """Plugin with no schema attr exercises AttributeError detail path; output has no ANSI."""
1844 from muse.plugins.registry import _REGISTRY
1845 from unittest.mock import MagicMock
1846 # spec omits schema → _run_validate_plugin catches AttributeError, writes detail string
1847 mock = MagicMock(spec=["snapshot", "diff", "merge", "drift", "apply"])
1848 with patch.dict(_REGISTRY, {"ansi_test": mock}):
1849 result = runner.invoke(cli, ["domains", "validate", "ansi_test"])
1850 assert "\x1b" not in result.output
1851
1852 def test_no_traceback_on_unknown_domain(self) -> None:
1853 result = runner.invoke(cli, ["domains", "validate", "totally_unknown_domain"])
1854 assert "Traceback" not in result.output
1855
1856 def test_multi_domain_validate_all_domains_present_in_output(self) -> None:
1857 """When validating all, every registered domain name appears in output."""
1858 from muse.plugins.registry import _REGISTRY
1859 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1860 result = runner.invoke(cli, ["domains", "validate", "-j"])
1861 assert result.exit_code == 0
1862 output = result.output
1863 for domain_name in _REGISTRY:
1864 assert domain_name in output, f"Missing domain {domain_name!r} in output"
1865
1866
1867 # ---------------------------------------------------------------------------
1868 # Stress — muse domains validate
1869 # ---------------------------------------------------------------------------
1870
1871
1872 class TestDomainsValidateStress:
1873 def test_50_sequential_validate_calls(self) -> None:
1874 """50 sequential validate calls all succeed."""
1875 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1876 for i in range(50):
1877 result = runner.invoke(cli, ["domains", "validate", "code", "-j"])
1878 assert result.exit_code == 0, f"Call {i} failed: {result.output}"
1879
1880 def test_alternate_code_scaffold_100_times(self) -> None:
1881 """Alternate between validating code and scaffold 100 times."""
1882 domains = ["code", "scaffold"]
1883 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1884 for i in range(100):
1885 target = domains[i % 2]
1886 result = runner.invoke(cli, ["domains", "validate", target, "-j"])
1887 assert result.exit_code == 0, f"Step {i}: {result.output}"
1888 data = json.loads(result.output.strip())
1889 assert data["domain"] == target
1890
1891 def test_concurrent_validate_8_threads(self) -> None:
1892 """8 threads each validate code concurrently using core function directly."""
1893 import argparse
1894 import threading
1895
1896 from muse.cli.commands.domains import run_validate
1897
1898 errors: list[str] = []
1899
1900 def worker(idx: int) -> None:
1901 args = argparse.Namespace(validate_name="code", as_json=True)
1902 try:
1903 with patch("muse.cli.commands.domains._find_repo_root", return_value=None):
1904 run_validate(args)
1905 except SystemExit as exc:
1906 if exc.code != 0:
1907 errors.append(f"Thread {idx}: exit {exc.code}")
1908 except Exception as exc:
1909 errors.append(f"Thread {idx}: {exc}")
1910
1911 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1912 for t in threads:
1913 t.start()
1914 for t in threads:
1915 t.join()
1916 assert not errors, f"Concurrent failures: {errors}"
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago