gabriel / muse public
test_plumbing_check_attr.py python
514 lines 19.3 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 check-attr``.
2
3 Audit findings addressed here
4 ------------------------------
5 Performance
6 - Default mode previously called ``resolve_strategy`` + ``_find_matching_rule``
7 separately → 2× rule-list iteration per path. Replaced with
8 ``_resolve_with_rule`` — single O(N) pass returning (strategy, rule).
9
10 Code quality
11 - ``--all-rules`` manual dim-match loop extracted into ``_all_matching_rules``.
12 - ``_dim_match`` helper eliminates repeated boolean expression.
13
14 Security
15 - Format error now goes to stderr (was stdout) — verified below.
16 - ANSI injection in text output stripped via sanitize_display().
17 - Null bytes in paths now rejected with USER_ERROR.
18
19 Agent UX
20 - ``--stdin`` — read paths from stdin (one per line, # comments skipped).
21 - ``--rules-only`` — emit loaded rules without testing paths.
22 - ``formatter_class`` added for clean --help rendering.
23 - ``nargs="*"`` on paths (was "+") to support --stdin and --rules-only.
24
25 Coverage tiers
26 --------------
27 - Unit: _dim_match, _resolve_with_rule, _all_matching_rules, _rule_to_dict,
28 _RuleDict / _PathResult schemas
29 - Integration: JSON/text formats, --dimension filter, --all-rules, --rules-only,
30 --stdin, empty .museattributes, multiple rules priority, negation
31 - Security: null bytes rejected, ANSI stripped in text, format error→stderr,
32 no tracebacks, bad TOML errors cleanly
33 - Stress: 1 000-path batch, 50-rule set, 200 sequential runs, stdin 1 000 paths
34 """
35 from __future__ import annotations
36
37 import json
38 import pathlib
39
40 import pytest
41
42 from muse.core.attributes import AttributeRule
43 from muse.core.errors import ExitCode
44 from tests.cli_test_helper import CliRunner, InvokeResult
45
46 runner = CliRunner()
47
48
49 # ---------------------------------------------------------------------------
50 # Helpers
51 # ---------------------------------------------------------------------------
52
53 def _make_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path:
54 repo = tmp_path / "repo"
55 muse = repo / ".muse"
56 for sub in ("objects", "commits", "snapshots", "refs/heads"):
57 (muse / sub).mkdir(parents=True)
58 (muse / "HEAD").write_text("ref: refs/heads/main")
59 (muse / "repo.json").write_text(json.dumps({"repo_id": "r1", "domain": domain}))
60 return repo
61
62
63 def _write_attrs(repo: pathlib.Path, content: str) -> None:
64 (repo / ".museattributes").write_text(content)
65
66
67 _SIMPLE_ATTRS = """
68 [meta]
69 domain = "code"
70
71 [[rules]]
72 path = "build/*"
73 dimension = "*"
74 strategy = "ours"
75 comment = "Build artifacts prefer ours."
76 priority = 10
77
78 [[rules]]
79 path = "*.md"
80 dimension = "*"
81 strategy = "union"
82 priority = 5
83 """
84
85
86 def _ca(repo: pathlib.Path, *args: str, stdin: str | None = None) -> InvokeResult:
87 from muse.cli.app import main as cli
88 return runner.invoke(
89 cli,
90 ["check-attr", *args],
91 env={"MUSE_REPO_ROOT": str(repo)},
92 input=stdin,
93 )
94
95
96 def _make_rule(
97 path_pattern: str = "*",
98 dimension: str = "*",
99 strategy: str = "ours",
100 priority: int = 0,
101 source_index: int = 0,
102 ) -> AttributeRule:
103 from dataclasses import dataclass
104 return AttributeRule(
105 path_pattern=path_pattern,
106 dimension=dimension,
107 strategy=strategy,
108 priority=priority,
109 source_index=source_index,
110 )
111
112
113 # ---------------------------------------------------------------------------
114 # Unit — private helpers
115 # ---------------------------------------------------------------------------
116
117
118 class TestDimMatch:
119 def test_wildcard_rule_matches_any(self) -> None:
120 from muse.cli.commands.plumbing.check_attr import _dim_match
121 rule = _make_rule(dimension="*")
122 assert _dim_match(rule, "notes")
123 assert _dim_match(rule, "pitch_bend")
124
125 def test_exact_dim_matches(self) -> None:
126 from muse.cli.commands.plumbing.check_attr import _dim_match
127 rule = _make_rule(dimension="notes")
128 assert _dim_match(rule, "notes")
129
130 def test_exact_dim_no_match(self) -> None:
131 from muse.cli.commands.plumbing.check_attr import _dim_match
132 rule = _make_rule(dimension="notes")
133 assert not _dim_match(rule, "pitch_bend")
134
135 def test_wildcard_query_matches_any_rule(self) -> None:
136 from muse.cli.commands.plumbing.check_attr import _dim_match
137 rule = _make_rule(dimension="notes")
138 assert _dim_match(rule, "*")
139
140
141 class TestResolveWithRule:
142 def test_no_rules_returns_auto(self) -> None:
143 from muse.cli.commands.plumbing.check_attr import _resolve_with_rule
144 strategy, rule = _resolve_with_rule([], "tracks/drums.mid", "*")
145 assert strategy == "auto"
146 assert rule is None
147
148 def test_matching_rule_returned(self) -> None:
149 from muse.cli.commands.plumbing.check_attr import _resolve_with_rule
150 rules = [_make_rule(path_pattern="build/*", strategy="ours")]
151 strategy, rule = _resolve_with_rule(rules, "build/out.bin", "*")
152 assert strategy == "ours"
153 assert rule is not None
154 assert rule.path_pattern == "build/*"
155
156 def test_non_matching_path_returns_auto(self) -> None:
157 from muse.cli.commands.plumbing.check_attr import _resolve_with_rule
158 rules = [_make_rule(path_pattern="build/*", strategy="ours")]
159 strategy, rule = _resolve_with_rule(rules, "tracks/drums.mid", "*")
160 assert strategy == "auto"
161 assert rule is None
162
163 def test_first_match_wins(self) -> None:
164 from muse.cli.commands.plumbing.check_attr import _resolve_with_rule
165 rules = [
166 _make_rule(path_pattern="*.mid", strategy="ours", priority=10),
167 _make_rule(path_pattern="tracks/*", strategy="theirs", priority=5),
168 ]
169 strategy, rule = _resolve_with_rule(rules, "tracks/drums.mid", "*")
170 assert strategy == "ours"
171
172 def test_dimension_filter_respected(self) -> None:
173 from muse.cli.commands.plumbing.check_attr import _resolve_with_rule
174 rules = [_make_rule(path_pattern="*", dimension="notes", strategy="union")]
175 strategy, rule = _resolve_with_rule(rules, "tracks/drums.mid", "pitch_bend")
176 assert strategy == "auto"
177 assert rule is None
178
179
180 class TestAllMatchingRules:
181 def test_returns_all_matches(self) -> None:
182 from muse.cli.commands.plumbing.check_attr import _all_matching_rules
183 rules = [
184 _make_rule(path_pattern="*.mid", strategy="ours"),
185 _make_rule(path_pattern="tracks/*", strategy="theirs"),
186 _make_rule(path_pattern="build/*", strategy="union"),
187 ]
188 matched = _all_matching_rules(rules, "tracks/drums.mid", "*")
189 assert len(matched) == 2
190
191 def test_empty_when_no_match(self) -> None:
192 from muse.cli.commands.plumbing.check_attr import _all_matching_rules
193 rules = [_make_rule(path_pattern="build/*", strategy="ours")]
194 assert _all_matching_rules(rules, "tracks/drums.mid", "*") == []
195
196
197 class TestRuleToDict:
198 def test_all_fields_present(self) -> None:
199 from muse.cli.commands.plumbing.check_attr import _rule_to_dict
200 rule = _make_rule(path_pattern="*.mid", dimension="notes", strategy="ours",
201 priority=5, source_index=2)
202 d = _rule_to_dict(rule)
203 assert d["path_pattern"] == "*.mid"
204 assert d["dimension"] == "notes"
205 assert d["strategy"] == "ours"
206 assert d["priority"] == 5
207 assert d["source_index"] == 2
208
209
210 class TestSchemas:
211 def test_path_result_fields(self) -> None:
212 from muse.cli.commands.plumbing.check_attr import _PathResult
213 fields = set(_PathResult.__annotations__)
214 assert fields == {"path", "dimension", "strategy", "rule"}
215
216 def test_rule_dict_fields(self) -> None:
217 from muse.cli.commands.plumbing.check_attr import _RuleDict
218 fields = set(_RuleDict.__annotations__)
219 assert {"path_pattern", "dimension", "strategy", "comment",
220 "priority", "source_index"} == fields
221
222
223 # ---------------------------------------------------------------------------
224 # Integration — JSON output (default mode)
225 # ---------------------------------------------------------------------------
226
227
228 class TestJsonOutput:
229 def test_no_attrs_file_returns_auto(self, tmp_path: pathlib.Path) -> None:
230 repo = _make_repo(tmp_path)
231 result = _ca(repo, "tracks/drums.mid")
232 assert result.exit_code == 0
233 data = json.loads(result.output)
234 assert data["rules_loaded"] == 0
235 assert data["results"][0]["strategy"] == "auto"
236 assert data["results"][0]["rule"] is None
237
238 def test_matching_rule_present(self, tmp_path: pathlib.Path) -> None:
239 repo = _make_repo(tmp_path)
240 _write_attrs(repo, _SIMPLE_ATTRS)
241 data = json.loads(_ca(repo, "build/out.bin").output)
242 assert data["results"][0]["strategy"] == "ours"
243 assert data["results"][0]["rule"] is not None
244 assert data["results"][0]["rule"]["path_pattern"] == "build/*"
245
246 def test_non_matching_path_auto(self, tmp_path: pathlib.Path) -> None:
247 repo = _make_repo(tmp_path)
248 _write_attrs(repo, _SIMPLE_ATTRS)
249 data = json.loads(_ca(repo, "tracks/drums.mid").output)
250 assert data["results"][0]["strategy"] == "auto"
251
252 def test_multiple_paths(self, tmp_path: pathlib.Path) -> None:
253 repo = _make_repo(tmp_path)
254 _write_attrs(repo, _SIMPLE_ATTRS)
255 data = json.loads(_ca(repo, "build/out.bin", "README.md", "main.py").output)
256 assert len(data["results"]) == 3
257 strategies = [r["strategy"] for r in data["results"]]
258 assert strategies == ["ours", "union", "auto"]
259
260 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
261 repo = _make_repo(tmp_path)
262 result = _ca(repo, "--json", "foo.py")
263 assert result.exit_code == 0
264 assert "results" in json.loads(result.output)
265
266 def test_dimension_filter(self, tmp_path: pathlib.Path) -> None:
267 repo = _make_repo(tmp_path)
268 _write_attrs(repo, """
269 [[rules]]
270 path = "tracks/*"
271 dimension = "notes"
272 strategy = "union"
273 """)
274 data = json.loads(_ca(repo, "--dimension", "notes", "tracks/drums.mid").output)
275 assert data["results"][0]["strategy"] == "union"
276
277 def test_dimension_no_match_auto(self, tmp_path: pathlib.Path) -> None:
278 repo = _make_repo(tmp_path)
279 _write_attrs(repo, """
280 [[rules]]
281 path = "tracks/*"
282 dimension = "notes"
283 strategy = "union"
284 """)
285 data = json.loads(_ca(repo, "--dimension", "pitch_bend", "tracks/drums.mid").output)
286 assert data["results"][0]["strategy"] == "auto"
287
288
289 # ---------------------------------------------------------------------------
290 # Integration — text output
291 # ---------------------------------------------------------------------------
292
293
294 class TestTextOutput:
295 def test_strategy_in_output(self, tmp_path: pathlib.Path) -> None:
296 repo = _make_repo(tmp_path)
297 _write_attrs(repo, _SIMPLE_ATTRS)
298 result = _ca(repo, "--format", "text", "build/out.bin")
299 assert result.exit_code == 0
300 assert "strategy=ours" in result.output
301
302 def test_no_matching_rule_label(self, tmp_path: pathlib.Path) -> None:
303 repo = _make_repo(tmp_path)
304 result = _ca(repo, "--format", "text", "main.py")
305 assert "no matching rule" in result.output
306
307
308 # ---------------------------------------------------------------------------
309 # Integration — --all-rules
310 # ---------------------------------------------------------------------------
311
312
313 class TestAllRules:
314 def test_all_matching_rules_returned(self, tmp_path: pathlib.Path) -> None:
315 repo = _make_repo(tmp_path)
316 _write_attrs(repo, """
317 [[rules]]
318 path = "tracks/*"
319 dimension = "*"
320 strategy = "ours"
321
322 [[rules]]
323 path = "*.mid"
324 dimension = "*"
325 strategy = "union"
326 """)
327 data = json.loads(_ca(repo, "--all-rules", "tracks/drums.mid").output)
328 result = data["results"][0]
329 assert len(result["matching_rules"]) == 2
330
331 def test_no_matching_rules(self, tmp_path: pathlib.Path) -> None:
332 repo = _make_repo(tmp_path)
333 _write_attrs(repo, _SIMPLE_ATTRS)
334 data = json.loads(_ca(repo, "--all-rules", "main.py").output)
335 assert data["results"][0]["matching_rules"] == []
336
337 def test_all_rules_text_output(self, tmp_path: pathlib.Path) -> None:
338 repo = _make_repo(tmp_path)
339 _write_attrs(repo, _SIMPLE_ATTRS)
340 result = _ca(repo, "--all-rules", "--format", "text", "build/a.bin")
341 assert "strategy=ours" in result.output
342
343
344 # ---------------------------------------------------------------------------
345 # Integration — --rules-only (new agent UX)
346 # ---------------------------------------------------------------------------
347
348
349 class TestRulesOnly:
350 def test_json_rules_list(self, tmp_path: pathlib.Path) -> None:
351 repo = _make_repo(tmp_path)
352 _write_attrs(repo, _SIMPLE_ATTRS)
353 data = json.loads(_ca(repo, "--rules-only").output)
354 assert "rules" in data
355 assert len(data["rules"]) == 2
356 assert data["domain"] == "code"
357
358 def test_empty_attrs_empty_list(self, tmp_path: pathlib.Path) -> None:
359 repo = _make_repo(tmp_path)
360 data = json.loads(_ca(repo, "--rules-only").output)
361 assert data["rules"] == []
362 assert data["rules_loaded"] == 0
363
364 def test_text_format(self, tmp_path: pathlib.Path) -> None:
365 repo = _make_repo(tmp_path)
366 _write_attrs(repo, _SIMPLE_ATTRS)
367 result = _ca(repo, "--rules-only", "--format", "text")
368 assert result.exit_code == 0
369 assert "strategy=ours" in result.output
370
371 def test_no_path_required(self, tmp_path: pathlib.Path) -> None:
372 repo = _make_repo(tmp_path)
373 result = _ca(repo, "--rules-only")
374 assert result.exit_code == 0
375
376
377 # ---------------------------------------------------------------------------
378 # Integration — --stdin (new agent UX)
379 # ---------------------------------------------------------------------------
380
381
382 class TestStdinMode:
383 def test_reads_paths_from_stdin(self, tmp_path: pathlib.Path) -> None:
384 repo = _make_repo(tmp_path)
385 _write_attrs(repo, _SIMPLE_ATTRS)
386 result = _ca(repo, "--stdin", stdin="build/out.bin\nmain.py\n")
387 data = json.loads(result.output)
388 assert len(data["results"]) == 2
389 assert data["results"][0]["strategy"] == "ours"
390 assert data["results"][1]["strategy"] == "auto"
391
392 def test_blank_lines_and_comments_skipped(self, tmp_path: pathlib.Path) -> None:
393 repo = _make_repo(tmp_path)
394 result = _ca(repo, "--stdin", stdin="\n# comment\nmain.py\n\n")
395 data = json.loads(result.output)
396 assert len(data["results"]) == 1
397
398 def test_stdin_combines_with_positional(self, tmp_path: pathlib.Path) -> None:
399 repo = _make_repo(tmp_path)
400 result = _ca(repo, "--stdin", "positional.py", stdin="from_stdin.py\n")
401 data = json.loads(result.output)
402 paths = [r["path"] for r in data["results"]]
403 assert "positional.py" in paths
404 assert "from_stdin.py" in paths
405
406 def test_empty_stdin_no_positional_errors(self, tmp_path: pathlib.Path) -> None:
407 repo = _make_repo(tmp_path)
408 result = _ca(repo, "--stdin", stdin="")
409 assert result.exit_code == ExitCode.USER_ERROR
410
411
412 # ---------------------------------------------------------------------------
413 # Security
414 # ---------------------------------------------------------------------------
415
416
417 class TestSecurity:
418 def test_null_byte_in_path_rejected(self, tmp_path: pathlib.Path) -> None:
419 repo = _make_repo(tmp_path)
420 result = _ca(repo, "tracks/\x00evil.mid")
421 assert result.exit_code == ExitCode.USER_ERROR
422 assert "null byte" in result.output.lower()
423
424 def test_ansi_in_path_stripped_text(self, tmp_path: pathlib.Path) -> None:
425 repo = _make_repo(tmp_path)
426 result = _ca(repo, "--format", "text", "\x1b[31mevil\x1b[0m.py")
427 assert "\x1b" not in result.output
428
429 def test_ansi_in_path_pattern_stripped_text(self, tmp_path: pathlib.Path) -> None:
430 repo = _make_repo(tmp_path)
431 _write_attrs(repo, """
432 [[rules]]
433 path = "\\u001b[31m*\\u001b[0m"
434 dimension = "*"
435 strategy = "ours"
436 """)
437 result = _ca(repo, "--format", "text", "tracks/drums.mid")
438 assert "\x1b" not in result.output
439
440 def test_format_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
441 repo = _make_repo(tmp_path)
442 result = _ca(repo, "--format", "yaml", "foo.py")
443 assert result.exit_code == ExitCode.USER_ERROR
444 assert "error" in result.stderr.lower()
445 assert result.stdout_bytes == b""
446
447 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
448 repo = _make_repo(tmp_path)
449 result = _ca(repo, "--format", "xml", "foo.py")
450 assert "Traceback" not in result.output
451
452 def test_invalid_toml_errors(self, tmp_path: pathlib.Path) -> None:
453 repo = _make_repo(tmp_path)
454 (repo / ".museattributes").write_text("[broken toml !!!")
455 result = _ca(repo, "foo.py")
456 assert result.exit_code == ExitCode.INTERNAL_ERROR
457 assert "Traceback" not in result.output
458
459 def test_no_paths_errors_to_stderr(self, tmp_path: pathlib.Path) -> None:
460 repo = _make_repo(tmp_path)
461 result = _ca(repo)
462 assert result.exit_code == ExitCode.USER_ERROR
463 assert "error" in result.stderr.lower()
464
465
466 # ---------------------------------------------------------------------------
467 # Stress
468 # ---------------------------------------------------------------------------
469
470
471 class TestStress:
472 def test_1000_paths(self, tmp_path: pathlib.Path) -> None:
473 repo = _make_repo(tmp_path)
474 _write_attrs(repo, _SIMPLE_ATTRS)
475 paths = [f"build/file_{i:04d}.bin" for i in range(500)]
476 paths += [f"src/file_{i:04d}.py" for i in range(500)]
477 result = _ca(repo, *paths)
478 assert result.exit_code == 0
479 data = json.loads(result.output)
480 assert len(data["results"]) == 1000
481 ours_count = sum(1 for r in data["results"] if r["strategy"] == "ours")
482 assert ours_count == 500
483
484 def test_50_rule_set(self, tmp_path: pathlib.Path) -> None:
485 repo = _make_repo(tmp_path)
486 rules_toml = "\n".join(
487 f'[[rules]]\npath = "dir_{i}/*"\ndimension = "*"\nstrategy = "ours"\npriority = {50 - i}\n'
488 for i in range(50)
489 )
490 _write_attrs(repo, rules_toml)
491 data = json.loads(_ca(repo, "dir_0/file.py", "dir_49/file.py", "other.py").output)
492 assert data["rules_loaded"] == 50
493 assert data["results"][0]["strategy"] == "ours"
494 assert data["results"][1]["strategy"] == "ours"
495 assert data["results"][2]["strategy"] == "auto"
496
497 def test_200_sequential_runs(self, tmp_path: pathlib.Path) -> None:
498 repo = _make_repo(tmp_path)
499 _write_attrs(repo, _SIMPLE_ATTRS)
500 for i in range(200):
501 result = _ca(repo, "build/out.bin")
502 assert result.exit_code == 0, f"failed at iteration {i}"
503 data = json.loads(result.output)
504 assert data["results"][0]["strategy"] == "ours"
505
506 def test_stdin_1000_paths(self, tmp_path: pathlib.Path) -> None:
507 repo = _make_repo(tmp_path)
508 _write_attrs(repo, _SIMPLE_ATTRS)
509 stdin_input = "\n".join(f"build/file_{i}.bin" for i in range(1000)) + "\n"
510 result = _ca(repo, "--stdin", stdin=stdin_input)
511 assert result.exit_code == 0
512 data = json.loads(result.output)
513 assert len(data["results"]) == 1000
514 assert all(r["strategy"] == "ours" for r in data["results"])
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