"""Comprehensive hardening tests for ``muse domains``. Covers: Unit tests: - _validate_domain_name: valid, path-traversal, reserved, bad chars - _validate_publish_url: http/https OK, file/ftp/data rejected - _active_domain: no root, missing repo.json, explicit domain, missing key - _build_entry: schema present, schema absent, active flag boolean - _post_json: correct wire format, response size cap, non-object JSON - _check_method / _run_validate_plugin: all checks pass, missing method Integration tests: - run (dashboard): text output, --json, --new validation + success - run_info: known domain, unknown domain, --json schema - run_use: no repo, unknown domain, known domain switches repo.json, --json - run_validate: all-pass, missing method, --json, multi-domain Security tests: - Path traversal in --new rejected - ANSI in domain name sanitized in dashboard - file:// publish hub rejected - Unsanitized server response never echoed raw E2E tests (full CLI invocation via CliRunner): - muse domains --json emits boolean 'active' field - muse domains --json includes module_path - muse domains info code --json schema present - muse domains validate --json ok - muse domains use code --json inside repo - muse domains --new myplug creates directory Stress tests: - 8 concurrent validate calls on isolated registry snapshots - 8 concurrent _active_domain reads on isolated tmp dirs """ from __future__ import annotations import argparse import http.client import json import pathlib import shutil import threading import urllib.error import urllib.request from contextlib import ExitStack from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest from muse.cli.commands.domains import ( _DomainEntryJson, _PublishResponse, _ScaffoldJson, _UseJson, _ValidateJson, ) from muse.domain import ( DriftReport, LiveState, MergeResult, StateSnapshot, StateDelta, SnapshotManifest, ) from muse.core.schema import DomainSchema from muse.core.paths import muse_dir, repo_json_path from tests.cli_test_helper import CliRunner, InvokeResult if TYPE_CHECKING: from muse.cli.commands.domains import _PublishPayload, _Capabilities, _DimensionDef from muse.core.transport import SigningIdentity cli = None # argparse migration — CliRunner ignores this argument runner = CliRunner() # --------------------------------------------------------------------------- # JSON helpers — each returns a specific TypedDict to satisfy typing_audit # --------------------------------------------------------------------------- def _first_json_blob(result: InvokeResult) -> str: """Return the first complete JSON blob string from result.output.""" output = result.output depth = 0 start: int | None = None for i, ch in enumerate(output): if ch in "{[": if start is None: start = i depth += 1 elif ch in "}]": depth -= 1 if depth == 0 and start is not None: return output[start : i + 1] raise AssertionError(f"No JSON found in output:\n{output!r}") def _parse_domains_list(result: InvokeResult) -> list[_DomainEntryJson]: """Parse a JSON array of domain entries from result.""" parsed = json.loads(_first_json_blob(result)) if isinstance(parsed, dict) and "domains" in parsed: parsed = parsed["domains"] assert isinstance(parsed, list) return parsed def _parse_domain_entry(result: InvokeResult) -> _DomainEntryJson: """Parse a single domain entry JSON dict from result.""" parsed: _DomainEntryJson = json.loads(_first_json_blob(result)) assert isinstance(parsed, dict) return parsed def _parse_scaffold(result: InvokeResult) -> _ScaffoldJson: parsed: _ScaffoldJson = json.loads(_first_json_blob(result)) assert isinstance(parsed, dict) return parsed def _parse_use(result: InvokeResult) -> _UseJson: parsed: _UseJson = json.loads(_first_json_blob(result)) assert isinstance(parsed, dict) return parsed def _parse_validate(result: InvokeResult) -> _ValidateJson: parsed: _ValidateJson = json.loads(_first_json_blob(result)) assert isinstance(parsed, dict) return parsed def _parse_validate_list(result: InvokeResult) -> list[_ValidateJson]: parsed = json.loads(_first_json_blob(result)) if isinstance(parsed, dict) and "results" in parsed: parsed = parsed["results"] assert isinstance(parsed, list) return parsed def _parse_publish(result: InvokeResult) -> _PublishResponse: parsed: _PublishResponse = json.loads(_first_json_blob(result)) assert isinstance(parsed, dict) return parsed # --------------------------------------------------------------------------- # Repository fixture # --------------------------------------------------------------------------- def _init_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path: """Create a minimal .muse repo structure under tmp_path.""" muse = muse_dir(tmp_path) muse.mkdir(parents=True) repo_json = { "repo_id": "test-repo", "schema_version": "0.1.5", "created_at": "2026-01-01T00:00:00+00:00", "domain": domain, } (muse / "repo.json").write_text(json.dumps(repo_json), encoding="utf-8") return tmp_path # --------------------------------------------------------------------------- # Unit — _validate_domain_name # --------------------------------------------------------------------------- class TestValidateDomainName: def _call(self, name: str) -> int: from muse.cli.commands.domains import _validate_domain_name with pytest.raises(SystemExit) as exc_info: _validate_domain_name(name) code = exc_info.value.code assert isinstance(code, int) return code def test_valid_lowercase(self) -> None: from muse.cli.commands.domains import _validate_domain_name _validate_domain_name("genomics") # must not raise def test_valid_with_hyphen(self) -> None: from muse.cli.commands.domains import _validate_domain_name _validate_domain_name("spatial-3d") def test_valid_with_underscore(self) -> None: from muse.cli.commands.domains import _validate_domain_name _validate_domain_name("my_domain") def test_path_traversal_dotdot_rejected(self) -> None: assert self._call("../traversal") == 1 def test_path_traversal_nested_rejected(self) -> None: assert self._call("../../etc/passwd") == 1 def test_uppercase_rejected(self) -> None: assert self._call("Genomics") == 1 def test_starts_with_digit_rejected(self) -> None: assert self._call("3d-scenes") == 1 def test_slash_rejected(self) -> None: assert self._call("malicious/path") == 1 def test_null_byte_rejected(self) -> None: assert self._call("malicious\x00name") == 1 def test_space_rejected(self) -> None: assert self._call("my domain") == 1 def test_reserved_scaffold_rejected(self) -> None: assert self._call("scaffold") == 1 def test_max_length_64_accepted(self) -> None: from muse.cli.commands.domains import _validate_domain_name _validate_domain_name("a" * 64) def test_max_length_65_rejected(self) -> None: assert self._call("a" * 65) == 1 def test_empty_rejected(self) -> None: assert self._call("") == 1 # --------------------------------------------------------------------------- # Unit — _validate_publish_url # --------------------------------------------------------------------------- class TestValidatePublishUrl: def _call(self, url: str) -> int: from muse.cli.commands.domains import _validate_publish_url with pytest.raises(SystemExit) as exc_info: _validate_publish_url(url) code = exc_info.value.code assert isinstance(code, int) return code def test_https_ok(self) -> None: from muse.cli.commands.domains import _validate_publish_url _validate_publish_url("https://musehub.ai") # must not raise def test_http_ok(self) -> None: from muse.cli.commands.domains import _validate_publish_url _validate_publish_url("https://localhost:1337") def test_file_scheme_rejected(self) -> None: assert self._call("file:///etc/passwd") == 1 def test_ftp_scheme_rejected(self) -> None: assert self._call("ftp://attacker.example.com") == 1 def test_data_uri_rejected(self) -> None: assert self._call("data:text/plain,malicious") == 1 def test_empty_scheme_rejected(self) -> None: assert self._call("://malicious") == 1 # --------------------------------------------------------------------------- # Unit — _active_domain # --------------------------------------------------------------------------- class TestActiveDomain: def test_none_root_returns_none(self) -> None: from muse.cli.commands.domains import _active_domain assert _active_domain(None) is None def test_missing_repo_json_returns_none(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.domains import _active_domain muse_dir(tmp_path).mkdir() assert _active_domain(tmp_path) is None def test_explicit_domain_returned(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.domains import _active_domain _init_repo(tmp_path, domain="code") assert _active_domain(tmp_path) == "code" def test_missing_domain_key_returns_default(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.domains import _active_domain, _DEFAULT_DOMAIN muse = muse_dir(tmp_path) muse.mkdir() (muse / "repo.json").write_text('{"repo_id": "x"}', encoding="utf-8") assert _active_domain(tmp_path) == _DEFAULT_DOMAIN def test_corrupt_json_returns_none(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.domains import _active_domain muse = muse_dir(tmp_path) muse.mkdir() (muse / "repo.json").write_text("NOT JSON", encoding="utf-8") assert _active_domain(tmp_path) is None def test_no_midi_fallback(self, tmp_path: pathlib.Path) -> None: """The old 'midi' fallback is gone — default is _DEFAULT_DOMAIN ('code').""" from muse.cli.commands.domains import _active_domain muse = muse_dir(tmp_path) muse.mkdir() (muse / "repo.json").write_text('{"domain": ""}', encoding="utf-8") result = _active_domain(tmp_path) assert result != "midi" # --------------------------------------------------------------------------- # Unit — _build_entry # --------------------------------------------------------------------------- class TestBuildEntry: def test_schema_present(self) -> None: from muse.cli.commands.domains import _build_entry from muse.plugins.registry import _REGISTRY plugin = _REGISTRY["code"] entry = _build_entry("code", plugin, "code") assert entry["domain"] == "code" assert entry["active"] is True assert isinstance(entry["active"], bool) assert "schema" in entry schema = entry["schema"] assert "schema_version" in schema assert "dimensions" in schema def test_schema_absent_when_not_implemented(self) -> None: from muse.cli.commands.domains import _build_entry mock_plugin = MagicMock() mock_plugin.schema.side_effect = NotImplementedError mock_plugin.__class__ = type("FakeDomain", (), {}) entry = _build_entry("fake", mock_plugin, None) assert "schema" not in entry assert entry["active"] is False def test_module_path_included(self) -> None: from muse.cli.commands.domains import _build_entry from muse.plugins.registry import _REGISTRY plugin = _REGISTRY["scaffold"] entry = _build_entry("scaffold", plugin, None) assert entry["module_path"] == "plugins/scaffold/plugin.py" def test_active_false_for_inactive(self) -> None: from muse.cli.commands.domains import _build_entry from muse.plugins.registry import _REGISTRY plugin = _REGISTRY["code"] entry = _build_entry("code", plugin, "scaffold") assert entry["active"] is False def test_capabilities_list_of_strings(self) -> None: from muse.cli.commands.domains import _build_entry from muse.plugins.registry import _REGISTRY plugin = _REGISTRY["code"] entry = _build_entry("code", plugin, None) caps = entry["capabilities"] assert isinstance(caps, list) assert all(isinstance(c, str) for c in caps) assert "Typed Deltas" in caps # --------------------------------------------------------------------------- # Unit — _post_json # --------------------------------------------------------------------------- def _make_caps_payload() -> "_PublishPayload": from muse.cli.commands.domains import _PublishPayload, _Capabilities, _DimensionDef caps = _Capabilities( dimensions=[_DimensionDef(name="notes", description="Note events")], artifact_types=["mid"], merge_semantics="three_way", supported_commands=["commit"], ) return _PublishPayload( author_slug="user", slug="music", display_name="Music", description="Desc", capabilities=caps, viewer_type="midi", version="0.1.0", ) def _make_signing() -> "SigningIdentity": from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from muse.core.transport import SigningIdentity return SigningIdentity(handle="testuser", private_key=Ed25519PrivateKey.generate()) class TestPostJson: def test_correct_method_and_headers(self) -> None: from muse.cli.commands.domains import _post_json captured: list[urllib.request.Request] = [] def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock: captured.append(req) resp = MagicMock() resp.read = MagicMock(return_value=b'{"domain_id":"1","scoped_id":"@u/s","manifest_hash":"abc"}') resp.__enter__ = lambda s: s resp.__exit__ = MagicMock(return_value=False) return resp with patch("urllib.request.urlopen", fake_urlopen): result = _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing()) assert len(captured) == 1 req = captured[0] assert req.get_method() == "POST" assert req.get_header("Content-type") == "application/json" assert req.get_header("Authorization").startswith("MSign ") assert result["scoped_id"] == "@u/s" def test_response_size_cap_applied(self) -> None: """resp.read() is called with _MAX_RESPONSE_BYTES, not unlimited.""" from muse.cli.commands.domains import _post_json, _MAX_RESPONSE_BYTES read_args: list[int] = [] def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock: resp = MagicMock() def _read(n: int) -> bytes: read_args.append(n) return b'{"domain_id":"1","scoped_id":"@u/s","manifest_hash":""}' resp.read = _read resp.__enter__ = lambda s: s resp.__exit__ = MagicMock(return_value=False) return resp with patch("urllib.request.urlopen", fake_urlopen): _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing()) assert read_args == [_MAX_RESPONSE_BYTES] def test_non_object_json_raises_value_error(self) -> None: from muse.cli.commands.domains import _post_json def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock: resp = MagicMock() resp.read = MagicMock(return_value=b"[1,2,3]") resp.__enter__ = lambda s: s resp.__exit__ = MagicMock(return_value=False) return resp with patch("urllib.request.urlopen", fake_urlopen): with pytest.raises(ValueError, match="Expected JSON object"): _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing()) def test_missing_keys_normalised_to_empty_string(self) -> None: from muse.cli.commands.domains import _post_json def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock: resp = MagicMock() resp.read = MagicMock(return_value=b"{}") resp.__enter__ = lambda s: s resp.__exit__ = MagicMock(return_value=False) return resp with patch("urllib.request.urlopen", fake_urlopen): result = _post_json("https://musehub.ai/api/v1/domains", _make_caps_payload(), _make_signing()) assert result["domain_id"] == "" assert result["scoped_id"] == "" assert result["manifest_hash"] == "" # --------------------------------------------------------------------------- # Unit — _run_validate_plugin # --------------------------------------------------------------------------- class _MinimalPlugin: """Stub implementing all required MuseDomainPlugin methods; schema raises NotImplementedError.""" def snapshot(self, live_state: LiveState) -> StateSnapshot: raise NotImplementedError def diff( self, base: StateSnapshot, target: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> StateDelta: raise NotImplementedError def merge( self, base: StateSnapshot, left: StateSnapshot, right: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> MergeResult: raise NotImplementedError def drift(self, committed: StateSnapshot, live: LiveState) -> DriftReport: raise NotImplementedError def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState: raise NotImplementedError def schema(self) -> DomainSchema: raise NotImplementedError class TestRunValidatePlugin: def test_all_checks_pass_for_code_plugin(self) -> None: from muse.cli.commands.domains import _run_validate_plugin from muse.plugins.registry import _REGISTRY result = _run_validate_plugin("code", _REGISTRY["code"], None) assert result["ok"] is True assert result["domain"] == "code" names = [c["name"] for c in result["checks"]] assert "has_method:snapshot" in names assert "schema()" in names def test_schema_not_implemented_flagged(self) -> None: """A plugin that has all methods but raises NotImplementedError for schema() must have the schema check fail while all method checks pass.""" from muse.cli.commands.domains import _run_validate_plugin result = _run_validate_plugin("min", _MinimalPlugin(), None) schema_check = next(c for c in result["checks"] if c["name"] == "schema()") assert schema_check["ok"] is False # All protocol method checks should pass since methods exist and are callable method_checks = [c for c in result["checks"] if c["name"].startswith("has_method:")] assert all(c["ok"] for c in method_checks) def test_scaffold_plugin_all_pass(self) -> None: from muse.cli.commands.domains import _run_validate_plugin from muse.plugins.registry import _REGISTRY result = _run_validate_plugin("scaffold", _REGISTRY["scaffold"], None) assert result["ok"] is True, [c for c in result["checks"] if not c["ok"]] # --------------------------------------------------------------------------- # Integration — muse domains (dashboard + --json) # --------------------------------------------------------------------------- class TestDomainsDashboard: def test_default_text_output_has_registered_domains(self) -> None: result = runner.invoke(cli, ["domains"]) assert result.exit_code == 0 assert "Registered domains:" in result.output def test_json_flag_emits_list(self) -> None: result = runner.invoke(cli, ["domains", "--json"]) assert result.exit_code == 0 domains = _parse_domains_list(result) assert len(domains) >= 1 def test_json_active_field_is_boolean(self) -> None: result = runner.invoke(cli, ["domains", "--json"]) assert result.exit_code == 0 for entry in _parse_domains_list(result): assert isinstance(entry.get("active"), bool), ( f"'active' must be bool, got {type(entry.get('active')).__name__}" ) def test_json_module_path_present(self) -> None: result = runner.invoke(cli, ["domains", "--json"]) assert result.exit_code == 0 for entry in _parse_domains_list(result): assert "module_path" in entry, f"missing 'module_path' in {entry}" def test_json_schema_present_for_code(self) -> None: result = runner.invoke(cli, ["domains", "--json"]) assert result.exit_code == 0 code_entry = next(e for e in _parse_domains_list(result) if e.get("domain") == "code") assert "schema" in code_entry schema = code_entry["schema"] assert "dimensions" in schema def test_json_no_string_true_false(self) -> None: """Agents must never see 'active': 'true' — only 'active': true.""" result = runner.invoke(cli, ["domains", "--json"]) raw = result.output assert '"active": "true"' not in raw assert '"active": "false"' not in raw def test_text_has_muse_domains_new_hint(self) -> None: result = runner.invoke(cli, ["domains"]) assert "muse domains --new" in result.output def test_text_has_info_hint(self) -> None: result = runner.invoke(cli, ["domains"]) assert "muse domains info" in result.output def test_text_has_validate_hint(self) -> None: result = runner.invoke(cli, ["domains"]) assert "muse domains validate" in result.output # --------------------------------------------------------------------------- # Integration — muse domains --new # --------------------------------------------------------------------------- class TestDomainsNew: def test_valid_name_creates_directory(self) -> None: plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins" dest = plugins_dir / "testdomain9" try: result = runner.invoke(cli, ["domains", "--new", "testdomain9"]) assert result.exit_code == 0, result.output assert dest.exists() assert (dest / "plugin.py").exists() finally: if dest.exists(): shutil.rmtree(str(dest)) def test_valid_name_json_flag_emits_scaffold_json(self) -> None: plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins" dest = plugins_dir / "testdomain10" try: result = runner.invoke(cli, ["domains", "--new", "testdomain10", "--json"]) assert result.exit_code == 0, result.output data = _parse_scaffold(result) assert data["name"] == "testdomain10" assert data["status"] == "ok" assert "class_name" in data assert "path" in data finally: if dest.exists(): shutil.rmtree(str(dest)) def test_path_traversal_rejected(self) -> None: result = runner.invoke(cli, ["domains", "--new", "../traversal"]) assert result.exit_code != 0 def test_uppercase_name_rejected(self) -> None: result = runner.invoke(cli, ["domains", "--new", "Genomics"]) assert result.exit_code != 0 def test_scaffold_reserved_rejected(self) -> None: result = runner.invoke(cli, ["domains", "--new", "scaffold"]) assert result.exit_code != 0 def test_duplicate_name_rejected(self) -> None: result = runner.invoke(cli, ["domains", "--new", "code"]) assert result.exit_code != 0 def test_no_pycache_in_created_directory(self) -> None: plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins" dest = plugins_dir / "testdomain11" try: result = runner.invoke(cli, ["domains", "--new", "testdomain11"]) assert result.exit_code == 0 pycache = dest / "__pycache__" assert not pycache.exists(), "__pycache__ must not be copied" finally: if dest.exists(): shutil.rmtree(str(dest)) def test_class_name_substituted(self) -> None: plugins_dir = pathlib.Path(__file__).parents[1] / "muse" / "plugins" dest = plugins_dir / "testdomain12" try: runner.invoke(cli, ["domains", "--new", "testdomain12"]) plugin_src = (dest / "plugin.py").read_text(encoding="utf-8") assert "Testdomain12Plugin" in plugin_src assert "ScaffoldPlugin" not in plugin_src finally: if dest.exists(): shutil.rmtree(str(dest)) # --------------------------------------------------------------------------- # Integration — muse domains info # --------------------------------------------------------------------------- class TestDomainsInfo: def test_known_domain_exits_zero(self) -> None: result = runner.invoke(cli, ["domains", "info", "code"]) assert result.exit_code == 0 def test_unknown_domain_exits_nonzero(self) -> None: result = runner.invoke(cli, ["domains", "info", "doesnotexist"]) assert result.exit_code != 0 def test_json_schema_has_required_keys(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) assert result.exit_code == 0 data = _parse_domain_entry(result) assert data["domain"] == "code" assert isinstance(data["active"], bool) assert "capabilities" in data assert "module_path" in data def test_scaffold_has_crdt_capability(self) -> None: result = runner.invoke(cli, ["domains", "info", "scaffold", "--json"]) assert result.exit_code == 0 data = _parse_domain_entry(result) assert "CRDT" in data["capabilities"] def test_text_output_includes_module_path(self) -> None: result = runner.invoke(cli, ["domains", "info", "code"]) assert "Module:" in result.output assert "plugins/code/plugin.py" in result.output def test_error_on_unknown_domain(self) -> None: result = runner.invoke(cli, ["domains", "info", "ghost"]) assert result.exit_code != 0 assert "not registered" in result.stderr # --------------------------------------------------------------------------- # Integration — muse domains use # --------------------------------------------------------------------------- class TestDomainsUse: def test_no_repo_exits_nonzero(self, tmp_path: pathlib.Path) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code != 0 def test_unknown_domain_exits_nonzero(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "doesnotexist"]) assert result.exit_code != 0 def test_known_domain_switches_repo_json(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 0, result.output data: Manifest = json.loads((repo_json_path(tmp_path)).read_text(encoding="utf-8")) assert data["domain"] == "code" def test_switch_is_idempotent(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 0 def test_json_output_has_required_keys(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) assert result.exit_code == 0, result.output data = _parse_use(result) assert data["domain"] == "code" assert data["status"] == "switched" assert "repo" in data def test_existing_repo_fields_preserved(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): runner.invoke(cli, ["domains", "use", "code"]) data: Manifest = json.loads((repo_json_path(tmp_path)).read_text(encoding="utf-8")) assert "repo_id" in data assert "schema_version" in data # --------------------------------------------------------------------------- # Integration — muse domains validate # --------------------------------------------------------------------------- class TestDomainsValidate: def test_code_plugin_passes(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code"]) assert result.exit_code == 0 def test_scaffold_plugin_passes(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "scaffold"]) assert result.exit_code == 0 def test_unknown_domain_exits_nonzero(self) -> None: result = runner.invoke(cli, ["domains", "validate", "ghost"]) assert result.exit_code != 0 def test_json_ok_field_is_boolean(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "--json"]) assert result.exit_code == 0 data = _parse_validate(result) assert isinstance(data["ok"], bool) assert data["ok"] is True def test_json_checks_list_present(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "scaffold", "--json"]) data = _parse_validate(result) assert isinstance(data["checks"], list) assert len(data["checks"]) >= 5 def test_broken_plugin_exits_nonzero(self) -> None: from muse.plugins.registry import _REGISTRY mock_plugin = MagicMock(spec=[]) # no methods at all with patch.dict(_REGISTRY, {"broken": mock_plugin}): result = runner.invoke(cli, ["domains", "validate", "broken"]) assert result.exit_code != 0 def test_no_name_validates_active_repo_domain(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "validate"]) assert result.exit_code == 0 assert "code" in result.output def test_no_name_no_repo_validates_all(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "--json"]) assert result.exit_code == 0 from muse.plugins.registry import _REGISTRY if len(_REGISTRY) > 1: _parse_validate_list(result) else: _parse_validate(result) # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestDomainsSecurity: def test_ansi_in_domain_name_sanitized_in_dashboard(self) -> None: """A registry entry with ANSI in its name must not bleed into output.""" from muse.plugins.registry import _REGISTRY from muse.plugins.scaffold.plugin import ScaffoldPlugin malicious_name = "\x1b[31mmalicious\x1b[0m" mock = ScaffoldPlugin() with patch.dict(_REGISTRY, {malicious_name: mock}): result = runner.invoke(cli, ["domains"]) assert "\x1b[31m" not in result.output def test_file_scheme_publish_hub_rejected(self, tmp_path: pathlib.Path) -> None: """file:// hub URL must be blocked before any network call.""" urlopen_called = False def _urlopen(req: urllib.request.Request, timeout: float) -> MagicMock: nonlocal urlopen_called urlopen_called = True raise AssertionError("urlopen must not be called with file:// URL") _init_repo(tmp_path, domain="code") with ExitStack() as stack: stack.enter_context(patch("urllib.request.urlopen", _urlopen)) stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value="tok") ) result = runner.invoke(cli, [ "domains", "publish", "--author", "user", "--slug", "music", "--name", "Music", "--description", "Desc", "--viewer-type", "midi", "--hub", "file:///etc/passwd", "--capabilities", '{"dimensions":[],"artifact_types":[],"merge_semantics":"three_way","supported_commands":[]}', ]) assert result.exit_code != 0 assert not urlopen_called def test_server_ansi_in_scoped_id_sanitized(self, tmp_path: pathlib.Path) -> None: """Server-returned scoped_id with ANSI (unicode-escaped in JSON) must not appear raw.""" _init_repo(tmp_path, domain="code") # JSON uses \u001b for the ESC character — valid JSON but contains ANSI. malicious_response = b'{"domain_id":"1","scoped_id":"\\u001b[31mhacked\\u001b[0m","manifest_hash":""}' def fake_urlopen(req: urllib.request.Request, timeout: float) -> MagicMock: resp = MagicMock() resp.read = MagicMock(return_value=malicious_response) resp.__enter__ = lambda s: s resp.__exit__ = MagicMock(return_value=False) return resp with ExitStack() as stack: stack.enter_context(patch("urllib.request.urlopen", fake_urlopen)) stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value=_make_signing()) ) result = runner.invoke(cli, [ "domains", "publish", "--author", "user", "--slug", "music", "--name", "Music", "--description", "Desc", "--viewer-type", "midi", "--capabilities", '{"dimensions":[],"artifact_types":[],"merge_semantics":"three_way","supported_commands":[]}', ]) assert result.exit_code == 0 assert "\x1b[31m" not in result.output def test_path_traversal_new_rejected_without_touching_fs(self) -> None: malicious = "../" * 5 + "malicious" malicious_path = pathlib.Path(__file__).parents[1] / "muse" / "plugins" / malicious result = runner.invoke(cli, ["domains", "--new", malicious]) assert result.exit_code != 0 assert not malicious_path.exists() def test_null_byte_name_rejected(self) -> None: result = runner.invoke(cli, ["domains", "--new", "malicious\x00name"]) assert result.exit_code != 0 # --------------------------------------------------------------------------- # E2E — verify the muse CLI entry point wires domains correctly # --------------------------------------------------------------------------- class TestDomainsE2E: def test_domains_help_exits_zero(self) -> None: result = runner.invoke(cli, ["domains", "--help"]) assert result.exit_code == 0 assert "domains" in result.output.lower() def test_domains_info_help(self) -> None: result = runner.invoke(cli, ["domains", "info", "--help"]) assert result.exit_code == 0 def test_domains_use_help(self) -> None: result = runner.invoke(cli, ["domains", "use", "--help"]) assert result.exit_code == 0 def test_domains_validate_help(self) -> None: result = runner.invoke(cli, ["domains", "validate", "--help"]) assert result.exit_code == 0 def test_domains_publish_help(self) -> None: result = runner.invoke(cli, ["domains", "publish", "--help"]) assert result.exit_code == 0 def test_domains_json_is_valid_json(self) -> None: result = runner.invoke(cli, ["domains", "--json"]) assert result.exit_code == 0 domains = _parse_domains_list(result) assert len(domains) >= 1 def test_info_json_is_valid_json(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) assert result.exit_code == 0 _parse_domain_entry(result) # must not raise def test_validate_json_is_valid_json(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "--json"]) assert result.exit_code == 0 _parse_validate(result) def test_use_requires_name_arg(self) -> None: result = runner.invoke(cli, ["domains", "use"]) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Publish E2E — happy path and error paths # --------------------------------------------------------------------------- class TestPublishE2E: _CAPS = json.dumps({ "dimensions": [{"name": "notes", "description": "Note events"}], "artifact_types": ["mid"], "merge_semantics": "three_way", "supported_commands": ["commit", "diff"], }) _BASE_ARGS = [ "domains", "publish", "--author", "user", "--slug", "music", "--name", "Music", "--description", "Desc", "--viewer-type", "midi", "--capabilities", _CAPS, ] def test_successful_publish_text_output(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") fake_resp = _PublishResponse(domain_id="1", scoped_id="@user/music", manifest_hash="abc") with ExitStack() as stack: stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value="tok") ) stack.enter_context( patch("muse.cli.commands.domains._post_json", return_value=fake_resp) ) result = runner.invoke(cli, self._BASE_ARGS) assert result.exit_code == 0, result.output assert "@user/music" in result.output def test_successful_publish_json_output(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") fake_resp = _PublishResponse(domain_id="1", scoped_id="@user/music", manifest_hash="abc") with ExitStack() as stack: stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value="tok") ) stack.enter_context( patch("muse.cli.commands.domains._post_json", return_value=fake_resp) ) result = runner.invoke(cli, [*self._BASE_ARGS, "--json"]) assert result.exit_code == 0 data = _parse_publish(result) assert data["scoped_id"] == "@user/music" def test_no_token_exits_nonzero(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with ExitStack() as stack: stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value=None) ) result = runner.invoke(cli, self._BASE_ARGS) assert result.exit_code != 0 def test_http_409_exits_nonzero(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") exc = urllib.error.HTTPError( url="", code=409, msg="Conflict", hdrs=http.client.HTTPMessage(), fp=None, ) with ExitStack() as stack: stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value="tok") ) stack.enter_context( patch("muse.cli.commands.domains._post_json", side_effect=exc) ) result = runner.invoke(cli, self._BASE_ARGS) assert result.exit_code != 0 def test_http_401_exits_nonzero(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") exc = urllib.error.HTTPError( url="", code=401, msg="Unauthorized", hdrs=http.client.HTTPMessage(), fp=None, ) with ExitStack() as stack: stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value="tok") ) stack.enter_context( patch("muse.cli.commands.domains._post_json", side_effect=exc) ) result = runner.invoke(cli, self._BASE_ARGS) assert result.exit_code != 0 def test_url_error_exits_nonzero(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") exc = urllib.error.URLError("connection refused") with ExitStack() as stack: stack.enter_context( patch("muse.cli.commands.domains.find_repo_root", return_value=tmp_path) ) stack.enter_context( patch("muse.cli.commands.domains.get_signing_identity", return_value="tok") ) stack.enter_context( patch("muse.cli.commands.domains._post_json", side_effect=exc) ) result = runner.invoke(cli, self._BASE_ARGS) assert result.exit_code != 0 # --------------------------------------------------------------------------- # Stress tests # --------------------------------------------------------------------------- class TestStress: def test_8_concurrent_validate_calls_isolated(self) -> None: """Concurrent validate calls on independent plugin instances must not interfere.""" from muse.cli.commands.domains import _run_validate_plugin from muse.plugins.registry import _REGISTRY errors: list[str] = [] def _validate(idx: int) -> None: try: plugin = _REGISTRY["code"] result = _run_validate_plugin("code", plugin, None) assert result["ok"] is True, f"Thread {idx}: validate failed" except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_validate, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent validate failures: {errors}" def test_8_concurrent_active_domain_reads(self, tmp_path: pathlib.Path) -> None: """Concurrent _active_domain reads on isolated dirs must all return correctly.""" from muse.cli.commands.domains import _active_domain roots: list[pathlib.Path] = [] for i in range(8): rd = tmp_path / f"repo{i}" _init_repo(rd, domain="code") roots.append(rd) errors: list[str] = [] def _read(idx: int) -> None: try: result = _active_domain(roots[idx]) assert result == "code", f"Thread {idx}: expected 'code', got {result!r}" except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_read, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent read failures: {errors}" def test_8_concurrent_build_entry_calls(self) -> None: """_build_entry must be re-entrant (no shared mutable state).""" from muse.cli.commands.domains import _build_entry from muse.plugins.registry import _REGISTRY errors: list[str] = [] def _build(idx: int) -> None: try: plugin = _REGISTRY["scaffold"] entry = _build_entry("scaffold", plugin, "code" if idx % 2 == 0 else "scaffold") assert entry["domain"] == "scaffold" assert isinstance(entry["active"], bool) except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=_build, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent build_entry failures: {errors}" # --------------------------------------------------------------------------- # Extended — muse domains info (deeper coverage) # --------------------------------------------------------------------------- class TestDomainsInfoExtended: def test_j_alias_works(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "-j"]) assert result.exit_code == 0 data = json.loads(result.output.strip()) assert data["domain"] == "code" def test_help_mentions_json_flag(self) -> None: result = runner.invoke(cli, ["domains", "info", "--help"]) assert "--json" in result.output or "-j" in result.output def test_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["domains", "info", "--help"]) assert "Exit code" in result.output or "exit code" in result.output def test_json_is_compact_single_line(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) assert result.exit_code == 0 lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) == 1, f"Expected 1 non-empty line, got {len(lines)}" def test_json_parses_cleanly(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert isinstance(data, dict) def test_json_domain_field_matches_arg(self) -> None: result = runner.invoke(cli, ["domains", "info", "scaffold", "--json"]) assert result.exit_code == 0 data = json.loads(result.output.strip()) assert data["domain"] == "scaffold" def test_json_active_is_bool(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert isinstance(data["active"], bool) def test_json_capabilities_is_list(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert isinstance(data["capabilities"], list) assert len(data["capabilities"]) >= 1 def test_json_module_path_is_string(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert isinstance(data["module_path"], str) assert len(data["module_path"]) > 0 def test_json_code_has_typed_deltas(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert "Typed Deltas" in data["capabilities"] def test_json_code_schema_present(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert "schema" in data def test_json_schema_has_required_keys(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) schema = data["schema"] for key in ("schema_version", "merge_mode", "description", "dimensions"): assert key in schema, f"Missing schema key: {key}" def test_json_schema_dimensions_is_list(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) assert isinstance(data["schema"]["dimensions"], list) assert len(data["schema"]["dimensions"]) >= 1 def test_json_schema_dimension_has_name_and_description(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "--json"]) data = json.loads(result.output.strip()) dim = data["schema"]["dimensions"][0] assert "name" in dim assert "description" in dim def test_text_shows_module_path(self) -> None: result = runner.invoke(cli, ["domains", "info", "code"]) assert "Module:" in result.output def test_text_shows_capabilities(self) -> None: result = runner.invoke(cli, ["domains", "info", "code"]) assert "Capabilities:" in result.output assert "Typed Deltas" in result.output def test_unknown_domain_error_mentions_name(self) -> None: result = runner.invoke(cli, ["domains", "info", "nonexistent_domain_xyz"]) assert result.exit_code == 1 assert "nonexistent_domain_xyz" in result.stderr or "not registered" in result.stderr def test_unknown_domain_lists_known_domains(self) -> None: result = runner.invoke(cli, ["domains", "info", "nope"]) assert result.exit_code == 1 # Should mention at least one known domain (e.g. "code") assert "code" in result.stderr # --------------------------------------------------------------------------- # Security — muse domains info # --------------------------------------------------------------------------- class TestDomainsInfoSecurity: def test_ansi_in_domain_name_stripped_error(self) -> None: """ANSI in the unknown domain name is stripped from the error message.""" result = runner.invoke(cli, ["domains", "info", "\x1b[31mmalicious\x1b[0m"]) assert result.exit_code == 1 assert "\x1b[" not in result.output def test_ansi_in_module_path_stripped_text(self) -> None: """module_path with ANSI injected by a plugin is sanitized in text output.""" from unittest.mock import patch as _patch malicious_entry = { "domain": "code", "module_path": "plugins/\x1b[31mmalicious\x1b[0m/plugin.py", "capabilities": ["Typed Deltas"], "active": False, } with _patch("muse.cli.commands.domains._build_entry", return_value=malicious_entry): result = runner.invoke(cli, ["domains", "info", "code"]) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_ansi_in_schema_description_stripped_text(self) -> None: """ANSI in schema description from a plugin is sanitized in text output.""" from unittest.mock import patch as _patch, MagicMock malicious_entry = { "domain": "code", "module_path": "plugins/code/plugin.py", "capabilities": ["Typed Deltas", "Domain Schema"], "active": False, "schema": { "schema_version": "1.0", "merge_mode": "structured", "description": "Good domain \x1b[31mmalicious\x1b[0m", "dimensions": [{"name": "symbols", "description": "sym"}], }, } with _patch("muse.cli.commands.domains._build_entry", return_value=malicious_entry): result = runner.invoke(cli, ["domains", "info", "code"]) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_ansi_in_dimension_name_stripped_text(self) -> None: """ANSI in a dimension name from a plugin schema is sanitized in text output.""" from unittest.mock import patch as _patch malicious_entry = { "domain": "code", "module_path": "plugins/code/plugin.py", "capabilities": ["Typed Deltas", "Domain Schema"], "active": False, "schema": { "schema_version": "1.0", "merge_mode": "structured", "description": "desc", "dimensions": [{"name": "\x1b[31mmalicious\x1b[0m", "description": "bad"}], }, } with _patch("muse.cli.commands.domains._build_entry", return_value=malicious_entry): result = runner.invoke(cli, ["domains", "info", "code"]) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_json_stdout_only_no_stray_text(self) -> None: """In JSON mode, stdout must be parseable JSON with no surrounding noise.""" result = runner.invoke(cli, ["domains", "info", "code", "--json"]) assert result.exit_code == 0 json.loads(result.output.strip()) def test_unknown_domain_no_traceback(self) -> None: result = runner.invoke(cli, ["domains", "info", "no_such_domain_xyz"]) assert result.exit_code == 1 assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress — muse domains info # --------------------------------------------------------------------------- class TestDomainsInfoStress: def test_50_sequential_info_calls(self) -> None: """50 sequential info invocations on 'code' all exit 0.""" for i in range(50): result = runner.invoke(cli, ["domains", "info", "code", "--json"]) assert result.exit_code == 0, f"Call {i} failed: {result.output}" def test_all_registered_domains_info_exits_zero(self) -> None: """Every domain currently in the registry responds to info with exit 0.""" from muse.plugins.registry import _REGISTRY for name in sorted(_REGISTRY): result = runner.invoke(cli, ["domains", "info", name, "--json"]) assert result.exit_code == 0, f"domains info {name!r} failed: {result.output}" def test_concurrent_info_calls_no_shared_state(self) -> None: """8 threads each call domains info in isolation — no race conditions.""" import threading errors: list[str] = [] from muse.cli.commands.domains import _build_entry, _active_domain, _find_repo_root from muse.plugins.registry import _REGISTRY def worker(idx: int) -> None: try: plugin = _REGISTRY.get("code") if plugin is None: errors.append(f"Thread {idx}: code plugin not found") return active = _active_domain(_find_repo_root()) entry = _build_entry("code", plugin, active) if entry["domain"] != "code": errors.append(f"Thread {idx}: wrong domain {entry['domain']!r}") if not isinstance(entry["active"], bool): errors.append(f"Thread {idx}: active is not bool") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent failures: {errors}" # --------------------------------------------------------------------------- # Extended — muse domains use (deeper coverage) # --------------------------------------------------------------------------- class TestDomainsUseExtended: def test_j_alias_works(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "-j"]) assert result.exit_code == 0, result.output data = json.loads(result.output.strip()) assert data["domain"] == "code" def test_help_mentions_json_flag(self) -> None: result = runner.invoke(cli, ["domains", "use", "--help"]) assert "--json" in result.output or "-j" in result.output def test_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["domains", "use", "--help"]) assert "Exit code" in result.output or "exit code" in result.output def test_json_is_compact_single_line(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) assert result.exit_code == 0, result.output lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) == 1, f"Expected 1 non-empty line, got {len(lines)}" def test_json_parses_cleanly(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) data = json.loads(result.output.strip()) assert isinstance(data, dict) def test_json_domain_matches_arg(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "scaffold", "--json"]) data = json.loads(result.output.strip()) assert data["domain"] == "scaffold" def test_json_status_is_switched(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) data = json.loads(result.output.strip()) assert data["status"] == "switched" def test_json_repo_ends_with_muse(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) data = json.loads(result.output.strip()) assert data["repo"].endswith(".muse") def test_repo_json_domain_actually_updated(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): runner.invoke(cli, ["domains", "use", "code"]) written = json.loads((repo_json_path(tmp_path)).read_text()) assert written["domain"] == "code" def test_existing_fields_preserved(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="scaffold") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): runner.invoke(cli, ["domains", "use", "code"]) written = json.loads((repo_json_path(tmp_path)).read_text()) assert "repo_id" in written assert "schema_version" in written def test_idempotent_same_domain_exits_zero(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 0 def test_switch_code_to_scaffold(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "scaffold"]) assert result.exit_code == 0 written = json.loads((repo_json_path(tmp_path)).read_text()) assert written["domain"] == "scaffold" def test_text_success_mentions_domain_name(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert "code" in result.output def test_text_success_mentions_repo_path(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert "Repo:" in result.output or ".muse" in result.output def test_unknown_domain_lists_known(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "no_such_domain"]) assert result.exit_code == 1 assert "code" in result.stderr # known domains listed def test_no_repo_exits_1(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 1 def test_corrupt_repo_json_exits_1(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (repo_json_path(tmp_path)).write_text("{{bad json}}", encoding="utf-8") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 1 def test_extra_repo_json_fields_preserved(self, tmp_path: pathlib.Path) -> None: """Custom extra fields in repo.json survive a domain switch.""" muse = muse_dir(tmp_path) muse.mkdir(parents=True, exist_ok=True) (muse / "repo.json").write_text( json.dumps({ "repo_id": "my-repo", "schema_version": "1.0", "domain": "scaffold", "custom_field": "keep_me", }), encoding="utf-8", ) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 0 written = json.loads((repo_json_path(tmp_path)).read_text()) assert written["custom_field"] == "keep_me" assert written["domain"] == "code" # --------------------------------------------------------------------------- # Security — muse domains use # --------------------------------------------------------------------------- class TestDomainsUseSecurity: def test_ansi_in_unknown_domain_name_stripped(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "\x1b[31mmalicious\x1b[0m"]) assert result.exit_code == 1 assert "\x1b[" not in result.output def test_unregistered_domain_never_written(self, tmp_path: pathlib.Path) -> None: """An unregistered domain name is rejected before repo.json is touched.""" _init_repo(tmp_path, domain="code") original = (repo_json_path(tmp_path)).read_text() with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "malicious_domain"]) assert result.exit_code == 1 # repo.json must be unchanged assert (repo_json_path(tmp_path)).read_text() == original def test_domain_name_sanitized_in_success_text(self, tmp_path: pathlib.Path) -> None: """Domain name in success text is sanitized (no raw ANSI from env).""" _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_repo_path_sanitized_in_success_text(self, tmp_path: pathlib.Path) -> None: """Repo path in success text is sanitized.""" _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 0 assert "\x1b[" not in result.output def test_json_stdout_only_no_stray_text(self, tmp_path: pathlib.Path) -> None: """In JSON mode, stdout must be valid JSON and nothing else.""" _init_repo(tmp_path) with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) assert result.exit_code == 0 json.loads(result.output.strip()) def test_no_traceback_on_corrupt_repo_json(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) (repo_json_path(tmp_path)).write_text("{bad}", encoding="utf-8") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "use", "code"]) assert result.exit_code == 1 assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress — muse domains use # --------------------------------------------------------------------------- class TestDomainsUseStress: def test_50_sequential_use_calls_all_succeed(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): for i in range(50): result = runner.invoke(cli, ["domains", "use", "code", "--json"]) assert result.exit_code == 0, f"Call {i} failed: {result.output}" def test_alternate_domains_100_times(self, tmp_path: pathlib.Path) -> None: """Switch between code and scaffold 100 times — file integrity preserved.""" _init_repo(tmp_path, domain="code") domains = ["code", "scaffold"] with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): for i in range(100): target = domains[i % 2] result = runner.invoke(cli, ["domains", "use", target]) assert result.exit_code == 0, f"Switch {i} to {target!r} failed" written = json.loads((repo_json_path(tmp_path)).read_text()) assert written["domain"] in domains def test_concurrent_use_isolated_repos(self, tmp_path: pathlib.Path) -> None: """8 threads each switch domain on their own isolated repo.""" import argparse import threading from muse.cli.commands.domains import run_use errors: list[str] = [] def worker(idx: int) -> None: repo = tmp_path / f"repo_{idx}" _init_repo(repo, domain="scaffold") try: args = argparse.Namespace(use_name="code", json_out=True) with patch("muse.cli.commands.domains._find_repo_root", return_value=repo): run_use(args) written = json.loads((repo_json_path(repo)).read_text()) if written["domain"] != "code": errors.append(f"Thread {idx}: domain is {written['domain']!r}") except SystemExit as exc: errors.append(f"Thread {idx}: exit {exc.code}") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent failures: {errors}" # --------------------------------------------------------------------------- # Extended — muse domains validate # --------------------------------------------------------------------------- class TestDomainsValidateExtended: def test_j_alias_works(self) -> None: """-j is equivalent to --json.""" with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "-j"]) assert result.exit_code == 0 data = json.loads(result.output.strip()) assert data["domain"] == "code" def test_help_flag(self) -> None: result = runner.invoke(cli, ["domains", "validate", "--help"]) assert result.exit_code == 0 assert "validate" in result.output.lower() def test_json_compact_no_indent(self) -> None: """JSON output must be a single line (compact, no indent=2).""" with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "-j"]) assert result.exit_code == 0 lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) == 1, f"Expected compact JSON, got {len(lines)} lines" def test_json_fields_present(self) -> None: """JSON object contains domain, ok, checks.""" with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "-j"]) data = json.loads(result.output.strip()) assert set(data.keys()) >= {"domain", "ok", "checks"} def test_json_domain_matches_arg(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "scaffold", "-j"]) data = json.loads(result.output.strip()) assert data["domain"] == "scaffold" def test_json_checks_are_objects(self) -> None: """Each entry in checks has name, ok, detail.""" with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "-j"]) data = json.loads(result.output.strip()) for c in data["checks"]: assert "name" in c assert "ok" in c assert "detail" in c def test_text_output_shows_checkmarks(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code"]) assert "✅" in result.output or "✓" in result.output def test_text_output_shows_domain_name(self) -> None: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "scaffold"]) assert "scaffold" in result.output def test_no_name_in_repo_validates_active_domain(self, tmp_path: pathlib.Path) -> None: """When inside a repo with domain=code, validate with no name validates code.""" _init_repo(tmp_path, domain="code") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "validate", "-j"]) assert result.exit_code == 0 data = json.loads(result.output.strip()) assert data["domain"] == "code" def test_no_name_no_repo_validates_all_json(self) -> None: """With no name and no repo, all registered domains are validated.""" from muse.plugins.registry import _REGISTRY with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "-j"]) assert result.exit_code == 0 if len(_REGISTRY) == 1: data = json.loads(result.output.strip()) assert "domain" in data else: data = json.loads(result.output.strip()) if isinstance(data, dict) and "results" in data: data = data["results"] assert isinstance(data, list) assert len(data) == len(_REGISTRY) def test_unregistered_domain_exit1(self) -> None: result = runner.invoke(cli, ["domains", "validate", "does-not-exist"]) assert result.exit_code == 1 def test_broken_plugin_ok_false(self) -> None: """A plugin with no required methods emits ok=false.""" from muse.plugins.registry import _REGISTRY from unittest.mock import MagicMock mock = MagicMock(spec=[]) with patch.dict(_REGISTRY, {"broken2": mock}): result = runner.invoke(cli, ["domains", "validate", "broken2", "-j"]) assert result.exit_code == 1 data = json.loads(result.output.strip()) assert data["ok"] is False def test_broken_plugin_checks_have_failed_entries(self) -> None: from muse.plugins.registry import _REGISTRY from unittest.mock import MagicMock mock = MagicMock(spec=[]) with patch.dict(_REGISTRY, {"broken3": mock}): result = runner.invoke(cli, ["domains", "validate", "broken3", "-j"]) data = json.loads(result.output.strip()) failed = [c for c in data["checks"] if not c["ok"]] assert len(failed) >= 1 def test_multi_domain_json_is_list(self) -> None: """When validating all (no name, no repo) with >=2 domains, output is a list.""" from muse.plugins.registry import _REGISTRY with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "-j"]) assert result.exit_code == 0 if len(_REGISTRY) >= 2: data = json.loads(result.output.strip()) if isinstance(data, dict) and "results" in data: data = data["results"] assert isinstance(data, list) def test_active_domain_not_in_registry_falls_back_to_all(self, tmp_path: pathlib.Path) -> None: """Active domain removed from registry → falls back to validating all.""" _init_repo(tmp_path, domain="ghost") with patch("muse.cli.commands.domains._find_repo_root", return_value=tmp_path): result = runner.invoke(cli, ["domains", "validate", "-j"]) assert result.exit_code == 0 # should validate registry domains, not error on missing ghost def test_help_shows_agent_quickstart(self) -> None: result = runner.invoke(cli, ["domains", "validate", "--help"]) assert result.exit_code == 0 assert "Agent quickstart" in result.output def test_help_shows_json_schema(self) -> None: result = runner.invoke(cli, ["domains", "validate", "--help"]) assert "JSON output schema" in result.output def test_help_shows_exit_codes(self) -> None: result = runner.invoke(cli, ["domains", "validate", "--help"]) assert "Exit codes" in result.output # --------------------------------------------------------------------------- # Security — muse domains validate # --------------------------------------------------------------------------- class TestDomainsValidateSecurity: def test_ansi_in_domain_name_stripped_in_error(self) -> None: """ANSI in unknown domain name must not bleed into stderr.""" result = runner.invoke(cli, ["domains", "validate", "\x1b[31mmalicious\x1b[0m"]) assert result.exit_code == 1 assert "\x1b" not in result.output def test_unregistered_domain_never_checked(self) -> None: """An unregistered name exits before any check logic runs.""" result = runner.invoke(cli, ["domains", "validate", "notareal_domain_xyz"]) assert result.exit_code == 1 assert "not registered" in result.stderr def test_json_output_only_on_stdout(self, capsys: pytest.CaptureFixture) -> None: """JSON is emitted to stdout; errors go to stderr only.""" with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "code", "-j"]) assert result.exit_code == 0 json.loads(result.output.strip()) # must parse cleanly def test_ansi_in_check_detail_sanitized_text_mode(self) -> None: """Plugin with no schema attr exercises AttributeError detail path; output has no ANSI.""" from muse.plugins.registry import _REGISTRY from unittest.mock import MagicMock # spec omits schema → _run_validate_plugin catches AttributeError, writes detail string mock = MagicMock(spec=["snapshot", "diff", "merge", "drift", "apply"]) with patch.dict(_REGISTRY, {"ansi_test": mock}): result = runner.invoke(cli, ["domains", "validate", "ansi_test"]) assert "\x1b" not in result.output def test_no_traceback_on_unknown_domain(self) -> None: result = runner.invoke(cli, ["domains", "validate", "totally_unknown_domain"]) assert "Traceback" not in result.output def test_multi_domain_validate_all_domains_present_in_output(self) -> None: """When validating all, every registered domain name appears in output.""" from muse.plugins.registry import _REGISTRY with patch("muse.cli.commands.domains._find_repo_root", return_value=None): result = runner.invoke(cli, ["domains", "validate", "-j"]) assert result.exit_code == 0 output = result.output for domain_name in _REGISTRY: assert domain_name in output, f"Missing domain {domain_name!r} in output" # --------------------------------------------------------------------------- # Stress — muse domains validate # --------------------------------------------------------------------------- class TestDomainsValidateStress: def test_50_sequential_validate_calls(self) -> None: """50 sequential validate calls all succeed.""" with patch("muse.cli.commands.domains._find_repo_root", return_value=None): for i in range(50): result = runner.invoke(cli, ["domains", "validate", "code", "-j"]) assert result.exit_code == 0, f"Call {i} failed: {result.output}" def test_alternate_code_scaffold_100_times(self) -> None: """Alternate between validating code and scaffold 100 times.""" domains = ["code", "scaffold"] with patch("muse.cli.commands.domains._find_repo_root", return_value=None): for i in range(100): target = domains[i % 2] result = runner.invoke(cli, ["domains", "validate", target, "-j"]) assert result.exit_code == 0, f"Step {i}: {result.output}" data = json.loads(result.output.strip()) assert data["domain"] == target def test_concurrent_validate_8_threads(self) -> None: """8 threads each validate code concurrently using core function directly.""" import argparse import threading from muse.cli.commands.domains import run_validate errors: list[str] = [] def worker(idx: int) -> None: args = argparse.Namespace(validate_name="code", json_out=True) try: with patch("muse.cli.commands.domains._find_repo_root", return_value=None): run_validate(args) except SystemExit as exc: if exc.code != 0: errors.append(f"Thread {idx}: exit {exc.code}") except Exception as exc: errors.append(f"Thread {idx}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent failures: {errors}" # --------------------------------------------------------------------------- # TestRegisterFlags — --json / -j normalized on every domains subparser # --------------------------------------------------------------------------- class TestRegisterFlags: """Every domains subparser must register --json with -j shorthand.""" def _make_parser(self) -> argparse.ArgumentParser: from muse.cli.commands.domains import register root = argparse.ArgumentParser() subs = root.add_subparsers() register(subs) return root def test_domains_j_alias_exits_zero(self) -> None: result = runner.invoke(cli, ["domains", "-j"]) assert result.exit_code == 0, result.output def test_domains_j_alias_valid_json(self) -> None: result = runner.invoke(cli, ["domains", "-j"]) json.loads(result.output) # must not raise def test_domains_j_alias_has_domains_key(self) -> None: result = runner.invoke(cli, ["domains", "-j"]) assert "domains" in json.loads(result.output) def test_domains_json_out_default_false(self) -> None: p = self._make_parser() ns = p.parse_args(["domains"]) assert ns.json_out is False def test_domains_json_out_true_with_json_flag(self) -> None: p = self._make_parser() ns = p.parse_args(["domains", "--json"]) assert ns.json_out is True def test_domains_json_out_true_with_j_flag(self) -> None: p = self._make_parser() ns = p.parse_args(["domains", "-j"]) assert ns.json_out is True def test_validate_j_alias_exits_zero(self) -> None: result = runner.invoke(cli, ["domains", "validate", "-j"]) assert result.exit_code == 0, result.output def test_validate_j_alias_valid_json(self) -> None: result = runner.invoke(cli, ["domains", "validate", "-j"]) json.loads(result.output) # must not raise def test_info_j_alias_exits_zero(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "-j"]) assert result.exit_code == 0, result.output def test_info_j_alias_valid_json(self) -> None: result = runner.invoke(cli, ["domains", "info", "code", "-j"]) json.loads(result.output) # must not raise