"""Comprehensive tests for ``muse 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: _CapabilitiesDict schema, flag registration - Integration: --all-domains JSON/text, --domain flag, --capabilities-only, active-repo mode, JSON output, registered_domains present - Security: ANSI in domain name stripped in text, unknown flag exits non-zero - Stress: 200 sequential --all-domains calls """ from __future__ import annotations import argparse import json import pathlib from typing import TYPE_CHECKING from muse.core.errors import ExitCode from muse.core.paths import muse_dir, repo_json_path 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" dot_muse = muse_dir(repo) for sub in ("objects", "commits", "snapshots", "refs/heads"): (dot_muse / sub).mkdir(parents=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main") (dot_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 TestRegisterFlags: def _parse(self, *args: str) -> "argparse.Namespace": import argparse from muse.cli.commands.domain_info import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["domain-info", *args]) def test_default_json_out_is_false(self) -> None: ns = self._parse() assert ns.json_out is False def test_json_flag_sets_json_out(self) -> None: ns = self._parse("--json") assert ns.json_out is True def test_j_shorthand_sets_json_out(self) -> None: ns = self._parse("-j") assert ns.json_out is True class TestUnit: def test_capabilities_dict_fields(self) -> None: from muse.cli.commands.domain_info import _CapabilitiesDict fields = set(_CapabilitiesDict.__annotations__.keys()) assert "structured_merge" in fields assert "crdt" in fields assert "harmony" 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", "--json") 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") 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", "--json").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", "--json") 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", "--json") assert result.exit_code == 0 data = json.loads(result.output) assert "capabilities" in data assert "domain_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", "--json") assert result.exit_code == 0 data = json.loads(result.output) assert data["domain"] == "code" # --------------------------------------------------------------------------- # Integration — --capabilities-only # --------------------------------------------------------------------------- class TestCapabilitiesOnly: def test_json_has_no_domain_schema_key(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--capabilities-only", "--json") assert result.exit_code == 0 data = json.loads(result.output) assert "domain_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", "--json").output) caps = data["capabilities"] for key in ("structured_merge", "crdt", "harmony"): 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") 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", "--json") 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, "--json") assert result.exit_code == 0 data = json.loads(result.output) for key in ("domain", "plugin_class", "capabilities", "domain_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, "--json").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, "--json").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) 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_json_path(repo)).write_text( '{"repo_id": "test", "domain": "\\u001b[31mmalicious\\u001b[0m"}' ) result = _di(repo) # Command will fail because "malicious" is not a registered domain, # but the important thing is no raw ANSI in output assert "\x1b" not in result.output def test_unknown_flag_exits_nonzero(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _di(repo, "--format", "msgpack", "--all-domains") assert result.exit_code != 0 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_unknown_flag(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", "--json") 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", "--json") assert result.exit_code == 0, f"failed at iteration {i}" assert "capabilities" in json.loads(result.output)