gabriel / muse public
test_cmd_annotate_hardening.py python
965 lines 34.7 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
1 """Comprehensive tests for ``muse annotate`` CLI hardening.
2
3 Audit findings addressed
4 ------------------------
5 Security
6 - ANSI injection in reviewer names (from disk and from user input) now
7 blocked by sanitize_display() at output time and sanitize_provenance()
8 at input validation time.
9 - Control characters in reviewer names rejected before storage.
10 - Commit references resolved via resolve_commit_ref (safe glob-prefix scan)
11 — raw user strings no longer passed directly to read_commit.
12 - Error messages routed to stderr; stdout carries only data.
13 - SystemExit(1) replaced with ExitCode enum values.
14
15 Performance
16 - ORSet reconstruction removed from mutation path — replaced with pure
17 set union (O(n) set operations vs. O(n) ORSet token generation for
18 every existing reviewer on every mutation).
19
20 Correctness
21 - Short commit IDs (prefix scan) now work: muse annotate abc1234.
22 - HEAD~N references work through resolve_commit_ref.
23 - Multiple --reviewed-by flags (action='append') work without comma sep.
24 - --remove-reviewer removes from stored list.
25 - --dry-run shows prospective state without writing.
26
27 Agent UX
28 - --json flag emits complete, stable _AnnotateJson schema.
29 - JSON always present and always contains all fields.
30
31 Coverage tiers
32 --------------
33 - Unit: _validate_reviewer, _parse_reviewer_list, _commit_to_json
34 - Integration: run show, add, remove, test-run, dry-run, combination
35 - Security: ANSI in names, control chars, stderr routing, no tracebacks
36 - E2E: full CLI invocation, JSON schema, exit codes
37 - Stress: 100 reviewers, 50 annotations, concurrent isolated repos
38 """
39 from __future__ import annotations
40
41 import argparse
42 import datetime
43 import json
44 import pathlib
45 import threading
46 from typing import TYPE_CHECKING
47 from unittest.mock import patch
48
49 import pytest
50
51 from muse.core.ids import hash_commit, hash_snapshot
52 from muse.core.commits import (
53 CommitRecord,
54 read_commit,
55 write_commit,
56 )
57 from muse.core.errors import ExitCode
58 from muse.core.types import NULL_COMMIT_ID
59 from muse.core.paths import heads_dir, muse_dir, ref_path
60 from tests.cli_test_helper import CliRunner, InvokeResult
61
62 if TYPE_CHECKING:
63 from muse.cli.commands.annotate import _AnnotateJson
64
65 runner = CliRunner()
66 cli = None # argparse migration — CliRunner ignores this
67
68
69 # ---------------------------------------------------------------------------
70 # Helpers
71 # ---------------------------------------------------------------------------
72
73
74 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
75 """Create a minimal Muse repo layout."""
76 muse = muse_dir(tmp_path)
77 for sub in ("commits", "snapshots", "refs/heads", "objects"):
78 (muse / sub).mkdir(parents=True, exist_ok=True)
79 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
80 (muse / "repo.json").write_text(
81 json.dumps({"repo_id": "test-repo"}), encoding="utf-8"
82 )
83 return tmp_path
84
85
86 def _write_commit(
87 root: pathlib.Path,
88 message: str = "test commit",
89 branch: str = "main",
90 ) -> CommitRecord:
91 """Write a content-addressed CommitRecord and update the branch ref."""
92 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
93 snap_id = hash_snapshot({})
94 cid = hash_commit(
95 parent_ids=[],
96 snapshot_id=snap_id,
97 message=message,
98 committed_at_iso=committed_at.isoformat(),
99 author="test-author",
100 )
101 record = CommitRecord(
102 commit_id=cid,
103 branch=branch,
104 snapshot_id=snap_id,
105 message=message,
106 committed_at=committed_at,
107 author="test-author",
108 )
109 write_commit(root, record)
110 (ref_path(root, branch)).write_text(cid, encoding="utf-8")
111 return record
112
113
114 def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
115 """Run ``muse annotate <args>`` inside *root*."""
116 return runner.invoke(
117 cli,
118 ["annotate", *args],
119 env={"MUSE_REPO_ROOT": str(root)},
120 )
121
122
123 def _parse_json(result: InvokeResult) -> "_AnnotateJson":
124 """Extract and parse the first JSON blob from *result.output*."""
125 from muse.cli.commands.annotate import _AnnotateJson
126
127 start = result.output.index("{")
128 blob = result.output[start:]
129 depth = 0
130 end = 0
131 for i, ch in enumerate(blob):
132 if ch == "{":
133 depth += 1
134 elif ch == "}":
135 depth -= 1
136 if depth == 0:
137 end = i + 1
138 break
139 raw = json.loads(blob[:end])
140 assert isinstance(raw, dict)
141 reviewed_by = raw.get("reviewed_by", [])
142 assert isinstance(reviewed_by, list)
143 str_reviewed: list[str] = [str(r) for r in reviewed_by]
144 return _AnnotateJson(
145 commit_id=str(raw.get("commit_id", "")),
146 message=str(raw.get("message", "")),
147 branch=str(raw.get("branch", "")),
148 author=str(raw.get("author", "")),
149 committed_at=str(raw.get("committed_at", "")),
150 reviewed_by=str_reviewed,
151 test_runs=int(raw.get("test_runs", 0)),
152 changed=bool(raw.get("changed", False)),
153 dry_run=bool(raw.get("dry_run", False)),
154 )
155
156
157 # ---------------------------------------------------------------------------
158 # Unit — _validate_reviewer
159 # ---------------------------------------------------------------------------
160
161
162 class TestValidateReviewer:
163 def test_valid_name_returns_unchanged(self) -> None:
164 from muse.cli.commands.annotate import _validate_reviewer
165
166 assert _validate_reviewer("alice") == "alice"
167 assert _validate_reviewer("agent-x") == "agent-x"
168 assert _validate_reviewer("ci-bot-v2") == "ci-bot-v2"
169
170 def test_empty_name_exits_user_error(self) -> None:
171 from muse.cli.commands.annotate import _validate_reviewer
172
173 with pytest.raises(SystemExit) as exc:
174 _validate_reviewer("")
175 assert exc.value.code == ExitCode.USER_ERROR.value
176
177 def test_name_with_ansi_exits_user_error(self) -> None:
178 from muse.cli.commands.annotate import _validate_reviewer
179
180 with pytest.raises(SystemExit) as exc:
181 _validate_reviewer("\x1b[31mmalicious\x1b[0m")
182 assert exc.value.code == ExitCode.USER_ERROR.value
183
184 def test_name_with_null_byte_exits_user_error(self) -> None:
185 from muse.cli.commands.annotate import _validate_reviewer
186
187 with pytest.raises(SystemExit) as exc:
188 _validate_reviewer("foo\x00bar")
189 assert exc.value.code == ExitCode.USER_ERROR.value
190
191 def test_name_with_newline_exits_user_error(self) -> None:
192 from muse.cli.commands.annotate import _validate_reviewer
193
194 with pytest.raises(SystemExit) as exc:
195 _validate_reviewer("alice\nbob")
196 assert exc.value.code == ExitCode.USER_ERROR.value
197
198 def test_overlong_name_exits_user_error(self) -> None:
199 from muse.cli.commands.annotate import _validate_reviewer, _MAX_REVIEWER_LEN
200
201 with pytest.raises(SystemExit) as exc:
202 _validate_reviewer("a" * (_MAX_REVIEWER_LEN + 1))
203 assert exc.value.code == ExitCode.USER_ERROR.value
204
205 def test_exact_max_length_accepted(self) -> None:
206 from muse.cli.commands.annotate import _validate_reviewer, _MAX_REVIEWER_LEN
207
208 name = "a" * _MAX_REVIEWER_LEN
209 assert _validate_reviewer(name) == name
210
211
212 # ---------------------------------------------------------------------------
213 # Unit — _parse_reviewer_list
214 # ---------------------------------------------------------------------------
215
216
217 class TestParseReviewerList:
218 def test_single_name(self) -> None:
219 from muse.cli.commands.annotate import _parse_reviewer_list
220
221 assert _parse_reviewer_list(["alice"]) == ["alice"]
222
223 def test_comma_separated(self) -> None:
224 from muse.cli.commands.annotate import _parse_reviewer_list
225
226 result = _parse_reviewer_list(["alice,bob"])
227 assert "alice" in result
228 assert "bob" in result
229
230 def test_multiple_flags(self) -> None:
231 from muse.cli.commands.annotate import _parse_reviewer_list
232
233 result = _parse_reviewer_list(["alice", "bob"])
234 assert "alice" in result
235 assert "bob" in result
236
237 def test_deduplication(self) -> None:
238 from muse.cli.commands.annotate import _parse_reviewer_list
239
240 result = _parse_reviewer_list(["alice", "alice"])
241 assert result.count("alice") == 1
242
243 def test_empty_list(self) -> None:
244 from muse.cli.commands.annotate import _parse_reviewer_list
245
246 assert _parse_reviewer_list([]) == []
247
248 def test_whitespace_stripped(self) -> None:
249 from muse.cli.commands.annotate import _parse_reviewer_list
250
251 result = _parse_reviewer_list([" alice , bob "])
252 assert "alice" in result
253 assert "bob" in result
254
255 def test_empty_segments_skipped(self) -> None:
256 from muse.cli.commands.annotate import _parse_reviewer_list
257
258 result = _parse_reviewer_list(["alice,,bob"])
259 assert "alice" in result
260 assert "bob" in result
261 assert "" not in result
262
263 def test_invalid_name_raises(self) -> None:
264 from muse.cli.commands.annotate import _parse_reviewer_list
265
266 with pytest.raises(SystemExit):
267 _parse_reviewer_list(["\x1b[31mmalicious\x1b[0m"])
268
269
270 # ---------------------------------------------------------------------------
271 # Unit — _commit_to_json
272 # ---------------------------------------------------------------------------
273
274
275 class TestCommitToJson:
276 def _record(self) -> CommitRecord:
277 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
278 snap_id = hash_snapshot({})
279 cid = hash_commit(
280 parent_ids=[],
281 snapshot_id=snap_id,
282 message="msg",
283 committed_at_iso=committed_at.isoformat(),
284 author="alice",
285 )
286 return CommitRecord(
287 commit_id=cid,
288 branch="main",
289 snapshot_id=snap_id,
290 message="msg",
291 committed_at=committed_at,
292 author="alice",
293 reviewed_by=["bob"],
294 test_runs=3,
295 )
296
297 def test_all_fields_present(self) -> None:
298 from muse.cli.commands.annotate import _commit_to_json
299
300 rec = self._record()
301 j = _commit_to_json(rec, changed=True, dry_run=False)
302 for field in (
303 "commit_id", "message", "branch", "author",
304 "committed_at", "reviewed_by", "test_runs", "changed", "dry_run",
305 ):
306 assert field in j, f"Missing field: {field}"
307
308 def test_changed_and_dry_run_flags(self) -> None:
309 from muse.cli.commands.annotate import _commit_to_json
310
311 rec = self._record()
312 j = _commit_to_json(rec, changed=True, dry_run=True)
313 assert j["changed"] is True
314 assert j["dry_run"] is True
315
316 def test_reviewed_by_is_list(self) -> None:
317 from muse.cli.commands.annotate import _commit_to_json
318
319 rec = self._record()
320 j = _commit_to_json(rec, changed=False, dry_run=False)
321 assert isinstance(j["reviewed_by"], list)
322 assert "bob" in j["reviewed_by"]
323
324
325 # ---------------------------------------------------------------------------
326 # Integration — show mode (no mutation flags)
327 # ---------------------------------------------------------------------------
328
329
330 class TestShowMode:
331 def test_show_no_reviewers(self, tmp_path: pathlib.Path) -> None:
332 repo = _make_repo(tmp_path)
333 c = _write_commit(repo)
334 result = _invoke(repo, c.commit_id)
335 assert result.exit_code == 0
336 assert "reviewed-by: (none)" in result.output
337
338 def test_show_existing_reviewers(self, tmp_path: pathlib.Path) -> None:
339 repo = _make_repo(tmp_path)
340 c = _write_commit(repo)
341 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
342 result = _invoke(repo, c.commit_id)
343 assert result.exit_code == 0
344 assert "alice" in result.output
345
346 def test_show_test_runs(self, tmp_path: pathlib.Path) -> None:
347 repo = _make_repo(tmp_path)
348 c = _write_commit(repo)
349 _invoke(repo, "--test-run", c.commit_id)
350 result = _invoke(repo, c.commit_id)
351 assert result.exit_code == 0
352 assert "test-runs: 1" in result.output
353
354 def test_show_head_default(self, tmp_path: pathlib.Path) -> None:
355 repo = _make_repo(tmp_path)
356 _write_commit(repo)
357 result = _invoke(repo)
358 assert result.exit_code == 0
359 assert "reviewed-by" in result.output
360
361 def test_show_json_schema(self, tmp_path: pathlib.Path) -> None:
362 repo = _make_repo(tmp_path)
363 c = _write_commit(repo)
364 result = _invoke(repo, "--json", c.commit_id)
365 assert result.exit_code == 0
366 data = _parse_json(result)
367 assert data["commit_id"] == c.commit_id
368 assert data["changed"] is False
369 assert data["dry_run"] is False
370 assert isinstance(data["reviewed_by"], list)
371 assert isinstance(data["test_runs"], int)
372
373 def test_show_json_has_message_and_branch(self, tmp_path: pathlib.Path) -> None:
374 repo = _make_repo(tmp_path)
375 c = _write_commit(repo, message="feat: some feature")
376 result = _invoke(repo, "--json", c.commit_id)
377 data = _parse_json(result)
378 assert data["message"] == "feat: some feature"
379 assert data["branch"] == "main"
380
381 def test_unknown_commit_exits_not_found(self, tmp_path: pathlib.Path) -> None:
382 repo = _make_repo(tmp_path)
383 result = _invoke(repo, "deadbeef")
384 assert result.exit_code == ExitCode.NOT_FOUND.value
385
386 def test_unknown_commit_no_traceback(self, tmp_path: pathlib.Path) -> None:
387 repo = _make_repo(tmp_path)
388 result = _invoke(repo, "deadbeef")
389 assert "Traceback" not in result.output
390
391
392 # ---------------------------------------------------------------------------
393 # Integration — add reviewer (--reviewed-by)
394 # ---------------------------------------------------------------------------
395
396
397 class TestAddReviewer:
398 def test_single_reviewer_added(self, tmp_path: pathlib.Path) -> None:
399 repo = _make_repo(tmp_path)
400 c = _write_commit(repo)
401 result = _invoke(repo, "--reviewed-by", "alice", c.commit_id)
402 assert result.exit_code == 0
403 rec = read_commit(repo, c.commit_id)
404 assert rec is not None
405 assert "alice" in rec.reviewed_by
406
407 def test_comma_separated_reviewers(self, tmp_path: pathlib.Path) -> None:
408 repo = _make_repo(tmp_path)
409 c = _write_commit(repo)
410 result = _invoke(repo, "--reviewed-by", "alice,bob", c.commit_id)
411 assert result.exit_code == 0
412 rec = read_commit(repo, c.commit_id)
413 assert rec is not None
414 assert "alice" in rec.reviewed_by
415 assert "bob" in rec.reviewed_by
416
417 def test_multiple_reviewed_by_flags(self, tmp_path: pathlib.Path) -> None:
418 repo = _make_repo(tmp_path)
419 c = _write_commit(repo)
420 result = _invoke(
421 repo,
422 "--reviewed-by", "alice",
423 "--reviewed-by", "bob",
424 c.commit_id,
425 )
426 assert result.exit_code == 0
427 rec = read_commit(repo, c.commit_id)
428 assert rec is not None
429 assert "alice" in rec.reviewed_by
430 assert "bob" in rec.reviewed_by
431
432 def test_orset_idempotent(self, tmp_path: pathlib.Path) -> None:
433 repo = _make_repo(tmp_path)
434 c = _write_commit(repo)
435 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
436 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
437 rec = read_commit(repo, c.commit_id)
438 assert rec is not None
439 assert rec.reviewed_by.count("alice") == 1
440
441 def test_reviewer_added_message_shown(self, tmp_path: pathlib.Path) -> None:
442 repo = _make_repo(tmp_path)
443 c = _write_commit(repo)
444 result = _invoke(repo, "--reviewed-by", "alice", c.commit_id)
445 assert "alice" in result.output
446 assert "Added" in result.output
447
448 def test_no_changes_when_reviewer_already_present(
449 self, tmp_path: pathlib.Path
450 ) -> None:
451 repo = _make_repo(tmp_path)
452 c = _write_commit(repo)
453 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
454 result = _invoke(repo, "--reviewed-by", "alice", c.commit_id)
455 assert result.exit_code == 0
456 assert "no changes" in result.output
457
458 def test_json_reflects_new_reviewer(self, tmp_path: pathlib.Path) -> None:
459 repo = _make_repo(tmp_path)
460 c = _write_commit(repo)
461 result = _invoke(repo, "--reviewed-by", "alice", "--json", c.commit_id)
462 assert result.exit_code == 0
463 data = _parse_json(result)
464 assert "alice" in data["reviewed_by"]
465 assert data["changed"] is True
466
467 def test_json_changed_false_when_no_change(self, tmp_path: pathlib.Path) -> None:
468 repo = _make_repo(tmp_path)
469 c = _write_commit(repo)
470 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
471 result = _invoke(repo, "--reviewed-by", "alice", "--json", c.commit_id)
472 data = _parse_json(result)
473 assert data["changed"] is False
474
475
476 # ---------------------------------------------------------------------------
477 # Integration — remove reviewer (--remove-reviewer)
478 # ---------------------------------------------------------------------------
479
480
481 class TestRemoveReviewer:
482 def test_remove_existing_reviewer(self, tmp_path: pathlib.Path) -> None:
483 repo = _make_repo(tmp_path)
484 c = _write_commit(repo)
485 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
486 result = _invoke(repo, "--remove-reviewer", "alice", c.commit_id)
487 assert result.exit_code == 0
488 rec = read_commit(repo, c.commit_id)
489 assert rec is not None
490 assert "alice" not in rec.reviewed_by
491
492 def test_remove_absent_reviewer_warns_to_stderr(
493 self, tmp_path: pathlib.Path
494 ) -> None:
495 repo = _make_repo(tmp_path)
496 c = _write_commit(repo)
497 result = _invoke(repo, "--remove-reviewer", "nobody", c.commit_id)
498 assert result.exit_code == 0
499 assert "⚠️" in result.stderr or "not present" in result.stderr
500
501 def test_remove_one_preserves_others(self, tmp_path: pathlib.Path) -> None:
502 repo = _make_repo(tmp_path)
503 c = _write_commit(repo)
504 _invoke(repo, "--reviewed-by", "alice,bob", c.commit_id)
505 _invoke(repo, "--remove-reviewer", "alice", c.commit_id)
506 rec = read_commit(repo, c.commit_id)
507 assert rec is not None
508 assert "alice" not in rec.reviewed_by
509 assert "bob" in rec.reviewed_by
510
511 def test_remove_json_reflects_change(self, tmp_path: pathlib.Path) -> None:
512 repo = _make_repo(tmp_path)
513 c = _write_commit(repo)
514 _invoke(repo, "--reviewed-by", "alice", c.commit_id)
515 result = _invoke(
516 repo, "--remove-reviewer", "alice", "--json", c.commit_id
517 )
518 data = _parse_json(result)
519 assert "alice" not in data["reviewed_by"]
520 assert data["changed"] is True
521
522
523 # ---------------------------------------------------------------------------
524 # Integration — test-run counter
525 # ---------------------------------------------------------------------------
526
527
528 class TestTestRun:
529 def test_test_run_increments(self, tmp_path: pathlib.Path) -> None:
530 repo = _make_repo(tmp_path)
531 c = _write_commit(repo)
532 _invoke(repo, "--test-run", c.commit_id)
533 _invoke(repo, "--test-run", c.commit_id)
534 rec = read_commit(repo, c.commit_id)
535 assert rec is not None
536 assert rec.test_runs == 2
537
538 def test_test_run_shown_in_output(self, tmp_path: pathlib.Path) -> None:
539 repo = _make_repo(tmp_path)
540 c = _write_commit(repo)
541 result = _invoke(repo, "--test-run", c.commit_id)
542 assert result.exit_code == 0
543 assert "Test run recorded" in result.output
544
545 def test_test_run_json_schema(self, tmp_path: pathlib.Path) -> None:
546 repo = _make_repo(tmp_path)
547 c = _write_commit(repo)
548 result = _invoke(repo, "--test-run", "--json", c.commit_id)
549 assert result.exit_code == 0
550 data = _parse_json(result)
551 assert data["test_runs"] == 1
552 assert data["changed"] is True
553
554
555 # ---------------------------------------------------------------------------
556 # Integration — dry-run
557 # ---------------------------------------------------------------------------
558
559
560 class TestDryRun:
561 def test_dry_run_does_not_write(self, tmp_path: pathlib.Path) -> None:
562 repo = _make_repo(tmp_path)
563 c = _write_commit(repo)
564 _invoke(repo, "--reviewed-by", "alice", "--dry-run", c.commit_id)
565 rec = read_commit(repo, c.commit_id)
566 assert rec is not None
567 assert "alice" not in rec.reviewed_by
568
569 def test_dry_run_shows_prospective_state(self, tmp_path: pathlib.Path) -> None:
570 repo = _make_repo(tmp_path)
571 c = _write_commit(repo)
572 result = _invoke(repo, "--reviewed-by", "alice", "--dry-run", c.commit_id)
573 assert result.exit_code == 0
574 assert "dry-run" in result.output.lower() or "[dry-run]" in result.output
575
576 def test_dry_run_json_reflects_prospective_state(
577 self, tmp_path: pathlib.Path
578 ) -> None:
579 repo = _make_repo(tmp_path)
580 c = _write_commit(repo)
581 result = _invoke(
582 repo, "--reviewed-by", "alice", "--dry-run", "--json", c.commit_id
583 )
584 data = _parse_json(result)
585 assert "alice" in data["reviewed_by"] # prospective
586 assert data["dry_run"] is True
587 assert data["changed"] is True
588 # Verify disk was NOT modified
589 rec = read_commit(repo, c.commit_id)
590 assert rec is not None
591 assert "alice" not in rec.reviewed_by
592
593 def test_dry_run_test_run_not_incremented(self, tmp_path: pathlib.Path) -> None:
594 repo = _make_repo(tmp_path)
595 c = _write_commit(repo)
596 _invoke(repo, "--test-run", "--dry-run", c.commit_id)
597 rec = read_commit(repo, c.commit_id)
598 assert rec is not None
599 assert rec.test_runs == 0
600
601
602 # ---------------------------------------------------------------------------
603 # Integration — commit reference resolution
604 # ---------------------------------------------------------------------------
605
606
607 class TestCommitRefResolution:
608 def test_full_commit_id(self, tmp_path: pathlib.Path) -> None:
609 repo = _make_repo(tmp_path)
610 c = _write_commit(repo)
611 result = _invoke(repo, c.commit_id)
612 assert result.exit_code == 0
613
614 def test_short_prefix_resolves(self, tmp_path: pathlib.Path) -> None:
615 repo = _make_repo(tmp_path)
616 c = _write_commit(repo)
617 prefix = c.commit_id.removeprefix("sha256:")[:8]
618 result = _invoke(repo, prefix)
619 assert result.exit_code == 0
620
621 def test_head_default_resolves(self, tmp_path: pathlib.Path) -> None:
622 repo = _make_repo(tmp_path)
623 _write_commit(repo)
624 result = _invoke(repo)
625 assert result.exit_code == 0
626
627 def test_head_tilde_resolves(self, tmp_path: pathlib.Path) -> None:
628 repo = _make_repo(tmp_path)
629 c1 = _write_commit(repo, message="first")
630 # Write a second commit with c1 as parent
631 committed_at = datetime.datetime(2026, 3, 2, tzinfo=datetime.timezone.utc)
632 snap_id = hash_snapshot({})
633 cid2 = hash_commit(
634 parent_ids=[c1.commit_id],
635 snapshot_id=snap_id,
636 message="second",
637 committed_at_iso=committed_at.isoformat(),
638 author="test-author",
639 )
640 c2 = CommitRecord(
641 commit_id=cid2,
642 branch="main",
643 snapshot_id=snap_id,
644 message="second",
645 committed_at=committed_at,
646 author="test-author",
647 parent_commit_id=c1.commit_id,
648 )
649 write_commit(repo, c2)
650 (heads_dir(repo) / "main").write_text(
651 cid2, encoding="utf-8"
652 )
653 result = _invoke(repo, "HEAD~1")
654 assert result.exit_code == 0
655
656 def test_nonexistent_commit_exits_not_found(
657 self, tmp_path: pathlib.Path
658 ) -> None:
659 repo = _make_repo(tmp_path)
660 result = _invoke(repo, NULL_COMMIT_ID)
661 assert result.exit_code == ExitCode.NOT_FOUND.value
662
663
664 # ---------------------------------------------------------------------------
665 # Security
666 # ---------------------------------------------------------------------------
667
668
669 class TestAnnotateSecurity:
670 _ANSI = "\x1b[31mmalicious\x1b[0m"
671
672 def test_ansi_in_reviewer_name_rejected(self, tmp_path: pathlib.Path) -> None:
673 repo = _make_repo(tmp_path)
674 c = _write_commit(repo)
675 result = _invoke(repo, "--reviewed-by", self._ANSI, c.commit_id)
676 assert result.exit_code == ExitCode.USER_ERROR.value
677
678 def test_ansi_not_stored_in_reviewed_by(self, tmp_path: pathlib.Path) -> None:
679 repo = _make_repo(tmp_path)
680 c = _write_commit(repo)
681 _invoke(repo, "--reviewed-by", self._ANSI, c.commit_id)
682 rec = read_commit(repo, c.commit_id)
683 assert rec is not None
684 for r in rec.reviewed_by:
685 assert "\x1b" not in r
686
687 def test_ansi_in_existing_reviewer_stripped_on_display(
688 self, tmp_path: pathlib.Path
689 ) -> None:
690 """Even if a legacy record has ANSI in reviewed_by, output is sanitized."""
691 repo = _make_repo(tmp_path)
692 c = _write_commit(repo)
693 # Force-write a record with ANSI in reviewed_by (simulating legacy data).
694 from muse.core.commits import overwrite_commit
695 c.reviewed_by = [self._ANSI]
696 overwrite_commit(repo, c)
697 result = _invoke(repo, c.commit_id)
698 assert result.exit_code == 0
699 assert "\x1b[" not in result.output
700
701 def test_null_byte_in_reviewer_rejected(self, tmp_path: pathlib.Path) -> None:
702 repo = _make_repo(tmp_path)
703 c = _write_commit(repo)
704 result = _invoke(repo, "--reviewed-by", "foo\x00bar", c.commit_id)
705 assert result.exit_code == ExitCode.USER_ERROR.value
706
707 def test_commit_not_found_error_to_stderr_not_traceback(
708 self, tmp_path: pathlib.Path
709 ) -> None:
710 repo = _make_repo(tmp_path)
711 result = _invoke(repo, "nosuchcommit")
712 assert result.exit_code != 0
713 assert "Traceback" not in result.output
714
715 def test_invalid_reviewer_control_char_rejected(
716 self, tmp_path: pathlib.Path
717 ) -> None:
718 repo = _make_repo(tmp_path)
719 c = _write_commit(repo)
720 result = _invoke(repo, "--reviewed-by", "alice\x07bell", c.commit_id)
721 assert result.exit_code == ExitCode.USER_ERROR.value
722
723 def test_overlong_reviewer_rejected(self, tmp_path: pathlib.Path) -> None:
724 from muse.cli.commands.annotate import _MAX_REVIEWER_LEN
725
726 repo = _make_repo(tmp_path)
727 c = _write_commit(repo)
728 result = _invoke(
729 repo, "--reviewed-by", "a" * (_MAX_REVIEWER_LEN + 1), c.commit_id
730 )
731 assert result.exit_code == ExitCode.USER_ERROR.value
732
733 def test_json_output_on_stdout_errors_on_stderr(
734 self, tmp_path: pathlib.Path
735 ) -> None:
736 """JSON consumers must not see errors on stdout."""
737 repo = _make_repo(tmp_path)
738 c = _write_commit(repo)
739 result = _invoke(repo, "--reviewed-by", "alice", "--json", c.commit_id)
740 assert result.exit_code == 0
741 # First non-whitespace char on stdout should be '{' (start of JSON)
742 stripped = result.output.lstrip()
743 assert stripped.startswith("{"), f"Expected JSON, got: {result.output[:80]!r}"
744
745
746 # ---------------------------------------------------------------------------
747 # E2E — full CLI invocations
748 # ---------------------------------------------------------------------------
749
750
751 class TestE2E:
752 def test_json_schema_all_fields_present(self, tmp_path: pathlib.Path) -> None:
753 repo = _make_repo(tmp_path)
754 c = _write_commit(repo)
755 result = _invoke(repo, "--json", c.commit_id)
756 assert result.exit_code == 0
757 data = _parse_json(result)
758 for field in (
759 "commit_id", "message", "branch", "author",
760 "committed_at", "reviewed_by", "test_runs", "changed", "dry_run",
761 ):
762 assert field in data, f"Missing field in JSON: {field}"
763
764 def test_combination_reviewer_and_test_run(
765 self, tmp_path: pathlib.Path
766 ) -> None:
767 repo = _make_repo(tmp_path)
768 c = _write_commit(repo)
769 result = _invoke(
770 repo,
771 "--reviewed-by", "alice",
772 "--test-run",
773 c.commit_id,
774 )
775 assert result.exit_code == 0
776 rec = read_commit(repo, c.commit_id)
777 assert rec is not None
778 assert "alice" in rec.reviewed_by
779 assert rec.test_runs == 1
780
781 def test_head_tilde_used_for_annotation(self, tmp_path: pathlib.Path) -> None:
782 repo = _make_repo(tmp_path)
783 c1 = _write_commit(repo, message="first")
784 committed_at = datetime.datetime(2026, 3, 2, tzinfo=datetime.timezone.utc)
785 snap_id = hash_snapshot({})
786 cid2 = hash_commit(
787 parent_ids=[c1.commit_id],
788 snapshot_id=snap_id,
789 message="second",
790 committed_at_iso=committed_at.isoformat(),
791 author="test-author",
792 )
793 c2 = CommitRecord(
794 commit_id=cid2,
795 branch="main",
796 snapshot_id=snap_id,
797 message="second",
798 committed_at=committed_at,
799 author="test-author",
800 parent_commit_id=c1.commit_id,
801 )
802 write_commit(repo, c2)
803 (heads_dir(repo) / "main").write_text(
804 cid2, encoding="utf-8"
805 )
806 # Annotate the first (parent) commit via HEAD~1
807 result = _invoke(repo, "HEAD~1", "--reviewed-by", "alice")
808 assert result.exit_code == 0
809 rec = read_commit(repo, c1.commit_id)
810 assert rec is not None
811 assert "alice" in rec.reviewed_by
812
813 def test_short_prefix_annotates_correct_commit(
814 self, tmp_path: pathlib.Path
815 ) -> None:
816 repo = _make_repo(tmp_path)
817 c = _write_commit(repo)
818 prefix = c.commit_id[:10]
819 result = _invoke(repo, prefix, "--reviewed-by", "bot")
820 assert result.exit_code == 0
821 rec = read_commit(repo, c.commit_id)
822 assert rec is not None
823 assert "bot" in rec.reviewed_by
824
825 def test_help_text_shows_new_flags(self, tmp_path: pathlib.Path) -> None:
826 result = runner.invoke(cli, ["annotate", "--help"])
827 assert result.exit_code == 0
828 assert "--remove-reviewer" in result.output
829 assert "--dry-run" in result.output
830 assert "--json" in result.output
831
832
833 # ---------------------------------------------------------------------------
834 # Stress
835 # ---------------------------------------------------------------------------
836
837
838 class TestStress:
839 def test_100_reviewers_stored_correctly(
840 self, tmp_path: pathlib.Path
841 ) -> None:
842 repo = _make_repo(tmp_path)
843 c = _write_commit(repo)
844 # Add 100 reviewers in a single comma-separated call
845 names = [f"reviewer-{i:03d}" for i in range(100)]
846 csv = ",".join(names)
847 result = _invoke(repo, "--reviewed-by", csv, c.commit_id)
848 assert result.exit_code == 0
849 rec = read_commit(repo, c.commit_id)
850 assert rec is not None
851 assert len(rec.reviewed_by) == 100
852 for name in names:
853 assert name in rec.reviewed_by
854
855 def test_50_test_run_increments(self, tmp_path: pathlib.Path) -> None:
856 repo = _make_repo(tmp_path)
857 c = _write_commit(repo)
858 for _ in range(50):
859 r = _invoke(repo, "--test-run", c.commit_id)
860 assert r.exit_code == 0
861 rec = read_commit(repo, c.commit_id)
862 assert rec is not None
863 assert rec.test_runs == 50
864
865 def test_concurrent_overwrite_isolated_repos(
866 self, tmp_path: pathlib.Path
867 ) -> None:
868 """Eight threads each call overwrite_commit on isolated repos.
869
870 Tests the core mutation layer for thread-safety without going through
871 the CliRunner (whose env patching is not thread-safe).
872 """
873 from muse.core.commits import overwrite_commit
874
875 errors: list[str] = []
876
877 def worker(idx: int) -> None:
878 try:
879 repo = _make_repo(tmp_path / f"repo{idx}")
880 c = _write_commit(repo)
881 c.reviewed_by = [f"agent-{idx}"]
882 overwrite_commit(repo, c)
883 # Read back and verify
884 rec = read_commit(repo, c.commit_id)
885 if rec is None:
886 errors.append(f"Thread {idx}: read_commit returned None")
887 return
888 if f"agent-{idx}" not in rec.reviewed_by:
889 errors.append(f"Thread {idx}: reviewer not found in {rec.reviewed_by!r}")
890 except Exception as exc:
891 errors.append(f"Thread {idx}: {exc}")
892
893 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
894 for t in threads:
895 t.start()
896 for t in threads:
897 t.join()
898
899 assert errors == [], f"Concurrent annotation failures: {errors}"
900
901 def test_concurrent_validate_reviewer(self) -> None:
902 """Eight threads validating names concurrently — no shared mutable state."""
903 from muse.cli.commands.annotate import _validate_reviewer
904
905 errors: list[str] = []
906
907 def worker(idx: int) -> None:
908 try:
909 name = f"reviewer-{idx:03d}"
910 result = _validate_reviewer(name)
911 if result != name:
912 errors.append(f"Thread {idx}: got {result!r}")
913 except Exception as exc:
914 errors.append(f"Thread {idx}: {exc}")
915
916 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
917 for t in threads:
918 t.start()
919 for t in threads:
920 t.join()
921
922 assert errors == [], f"Concurrent validation failures: {errors}"
923
924
925 class TestRegisterFlags:
926 """Argparse registration tests for ``muse annotate``."""
927
928 def _parse(self, *args: str) -> argparse.Namespace:
929 from muse.cli.commands.annotate import register
930 p = argparse.ArgumentParser()
931 sub = p.add_subparsers()
932 register(sub)
933 return p.parse_args(["annotate", *args])
934
935 def test_default_json_out_is_false(self) -> None:
936 ns = self._parse()
937 assert ns.json_out is False
938
939 def test_json_flag_sets_json_out(self) -> None:
940 ns = self._parse("--json")
941 assert ns.json_out is True
942
943 def test_j_shorthand_sets_json_out(self) -> None:
944 ns = self._parse("-j")
945 assert ns.json_out is True
946
947 def test_dry_run_default(self) -> None:
948 ns = self._parse()
949 assert ns.dry_run is False
950
951 def test_dry_run_flag(self) -> None:
952 ns = self._parse("--dry-run")
953 assert ns.dry_run is True
954
955 def test_dry_run_n_shorthand(self) -> None:
956 ns = self._parse("-n")
957 assert ns.dry_run is True
958
959 def test_commit_arg_default(self) -> None:
960 ns = self._parse()
961 assert ns.commit_arg is None
962
963 def test_reviewed_by_default(self) -> None:
964 ns = self._parse()
965 assert ns.reviewed_by is None
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 28 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 29 days ago