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