gabriel / muse public
test_plumbing_domain_info.py python
288 lines 11.5 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive tests for ``muse plumbing domain-info``.
2
3 Audit findings addressed here
4 ------------------------------
5 Security
6 - Format error now goes to stderr (was stdout) — verified below.
7 - ANSI injection in domain name stripped in text mode.
8
9 Agent UX
10 - ``--domain <name>`` — inspect any domain without entering its repo.
11 - ``--capabilities-only`` — lightweight capability check.
12 - ``--all-domains`` — enumerate the registry.
13
14 Docs
15 - Capabilities section added to module docstring.
16
17 Coverage tiers
18 --------------
19 - Unit: _FORMAT_CHOICES, _CapabilitiesDict schema
20 - Integration: --all-domains JSON/text, --domain flag, --capabilities-only,
21 active-repo mode, JSON/text format, registered_domains present
22 - Security: ANSI in domain name stripped in text, format error to stderr
23 - Stress: 200 sequential --all-domains calls
24 """
25 from __future__ import annotations
26
27 import json
28 import pathlib
29
30 from muse.core.errors import ExitCode
31 from tests.cli_test_helper import CliRunner, InvokeResult
32
33 runner = CliRunner()
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40 def _make_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path:
41 repo = tmp_path / "repo"
42 muse = repo / ".muse"
43 for sub in ("objects", "commits", "snapshots", "refs/heads"):
44 (muse / sub).mkdir(parents=True)
45 (muse / "HEAD").write_text("ref: refs/heads/main")
46 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": domain}))
47 return repo
48
49
50 def _di(repo: pathlib.Path | None, *args: str) -> InvokeResult:
51 from muse.cli.app import main as cli
52 env = {"MUSE_REPO_ROOT": str(repo)} if repo is not None else {}
53 return runner.invoke(cli, ["domain-info", *args], env=env)
54
55
56 # ---------------------------------------------------------------------------
57 # Unit
58 # ---------------------------------------------------------------------------
59
60
61 class TestUnit:
62 def test_format_choices(self) -> None:
63 from muse.cli.commands.plumbing.domain_info import _FORMAT_CHOICES
64 assert "json" in _FORMAT_CHOICES
65 assert "text" in _FORMAT_CHOICES
66
67 def test_capabilities_dict_fields(self) -> None:
68 from muse.cli.commands.plumbing.domain_info import _CapabilitiesDict
69 fields = set(_CapabilitiesDict.__annotations__.keys())
70 assert "structured_merge" in fields
71 assert "crdt" in fields
72 assert "rerere" in fields
73
74
75 # ---------------------------------------------------------------------------
76 # Integration — --all-domains
77 # ---------------------------------------------------------------------------
78
79
80 class TestAllDomains:
81 def test_json_returns_list(self, tmp_path: pathlib.Path) -> None:
82 repo = _make_repo(tmp_path)
83 result = _di(repo, "--all-domains")
84 assert result.exit_code == 0
85 data = json.loads(result.output)
86 assert "registered_domains" in data
87 assert isinstance(data["registered_domains"], list)
88 assert len(data["registered_domains"]) > 0
89
90 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
91 repo = _make_repo(tmp_path)
92 result = _di(repo, "--all-domains", "--json")
93 assert result.exit_code == 0
94 assert "registered_domains" in json.loads(result.output)
95
96 def test_text_one_per_line(self, tmp_path: pathlib.Path) -> None:
97 repo = _make_repo(tmp_path)
98 result = _di(repo, "--all-domains", "--format", "text")
99 assert result.exit_code == 0
100 lines = [l for l in result.output.splitlines() if l.strip()]
101 assert len(lines) > 0
102
103 def test_no_repo_required(self, tmp_path: pathlib.Path) -> None:
104 """--all-domains must not require a Muse repository."""
105 result = _di(None, "--all-domains",
106 "--json")
107 assert result.exit_code == 0
108 data = json.loads(result.output)
109 assert "registered_domains" in data
110
111 def test_code_domain_present(self, tmp_path: pathlib.Path) -> None:
112 repo = _make_repo(tmp_path)
113 data = json.loads(_di(repo, "--all-domains").output)
114 assert "code" in data["registered_domains"]
115
116
117 # ---------------------------------------------------------------------------
118 # Integration — --domain flag (new agent UX)
119 # ---------------------------------------------------------------------------
120
121
122 class TestDomainFlag:
123 def test_inspect_code_domain_without_repo(self, tmp_path: pathlib.Path) -> None:
124 """Agents can inspect a domain without being inside its repo."""
125 result = _di(None, "--domain", "code")
126 assert result.exit_code == 0
127 data = json.loads(result.output)
128 assert data["domain"] == "code"
129 assert "capabilities" in data
130
131 def test_inspect_code_capabilities_only(self, tmp_path: pathlib.Path) -> None:
132 result = _di(None, "--domain", "code", "--capabilities-only")
133 assert result.exit_code == 0
134 data = json.loads(result.output)
135 assert "capabilities" in data
136 assert "schema" not in data
137
138 def test_unknown_domain_errors(self, tmp_path: pathlib.Path) -> None:
139 result = _di(None, "--domain", "nonexistent-domain")
140 assert result.exit_code == ExitCode.USER_ERROR
141
142 def test_invalid_domain_name_rejected(self, tmp_path: pathlib.Path) -> None:
143 """Domain names must match the validation regex."""
144 result = _di(None, "--domain", "UPPERCASE")
145 assert result.exit_code == ExitCode.USER_ERROR
146
147 def test_domain_flag_overrides_repo_domain(self, tmp_path: pathlib.Path) -> None:
148 """--domain should be used even when inside a repo with a different domain."""
149 repo = _make_repo(tmp_path, domain="code")
150 result = _di(repo, "--domain", "code")
151 assert result.exit_code == 0
152 data = json.loads(result.output)
153 assert data["domain"] == "code"
154
155
156 # ---------------------------------------------------------------------------
157 # Integration — --capabilities-only
158 # ---------------------------------------------------------------------------
159
160
161 class TestCapabilitiesOnly:
162 def test_json_has_no_schema_key(self, tmp_path: pathlib.Path) -> None:
163 repo = _make_repo(tmp_path)
164 result = _di(repo, "--capabilities-only")
165 assert result.exit_code == 0
166 data = json.loads(result.output)
167 assert "schema" not in data
168 assert "capabilities" in data
169 assert "domain" in data
170
171 def test_capabilities_are_booleans(self, tmp_path: pathlib.Path) -> None:
172 repo = _make_repo(tmp_path)
173 data = json.loads(_di(repo, "--capabilities-only").output)
174 caps = data["capabilities"]
175 for key in ("structured_merge", "crdt", "rerere"):
176 assert key in caps
177 assert isinstance(caps[key], bool)
178
179 def test_text_format_capabilities_only(self, tmp_path: pathlib.Path) -> None:
180 repo = _make_repo(tmp_path)
181 result = _di(repo, "--capabilities-only", "--format", "text")
182 assert result.exit_code == 0
183 assert "Domain:" in result.output
184 assert "Capabilities:" in result.output
185 assert "Plugin:" not in result.output
186
187 def test_domain_flag_and_capabilities_only(self, tmp_path: pathlib.Path) -> None:
188 """Agents use this combo constantly for merge-strategy negotiation."""
189 result = _di(None, "--domain", "code", "--capabilities-only")
190 assert result.exit_code == 0
191 data = json.loads(result.output)
192 assert data["domain"] == "code"
193 assert "capabilities" in data
194
195
196 # ---------------------------------------------------------------------------
197 # Integration — active-repo mode
198 # ---------------------------------------------------------------------------
199
200
201 class TestActiveRepoMode:
202 def test_json_output_keys(self, tmp_path: pathlib.Path) -> None:
203 repo = _make_repo(tmp_path)
204 result = _di(repo)
205 assert result.exit_code == 0
206 data = json.loads(result.output)
207 for key in ("domain", "plugin_class", "capabilities", "schema", "registered_domains"):
208 assert key in data, f"missing key: {key}"
209
210 def test_domain_matches_repo(self, tmp_path: pathlib.Path) -> None:
211 repo = _make_repo(tmp_path, domain="code")
212 data = json.loads(_di(repo).output)
213 assert data["domain"] == "code"
214
215 def test_registered_domains_in_output(self, tmp_path: pathlib.Path) -> None:
216 repo = _make_repo(tmp_path)
217 data = json.loads(_di(repo).output)
218 assert isinstance(data["registered_domains"], list)
219 assert "code" in data["registered_domains"]
220
221 def test_text_format_shows_domain(self, tmp_path: pathlib.Path) -> None:
222 repo = _make_repo(tmp_path)
223 result = _di(repo, "--format", "text")
224 assert result.exit_code == 0
225 assert "Domain:" in result.output
226 assert "Plugin:" in result.output
227 assert "Capabilities:" in result.output
228
229 def test_unknown_domain_in_repo_errors(self, tmp_path: pathlib.Path) -> None:
230 repo = _make_repo(tmp_path, domain="unknown-domain-xyz")
231 result = _di(repo)
232 assert result.exit_code == ExitCode.USER_ERROR
233
234
235 # ---------------------------------------------------------------------------
236 # Security
237 # ---------------------------------------------------------------------------
238
239
240 class TestSecurity:
241 def test_ansi_in_domain_stripped_text(self, tmp_path: pathlib.Path) -> None:
242 """A maliciously crafted repo.json with ANSI in domain is safe in text mode."""
243 repo = _make_repo(tmp_path)
244 (repo / ".muse" / "repo.json").write_text(
245 '{"repo_id": "test", "domain": "\\u001b[31mevil\\u001b[0m"}'
246 )
247 result = _di(repo, "--format", "text")
248 # Command will fail because "evil" is not a registered domain,
249 # but the important thing is no raw ANSI in output
250 assert "\x1b" not in result.output
251
252 def test_format_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
253 repo = _make_repo(tmp_path)
254 result = _di(repo, "--format", "msgpack", "--all-domains")
255 assert result.exit_code == ExitCode.USER_ERROR
256 # The error must be on stderr — check stderr has the message
257 assert "error" in result.stderr.lower()
258 # stdout (binary) must be empty — the error didn't go there
259 assert result.stdout_bytes == b""
260
261 def test_no_traceback_on_bad_domain(self, tmp_path: pathlib.Path) -> None:
262 result = _di(None, "--domain", "nonexistent-domain")
263 assert "Traceback" not in result.output
264
265 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
266 repo = _make_repo(tmp_path)
267 result = _di(repo, "--format", "yaml", "--all-domains")
268 assert "Traceback" not in result.output
269
270
271 # ---------------------------------------------------------------------------
272 # Stress
273 # ---------------------------------------------------------------------------
274
275
276 class TestStress:
277 def test_200_all_domains_calls(self, tmp_path: pathlib.Path) -> None:
278 for i in range(200):
279 result = _di(None, "--all-domains")
280 assert result.exit_code == 0, f"failed at iteration {i}"
281 data = json.loads(result.output)
282 assert "registered_domains" in data
283
284 def test_200_capabilities_only_calls(self, tmp_path: pathlib.Path) -> None:
285 for i in range(200):
286 result = _di(None, "--domain", "code", "--capabilities-only")
287 assert result.exit_code == 0, f"failed at iteration {i}"
288 assert "capabilities" in json.loads(result.output)
File History 3 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago