"""Comprehensive tests for ``muse plumbing domain-info``. Audit findings addressed here ------------------------------ Security - Format error now goes to stderr (was stdout) — verified below. - ANSI injection in domain name stripped in text mode. Agent UX - ``--domain `` — inspect any domain without entering its repo. - ``--capabilities-only`` — lightweight capability check. - ``--all-domains`` — enumerate the registry. Docs - Capabilities section added to module docstring. Coverage tiers -------------- - Unit: _FORMAT_CHOICES, _CapabilitiesDict schema - Integration: --all-domains JSON/text, --domain flag, --capabilities-only, active-repo mode, JSON/text format, registered_domains present - Security: ANSI in domain name stripped in text, format error to stderr - Stress: 200 sequential --all-domains calls """ from __future__ import annotations import json import pathlib from muse.core.errors import ExitCode from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path: repo = tmp_path / "repo" muse = repo / ".muse" for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": domain})) return repo def _di(repo: pathlib.Path | None, *args: str) -> InvokeResult: from muse.cli.app import main as cli env = {"MUSE_REPO_ROOT": str(repo)} if repo is not None else {} return runner.invoke(cli, ["domain-info", *args], env=env) # --------------------------------------------------------------------------- # Unit # --------------------------------------------------------------------------- class TestUnit: def test_format_choices(self) -> None: from muse.cli.commands.plumbing.domain_info import _FORMAT_CHOICES assert "json" in _FORMAT_CHOICES assert "text" in _FORMAT_CHOICES def test_capabilities_dict_fields(self) -> None: from muse.cli.commands.plumbing.domain_info import _CapabilitiesDict fields = set(_CapabilitiesDict.__annotations__.keys()) assert "structured_merge" in fields assert "crdt" in fields assert "rerere" in fields # --------------------------------------------------------------------------- # Integration — --all-domains # --------------------------------------------------------------------------- class TestAllDomains: def test_json_returns_list(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--all-domains") assert result.exit_code == 0 data = json.loads(result.output) assert "registered_domains" in data assert isinstance(data["registered_domains"], list) assert len(data["registered_domains"]) > 0 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--all-domains", "--json") assert result.exit_code == 0 assert "registered_domains" in json.loads(result.output) def test_text_one_per_line(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--all-domains", "--format", "text") assert result.exit_code == 0 lines = [l for l in result.output.splitlines() if l.strip()] assert len(lines) > 0 def test_no_repo_required(self, tmp_path: pathlib.Path) -> None: """--all-domains must not require a Muse repository.""" result = _di(None, "--all-domains", "--json") assert result.exit_code == 0 data = json.loads(result.output) assert "registered_domains" in data def test_code_domain_present(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) data = json.loads(_di(repo, "--all-domains").output) assert "code" in data["registered_domains"] # --------------------------------------------------------------------------- # Integration — --domain flag (new agent UX) # --------------------------------------------------------------------------- class TestDomainFlag: def test_inspect_code_domain_without_repo(self, tmp_path: pathlib.Path) -> None: """Agents can inspect a domain without being inside its repo.""" result = _di(None, "--domain", "code") assert result.exit_code == 0 data = json.loads(result.output) assert data["domain"] == "code" assert "capabilities" in data def test_inspect_code_capabilities_only(self, tmp_path: pathlib.Path) -> None: result = _di(None, "--domain", "code", "--capabilities-only") assert result.exit_code == 0 data = json.loads(result.output) assert "capabilities" in data assert "schema" not in data def test_unknown_domain_errors(self, tmp_path: pathlib.Path) -> None: result = _di(None, "--domain", "nonexistent-domain") assert result.exit_code == ExitCode.USER_ERROR def test_invalid_domain_name_rejected(self, tmp_path: pathlib.Path) -> None: """Domain names must match the validation regex.""" result = _di(None, "--domain", "UPPERCASE") assert result.exit_code == ExitCode.USER_ERROR def test_domain_flag_overrides_repo_domain(self, tmp_path: pathlib.Path) -> None: """--domain should be used even when inside a repo with a different domain.""" repo = _make_repo(tmp_path, domain="code") result = _di(repo, "--domain", "code") assert result.exit_code == 0 data = json.loads(result.output) assert data["domain"] == "code" # --------------------------------------------------------------------------- # Integration — --capabilities-only # --------------------------------------------------------------------------- class TestCapabilitiesOnly: def test_json_has_no_schema_key(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--capabilities-only") assert result.exit_code == 0 data = json.loads(result.output) assert "schema" not in data assert "capabilities" in data assert "domain" in data def test_capabilities_are_booleans(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) data = json.loads(_di(repo, "--capabilities-only").output) caps = data["capabilities"] for key in ("structured_merge", "crdt", "rerere"): assert key in caps assert isinstance(caps[key], bool) def test_text_format_capabilities_only(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--capabilities-only", "--format", "text") assert result.exit_code == 0 assert "Domain:" in result.output assert "Capabilities:" in result.output assert "Plugin:" not in result.output def test_domain_flag_and_capabilities_only(self, tmp_path: pathlib.Path) -> None: """Agents use this combo constantly for merge-strategy negotiation.""" result = _di(None, "--domain", "code", "--capabilities-only") assert result.exit_code == 0 data = json.loads(result.output) assert data["domain"] == "code" assert "capabilities" in data # --------------------------------------------------------------------------- # Integration — active-repo mode # --------------------------------------------------------------------------- class TestActiveRepoMode: def test_json_output_keys(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo) assert result.exit_code == 0 data = json.loads(result.output) for key in ("domain", "plugin_class", "capabilities", "schema", "registered_domains"): assert key in data, f"missing key: {key}" def test_domain_matches_repo(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, domain="code") data = json.loads(_di(repo).output) assert data["domain"] == "code" def test_registered_domains_in_output(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) data = json.loads(_di(repo).output) assert isinstance(data["registered_domains"], list) assert "code" in data["registered_domains"] def test_text_format_shows_domain(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--format", "text") assert result.exit_code == 0 assert "Domain:" in result.output assert "Plugin:" in result.output assert "Capabilities:" in result.output def test_unknown_domain_in_repo_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path, domain="unknown-domain-xyz") result = _di(repo) assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_domain_stripped_text(self, tmp_path: pathlib.Path) -> None: """A maliciously crafted repo.json with ANSI in domain is safe in text mode.""" repo = _make_repo(tmp_path) (repo / ".muse" / "repo.json").write_text( '{"repo_id": "test", "domain": "\\u001b[31mevil\\u001b[0m"}' ) result = _di(repo, "--format", "text") # Command will fail because "evil" is not a registered domain, # but the important thing is no raw ANSI in output assert "\x1b" not in result.output def test_format_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--format", "msgpack", "--all-domains") assert result.exit_code == ExitCode.USER_ERROR # The error must be on stderr — check stderr has the message assert "error" in result.stderr.lower() # stdout (binary) must be empty — the error didn't go there assert result.stdout_bytes == b"" def test_no_traceback_on_bad_domain(self, tmp_path: pathlib.Path) -> None: result = _di(None, "--domain", "nonexistent-domain") assert "Traceback" not in result.output def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--format", "yaml", "--all-domains") assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_200_all_domains_calls(self, tmp_path: pathlib.Path) -> None: for i in range(200): result = _di(None, "--all-domains") assert result.exit_code == 0, f"failed at iteration {i}" data = json.loads(result.output) assert "registered_domains" in data def test_200_capabilities_only_calls(self, tmp_path: pathlib.Path) -> None: for i in range(200): result = _di(None, "--domain", "code", "--capabilities-only") assert result.exit_code == 0, f"failed at iteration {i}" assert "capabilities" in json.loads(result.output)