gabriel / muse public
test_cmd_attributes_hardening.py python
2,221 lines 85.0 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago
1 """Comprehensive tests for ``muse attributes`` CLI hardening.
2
3 Audit findings addressed
4 ------------------------
5 Security
6 - ANSI injection in domain, path_pattern, dimension, strategy, comment
7 fields stripped via sanitize_display() before text-mode output.
8 - Null bytes in paths supplied to ``check`` rejected with USER_ERROR.
9 - No raw Python tracebacks exposed on ValueError from load_attributes_full.
10 - All diagnostic messages routed to stderr; stdout carries data only.
11
12 Performance
13 - Single-pass parsing (load_attributes_full) instead of two separate
14 _parse_raw calls (read_attributes_meta + load_attributes).
15 - File size cap (_MAX_ATTRIBUTES_BYTES) prevents OOM.
16
17 Correctness
18 - ``list`` now shows comment and priority columns.
19 - JSON always includes ``domain`` (empty string when unset).
20 - JSON includes ``comment`` and ``priority`` fields on every rule.
21 - Empty-file vs. missing-file messages are distinct.
22 - validate exit-codes 0 on valid, USER_ERROR on invalid.
23
24 Agent UX
25 - ``muse attributes list --json`` stable schema.
26 - ``muse attributes check --json`` stable schema with rule_index.
27 - ``muse attributes validate --json`` stable schema.
28 - subcommand structure mirrors hub/config/auth pattern.
29
30 Coverage tiers
31 --------------
32 - Unit: _resolve_with_index, _rule_to_json, TypedDict schemas
33 - Integration: run_list, run_check, run_validate via CliRunner
34 - Security: ANSI injection, null bytes, stderr routing, no tracebacks
35 - E2E: full CLI invocation, JSON round-trips, exit codes
36 - Stress: 1 000 paths, 200 rules, concurrent isolated parses
37 """
38 from __future__ import annotations
39
40 import json
41 import pathlib
42 import threading
43 from typing import TYPE_CHECKING
44 from unittest.mock import patch
45
46 import pytest
47 from unittest.mock import patch
48
49 from muse.core.attributes import (
50 AttributeRule,
51 AttributesMeta,
52 _MAX_ATTRIBUTES_BYTES,
53 load_attributes_full,
54 )
55 from muse.core.errors import ExitCode
56 from tests.cli_test_helper import CliRunner, InvokeResult
57 from muse.core.paths import muse_dir
58
59 if TYPE_CHECKING:
60 from muse.cli.commands.attributes import (
61 _CheckJson,
62 _CheckResultJson,
63 _ListJson,
64 _RuleJson,
65 _ValidateJson,
66 )
67
68 runner = CliRunner()
69 cli = None # argparse-based; CliRunner ignores this
70
71
72 # ---------------------------------------------------------------------------
73 # Helpers
74 # ---------------------------------------------------------------------------
75
76
77 def _make_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path:
78 """Create a minimal Muse repo layout at *tmp_path*."""
79 repo = tmp_path / "repo"
80 dot_muse = muse_dir(repo)
81 for sub in ("objects", "commits", "snapshots", "refs/heads"):
82 (dot_muse / sub).mkdir(parents=True)
83 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
84 (dot_muse / "repo.json").write_text(
85 json.dumps({"repo_id": "r1", "domain": domain}),
86 encoding="utf-8",
87 )
88 return repo
89
90
91 def _write_attrs(repo: pathlib.Path, content: str) -> None:
92 (repo / ".museattributes").write_text(content, encoding="utf-8")
93
94
95 _SIMPLE_ATTRS = """\
96 [meta]
97 domain = "code"
98
99 [[rules]]
100 path = "build/*"
101 dimension = "*"
102 strategy = "ours"
103 comment = "Build artifacts prefer ours."
104 priority = 10
105
106 [[rules]]
107 path = "*.md"
108 dimension = "*"
109 strategy = "theirs"
110 comment = "Docs from incoming branch."
111 priority = 5
112
113 [[rules]]
114 path = "*"
115 dimension = "*"
116 strategy = "auto"
117 comment = ""
118 priority = 0
119 """
120
121 _ANSI = "\x1b[31mmalicious\x1b[0m"
122
123
124 def _invoke(repo: pathlib.Path, *args: str) -> InvokeResult:
125 """Run ``muse attributes <args>`` inside *repo* via MUSE_REPO_ROOT."""
126 return runner.invoke(
127 cli,
128 ["attributes", *args],
129 env={"MUSE_REPO_ROOT": str(repo)},
130 )
131
132
133 # ---------------------------------------------------------------------------
134 # Unit — TypedDicts exist with correct keys
135 # ---------------------------------------------------------------------------
136
137
138 class TestTypedDicts:
139 def test_rule_json_keys(self) -> None:
140 from muse.cli.commands.attributes import _RuleJson
141
142 rule: _RuleJson = {
143 "path_pattern": "x",
144 "dimension": "y",
145 "strategy": "auto",
146 "comment": "",
147 "priority": 0,
148 "source_index": 0,
149 }
150 assert set(rule.keys()) == {
151 "path_pattern",
152 "dimension",
153 "strategy",
154 "comment",
155 "priority",
156 "source_index",
157 }
158
159 def test_list_json_keys(self) -> None:
160 from muse.cli.commands.attributes import _ListJson
161
162 payload: _ListJson = {"domain": "", "rules": []}
163 assert set(payload.keys()) == {"domain", "rules"}
164
165 def test_check_json_keys(self) -> None:
166 from muse.cli.commands.attributes import _CheckJson, _CheckResultJson
167
168 item: _CheckResultJson = {
169 "path": "x.mid",
170 "dimension": "*",
171 "strategy": "auto",
172 "rule_index": -1,
173 }
174 payload: _CheckJson = {"results": [item]}
175 assert "results" in payload
176
177 def test_validate_json_keys(self) -> None:
178 from muse.cli.commands.attributes import _ValidateJson, _ValidateErrorJson
179
180 err: _ValidateErrorJson = {"kind": "missing", "message": "no file"}
181 payload: _ValidateJson = {"valid": False, "errors": [err]}
182 assert set(payload.keys()) == {"valid", "errors"}
183
184
185 # ---------------------------------------------------------------------------
186 # Unit — _resolve_with_index
187 # ---------------------------------------------------------------------------
188
189
190 class TestResolveWithIndex:
191 def _rules(self) -> list[AttributeRule]:
192 return [
193 AttributeRule("build/*", "*", "ours", "x", 10, 0),
194 AttributeRule("*.md", "*", "theirs", "", 5, 1),
195 AttributeRule("*", "*", "auto", "", 0, 2),
196 ]
197
198 def test_first_rule_match(self) -> None:
199 from muse.cli.commands.attributes import _resolve_with_index
200
201 strategy, idx = _resolve_with_index(self._rules(), "build/foo.bin", "*")
202 assert strategy == "ours"
203 assert idx == 0
204
205 def test_second_rule_match(self) -> None:
206 from muse.cli.commands.attributes import _resolve_with_index
207
208 strategy, idx = _resolve_with_index(self._rules(), "README.md", "*")
209 assert strategy == "theirs"
210 assert idx == 1
211
212 def test_fallthrough_to_wildcard(self) -> None:
213 from muse.cli.commands.attributes import _resolve_with_index
214
215 strategy, idx = _resolve_with_index(self._rules(), "src/main.py", "*")
216 assert strategy == "auto"
217 assert idx == 2
218
219 def test_no_match_returns_auto_neg1(self) -> None:
220 from muse.cli.commands.attributes import _resolve_with_index
221
222 rules: list[AttributeRule] = [
223 AttributeRule("build/*", "notes", "ours", "", 0, 0),
224 ]
225 strategy, idx = _resolve_with_index(rules, "src/main.py", "notes")
226 assert strategy == "auto"
227 assert idx == -1
228
229 def test_dimension_filter_respected(self) -> None:
230 from muse.cli.commands.attributes import _resolve_with_index
231
232 rules: list[AttributeRule] = [
233 AttributeRule("*.mid", "pitch_bend", "manual", "", 0, 0),
234 AttributeRule("*.mid", "*", "auto", "", 0, 1),
235 ]
236 strategy, idx = _resolve_with_index(rules, "track.mid", "notes")
237 assert strategy == "auto"
238 assert idx == 1
239
240 def test_empty_rules_returns_auto(self) -> None:
241 from muse.cli.commands.attributes import _resolve_with_index
242
243 strategy, idx = _resolve_with_index([], "anything.mid", "*")
244 assert strategy == "auto"
245 assert idx == -1
246
247
248 # ---------------------------------------------------------------------------
249 # Unit — _rule_to_json
250 # ---------------------------------------------------------------------------
251
252
253 class TestRuleToJson:
254 def test_all_fields_present(self) -> None:
255 from muse.cli.commands.attributes import _rule_to_json
256
257 rule = AttributeRule("drums/*", "*", "ours", "Drums are ours", 10, 3)
258 j = _rule_to_json(rule)
259 assert j["path_pattern"] == "drums/*"
260 assert j["dimension"] == "*"
261 assert j["strategy"] == "ours"
262 assert j["comment"] == "Drums are ours"
263 assert j["priority"] == 10
264 assert j["source_index"] == 3
265
266 def test_defaults_preserved(self) -> None:
267 from muse.cli.commands.attributes import _rule_to_json
268
269 rule = AttributeRule("*", "*", "auto")
270 j = _rule_to_json(rule)
271 assert j["comment"] == ""
272 assert j["priority"] == 0
273 assert j["source_index"] == 0
274
275
276 # ---------------------------------------------------------------------------
277 # Unit — load_attributes_full single-pass
278 # ---------------------------------------------------------------------------
279
280
281 class TestLoadAttributesFull:
282 def test_missing_file_returns_empty_meta_and_rules(
283 self, tmp_path: pathlib.Path
284 ) -> None:
285 meta, rules = load_attributes_full(tmp_path)
286 assert meta == {}
287 assert rules == []
288
289 def test_returns_meta_and_rules_together(self, tmp_path: pathlib.Path) -> None:
290 _write_attrs(tmp_path, _SIMPLE_ATTRS)
291 meta, rules = load_attributes_full(tmp_path)
292 assert meta.get("domain") == "code"
293 assert len(rules) == 3
294
295 def test_priority_sort_applied(self, tmp_path: pathlib.Path) -> None:
296 _write_attrs(tmp_path, _SIMPLE_ATTRS)
297 _, rules = load_attributes_full(tmp_path)
298 priorities = [r.priority for r in rules]
299 assert priorities == sorted(priorities, reverse=True)
300
301 def test_invalid_strategy_raises_value_error(
302 self, tmp_path: pathlib.Path
303 ) -> None:
304 _write_attrs(
305 tmp_path,
306 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "bogus"\n',
307 )
308 with pytest.raises(ValueError, match="unknown strategy"):
309 load_attributes_full(tmp_path)
310
311 def test_file_too_large_raises_value_error(
312 self, tmp_path: pathlib.Path
313 ) -> None:
314 giant = tmp_path / ".museattributes"
315 giant.write_bytes(b"# " + b"x" * (_MAX_ATTRIBUTES_BYTES + 1))
316 with pytest.raises(ValueError, match="file too large"):
317 load_attributes_full(tmp_path)
318
319 def test_bad_toml_raises_value_error(self, tmp_path: pathlib.Path) -> None:
320 _write_attrs(tmp_path, "[[rules]]\npath = <<<invalid\n")
321 with pytest.raises(ValueError, match="TOML parse error"):
322 load_attributes_full(tmp_path)
323
324 def test_single_parse_not_double(self, tmp_path: pathlib.Path) -> None:
325 """load_attributes_full must invoke _parse_raw exactly once."""
326 _write_attrs(tmp_path, _SIMPLE_ATTRS)
327 call_count = 0
328
329 from muse.core import attributes as attrs_mod
330
331 original = attrs_mod._parse_raw
332
333 def counting_parse(root: pathlib.Path) -> attrs_mod.MuseAttributesFile:
334 nonlocal call_count
335 call_count += 1
336 return original(root)
337
338 with patch.object(attrs_mod, "_parse_raw", counting_parse):
339 load_attributes_full(tmp_path)
340
341 assert call_count == 1, f"Expected 1 parse, got {call_count}"
342
343
344 # ---------------------------------------------------------------------------
345 # Integration — muse attributes list
346 # ---------------------------------------------------------------------------
347
348
349 class TestRunList:
350 def test_missing_file_exits_zero_text_to_stderr(
351 self, tmp_path: pathlib.Path
352 ) -> None:
353 repo = _make_repo(tmp_path)
354 result = _invoke(repo, "list")
355 assert result.exit_code == 0
356 assert "No .museattributes" in result.stderr
357
358 def test_empty_file_exits_zero_no_rules_message(
359 self, tmp_path: pathlib.Path
360 ) -> None:
361 repo = _make_repo(tmp_path)
362 _write_attrs(repo, "")
363 result = _invoke(repo, "list")
364 assert result.exit_code == 0
365 assert "no rules" in result.stderr.lower() or "empty" in result.stderr.lower()
366
367 def test_table_shows_all_three_rules(self, tmp_path: pathlib.Path) -> None:
368 repo = _make_repo(tmp_path)
369 _write_attrs(repo, _SIMPLE_ATTRS)
370 result = _invoke(repo, "list")
371 assert result.exit_code == 0
372 assert "build/*" in result.output
373 assert "*.md" in result.output
374
375 def test_table_shows_domain(self, tmp_path: pathlib.Path) -> None:
376 repo = _make_repo(tmp_path)
377 _write_attrs(repo, _SIMPLE_ATTRS)
378 result = _invoke(repo, "list")
379 assert "Domain: code" in result.output
380
381 def test_table_shows_comment_column(self, tmp_path: pathlib.Path) -> None:
382 repo = _make_repo(tmp_path)
383 _write_attrs(repo, _SIMPLE_ATTRS)
384 result = _invoke(repo, "list")
385 assert "Build artifacts prefer ours." in result.output
386
387 def test_table_shows_priority_column(self, tmp_path: pathlib.Path) -> None:
388 repo = _make_repo(tmp_path)
389 _write_attrs(repo, _SIMPLE_ATTRS)
390 result = _invoke(repo, "list")
391 assert "10" in result.output # priority for build/* rule
392
393 def test_table_shows_pri_header(self, tmp_path: pathlib.Path) -> None:
394 repo = _make_repo(tmp_path)
395 _write_attrs(repo, _SIMPLE_ATTRS)
396 result = _invoke(repo, "list")
397 assert "Pri" in result.output
398
399 def test_invalid_strategy_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
400 repo = _make_repo(tmp_path)
401 _write_attrs(
402 repo,
403 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "bogus"\n',
404 )
405 result = _invoke(repo, "list")
406 assert result.exit_code != 0
407
408 def test_bad_toml_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
409 repo = _make_repo(tmp_path)
410 _write_attrs(repo, "[[broken\n")
411 result = _invoke(repo, "list")
412 assert result.exit_code != 0
413
414 def test_bad_toml_no_traceback(self, tmp_path: pathlib.Path) -> None:
415 repo = _make_repo(tmp_path)
416 _write_attrs(repo, "[[broken\n")
417 result = _invoke(repo, "list")
418 assert "Traceback" not in result.output
419 assert "Traceback" not in (result.output or "")
420
421
422 class TestRunListJson:
423 def _parse(self, result: InvokeResult) -> "_ListJson":
424 from muse.cli.commands.attributes import _ListJson, _RuleJson
425
426 start = next(
427 i for i, l in enumerate(result.output.splitlines())
428 if l.strip().startswith("{")
429 )
430 blob = "\n".join(result.output.splitlines()[start:])
431 depth = 0
432 end = 0
433 for i, ch in enumerate(blob):
434 if ch == "{":
435 depth += 1
436 elif ch == "}":
437 depth -= 1
438 if depth == 0:
439 end = i + 1
440 break
441 raw = json.loads(blob[:end])
442 assert isinstance(raw, dict)
443 domain = raw.get("domain", "")
444 assert isinstance(domain, str)
445 rules_raw = raw.get("rules", [])
446 assert isinstance(rules_raw, list)
447 rules: list[_RuleJson] = []
448 for r in rules_raw:
449 assert isinstance(r, dict)
450 rules.append(
451 _RuleJson(
452 path_pattern=str(r.get("path_pattern", "")),
453 dimension=str(r.get("dimension", "")),
454 strategy=str(r.get("strategy", "")),
455 comment=str(r.get("comment", "")),
456 priority=int(r.get("priority", 0)),
457 source_index=int(r.get("source_index", 0)),
458 )
459 )
460 return _ListJson(domain=domain, rules=rules)
461
462 def test_json_schema_domain_always_present(
463 self, tmp_path: pathlib.Path
464 ) -> None:
465 repo = _make_repo(tmp_path)
466 _write_attrs(repo, _SIMPLE_ATTRS)
467 result = _invoke(repo, "list", "--json")
468 data = self._parse(result)
469 assert "domain" in data
470 assert data["domain"] == "code"
471
472 def test_json_domain_empty_string_when_no_meta(
473 self, tmp_path: pathlib.Path
474 ) -> None:
475 repo = _make_repo(tmp_path)
476 _write_attrs(
477 repo,
478 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\n',
479 )
480 result = _invoke(repo, "list", "--json")
481 data = self._parse(result)
482 assert data["domain"] == ""
483
484 def test_json_rules_is_list(self, tmp_path: pathlib.Path) -> None:
485 repo = _make_repo(tmp_path)
486 _write_attrs(repo, _SIMPLE_ATTRS)
487 result = _invoke(repo, "list", "--json")
488 data = self._parse(result)
489 assert isinstance(data["rules"], list)
490 assert len(data["rules"]) == 3
491
492 def test_json_rule_has_all_fields(self, tmp_path: pathlib.Path) -> None:
493 repo = _make_repo(tmp_path)
494 _write_attrs(repo, _SIMPLE_ATTRS)
495 result = _invoke(repo, "list", "--json")
496 data = self._parse(result)
497 rule = data["rules"][0]
498 for field in ("path_pattern", "dimension", "strategy", "comment", "priority", "source_index"):
499 assert field in rule, f"Missing field: {field}"
500
501 def test_json_includes_comment(self, tmp_path: pathlib.Path) -> None:
502 repo = _make_repo(tmp_path)
503 _write_attrs(repo, _SIMPLE_ATTRS)
504 result = _invoke(repo, "list", "--json")
505 data = self._parse(result)
506 comments = [r["comment"] for r in data["rules"]]
507 assert any("Build artifacts" in c for c in comments)
508
509 def test_json_includes_priority(self, tmp_path: pathlib.Path) -> None:
510 repo = _make_repo(tmp_path)
511 _write_attrs(repo, _SIMPLE_ATTRS)
512 result = _invoke(repo, "list", "--json")
513 data = self._parse(result)
514 priorities = [r["priority"] for r in data["rules"]]
515 assert 10 in priorities
516
517 def test_json_missing_file_exits_zero(self, tmp_path: pathlib.Path) -> None:
518 repo = _make_repo(tmp_path)
519 result = _invoke(repo, "list", "--json")
520 assert result.exit_code == 0
521
522 def test_json_missing_file_has_empty_rules(
523 self, tmp_path: pathlib.Path
524 ) -> None:
525 repo = _make_repo(tmp_path)
526 result = _invoke(repo, "list", "--json")
527 data = self._parse(result)
528 assert data["rules"] == []
529 assert data["domain"] == ""
530
531
532 # ---------------------------------------------------------------------------
533 # Integration — muse attributes check
534 # ---------------------------------------------------------------------------
535
536
537 class TestRunCheck:
538 def _parse_check(self, result: InvokeResult) -> "_CheckJson":
539 from muse.cli.commands.attributes import _CheckJson, _CheckResultJson
540
541 start = result.output.index("{")
542 blob = result.output[start:]
543 depth = 0
544 end = 0
545 for i, ch in enumerate(blob):
546 if ch == "{":
547 depth += 1
548 elif ch == "}":
549 depth -= 1
550 if depth == 0:
551 end = i + 1
552 break
553 raw = json.loads(blob[:end])
554 assert isinstance(raw, dict)
555 results_raw = raw.get("results", [])
556 assert isinstance(results_raw, list)
557 results: list[_CheckResultJson] = []
558 for item in results_raw:
559 assert isinstance(item, dict)
560 results.append(
561 _CheckResultJson(
562 path=str(item.get("path", "")),
563 dimension=str(item.get("dimension", "")),
564 strategy=str(item.get("strategy", "")),
565 rule_index=int(item.get("rule_index", -1)),
566 )
567 )
568 return _CheckJson(results=results)
569
570 def test_check_resolves_single_path(self, tmp_path: pathlib.Path) -> None:
571 repo = _make_repo(tmp_path)
572 _write_attrs(repo, _SIMPLE_ATTRS)
573 result = _invoke(repo, "check", "build/foo.o")
574 assert result.exit_code == 0
575 assert "ours" in result.output
576
577 def test_check_shows_rule_index(self, tmp_path: pathlib.Path) -> None:
578 repo = _make_repo(tmp_path)
579 _write_attrs(repo, _SIMPLE_ATTRS)
580 result = _invoke(repo, "check", "build/foo.o")
581 assert "rule #" in result.output
582
583 def test_check_unmatched_shows_default(self, tmp_path: pathlib.Path) -> None:
584 repo = _make_repo(tmp_path)
585 _write_attrs(
586 repo,
587 '[[rules]]\npath = "build/*"\ndimension = "*"\nstrategy = "ours"\n',
588 )
589 result = _invoke(repo, "check", "src/main.py")
590 assert result.exit_code == 0
591 assert "auto" in result.output
592 assert "default" in result.output
593
594 def test_check_multiple_paths(self, tmp_path: pathlib.Path) -> None:
595 repo = _make_repo(tmp_path)
596 _write_attrs(repo, _SIMPLE_ATTRS)
597 result = _invoke(repo, "check", "build/x", "README.md", "src/a.py")
598 assert result.exit_code == 0
599 assert "build/x" in result.output
600 assert "README.md" in result.output
601 assert "src/a.py" in result.output
602
603 def test_check_dimension_filter(self, tmp_path: pathlib.Path) -> None:
604 repo = _make_repo(tmp_path)
605 content = (
606 '[[rules]]\npath = "*.mid"\ndimension = "pitch_bend"\n'
607 'strategy = "manual"\n\n'
608 '[[rules]]\npath = "*.mid"\ndimension = "*"\nstrategy = "auto"\n'
609 )
610 _write_attrs(repo, content)
611 result = _invoke(repo, "check", "track.mid", "--dimension", "pitch_bend")
612 assert result.exit_code == 0
613 assert "manual" in result.output
614
615 def test_check_dimension_star_matches_any(
616 self, tmp_path: pathlib.Path
617 ) -> None:
618 repo = _make_repo(tmp_path)
619 _write_attrs(repo, _SIMPLE_ATTRS)
620 result = _invoke(repo, "check", "build/x", "--dimension", "notes")
621 assert "ours" in result.output
622
623 def test_check_json_schema(self, tmp_path: pathlib.Path) -> None:
624 repo = _make_repo(tmp_path)
625 _write_attrs(repo, _SIMPLE_ATTRS)
626 result = _invoke(repo, "check", "build/foo.o", "--json")
627 assert result.exit_code == 0
628 data = self._parse_check(result)
629 assert "results" in data
630 item = data["results"][0]
631 for field in ("path", "dimension", "strategy", "rule_index"):
632 assert field in item, f"Missing: {field}"
633
634 def test_check_json_rule_index_positive_on_match(
635 self, tmp_path: pathlib.Path
636 ) -> None:
637 repo = _make_repo(tmp_path)
638 _write_attrs(repo, _SIMPLE_ATTRS)
639 result = _invoke(repo, "check", "build/foo.o", "--json")
640 data = self._parse_check(result)
641 assert data["results"][0]["rule_index"] >= 0
642
643 def test_check_json_rule_index_neg1_on_no_match(
644 self, tmp_path: pathlib.Path
645 ) -> None:
646 repo = _make_repo(tmp_path)
647 _write_attrs(
648 repo,
649 '[[rules]]\npath = "build/*"\ndimension = "*"\nstrategy = "ours"\n',
650 )
651 result = _invoke(repo, "check", "src/a.py", "--json")
652 data = self._parse_check(result)
653 assert data["results"][0]["rule_index"] == -1
654
655 def test_check_invalid_strategy_in_file_exits_nonzero(
656 self, tmp_path: pathlib.Path
657 ) -> None:
658 repo = _make_repo(tmp_path)
659 _write_attrs(
660 repo,
661 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "bogus"\n',
662 )
663 result = _invoke(repo, "check", "foo.mid")
664 assert result.exit_code != 0
665
666 def test_check_no_file_exits_zero_no_match(
667 self, tmp_path: pathlib.Path
668 ) -> None:
669 repo = _make_repo(tmp_path)
670 result = _invoke(repo, "check", "any/path.mid")
671 assert result.exit_code == 0
672 assert "auto" in result.output
673
674
675 # ---------------------------------------------------------------------------
676 # Integration — muse attributes validate
677 # ---------------------------------------------------------------------------
678
679
680 class TestRunValidate:
681 def _parse_validate(self, result: InvokeResult) -> "_ValidateJson":
682 from muse.cli.commands.attributes import _ValidateErrorJson, _ValidateJson
683
684 start = result.output.index("{")
685 blob = result.output[start:]
686 depth = 0
687 end = 0
688 for i, ch in enumerate(blob):
689 if ch == "{":
690 depth += 1
691 elif ch == "}":
692 depth -= 1
693 if depth == 0:
694 end = i + 1
695 break
696 raw = json.loads(blob[:end])
697 assert isinstance(raw, dict)
698 valid_val = raw.get("valid", False)
699 assert isinstance(valid_val, bool)
700 errors_raw = raw.get("errors", [])
701 assert isinstance(errors_raw, list)
702 errors: list[_ValidateErrorJson] = []
703 for e in errors_raw:
704 assert isinstance(e, dict)
705 errors.append(
706 _ValidateErrorJson(
707 kind=str(e.get("kind", "")),
708 message=str(e.get("message", "")),
709 )
710 )
711 return _ValidateJson(valid=valid_val, errors=errors)
712
713 def test_valid_file_exits_zero(self, tmp_path: pathlib.Path) -> None:
714 repo = _make_repo(tmp_path)
715 _write_attrs(repo, _SIMPLE_ATTRS)
716 result = _invoke(repo, "validate")
717 assert result.exit_code == 0
718
719 def test_valid_file_shows_success_message(
720 self, tmp_path: pathlib.Path
721 ) -> None:
722 repo = _make_repo(tmp_path)
723 _write_attrs(repo, _SIMPLE_ATTRS)
724 result = _invoke(repo, "validate")
725 assert "valid" in result.output.lower() or "✅" in result.output
726
727 def test_valid_file_shows_rule_count(self, tmp_path: pathlib.Path) -> None:
728 repo = _make_repo(tmp_path)
729 _write_attrs(repo, _SIMPLE_ATTRS)
730 result = _invoke(repo, "validate")
731 assert "3 rule" in result.output
732
733 def test_missing_file_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
734 repo = _make_repo(tmp_path)
735 result = _invoke(repo, "validate")
736 assert result.exit_code != 0
737
738 def test_bad_strategy_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
739 repo = _make_repo(tmp_path)
740 _write_attrs(
741 repo,
742 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "zap"\n',
743 )
744 result = _invoke(repo, "validate")
745 assert result.exit_code != 0
746
747 def test_bad_toml_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
748 repo = _make_repo(tmp_path)
749 _write_attrs(repo, "[[broken\n")
750 result = _invoke(repo, "validate")
751 assert result.exit_code != 0
752
753 def test_bad_toml_no_traceback(self, tmp_path: pathlib.Path) -> None:
754 repo = _make_repo(tmp_path)
755 _write_attrs(repo, "[[broken\n")
756 result = _invoke(repo, "validate")
757 assert "Traceback" not in result.output
758
759 def test_json_valid_schema(self, tmp_path: pathlib.Path) -> None:
760 repo = _make_repo(tmp_path)
761 _write_attrs(repo, _SIMPLE_ATTRS)
762 result = _invoke(repo, "validate", "--json")
763 assert result.exit_code == 0
764 data = self._parse_validate(result)
765 assert data["valid"] is True
766 assert data["errors"] == []
767
768 def test_json_invalid_has_errors(self, tmp_path: pathlib.Path) -> None:
769 repo = _make_repo(tmp_path)
770 _write_attrs(
771 repo,
772 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "oops"\n',
773 )
774 result = _invoke(repo, "validate", "--json")
775 assert result.exit_code != 0
776 data = self._parse_validate(result)
777 assert data["valid"] is False
778 assert len(data["errors"]) > 0
779
780 def test_json_missing_file_error_kind(self, tmp_path: pathlib.Path) -> None:
781 repo = _make_repo(tmp_path)
782 result = _invoke(repo, "validate", "--json")
783 assert result.exit_code != 0
784 data = self._parse_validate(result)
785 assert data["errors"][0]["kind"] == "missing"
786
787 def test_json_semantic_error_kind(self, tmp_path: pathlib.Path) -> None:
788 repo = _make_repo(tmp_path)
789 _write_attrs(
790 repo,
791 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "zap"\n',
792 )
793 result = _invoke(repo, "validate", "--json")
794 data = self._parse_validate(result)
795 assert data["errors"][0]["kind"] == "semantic"
796
797
798 # ---------------------------------------------------------------------------
799 # Security
800 # ---------------------------------------------------------------------------
801
802
803 class TestAttributesSecurity:
804 def test_ansi_in_domain_stripped_from_text_output(
805 self, tmp_path: pathlib.Path
806 ) -> None:
807 repo = _make_repo(tmp_path)
808 _write_attrs(
809 repo,
810 f'[meta]\ndomain = "{_ANSI}"\n'
811 '[[rules]]\npath="*"\ndimension="*"\nstrategy="auto"\n',
812 )
813 result = _invoke(repo, "list")
814 assert "\x1b[" not in result.output
815
816 def test_ansi_in_path_pattern_stripped(self, tmp_path: pathlib.Path) -> None:
817 repo = _make_repo(tmp_path)
818 _write_attrs(
819 repo,
820 f'[[rules]]\npath = "{_ANSI}"\ndimension = "*"\nstrategy = "auto"\n',
821 )
822 result = _invoke(repo, "list")
823 assert "\x1b[" not in result.output
824
825 def test_ansi_in_comment_stripped(self, tmp_path: pathlib.Path) -> None:
826 repo = _make_repo(tmp_path)
827 _write_attrs(
828 repo,
829 f'[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\n'
830 f'comment = "{_ANSI}"\n',
831 )
832 result = _invoke(repo, "list")
833 assert "\x1b[" not in result.output
834
835 def test_ansi_in_check_path_stripped_in_output(
836 self, tmp_path: pathlib.Path
837 ) -> None:
838 repo = _make_repo(tmp_path)
839 _write_attrs(repo, _SIMPLE_ATTRS)
840 ansi_path = f"{_ANSI}/foo.mid"
841 result = _invoke(repo, "check", ansi_path)
842 assert result.exit_code == 0
843 assert "\x1b[" not in result.output
844
845 def test_null_byte_in_check_path_exits_user_error(
846 self, tmp_path: pathlib.Path
847 ) -> None:
848 repo = _make_repo(tmp_path)
849 _write_attrs(repo, _SIMPLE_ATTRS)
850 result = _invoke(repo, "check", "foo\x00bar")
851 assert result.exit_code == ExitCode.USER_ERROR.value
852
853 def test_null_byte_rejected_even_without_attrs_file(
854 self, tmp_path: pathlib.Path
855 ) -> None:
856 repo = _make_repo(tmp_path)
857 result = _invoke(repo, "check", "foo\x00bar")
858 assert result.exit_code == ExitCode.USER_ERROR.value
859
860 def test_error_messages_to_stderr_not_only_stdout(
861 self, tmp_path: pathlib.Path
862 ) -> None:
863 """Errors for bad TOML must not produce raw tracebacks."""
864 repo = _make_repo(tmp_path)
865 _write_attrs(repo, "[[broken\n")
866 result = _invoke(repo, "list")
867 assert result.exit_code != 0
868 assert "Traceback" not in result.output
869
870 def test_oversized_file_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
871 repo = _make_repo(tmp_path)
872 giant = repo / ".museattributes"
873 giant.write_bytes(b"# " + b"x" * (_MAX_ATTRIBUTES_BYTES + 1))
874 result = _invoke(repo, "list")
875 assert result.exit_code != 0
876
877 def test_oversized_file_no_oom(self, tmp_path: pathlib.Path) -> None:
878 """Validate the size cap fires before TOML parsing."""
879 repo = _make_repo(tmp_path)
880 giant = repo / ".museattributes"
881 giant.write_bytes(b"# " + b"x" * (_MAX_ATTRIBUTES_BYTES + 1))
882 result = _invoke(repo, "validate")
883 assert result.exit_code != 0
884 assert "Traceback" not in result.output
885
886 def test_json_output_on_stdout_error_on_stderr_list(
887 self, tmp_path: pathlib.Path
888 ) -> None:
889 """When --json used for list, valid JSON goes to stdout."""
890 repo = _make_repo(tmp_path)
891 _write_attrs(repo, _SIMPLE_ATTRS)
892 result = _invoke(repo, "list", "--json")
893 assert result.exit_code == 0
894 first_json_line = next(
895 (l for l in result.output.splitlines() if l.strip().startswith("{")),
896 None,
897 )
898 assert first_json_line is not None, "No JSON on stdout"
899
900
901 # ---------------------------------------------------------------------------
902 # E2E — full CLI invocation with CliRunner
903 # ---------------------------------------------------------------------------
904
905
906 class TestE2E:
907 def test_list_subcommand_is_required_without_subcommand(
908 self, tmp_path: pathlib.Path
909 ) -> None:
910 """muse attributes (no subcommand) should exit non-zero (subcommand required)."""
911 repo = _make_repo(tmp_path)
912 result = runner.invoke(
913 cli,
914 ["attributes"],
915 env={"MUSE_REPO_ROOT": str(repo)},
916 )
917 # subcommand is required — argparse exits 2
918 assert result.exit_code != 0
919
920 def test_list_json_round_trips(self, tmp_path: pathlib.Path) -> None:
921 repo = _make_repo(tmp_path)
922 _write_attrs(repo, _SIMPLE_ATTRS)
923 result = _invoke(repo, "list", "--json")
924 assert result.exit_code == 0
925 raw = next(
926 i for i, l in enumerate(result.output.splitlines())
927 if l.strip().startswith("{")
928 )
929 blob = "\n".join(result.output.splitlines()[raw:])
930 depth = 0
931 end = 0
932 for i, ch in enumerate(blob):
933 if ch == "{":
934 depth += 1
935 elif ch == "}":
936 depth -= 1
937 if depth == 0:
938 end = i + 1
939 break
940 data = json.loads(blob[:end])
941 assert data["domain"] == "code"
942 assert len(data["rules"]) == 3
943
944 def test_check_json_round_trips(self, tmp_path: pathlib.Path) -> None:
945 repo = _make_repo(tmp_path)
946 _write_attrs(repo, _SIMPLE_ATTRS)
947 result = _invoke(repo, "check", "build/x", "--json")
948 assert result.exit_code == 0
949 start = result.output.index("{")
950 blob = result.output[start:]
951 depth = 0
952 end = 0
953 for i, ch in enumerate(blob):
954 if ch == "{":
955 depth += 1
956 elif ch == "}":
957 depth -= 1
958 if depth == 0:
959 end = i + 1
960 break
961 data = json.loads(blob[:end])
962 assert data["results"][0]["strategy"] == "ours"
963
964 def test_validate_exits_0_valid_file(self, tmp_path: pathlib.Path) -> None:
965 repo = _make_repo(tmp_path)
966 _write_attrs(repo, _SIMPLE_ATTRS)
967 result = _invoke(repo, "validate")
968 assert result.exit_code == 0
969
970 def test_validate_exits_nonzero_missing_file(
971 self, tmp_path: pathlib.Path
972 ) -> None:
973 repo = _make_repo(tmp_path)
974 result = _invoke(repo, "validate")
975 assert result.exit_code != 0
976
977 def test_help_list_available(self, tmp_path: pathlib.Path) -> None:
978 result = runner.invoke(cli, ["attributes", "--help"])
979 assert result.exit_code == 0
980 assert "list" in result.output
981 assert "check" in result.output
982 assert "validate" in result.output
983
984 def test_all_valid_strategies_accepted(self, tmp_path: pathlib.Path) -> None:
985 from muse.core.attributes import VALID_STRATEGIES
986
987 repo = _make_repo(tmp_path)
988 for strategy in sorted(VALID_STRATEGIES):
989 attrs = (
990 f'[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "{strategy}"\n'
991 )
992 _write_attrs(repo, attrs)
993 result = _invoke(repo, "validate")
994 assert result.exit_code == 0, f"Strategy {strategy!r} should be valid"
995
996 def test_check_text_format_includes_path_and_strategy(
997 self, tmp_path: pathlib.Path
998 ) -> None:
999 repo = _make_repo(tmp_path)
1000 _write_attrs(repo, _SIMPLE_ATTRS)
1001 result = _invoke(repo, "check", "README.md")
1002 assert "README.md" in result.output
1003 assert "theirs" in result.output
1004
1005
1006 # ---------------------------------------------------------------------------
1007 # Stress
1008 # ---------------------------------------------------------------------------
1009
1010
1011 class TestStress:
1012 def test_1000_paths_resolved_efficiently(
1013 self, tmp_path: pathlib.Path
1014 ) -> None:
1015 repo = _make_repo(tmp_path)
1016 _write_attrs(repo, _SIMPLE_ATTRS)
1017 paths = [f"build/file_{i}.o" for i in range(1000)]
1018 result = _invoke(repo, "check", *paths)
1019 assert result.exit_code == 0
1020 # All 1 000 paths should resolve to "ours"
1021 ours_count = result.output.count("ours")
1022 assert ours_count == 1000
1023
1024 def test_200_rules_loads_correctly(self, tmp_path: pathlib.Path) -> None:
1025 repo = _make_repo(tmp_path)
1026 rules_toml = "".join(
1027 f'[[rules]]\npath = "dir{i}/*"\ndimension = "*"\nstrategy = "auto"\n\n'
1028 for i in range(200)
1029 )
1030 _write_attrs(repo, rules_toml)
1031 result = _invoke(repo, "list", "--json")
1032 assert result.exit_code == 0
1033 start = result.output.index("{")
1034 blob = result.output[start:]
1035 depth = 0
1036 end = 0
1037 for i, ch in enumerate(blob):
1038 if ch == "{":
1039 depth += 1
1040 elif ch == "}":
1041 depth -= 1
1042 if depth == 0:
1043 end = i + 1
1044 break
1045 data = json.loads(blob[:end])
1046 assert len(data["rules"]) == 200
1047
1048 def test_concurrent_load_isolated_repos(
1049 self, tmp_path: pathlib.Path
1050 ) -> None:
1051 """Eight threads each call load_attributes_full on isolated repos.
1052
1053 Tests the core parsing layer for thread-safety without going through
1054 the CliRunner (whose env patching is not thread-safe).
1055 """
1056 errors: list[str] = []
1057
1058 def worker(idx: int) -> None:
1059 try:
1060 repo_dir = tmp_path / f"repo{idx}"
1061 repo_dir.mkdir(parents=True, exist_ok=True)
1062 _write_attrs(repo_dir, _SIMPLE_ATTRS)
1063 meta, rules = load_attributes_full(repo_dir)
1064 if len(rules) != 3:
1065 errors.append(f"Thread {idx}: got {len(rules)} rules, want 3")
1066 if meta.get("domain") != "code":
1067 errors.append(f"Thread {idx}: wrong domain {meta.get('domain')!r}")
1068 except Exception as exc:
1069 errors.append(f"Thread {idx}: {exc}")
1070
1071 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1072 for t in threads:
1073 t.start()
1074 for t in threads:
1075 t.join()
1076
1077 assert errors == [], f"Concurrent failures: {errors}"
1078
1079 def test_concurrent_resolve_with_index(self, tmp_path: pathlib.Path) -> None:
1080 """Eight threads each call _resolve_with_index on independent rule lists.
1081
1082 Tests the resolution helper for thread-safety without shared state.
1083 """
1084 from muse.core.attributes import AttributeRule
1085 from muse.cli.commands.attributes import _resolve_with_index
1086
1087 errors: list[str] = []
1088 rules = [
1089 AttributeRule("build/*", "*", "ours", "", 10, 0),
1090 AttributeRule("*.md", "*", "theirs", "", 5, 1),
1091 AttributeRule("*", "*", "auto", "", 0, 2),
1092 ]
1093
1094 def worker(idx: int) -> None:
1095 try:
1096 # Each thread has its own path — no shared mutable state
1097 path = f"build/file_{idx}.o"
1098 strategy, rule_idx = _resolve_with_index(rules, path, "*")
1099 if strategy != "ours":
1100 errors.append(f"Thread {idx}: got {strategy!r}, want 'ours'")
1101 if rule_idx != 0:
1102 errors.append(f"Thread {idx}: rule_index {rule_idx}, want 0")
1103 except Exception as exc:
1104 errors.append(f"Thread {idx}: {exc}")
1105
1106 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1107 for t in threads:
1108 t.start()
1109 for t in threads:
1110 t.join()
1111
1112 assert errors == [], f"Concurrent failures: {errors}"
1113
1114
1115 # ---------------------------------------------------------------------------
1116 # Extended — muse attributes list (deeper coverage)
1117 # ---------------------------------------------------------------------------
1118
1119
1120 class TestRunListExtended:
1121 def test_help_mentions_json_flag(self, tmp_path: pathlib.Path) -> None:
1122 result = _invoke(tmp_path, "list", "--help")
1123 assert "--json" in result.output or "-j" in result.output
1124
1125 def test_j_alias_works(self, tmp_path: pathlib.Path) -> None:
1126 repo = _make_repo(tmp_path)
1127 _write_attrs(repo, _SIMPLE_ATTRS)
1128 result = _invoke(repo, "list", "-j")
1129 assert result.exit_code == 0
1130 data = json.loads(result.output.strip())
1131 assert "domain" in data
1132 assert "rules" in data
1133
1134 def test_json_is_valid_and_parseable(self, tmp_path: pathlib.Path) -> None:
1135 repo = _make_repo(tmp_path)
1136 _write_attrs(repo, _SIMPLE_ATTRS)
1137 result = _invoke(repo, "list", "--json")
1138 assert result.exit_code == 0
1139 data = json.loads(result.output)
1140 assert "domain" in data and "rules" in data and "rule_count" in data
1141
1142 def test_json_parses_cleanly(self, tmp_path: pathlib.Path) -> None:
1143 repo = _make_repo(tmp_path)
1144 _write_attrs(repo, _SIMPLE_ATTRS)
1145 result = _invoke(repo, "list", "--json")
1146 data = json.loads(result.output.strip())
1147 assert isinstance(data, dict)
1148
1149 def test_json_rule_count_matches_attrs(self, tmp_path: pathlib.Path) -> None:
1150 repo = _make_repo(tmp_path)
1151 _write_attrs(repo, _SIMPLE_ATTRS)
1152 result = _invoke(repo, "list", "--json")
1153 data = json.loads(result.output.strip())
1154 assert len(data["rules"]) == 3
1155
1156 def test_json_priority_is_int(self, tmp_path: pathlib.Path) -> None:
1157 repo = _make_repo(tmp_path)
1158 _write_attrs(repo, _SIMPLE_ATTRS)
1159 result = _invoke(repo, "list", "--json")
1160 data = json.loads(result.output.strip())
1161 for rule in data["rules"]:
1162 assert isinstance(rule["priority"], int)
1163
1164 def test_json_source_index_is_int(self, tmp_path: pathlib.Path) -> None:
1165 repo = _make_repo(tmp_path)
1166 _write_attrs(repo, _SIMPLE_ATTRS)
1167 result = _invoke(repo, "list", "--json")
1168 data = json.loads(result.output.strip())
1169 for rule in data["rules"]:
1170 assert isinstance(rule["source_index"], int)
1171
1172 def test_json_source_index_sequential(self, tmp_path: pathlib.Path) -> None:
1173 """source_index values form a complete 0..N-1 set."""
1174 repo = _make_repo(tmp_path)
1175 _write_attrs(repo, _SIMPLE_ATTRS)
1176 result = _invoke(repo, "list", "--json")
1177 data = json.loads(result.output.strip())
1178 indices = sorted(r["source_index"] for r in data["rules"])
1179 assert indices == list(range(len(data["rules"])))
1180
1181 def test_json_strategy_strings_valid(self, tmp_path: pathlib.Path) -> None:
1182 repo = _make_repo(tmp_path)
1183 _write_attrs(repo, _SIMPLE_ATTRS)
1184 result = _invoke(repo, "list", "--json")
1185 data = json.loads(result.output.strip())
1186 from muse.core.attributes import VALID_STRATEGIES
1187 for rule in data["rules"]:
1188 assert rule["strategy"] in VALID_STRATEGIES
1189
1190 def test_text_header_always_present(self, tmp_path: pathlib.Path) -> None:
1191 repo = _make_repo(tmp_path)
1192 _write_attrs(repo, _SIMPLE_ATTRS)
1193 result = _invoke(repo, "list")
1194 assert "Path pattern" in result.output
1195 assert "Strategy" in result.output
1196
1197 def test_text_separator_row_present(self, tmp_path: pathlib.Path) -> None:
1198 repo = _make_repo(tmp_path)
1199 _write_attrs(repo, _SIMPLE_ATTRS)
1200 result = _invoke(repo, "list")
1201 assert "---" in result.output
1202
1203 def test_text_no_comment_column_when_no_comments(self, tmp_path: pathlib.Path) -> None:
1204 repo = _make_repo(tmp_path)
1205 _write_attrs(
1206 repo,
1207 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\n',
1208 )
1209 result = _invoke(repo, "list")
1210 assert result.exit_code == 0
1211 assert "Comment" not in result.output
1212
1213 def test_text_comment_column_present_when_any_comment(self, tmp_path: pathlib.Path) -> None:
1214 repo = _make_repo(tmp_path)
1215 _write_attrs(repo, _SIMPLE_ATTRS)
1216 result = _invoke(repo, "list")
1217 assert "Comment" in result.output
1218
1219 def test_json_missing_file_exits_zero_empty_rules(self, tmp_path: pathlib.Path) -> None:
1220 repo = _make_repo(tmp_path)
1221 result = _invoke(repo, "list", "--json")
1222 assert result.exit_code == 0
1223 data = json.loads(result.output.strip())
1224 assert data["rules"] == []
1225 assert data["domain"] == ""
1226
1227 def test_text_missing_file_message_present(self, tmp_path: pathlib.Path) -> None:
1228 repo = _make_repo(tmp_path)
1229 result = _invoke(repo, "list")
1230 assert result.exit_code == 0
1231 assert ".museattributes" in result.stderr
1232
1233 def test_json_single_rule_roundtrip(self, tmp_path: pathlib.Path) -> None:
1234 repo = _make_repo(tmp_path)
1235 _write_attrs(
1236 repo,
1237 '[[rules]]\npath = "src/*.py"\ndimension = "code"\nstrategy = "ours"\ncomment = "Python wins"\npriority = 7\n',
1238 )
1239 result = _invoke(repo, "list", "-j")
1240 assert result.exit_code == 0
1241 data = json.loads(result.output.strip())
1242 rule = data["rules"][0]
1243 assert rule["path_pattern"] == "src/*.py"
1244 assert rule["dimension"] == "code"
1245 assert rule["strategy"] == "ours"
1246 assert rule["comment"] == "Python wins"
1247 assert rule["priority"] == 7
1248
1249 def test_help_shows_exit_codes(self, tmp_path: pathlib.Path) -> None:
1250 result = _invoke(tmp_path, "list", "--help")
1251 assert "Exit code" in result.output or "exit code" in result.output
1252
1253 def test_domain_in_text_output(self, tmp_path: pathlib.Path) -> None:
1254 repo = _make_repo(tmp_path)
1255 _write_attrs(repo, _SIMPLE_ATTRS)
1256 result = _invoke(repo, "list")
1257 assert "Domain: code" in result.output
1258
1259
1260 # ---------------------------------------------------------------------------
1261 # Security — muse attributes list
1262 # ---------------------------------------------------------------------------
1263
1264
1265 class TestRunListSecurity:
1266 def test_ansi_in_domain_stripped_text(self, tmp_path: pathlib.Path) -> None:
1267 repo = _make_repo(tmp_path)
1268 _write_attrs(
1269 repo,
1270 f'[meta]\ndomain = "{_ANSI}"\n\n[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\n',
1271 )
1272 result = _invoke(repo, "list")
1273 assert "\x1b[" not in result.output
1274
1275 def test_ansi_in_path_pattern_stripped_text(self, tmp_path: pathlib.Path) -> None:
1276 repo = _make_repo(tmp_path)
1277 ansi_path = _ANSI + "/*"
1278 _write_attrs(
1279 repo,
1280 f'[[rules]]\npath = "{ansi_path}"\ndimension = "*"\nstrategy = "auto"\n',
1281 )
1282 result = _invoke(repo, "list")
1283 assert "\x1b[" not in result.output
1284
1285 def test_ansi_in_comment_stripped_text(self, tmp_path: pathlib.Path) -> None:
1286 repo = _make_repo(tmp_path)
1287 _write_attrs(
1288 repo,
1289 f'[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\ncomment = "{_ANSI}"\n',
1290 )
1291 result = _invoke(repo, "list")
1292 assert "\x1b[" not in result.output
1293
1294 def test_unicode_comment_preserved_verbatim_in_json(self, tmp_path: pathlib.Path) -> None:
1295 """Unicode in comments is preserved verbatim in JSON (sanitization is text-only).
1296
1297 TOML rejects control characters (including ANSI ESC \x1b) as illegal,
1298 so only valid Unicode can appear in TOML-sourced comments.
1299 """
1300 repo = _make_repo(tmp_path)
1301 _write_attrs(
1302 repo,
1303 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\ncomment = "caf\u00e9 note"\n',
1304 )
1305 result = _invoke(repo, "list", "--json")
1306 assert result.exit_code == 0
1307 data = json.loads(result.output.strip())
1308 comments = [r["comment"] for r in data["rules"]]
1309 assert any("café" in c for c in comments)
1310
1311 def test_oversized_file_exits_nonzero_list(self, tmp_path: pathlib.Path) -> None:
1312 repo = _make_repo(tmp_path)
1313 from muse.core.attributes import _MAX_ATTRIBUTES_BYTES
1314 (repo / ".museattributes").write_bytes(b"x" * (_MAX_ATTRIBUTES_BYTES + 1))
1315 result = _invoke(repo, "list")
1316 assert result.exit_code != 0
1317
1318 def test_json_stdout_only_no_stray_text(self, tmp_path: pathlib.Path) -> None:
1319 """In JSON mode, stdout must be valid JSON and nothing else."""
1320 repo = _make_repo(tmp_path)
1321 _write_attrs(repo, _SIMPLE_ATTRS)
1322 result = _invoke(repo, "list", "--json")
1323 assert result.exit_code == 0
1324 json.loads(result.output.strip())
1325
1326
1327 # ---------------------------------------------------------------------------
1328 # Stress — muse attributes list
1329 # ---------------------------------------------------------------------------
1330
1331
1332 class TestRunListStress:
1333 def _make_attrs_with_n_rules(self, n: int) -> str:
1334 lines = ['[meta]\ndomain = "stress"\n']
1335 for i in range(n):
1336 lines.append(
1337 f'[[rules]]\npath = "dir_{i}/*.bin"\ndimension = "*"\n'
1338 f'strategy = "ours"\ncomment = "rule {i}"\npriority = {i}\n'
1339 )
1340 return "\n".join(lines)
1341
1342 def test_500_rules_text_renders(self, tmp_path: pathlib.Path) -> None:
1343 repo = _make_repo(tmp_path)
1344 _write_attrs(repo, self._make_attrs_with_n_rules(500))
1345 result = _invoke(repo, "list")
1346 assert result.exit_code == 0
1347 assert "dir_0" in result.output
1348
1349 def test_500_rules_json_serializes(self, tmp_path: pathlib.Path) -> None:
1350 repo = _make_repo(tmp_path)
1351 _write_attrs(repo, self._make_attrs_with_n_rules(500))
1352 result = _invoke(repo, "list", "--json")
1353 assert result.exit_code == 0
1354 data = json.loads(result.output.strip())
1355 assert len(data["rules"]) == 500
1356
1357 def test_concurrent_list_isolated_repos(self, tmp_path: pathlib.Path) -> None:
1358 """Ten threads each call load_attributes_full on isolated repos — no shared state.
1359
1360 Uses the core layer directly to avoid CliRunner stdout-interleaving.
1361 """
1362 errors: list[str] = []
1363
1364 def worker(idx: int) -> None:
1365 try:
1366 repo = _make_repo(tmp_path / f"repo_{idx}")
1367 _write_attrs(repo, _SIMPLE_ATTRS)
1368 meta, rules = load_attributes_full(repo)
1369 if len(rules) != 3:
1370 errors.append(f"Thread {idx}: expected 3 rules, got {len(rules)}")
1371 if meta.get("domain") != "code":
1372 errors.append(f"Thread {idx}: wrong domain {meta.get('domain')!r}")
1373 except Exception as exc:
1374 errors.append(f"Thread {idx}: {exc}")
1375
1376 threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
1377 for t in threads:
1378 t.start()
1379 for t in threads:
1380 t.join()
1381 assert errors == [], f"Concurrent failures: {errors}"
1382
1383
1384 # ---------------------------------------------------------------------------
1385 # Extended — muse attributes check (deeper coverage)
1386 # ---------------------------------------------------------------------------
1387
1388
1389 class TestRunCheckExtended:
1390 def test_j_alias_works(self, tmp_path: pathlib.Path) -> None:
1391 repo = _make_repo(tmp_path)
1392 _write_attrs(repo, _SIMPLE_ATTRS)
1393 result = _invoke(repo, "check", "build/foo.o", "-j")
1394 assert result.exit_code == 0
1395 data = json.loads(result.output.strip())
1396 assert "results" in data
1397
1398 def test_help_mentions_json_flag(self, tmp_path: pathlib.Path) -> None:
1399 result = _invoke(tmp_path, "check", "--help")
1400 assert "--json" in result.output or "-j" in result.output
1401
1402 def test_help_shows_exit_codes(self, tmp_path: pathlib.Path) -> None:
1403 result = _invoke(tmp_path, "check", "--help")
1404 assert "Exit code" in result.output or "exit code" in result.output
1405
1406 def test_json_is_valid_and_parseable(self, tmp_path: pathlib.Path) -> None:
1407 repo = _make_repo(tmp_path)
1408 _write_attrs(repo, _SIMPLE_ATTRS)
1409 result = _invoke(repo, "check", "build/foo.o", "--json")
1410 assert result.exit_code == 0
1411 data = json.loads(result.output)
1412 assert "results" in data and isinstance(data["results"], list)
1413
1414 def test_json_parses_cleanly(self, tmp_path: pathlib.Path) -> None:
1415 repo = _make_repo(tmp_path)
1416 _write_attrs(repo, _SIMPLE_ATTRS)
1417 result = _invoke(repo, "check", "build/foo.o", "--json")
1418 data = json.loads(result.output.strip())
1419 assert isinstance(data, dict)
1420
1421 def test_json_result_count_matches_path_count(self, tmp_path: pathlib.Path) -> None:
1422 repo = _make_repo(tmp_path)
1423 _write_attrs(repo, _SIMPLE_ATTRS)
1424 result = _invoke(repo, "check", "build/x", "README.md", "src/a.py", "--json")
1425 data = json.loads(result.output.strip())
1426 assert len(data["results"]) == 3
1427
1428 def test_json_rule_index_is_int(self, tmp_path: pathlib.Path) -> None:
1429 repo = _make_repo(tmp_path)
1430 _write_attrs(repo, _SIMPLE_ATTRS)
1431 result = _invoke(repo, "check", "build/foo.o", "--json")
1432 data = json.loads(result.output.strip())
1433 assert isinstance(data["results"][0]["rule_index"], int)
1434
1435 def test_json_all_four_fields_present(self, tmp_path: pathlib.Path) -> None:
1436 repo = _make_repo(tmp_path)
1437 _write_attrs(repo, _SIMPLE_ATTRS)
1438 result = _invoke(repo, "check", "build/foo.o", "--json")
1439 data = json.loads(result.output.strip())
1440 item = data["results"][0]
1441 for field in ("path", "dimension", "strategy", "rule_index"):
1442 assert field in item, f"Missing field: {field}"
1443
1444 def test_json_dimension_echoed_from_arg(self, tmp_path: pathlib.Path) -> None:
1445 repo = _make_repo(tmp_path)
1446 _write_attrs(repo, _SIMPLE_ATTRS)
1447 result = _invoke(repo, "check", "build/foo.o", "--dimension", "notes", "--json")
1448 data = json.loads(result.output.strip())
1449 assert data["results"][0]["dimension"] == "notes"
1450
1451 def test_json_path_echoed_verbatim(self, tmp_path: pathlib.Path) -> None:
1452 repo = _make_repo(tmp_path)
1453 _write_attrs(repo, _SIMPLE_ATTRS)
1454 result = _invoke(repo, "check", "build/foo.o", "--json")
1455 data = json.loads(result.output.strip())
1456 assert data["results"][0]["path"] == "build/foo.o"
1457
1458 def test_json_no_match_rule_index_neg1(self, tmp_path: pathlib.Path) -> None:
1459 repo = _make_repo(tmp_path)
1460 _write_attrs(
1461 repo,
1462 '[[rules]]\npath = "build/*"\ndimension = "*"\nstrategy = "ours"\n',
1463 )
1464 result = _invoke(repo, "check", "src/unmatched.py", "--json")
1465 data = json.loads(result.output.strip())
1466 assert data["results"][0]["rule_index"] == -1
1467 assert data["results"][0]["strategy"] == "auto"
1468
1469 def test_json_no_file_exits_zero_returns_auto(self, tmp_path: pathlib.Path) -> None:
1470 repo = _make_repo(tmp_path)
1471 result = _invoke(repo, "check", "any/path.mid", "--json")
1472 assert result.exit_code == 0
1473 data = json.loads(result.output.strip())
1474 assert data["results"][0]["strategy"] == "auto"
1475 assert data["results"][0]["rule_index"] == -1
1476
1477 def test_d_alias_for_dimension(self, tmp_path: pathlib.Path) -> None:
1478 repo = _make_repo(tmp_path)
1479 content = (
1480 '[[rules]]\npath = "*.mid"\ndimension = "pitch_bend"\n'
1481 'strategy = "manual"\n'
1482 )
1483 _write_attrs(repo, content)
1484 result = _invoke(repo, "check", "track.mid", "-d", "pitch_bend", "--json")
1485 data = json.loads(result.output.strip())
1486 assert data["results"][0]["strategy"] == "manual"
1487
1488 def test_text_format_path_colon_strategy(self, tmp_path: pathlib.Path) -> None:
1489 repo = _make_repo(tmp_path)
1490 _write_attrs(repo, _SIMPLE_ATTRS)
1491 result = _invoke(repo, "check", "build/foo.o")
1492 assert result.exit_code == 0
1493 assert "build/foo.o:" in result.output
1494 assert "ours" in result.output
1495
1496 def test_text_no_match_shows_default(self, tmp_path: pathlib.Path) -> None:
1497 repo = _make_repo(tmp_path)
1498 result = _invoke(repo, "check", "unmatched.txt")
1499 assert "default" in result.output
1500
1501 def test_text_match_shows_rule_number(self, tmp_path: pathlib.Path) -> None:
1502 repo = _make_repo(tmp_path)
1503 _write_attrs(repo, _SIMPLE_ATTRS)
1504 result = _invoke(repo, "check", "build/foo.o")
1505 assert "rule #" in result.output
1506
1507 def test_invalid_strategy_exits_nonzero_json_mode(self, tmp_path: pathlib.Path) -> None:
1508 repo = _make_repo(tmp_path)
1509 _write_attrs(
1510 repo,
1511 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "bogus"\n',
1512 )
1513 result = _invoke(repo, "check", "foo.mid", "--json")
1514 assert result.exit_code != 0
1515
1516 def test_multiple_paths_results_in_order(self, tmp_path: pathlib.Path) -> None:
1517 repo = _make_repo(tmp_path)
1518 _write_attrs(repo, _SIMPLE_ATTRS)
1519 paths = ["build/a.o", "README.md", "src/b.py"]
1520 result = _invoke(repo, "check", *paths, "--json")
1521 data = json.loads(result.output.strip())
1522 assert [r["path"] for r in data["results"]] == paths
1523
1524
1525 # ---------------------------------------------------------------------------
1526 # Security — muse attributes check
1527 # ---------------------------------------------------------------------------
1528
1529
1530 class TestRunCheckSecurity:
1531 def test_null_byte_in_path_exits_user_error(self, tmp_path: pathlib.Path) -> None:
1532 repo = _make_repo(tmp_path)
1533 _write_attrs(repo, _SIMPLE_ATTRS)
1534 result = _invoke(repo, "check", "path\x00malicious")
1535 assert result.exit_code == ExitCode.USER_ERROR.value
1536
1537 def test_null_byte_in_json_mode_exits_user_error(self, tmp_path: pathlib.Path) -> None:
1538 repo = _make_repo(tmp_path)
1539 result = _invoke(repo, "check", "path\x00malicious", "--json")
1540 assert result.exit_code == ExitCode.USER_ERROR.value
1541
1542 def test_null_byte_error_to_stderr_not_stdout(self, tmp_path: pathlib.Path) -> None:
1543 repo = _make_repo(tmp_path)
1544 result = _invoke(repo, "check", "path\x00malicious")
1545 # CliRunner merges streams; error should mention null byte
1546 assert "null byte" in result.stderr or "null" in result.stderr
1547
1548 def test_ansi_in_caller_path_stripped_text(self, tmp_path: pathlib.Path) -> None:
1549 """Caller-supplied path with ANSI is sanitized in text output."""
1550 repo = _make_repo(tmp_path)
1551 result = _invoke(repo, "check", f"{_ANSI}/file.py")
1552 assert result.exit_code == 0
1553 assert "\x1b[" not in result.output
1554
1555 def test_multiple_paths_null_byte_in_second_exits_error(self, tmp_path: pathlib.Path) -> None:
1556 """Null byte anywhere in the path list aborts processing."""
1557 repo = _make_repo(tmp_path)
1558 _write_attrs(repo, _SIMPLE_ATTRS)
1559 result = _invoke(repo, "check", "build/ok.o", "bad\x00path")
1560 assert result.exit_code == ExitCode.USER_ERROR.value
1561
1562 def test_json_stdout_only_no_stray_text(self, tmp_path: pathlib.Path) -> None:
1563 """In JSON mode, stdout must be valid JSON and nothing else."""
1564 repo = _make_repo(tmp_path)
1565 _write_attrs(repo, _SIMPLE_ATTRS)
1566 result = _invoke(repo, "check", "build/foo.o", "--json")
1567 assert result.exit_code == 0
1568 json.loads(result.output.strip())
1569
1570
1571 # ---------------------------------------------------------------------------
1572 # Stress — muse attributes check
1573 # ---------------------------------------------------------------------------
1574
1575
1576 class TestRunCheckStress:
1577 def _make_many_rules(self, n: int) -> str:
1578 lines: list[str] = []
1579 for i in range(n):
1580 lines.append(
1581 f'[[rules]]\npath = "dir_{i}/*.bin"\ndimension = "*"\n'
1582 f'strategy = "ours"\npriority = {i}\n'
1583 )
1584 return "\n".join(lines)
1585
1586 def test_1000_paths_against_500_rules_core(self, tmp_path: pathlib.Path) -> None:
1587 """1000 path resolutions against 500 rules — core layer, no CLI overhead."""
1588 repo = _make_repo(tmp_path)
1589 _write_attrs(repo, self._make_many_rules(500))
1590 _, rules = load_attributes_full(repo)
1591 from muse.cli.commands.attributes import _resolve_with_index
1592 for i in range(1000):
1593 strategy, _ = _resolve_with_index(rules, f"dir_{i % 500}/file.bin", "*")
1594 assert strategy == "ours"
1595
1596 def test_500_paths_json_serializes(self, tmp_path: pathlib.Path) -> None:
1597 repo = _make_repo(tmp_path)
1598 _write_attrs(repo, _SIMPLE_ATTRS)
1599 paths = [f"build/file_{i}.o" for i in range(500)]
1600 result = _invoke(repo, "check", *paths, "--json")
1601 assert result.exit_code == 0
1602 data = json.loads(result.output.strip())
1603 assert len(data["results"]) == 500
1604
1605 def test_concurrent_check_core_no_shared_state(self, tmp_path: pathlib.Path) -> None:
1606 """Eight threads resolve paths concurrently using the core layer."""
1607 from muse.cli.commands.attributes import _resolve_with_index
1608
1609 errors: list[str] = []
1610 repo = _make_repo(tmp_path)
1611 _write_attrs(repo, _SIMPLE_ATTRS)
1612 _, rules = load_attributes_full(repo)
1613
1614 def worker(idx: int) -> None:
1615 try:
1616 strategy, rule_idx = _resolve_with_index(rules, f"build/file_{idx}.o", "*")
1617 if strategy != "ours":
1618 errors.append(f"Thread {idx}: expected 'ours', got {strategy!r}")
1619 if rule_idx < 0:
1620 errors.append(f"Thread {idx}: expected rule_index >= 0, got {rule_idx}")
1621 except Exception as exc:
1622 errors.append(f"Thread {idx}: {exc}")
1623
1624 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1625 for t in threads:
1626 t.start()
1627 for t in threads:
1628 t.join()
1629 assert errors == [], f"Concurrent failures: {errors}"
1630
1631
1632 # ---------------------------------------------------------------------------
1633 # Extended — muse attributes validate (deeper coverage)
1634 # ---------------------------------------------------------------------------
1635
1636
1637 class TestRunValidateExtended:
1638 def test_j_alias_works_valid_file(self, tmp_path: pathlib.Path) -> None:
1639 repo = _make_repo(tmp_path)
1640 _write_attrs(repo, _SIMPLE_ATTRS)
1641 result = _invoke(repo, "validate", "-j")
1642 assert result.exit_code == 0
1643 data = json.loads(result.output.strip())
1644 assert data["valid"] is True
1645
1646 def test_help_mentions_json_flag(self, tmp_path: pathlib.Path) -> None:
1647 result = _invoke(tmp_path, "validate", "--help")
1648 assert "--json" in result.output or "-j" in result.output
1649
1650 def test_help_shows_exit_codes(self, tmp_path: pathlib.Path) -> None:
1651 result = _invoke(tmp_path, "validate", "--help")
1652 assert "Exit code" in result.output or "exit code" in result.output
1653
1654 def test_json_is_valid_and_parseable(self, tmp_path: pathlib.Path) -> None:
1655 repo = _make_repo(tmp_path)
1656 _write_attrs(repo, _SIMPLE_ATTRS)
1657 result = _invoke(repo, "validate", "--json")
1658 assert result.exit_code == 0
1659 data = json.loads(result.output)
1660 assert "valid" in data and "rule_count" in data and "errors" in data
1661
1662 def test_json_valid_has_true_and_empty_errors(self, tmp_path: pathlib.Path) -> None:
1663 repo = _make_repo(tmp_path)
1664 _write_attrs(repo, _SIMPLE_ATTRS)
1665 result = _invoke(repo, "validate", "--json")
1666 data = json.loads(result.output.strip())
1667 assert data["valid"] is True
1668 assert data["errors"] == []
1669
1670 def test_json_invalid_has_false_and_nonempty_errors(self, tmp_path: pathlib.Path) -> None:
1671 repo = _make_repo(tmp_path)
1672 _write_attrs(
1673 repo,
1674 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "bogus"\n',
1675 )
1676 result = _invoke(repo, "validate", "--json")
1677 assert result.exit_code != 0
1678 data = json.loads(result.output.strip())
1679 assert data["valid"] is False
1680 assert len(data["errors"]) > 0
1681
1682 def test_json_missing_file_kind_is_missing(self, tmp_path: pathlib.Path) -> None:
1683 repo = _make_repo(tmp_path)
1684 result = _invoke(repo, "validate", "--json")
1685 assert result.exit_code != 0
1686 data = json.loads(result.output.strip())
1687 assert data["errors"][0]["kind"] == "missing"
1688
1689 def test_json_bad_strategy_kind_is_semantic(self, tmp_path: pathlib.Path) -> None:
1690 repo = _make_repo(tmp_path)
1691 _write_attrs(
1692 repo,
1693 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "zap"\n',
1694 )
1695 result = _invoke(repo, "validate", "--json")
1696 data = json.loads(result.output.strip())
1697 assert data["errors"][0]["kind"] == "semantic"
1698
1699 def test_json_bad_toml_kind_is_semantic(self, tmp_path: pathlib.Path) -> None:
1700 repo = _make_repo(tmp_path)
1701 _write_attrs(repo, "[[broken\n")
1702 result = _invoke(repo, "validate", "--json")
1703 data = json.loads(result.output.strip())
1704 assert data["errors"][0]["kind"] == "semantic"
1705
1706 def test_json_errors_is_always_array(self, tmp_path: pathlib.Path) -> None:
1707 repo = _make_repo(tmp_path)
1708 _write_attrs(repo, _SIMPLE_ATTRS)
1709 result = _invoke(repo, "validate", "--json")
1710 data = json.loads(result.output.strip())
1711 assert isinstance(data["errors"], list)
1712
1713 def test_json_valid_is_bool(self, tmp_path: pathlib.Path) -> None:
1714 repo = _make_repo(tmp_path)
1715 _write_attrs(repo, _SIMPLE_ATTRS)
1716 result = _invoke(repo, "validate", "--json")
1717 data = json.loads(result.output.strip())
1718 assert isinstance(data["valid"], bool)
1719
1720 def test_json_error_message_is_string(self, tmp_path: pathlib.Path) -> None:
1721 repo = _make_repo(tmp_path)
1722 result = _invoke(repo, "validate", "--json")
1723 data = json.loads(result.output.strip())
1724 assert isinstance(data["errors"][0]["message"], str)
1725 assert len(data["errors"][0]["message"]) > 0
1726
1727 def test_text_success_shows_rule_count(self, tmp_path: pathlib.Path) -> None:
1728 repo = _make_repo(tmp_path)
1729 _write_attrs(repo, _SIMPLE_ATTRS)
1730 result = _invoke(repo, "validate")
1731 assert "3 rule" in result.output
1732
1733 def test_text_success_shows_domain(self, tmp_path: pathlib.Path) -> None:
1734 repo = _make_repo(tmp_path)
1735 _write_attrs(repo, _SIMPLE_ATTRS)
1736 result = _invoke(repo, "validate")
1737 assert "code" in result.output
1738
1739 def test_text_no_traceback_on_bad_toml(self, tmp_path: pathlib.Path) -> None:
1740 repo = _make_repo(tmp_path)
1741 _write_attrs(repo, "[[broken\n")
1742 result = _invoke(repo, "validate")
1743 assert "Traceback" not in result.output
1744
1745 def test_json_exit_zero_on_valid(self, tmp_path: pathlib.Path) -> None:
1746 repo = _make_repo(tmp_path)
1747 _write_attrs(repo, _SIMPLE_ATTRS)
1748 result = _invoke(repo, "validate", "--json")
1749 assert result.exit_code == 0
1750
1751 def test_json_exit_nonzero_on_invalid(self, tmp_path: pathlib.Path) -> None:
1752 repo = _make_repo(tmp_path)
1753 _write_attrs(
1754 repo,
1755 '[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "bad"\n',
1756 )
1757 result = _invoke(repo, "validate", "--json")
1758 assert result.exit_code != 0
1759
1760 def test_empty_rules_file_exits_zero(self, tmp_path: pathlib.Path) -> None:
1761 """A .museattributes with no rules but valid TOML is still valid."""
1762 repo = _make_repo(tmp_path)
1763 _write_attrs(repo, '[meta]\ndomain = "x"\n')
1764 result = _invoke(repo, "validate")
1765 assert result.exit_code == 0
1766
1767
1768 # ---------------------------------------------------------------------------
1769 # Security — muse attributes validate
1770 # ---------------------------------------------------------------------------
1771
1772
1773 class TestRunValidateSecurity:
1774 def test_ansi_in_domain_stripped_text_success(self, tmp_path: pathlib.Path) -> None:
1775 """Domain with ANSI sequences is sanitized in the success message."""
1776 repo = _make_repo(tmp_path)
1777 _write_attrs(
1778 repo,
1779 f'[meta]\ndomain = "safe"\n\n[[rules]]\npath = "*"\ndimension = "*"\nstrategy = "auto"\n',
1780 )
1781 result = _invoke(repo, "validate")
1782 assert result.exit_code == 0
1783 assert "\x1b[" not in result.output
1784
1785 def test_oversized_file_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
1786 repo = _make_repo(tmp_path)
1787 from muse.core.attributes import _MAX_ATTRIBUTES_BYTES
1788 (repo / ".museattributes").write_bytes(b"x" * (_MAX_ATTRIBUTES_BYTES + 1))
1789 result = _invoke(repo, "validate")
1790 assert result.exit_code != 0
1791
1792 def test_oversized_file_json_valid_false(self, tmp_path: pathlib.Path) -> None:
1793 repo = _make_repo(tmp_path)
1794 from muse.core.attributes import _MAX_ATTRIBUTES_BYTES
1795 (repo / ".museattributes").write_bytes(b"x" * (_MAX_ATTRIBUTES_BYTES + 1))
1796 result = _invoke(repo, "validate", "--json")
1797 assert result.exit_code != 0
1798 data = json.loads(result.output.strip())
1799 assert data["valid"] is False
1800
1801 def test_bad_toml_no_traceback_json_mode(self, tmp_path: pathlib.Path) -> None:
1802 repo = _make_repo(tmp_path)
1803 _write_attrs(repo, "[[broken\n")
1804 result = _invoke(repo, "validate", "--json")
1805 assert "Traceback" not in result.output
1806
1807 def test_json_stdout_only_on_success(self, tmp_path: pathlib.Path) -> None:
1808 """In JSON mode on success, stdout is exactly one valid JSON line."""
1809 repo = _make_repo(tmp_path)
1810 _write_attrs(repo, _SIMPLE_ATTRS)
1811 result = _invoke(repo, "validate", "--json")
1812 assert result.exit_code == 0
1813 json.loads(result.output.strip())
1814
1815 def test_json_stdout_only_on_failure(self, tmp_path: pathlib.Path) -> None:
1816 """In JSON mode on failure, stdout is exactly one valid JSON line."""
1817 repo = _make_repo(tmp_path)
1818 result = _invoke(repo, "validate", "--json")
1819 assert result.exit_code != 0
1820 json.loads(result.output.strip())
1821
1822
1823 # ---------------------------------------------------------------------------
1824 # Stress — muse attributes validate
1825 # ---------------------------------------------------------------------------
1826
1827
1828 class TestRunValidateStress:
1829 def _make_attrs_with_n_rules(self, n: int) -> str:
1830 lines = ['[meta]\ndomain = "stress"\n']
1831 for i in range(n):
1832 lines.append(
1833 f'[[rules]]\npath = "dir_{i}/*.bin"\ndimension = "*"\n'
1834 f'strategy = "ours"\npriority = {i}\n'
1835 )
1836 return "\n".join(lines)
1837
1838 def test_500_rule_file_validates_successfully(self, tmp_path: pathlib.Path) -> None:
1839 repo = _make_repo(tmp_path)
1840 _write_attrs(repo, self._make_attrs_with_n_rules(500))
1841 result = _invoke(repo, "validate", "--json")
1842 assert result.exit_code == 0
1843 data = json.loads(result.output.strip())
1844 assert data["valid"] is True
1845
1846 def test_file_with_one_bad_rule_among_200_reports_error(self, tmp_path: pathlib.Path) -> None:
1847 content = self._make_attrs_with_n_rules(200)
1848 content += '\n[[rules]]\npath = "bad/*"\ndimension = "*"\nstrategy = "INVALID"\n'
1849 repo = _make_repo(tmp_path)
1850 _write_attrs(repo, content)
1851 result = _invoke(repo, "validate", "--json")
1852 assert result.exit_code != 0
1853 data = json.loads(result.output.strip())
1854 assert data["valid"] is False
1855
1856 def test_concurrent_validate_core_no_shared_state(self, tmp_path: pathlib.Path) -> None:
1857 """Eight threads validate isolated repos via the core layer."""
1858 errors: list[str] = []
1859
1860 def worker(idx: int) -> None:
1861 try:
1862 repo = _make_repo(tmp_path / f"repo_{idx}")
1863 _write_attrs(repo, _SIMPLE_ATTRS)
1864 meta, rules = load_attributes_full(repo)
1865 if len(rules) != 3:
1866 errors.append(f"Thread {idx}: got {len(rules)} rules, want 3")
1867 if not meta.get("domain"):
1868 errors.append(f"Thread {idx}: missing domain")
1869 except Exception as exc:
1870 errors.append(f"Thread {idx}: {exc}")
1871
1872 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
1873 for t in threads:
1874 t.start()
1875 for t in threads:
1876 t.join()
1877 assert errors == [], f"Concurrent failures: {errors}"
1878
1879
1880 # ---------------------------------------------------------------------------
1881 # New-flag tests — list --strategy, list --dimension, list rule_count,
1882 # check --match-required, validate rule_count, indent=2
1883 # ---------------------------------------------------------------------------
1884
1885
1886 _MIXED_ATTRS = """\
1887 [meta]
1888 domain = "midi"
1889
1890 [[rules]]
1891 path = "drums/*"
1892 dimension = "notes"
1893 strategy = "ours"
1894 comment = "Drum notes stay ours."
1895 priority = 20
1896
1897 [[rules]]
1898 path = "keys/*"
1899 dimension = "pitch_bend"
1900 strategy = "theirs"
1901 comment = "Keys pitch-bend from remote."
1902 priority = 15
1903
1904 [[rules]]
1905 path = "stems/*"
1906 dimension = "notes"
1907 strategy = "union"
1908 comment = "Union note additions."
1909 priority = 10
1910
1911 [[rules]]
1912 path = "*"
1913 dimension = "*"
1914 strategy = "auto"
1915 comment = ""
1916 priority = 0
1917 """
1918
1919
1920 class TestListStrategyFilter:
1921 def test_strategy_filter_returns_only_matching(self, tmp_path: pathlib.Path) -> None:
1922 repo = _make_repo(tmp_path)
1923 _write_attrs(repo, _MIXED_ATTRS)
1924 result = _invoke(repo, "list", "--strategy", "ours", "--json")
1925 assert result.exit_code == 0
1926 data = json.loads(result.output)
1927 assert all(r["strategy"] == "ours" for r in data["rules"])
1928 assert data["rule_count"] == 1
1929
1930 def test_strategy_filter_excludes_others(self, tmp_path: pathlib.Path) -> None:
1931 repo = _make_repo(tmp_path)
1932 _write_attrs(repo, _MIXED_ATTRS)
1933 result = _invoke(repo, "list", "--strategy", "theirs", "--json")
1934 data = json.loads(result.output)
1935 for r in data["rules"]:
1936 assert r["strategy"] == "theirs"
1937
1938 def test_strategy_filter_no_match_returns_empty(self, tmp_path: pathlib.Path) -> None:
1939 repo = _make_repo(tmp_path)
1940 _write_attrs(repo, _MIXED_ATTRS)
1941 result = _invoke(repo, "list", "--strategy", "manual", "--json")
1942 assert result.exit_code == 0
1943 data = json.loads(result.output)
1944 assert data["rules"] == []
1945 assert data["rule_count"] == 0
1946
1947 def test_strategy_filter_text_mode(self, tmp_path: pathlib.Path) -> None:
1948 repo = _make_repo(tmp_path)
1949 _write_attrs(repo, _MIXED_ATTRS)
1950 result = _invoke(repo, "list", "--strategy", "union")
1951 assert result.exit_code == 0
1952 assert "union" in result.output
1953 assert "theirs" not in result.output
1954
1955 def test_strategy_short_flag(self, tmp_path: pathlib.Path) -> None:
1956 repo = _make_repo(tmp_path)
1957 _write_attrs(repo, _MIXED_ATTRS)
1958 result = _invoke(repo, "list", "-s", "ours", "--json")
1959 assert result.exit_code == 0
1960 data = json.loads(result.output)
1961 assert data["rule_count"] == 1
1962
1963
1964 class TestListDimensionFilter:
1965 def test_dimension_filter_returns_only_matching(self, tmp_path: pathlib.Path) -> None:
1966 repo = _make_repo(tmp_path)
1967 _write_attrs(repo, _MIXED_ATTRS)
1968 result = _invoke(repo, "list", "--dimension", "notes", "--json")
1969 assert result.exit_code == 0
1970 data = json.loads(result.output)
1971 assert all(r["dimension"] == "notes" for r in data["rules"])
1972 assert data["rule_count"] == 2
1973
1974 def test_dimension_filter_no_match_empty(self, tmp_path: pathlib.Path) -> None:
1975 repo = _make_repo(tmp_path)
1976 _write_attrs(repo, _MIXED_ATTRS)
1977 result = _invoke(repo, "list", "--dimension", "reverb", "--json")
1978 assert result.exit_code == 0
1979 data = json.loads(result.output)
1980 assert data["rules"] == []
1981 assert data["rule_count"] == 0
1982
1983 def test_strategy_and_dimension_combined(self, tmp_path: pathlib.Path) -> None:
1984 repo = _make_repo(tmp_path)
1985 _write_attrs(repo, _MIXED_ATTRS)
1986 result = _invoke(repo, "list", "--strategy", "ours", "--dimension", "notes", "--json")
1987 assert result.exit_code == 0
1988 data = json.loads(result.output)
1989 assert data["rule_count"] == 1
1990 assert data["rules"][0]["strategy"] == "ours"
1991 assert data["rules"][0]["dimension"] == "notes"
1992
1993 def test_combined_no_match(self, tmp_path: pathlib.Path) -> None:
1994 repo = _make_repo(tmp_path)
1995 _write_attrs(repo, _MIXED_ATTRS)
1996 result = _invoke(repo, "list", "--strategy", "ours", "--dimension", "pitch_bend", "--json")
1997 assert result.exit_code == 0
1998 data = json.loads(result.output)
1999 assert data["rule_count"] == 0
2000
2001
2002 class TestListRuleCount:
2003 def test_rule_count_present_in_json(self, tmp_path: pathlib.Path) -> None:
2004 repo = _make_repo(tmp_path)
2005 _write_attrs(repo, _MIXED_ATTRS)
2006 result = _invoke(repo, "list", "--json")
2007 data = json.loads(result.output)
2008 assert "rule_count" in data
2009
2010 def test_rule_count_matches_rules_length(self, tmp_path: pathlib.Path) -> None:
2011 repo = _make_repo(tmp_path)
2012 _write_attrs(repo, _MIXED_ATTRS)
2013 result = _invoke(repo, "list", "--json")
2014 data = json.loads(result.output)
2015 assert data["rule_count"] == len(data["rules"])
2016
2017 def test_rule_count_zero_on_empty_file(self, tmp_path: pathlib.Path) -> None:
2018 repo = _make_repo(tmp_path)
2019 _write_attrs(repo, "[meta]\ndomain = \"code\"\n")
2020 result = _invoke(repo, "list", "--json")
2021 data = json.loads(result.output)
2022 assert data["rule_count"] == 0
2023 assert data["rules"] == []
2024
2025
2026 class TestCheckMatchRequired:
2027 def test_match_required_exits_0_when_all_matched(self, tmp_path: pathlib.Path) -> None:
2028 repo = _make_repo(tmp_path)
2029 _write_attrs(repo, _MIXED_ATTRS)
2030 result = _invoke(repo, "check", "drums/kick.mid", "--match-required")
2031 assert result.exit_code == 0
2032
2033 def test_match_required_exits_1_when_unmatched(self, tmp_path: pathlib.Path) -> None:
2034 repo = _make_repo(tmp_path)
2035 # Only rule: drums/* → ours. Check a path that won't match.
2036 attrs = "[meta]\ndomain=\"midi\"\n[[rules]]\npath=\"drums/*\"\ndimension=\"*\"\nstrategy=\"ours\"\n"
2037 _write_attrs(repo, attrs)
2038 result = _invoke(repo, "check", "untracked/file.mid", "--match-required")
2039 assert result.exit_code != 0
2040
2041 def test_match_required_error_on_stderr(self, tmp_path: pathlib.Path) -> None:
2042 repo = _make_repo(tmp_path)
2043 attrs = "[meta]\ndomain=\"midi\"\n[[rules]]\npath=\"drums/*\"\ndimension=\"*\"\nstrategy=\"ours\"\n"
2044 _write_attrs(repo, attrs)
2045 result = _invoke(repo, "check", "other/file.mid", "--match-required")
2046 assert result.exit_code != 0
2047 assert "❌" in result.stderr
2048
2049 def test_match_required_json_still_prints_before_exit(self, tmp_path: pathlib.Path) -> None:
2050 repo = _make_repo(tmp_path)
2051 attrs = "[meta]\ndomain=\"midi\"\n[[rules]]\npath=\"drums/*\"\ndimension=\"*\"\nstrategy=\"ours\"\n"
2052 _write_attrs(repo, attrs)
2053 result = _invoke(repo, "check", "other/file.mid", "--match-required", "--json")
2054 assert result.exit_code != 0
2055 # Error lands on stderr; JSON is on stdout. CliRunner merges them in
2056 # result.output, so split on the first closing brace to isolate JSON.
2057 assert "❌" in result.stderr
2058 json_part = result.output[: result.output.rfind("}") + 1]
2059 data = json.loads(json_part)
2060 assert "results" in data
2061
2062 def test_match_required_with_mixed_paths(self, tmp_path: pathlib.Path) -> None:
2063 repo = _make_repo(tmp_path)
2064 attrs = "[meta]\ndomain=\"midi\"\n[[rules]]\npath=\"drums/*\"\ndimension=\"*\"\nstrategy=\"ours\"\n"
2065 _write_attrs(repo, attrs)
2066 # One matched, one unmatched → exit 1
2067 result = _invoke(repo, "check", "drums/kick.mid", "other/file.mid", "--match-required")
2068 assert result.exit_code != 0
2069
2070 def test_match_required_false_by_default(self, tmp_path: pathlib.Path) -> None:
2071 repo = _make_repo(tmp_path)
2072 attrs = "[meta]\ndomain=\"midi\"\n[[rules]]\npath=\"drums/*\"\ndimension=\"*\"\nstrategy=\"ours\"\n"
2073 _write_attrs(repo, attrs)
2074 # Without --match-required, unmatched path is still exit 0
2075 result = _invoke(repo, "check", "other/file.mid")
2076 assert result.exit_code == 0
2077
2078
2079 class TestValidateRuleCount:
2080 def test_rule_count_present_on_success(self, tmp_path: pathlib.Path) -> None:
2081 repo = _make_repo(tmp_path)
2082 _write_attrs(repo, _MIXED_ATTRS)
2083 result = _invoke(repo, "validate", "--json")
2084 assert result.exit_code == 0
2085 data = json.loads(result.output)
2086 assert "rule_count" in data
2087 assert data["rule_count"] == 4
2088
2089 def test_rule_count_zero_on_missing(self, tmp_path: pathlib.Path) -> None:
2090 repo = _make_repo(tmp_path)
2091 result = _invoke(repo, "validate", "--json")
2092 assert result.exit_code != 0
2093 data = json.loads(result.output)
2094 assert data["rule_count"] == 0
2095
2096 def test_rule_count_zero_on_parse_error(self, tmp_path: pathlib.Path) -> None:
2097 repo = _make_repo(tmp_path)
2098 _write_attrs(repo, "not valid toml ][[[")
2099 result = _invoke(repo, "validate", "--json")
2100 assert result.exit_code != 0
2101 data = json.loads(result.output)
2102 assert data["rule_count"] == 0
2103
2104
2105 class TestJsonCompact:
2106 def test_list_json_valid(self, tmp_path: pathlib.Path) -> None:
2107 repo = _make_repo(tmp_path)
2108 _write_attrs(repo, _SIMPLE_ATTRS)
2109 result = _invoke(repo, "list", "--json")
2110 assert result.exit_code == 0
2111 data = json.loads(result.output)
2112 assert "rules" in data
2113
2114 def test_check_json_valid(self, tmp_path: pathlib.Path) -> None:
2115 repo = _make_repo(tmp_path)
2116 _write_attrs(repo, _SIMPLE_ATTRS)
2117 result = _invoke(repo, "check", "build/foo.o", "--json")
2118 assert result.exit_code == 0
2119 data = json.loads(result.output)
2120 assert "results" in data
2121
2122 def test_validate_json_valid(self, tmp_path: pathlib.Path) -> None:
2123 repo = _make_repo(tmp_path)
2124 _write_attrs(repo, _SIMPLE_ATTRS)
2125 result = _invoke(repo, "validate", "--json")
2126 assert result.exit_code == 0
2127 data = json.loads(result.output)
2128 assert "valid" in data
2129
2130
2131 # ---------------------------------------------------------------------------
2132 # Flag registration tests
2133 # ---------------------------------------------------------------------------
2134
2135 import argparse as _argparse
2136 from muse.cli.commands.attributes import register as _register_attributes
2137
2138
2139 def _parse_attrs(*args: str) -> _argparse.Namespace:
2140 """Build an argument parser via register() and parse args."""
2141 root_p = _argparse.ArgumentParser()
2142 subs = root_p.add_subparsers(dest="cmd")
2143 _register_attributes(subs)
2144 return root_p.parse_args(["attributes", *args])
2145
2146
2147 class TestRegisterFlags:
2148 # ── check subcommand ────────────────────────────────────────────────────
2149 def test_check_default_json_out_is_false(self) -> None:
2150 ns = _parse_attrs("check", "src/foo.py")
2151 assert ns.json_out is False
2152
2153 def test_check_json_flag_sets_json_out(self) -> None:
2154 ns = _parse_attrs("check", "src/foo.py", "--json")
2155 assert ns.json_out is True
2156
2157 def test_check_j_shorthand_sets_json_out(self) -> None:
2158 ns = _parse_attrs("check", "src/foo.py", "-j")
2159 assert ns.json_out is True
2160
2161 def test_check_dimension_default_is_star(self) -> None:
2162 ns = _parse_attrs("check", "src/foo.py")
2163 assert ns.dimension == "*"
2164
2165 def test_check_dimension_flag(self) -> None:
2166 ns = _parse_attrs("check", "src/foo.py", "--dimension", "notes")
2167 assert ns.dimension == "notes"
2168
2169 def test_check_d_shorthand(self) -> None:
2170 ns = _parse_attrs("check", "src/foo.py", "-d", "pitch_bend")
2171 assert ns.dimension == "pitch_bend"
2172
2173 def test_check_match_required_default_false(self) -> None:
2174 ns = _parse_attrs("check", "src/foo.py")
2175 assert ns.match_required is False
2176
2177 def test_check_match_required_flag(self) -> None:
2178 ns = _parse_attrs("check", "src/foo.py", "--match-required")
2179 assert ns.match_required is True
2180
2181 # ── list subcommand ─────────────────────────────────────────────────────
2182 def test_list_default_json_out_is_false(self) -> None:
2183 ns = _parse_attrs("list")
2184 assert ns.json_out is False
2185
2186 def test_list_json_flag_sets_json_out(self) -> None:
2187 ns = _parse_attrs("list", "--json")
2188 assert ns.json_out is True
2189
2190 def test_list_j_shorthand_sets_json_out(self) -> None:
2191 ns = _parse_attrs("list", "-j")
2192 assert ns.json_out is True
2193
2194 def test_list_strategy_default_none(self) -> None:
2195 ns = _parse_attrs("list")
2196 assert ns.strategy is None
2197
2198 def test_list_strategy_flag(self) -> None:
2199 ns = _parse_attrs("list", "--strategy", "ours")
2200 assert ns.strategy == "ours"
2201
2202 def test_list_dimension_default_none(self) -> None:
2203 ns = _parse_attrs("list")
2204 assert ns.dimension is None
2205
2206 def test_list_dimension_flag(self) -> None:
2207 ns = _parse_attrs("list", "--dimension", "notes")
2208 assert ns.dimension == "notes"
2209
2210 # ── validate subcommand ─────────────────────────────────────────────────
2211 def test_validate_default_json_out_is_false(self) -> None:
2212 ns = _parse_attrs("validate")
2213 assert ns.json_out is False
2214
2215 def test_validate_json_flag_sets_json_out(self) -> None:
2216 ns = _parse_attrs("validate", "--json")
2217 assert ns.json_out is True
2218
2219 def test_validate_j_shorthand_sets_json_out(self) -> None:
2220 ns = _parse_attrs("validate", "-j")
2221 assert ns.json_out is True
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago