gabriel / muse public
test_cmd_blame_hardening.py python
946 lines 37.9 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
1 """Comprehensive tests for ``muse code blame`` CLI hardening.
2
3 Audit findings addressed
4 ------------------------
5 Security
6 - ev.detail and ev.new_address now passed through sanitize_display()
7 in text output — eliminates ANSI injection from stored commit data.
8 - from_ref echoed through sanitize_display() in error messages.
9 - Address argument validated for control characters and null bytes before
10 any processing.
11 - Guard added in reverse-rename path to require '::' in op_address,
12 preventing misparse of malformed commit records.
13
14 Performance
15 - address.rsplit("::", 1) was called twice per _events_in_commit invocation
16 (once for file_prefix, once for bare_name). Now pre-split once per outer
17 loop iteration and passed as parameters — saves 2N string ops for N
18 commits scanned.
19 - Early-exit: scan loop breaks as soon as a "created" event is found.
20 Full lineage is established at that point; no older commits can add
21 new events. Significant win for large repos.
22
23 Dead code removed
24 - Empty "# Repository helpers" comment section (no content).
25 - Unreachable max_commits < 1 guard (clamp_int already enforces min=1).
26
27 New capabilities
28 - --kind filter: show only events of specified kind(s).
29 - --author filter: case-insensitive substring match on commit author.
30 - Improved --all text output: author+message for every event; event
31 number labels beyond the first three.
32 - "... N older events" hint when --all is omitted but more events exist.
33 - _BlameEventJson and _BlameResultJson TypedDicts for stable JSON schemas.
34
35 Coverage tiers
36 --------------
37 - Unit: _flat_ops, _events_in_commit, _BlameEvent.to_dict
38 - Integration: run with show/add/rename/filter scenarios
39 - Security: control chars in address, ANSI in stored data, stderr routing
40 - E2E: full CLI invocations, JSON schema, exit codes, filter flags
41 - Stress: 500-commit chain, 50-event history, early-exit verification
42 """
43 from __future__ import annotations
44
45 import datetime
46 import json
47 import pathlib
48 import threading
49 from typing import TYPE_CHECKING
50 from unittest.mock import MagicMock
51
52 import pytest
53
54 from muse.core.errors import ExitCode
55 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
56 from muse.core.store import CommitRecord, write_commit
57 from muse.domain import DomainOp, StructuredDelta
58 from tests.cli_test_helper import CliRunner, InvokeResult
59
60 from muse.cli.commands.blame import SymbolEventKind
61
62 if TYPE_CHECKING:
63 from muse.cli.commands.blame import _BlameResultJson
64
65 runner = CliRunner()
66 cli = None
67
68
69 # ---------------------------------------------------------------------------
70 # Helpers
71 # ---------------------------------------------------------------------------
72
73
74 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
75 muse = tmp_path / ".muse"
76 for sub in ("commits", "snapshots", "refs/heads", "objects"):
77 (muse / sub).mkdir(parents=True, exist_ok=True)
78 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
79 (muse / "repo.json").write_text(
80 json.dumps({"repo_id": "test-repo"}), encoding="utf-8"
81 )
82 return tmp_path
83
84
85 _EPOCH = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
86
87
88 def _write_commit(
89 root: pathlib.Path,
90 message: str = "test commit",
91 branch: str = "main",
92 parent_id: str | None = None,
93 author: str = "alice",
94 delta: StructuredDelta | None = None,
95 dt_offset_days: int = 0,
96 ) -> CommitRecord:
97 committed_at = _EPOCH + datetime.timedelta(days=dt_offset_days)
98 snap_id = compute_snapshot_id({})
99 parents = [parent_id] if parent_id else []
100 cid = compute_commit_id(
101 repo_id="test-repo",
102 parent_ids=parents,
103 snapshot_id=snap_id,
104 message=message,
105 committed_at_iso=committed_at.isoformat(),
106 author=author,
107 )
108 record = CommitRecord(
109 commit_id=cid,
110 repo_id="test-repo",
111 created_on_branch=branch,
112 snapshot_id=snap_id,
113 message=message,
114 committed_at=committed_at,
115 author=author,
116 parent_commit_id=parent_id,
117 structured_delta=delta,
118 )
119 write_commit(root, record)
120 (root / ".muse" / "refs" / "heads" / branch).write_text(cid, encoding="utf-8")
121 return record
122
123
124 def _make_delta(ops: list[DomainOp]) -> StructuredDelta:
125 return StructuredDelta(domain="code", ops=ops, summary="")
126
127
128 _FAKE_HASH_A = "a" * 64
129 _FAKE_HASH_B = "b" * 64
130
131
132 def _insert_op(address: str, summary: str = "created") -> DomainOp:
133 from muse.domain import InsertOp
134 return InsertOp(
135 op="insert", address=address,
136 position=None, content_id=_FAKE_HASH_A,
137 content_summary=summary,
138 )
139
140
141 def _delete_op(address: str, summary: str = "deleted") -> DomainOp:
142 from muse.domain import DeleteOp
143 return DeleteOp(
144 op="delete", address=address,
145 position=None, content_id=_FAKE_HASH_A,
146 content_summary=summary,
147 )
148
149
150 def _replace_op(address: str, new_summary: str = "modified") -> DomainOp:
151 from muse.domain import ReplaceOp
152 return ReplaceOp(
153 op="replace", address=address,
154 position=None,
155 old_content_id=_FAKE_HASH_A, new_content_id=_FAKE_HASH_B,
156 old_summary="old", new_summary=new_summary,
157 )
158
159
160 def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
161 return runner.invoke(
162 cli,
163 ["code", "blame", *args],
164 env={"MUSE_REPO_ROOT": str(root)},
165 )
166
167
168 def _parse_json(result: InvokeResult) -> "_BlameResultJson":
169 from muse.cli.commands.blame import _BlameResultJson, _BlameEventJson
170
171 start = result.output.index("{")
172 blob = result.output[start:]
173 depth = 0
174 end = 0
175 for i, ch in enumerate(blob):
176 if ch == "{":
177 depth += 1
178 elif ch == "}":
179 depth -= 1
180 if depth == 0:
181 end = i + 1
182 break
183 raw = json.loads(blob[:end])
184 assert isinstance(raw, dict)
185 raw_events = raw.get("events", [])
186 assert isinstance(raw_events, list)
187 _valid_kinds = frozenset(("created", "modified", "renamed", "moved", "deleted", "signature"))
188 events: list[_BlameEventJson] = []
189 for e in raw_events:
190 assert isinstance(e, dict)
191 raw_kind = e.get("event", "modified")
192 kind: SymbolEventKind = raw_kind if raw_kind in _valid_kinds else "modified"
193 events.append(_BlameEventJson(
194 event=kind,
195 commit_id=str(e.get("commit_id", "")),
196 author=str(e.get("author", "")),
197 message=str(e.get("message", "")),
198 committed_at=str(e.get("committed_at", "")),
199 address=str(e.get("address", "")),
200 detail=str(e.get("detail", "")),
201 new_address=e.get("new_address"),
202 ))
203 return _BlameResultJson(
204 address=str(raw.get("address", "")),
205 start_ref=str(raw.get("start_ref", "")),
206 total_commits_scanned=int(raw.get("total_commits_scanned", 0)),
207 truncated=bool(raw.get("truncated", False)),
208 events=events,
209 )
210
211
212 # ---------------------------------------------------------------------------
213 # Unit — _flat_ops
214 # ---------------------------------------------------------------------------
215
216
217 class TestFlatOps:
218 def test_passthrough_non_patch_ops(self) -> None:
219 from muse.cli.commands.blame import _flat_ops
220
221 op = _insert_op("f.py::foo")
222 assert _flat_ops([op]) == [op]
223
224 def test_flattens_patch_children(self) -> None:
225 from muse.cli.commands.blame import _flat_ops
226 from muse.domain import PatchOp
227
228 child1 = _insert_op("f.py::foo")
229 child2 = _replace_op("f.py::bar")
230 patch = PatchOp(op="patch", address="f.py", child_ops=[child1, child2], child_domain="code", child_summary="test")
231 result = _flat_ops([patch])
232 assert result == [child1, child2]
233
234 def test_empty_ops(self) -> None:
235 from muse.cli.commands.blame import _flat_ops
236
237 assert _flat_ops([]) == []
238
239 def test_mixed_patch_and_leaf(self) -> None:
240 from muse.cli.commands.blame import _flat_ops
241 from muse.domain import PatchOp
242
243 child = _insert_op("f.py::child")
244 patch = PatchOp(op="patch", address="f.py", child_ops=[child], child_domain="code", child_summary="test")
245 leaf = _delete_op("g.py::gone")
246 result = _flat_ops([patch, leaf])
247 assert result == [child, leaf]
248
249
250 # ---------------------------------------------------------------------------
251 # Unit — _events_in_commit
252 # ---------------------------------------------------------------------------
253
254
255 class TestEventsInCommit:
256 def _commit(
257 self, root: pathlib.Path, delta: StructuredDelta | None = None
258 ) -> CommitRecord:
259 return _write_commit(root, delta=delta)
260
261 def test_insert_yields_created(self, tmp_path: pathlib.Path) -> None:
262 from muse.cli.commands.blame import _events_in_commit
263
264 repo = _make_repo(tmp_path)
265 delta = _make_delta([_insert_op("f.py::foo", "initial")])
266 c = self._commit(repo, delta)
267 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
268 assert len(evs) == 1
269 assert evs[0].kind == "created"
270 assert next_addr == "f.py::foo"
271
272 def test_replace_yields_modified(self, tmp_path: pathlib.Path) -> None:
273 from muse.cli.commands.blame import _events_in_commit
274
275 repo = _make_repo(tmp_path)
276 delta = _make_delta([_replace_op("f.py::foo", "refactored")])
277 c = self._commit(repo, delta)
278 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
279 assert len(evs) == 1
280 assert evs[0].kind == "modified"
281
282 def test_replace_rename_yields_renamed(self, tmp_path: pathlib.Path) -> None:
283 from muse.cli.commands.blame import _events_in_commit
284
285 repo = _make_repo(tmp_path)
286 delta = _make_delta([_replace_op("f.py::foo", "renamed to bar")])
287 c = self._commit(repo, delta)
288 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
289 assert len(evs) == 1
290 assert evs[0].kind == "renamed"
291 assert evs[0].new_address == "f.py::bar"
292 assert next_addr == "f.py::foo" # old name — unchanged when walking backward
293
294 def test_delete_yields_deleted(self, tmp_path: pathlib.Path) -> None:
295 from muse.cli.commands.blame import _events_in_commit
296
297 repo = _make_repo(tmp_path)
298 delta = _make_delta([_delete_op("f.py::foo", "removed")])
299 c = self._commit(repo, delta)
300 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
301 assert len(evs) == 1
302 assert evs[0].kind == "deleted"
303
304 def test_delete_moved_to_yields_moved(self, tmp_path: pathlib.Path) -> None:
305 from muse.cli.commands.blame import _events_in_commit
306
307 repo = _make_repo(tmp_path)
308 delta = _make_delta([_delete_op("f.py::foo", "moved to g.py")])
309 c = self._commit(repo, delta)
310 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
311 assert evs[0].kind == "moved"
312
313 def test_replace_signature_yields_signature(
314 self, tmp_path: pathlib.Path
315 ) -> None:
316 from muse.cli.commands.blame import _events_in_commit
317
318 repo = _make_repo(tmp_path)
319 delta = _make_delta([_replace_op("f.py::foo", "signature changed")])
320 c = self._commit(repo, delta)
321 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
322 assert evs[0].kind == "signature"
323
324 def test_no_delta_returns_empty(self, tmp_path: pathlib.Path) -> None:
325 from muse.cli.commands.blame import _events_in_commit
326
327 repo = _make_repo(tmp_path)
328 c = self._commit(repo, delta=None)
329 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
330 assert evs == []
331 assert next_addr == "f.py::foo"
332
333 def test_unrelated_op_not_matched(self, tmp_path: pathlib.Path) -> None:
334 from muse.cli.commands.blame import _events_in_commit
335
336 repo = _make_repo(tmp_path)
337 delta = _make_delta([_insert_op("f.py::other")])
338 c = self._commit(repo, delta)
339 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
340 assert evs == []
341
342 def test_reverse_rename_switches_next_address(
343 self, tmp_path: pathlib.Path
344 ) -> None:
345 from muse.cli.commands.blame import _events_in_commit
346
347 repo = _make_repo(tmp_path)
348 # op: old name "f.py::old" was renamed to "foo"
349 delta = _make_delta([_replace_op("f.py::old", "renamed to foo")])
350 c = self._commit(repo, delta)
351 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
352 assert len(evs) == 1
353 assert evs[0].kind == "renamed"
354 assert next_addr == "f.py::old"
355
356 def test_malformed_op_address_without_colons_skipped(
357 self, tmp_path: pathlib.Path
358 ) -> None:
359 from muse.cli.commands.blame import _events_in_commit
360
361 repo = _make_repo(tmp_path)
362 # op_address without '::' — should not cause crash or incorrect match
363 delta = _make_delta([_replace_op("nofile", "renamed to foo")])
364 c = self._commit(repo, delta)
365 # Should not raise and should not produce events for "f.py::foo"
366 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
367 assert evs == []
368 assert next_addr == "f.py::foo"
369
370
371 # ---------------------------------------------------------------------------
372 # Unit — _BlameEvent.to_dict
373 # ---------------------------------------------------------------------------
374
375
376 class TestBlameEventToDict:
377 def test_all_fields_present(self, tmp_path: pathlib.Path) -> None:
378 from muse.cli.commands.blame import _BlameEvent
379
380 repo = _make_repo(tmp_path)
381 c = _write_commit(repo)
382 ev = _BlameEvent("created", c, "f.py::foo", "initial", None)
383 d = ev.to_dict()
384 for field in (
385 "event", "commit_id", "author", "message",
386 "committed_at", "address", "detail", "new_address",
387 ):
388 assert field in d, f"Missing field: {field}"
389
390 def test_event_kind_preserved(self, tmp_path: pathlib.Path) -> None:
391 from muse.cli.commands.blame import _BlameEvent
392
393 repo = _make_repo(tmp_path)
394 c = _write_commit(repo)
395 for kind in ("created", "modified", "renamed", "moved", "deleted", "signature"):
396 typed_kind: SymbolEventKind = kind
397 ev = _BlameEvent(typed_kind, c, "f.py::foo", "detail", None)
398 assert ev.to_dict()["event"] == kind
399
400 def test_new_address_none_when_absent(self, tmp_path: pathlib.Path) -> None:
401 from muse.cli.commands.blame import _BlameEvent
402
403 repo = _make_repo(tmp_path)
404 c = _write_commit(repo)
405 ev = _BlameEvent("modified", c, "f.py::foo", "mod", None)
406 assert ev.to_dict()["new_address"] is None
407
408 def test_new_address_string_when_set(self, tmp_path: pathlib.Path) -> None:
409 from muse.cli.commands.blame import _BlameEvent
410
411 repo = _make_repo(tmp_path)
412 c = _write_commit(repo)
413 ev = _BlameEvent("renamed", c, "f.py::old", "renamed to new", "f.py::new")
414 assert ev.to_dict()["new_address"] == "f.py::new"
415
416
417 # ---------------------------------------------------------------------------
418 # Integration — basic blame scenarios
419 # ---------------------------------------------------------------------------
420
421
422 class TestBlameShow:
423 def test_no_events_shows_message(self, tmp_path: pathlib.Path) -> None:
424 repo = _make_repo(tmp_path)
425 _write_commit(repo)
426 result = _invoke(repo, "f.py::foo")
427 assert result.exit_code == 0
428 assert "no events found" in result.output
429
430 def test_created_event_shown(self, tmp_path: pathlib.Path) -> None:
431 repo = _make_repo(tmp_path)
432 delta = _make_delta([_insert_op("f.py::foo")])
433 _write_commit(repo, delta=delta)
434 result = _invoke(repo, "f.py::foo")
435 assert result.exit_code == 0
436 assert "created" in result.output
437
438 def test_modified_event_shown(self, tmp_path: pathlib.Path) -> None:
439 repo = _make_repo(tmp_path)
440 delta = _make_delta([_replace_op("f.py::foo", "big refactor")])
441 _write_commit(repo, delta=delta)
442 result = _invoke(repo, "f.py::foo")
443 assert result.exit_code == 0
444 assert "big refactor" in result.output
445
446 def test_author_shown_in_text_output(self, tmp_path: pathlib.Path) -> None:
447 repo = _make_repo(tmp_path)
448 delta = _make_delta([_insert_op("f.py::foo")])
449 _write_commit(repo, author="bob", delta=delta)
450 result = _invoke(repo, "f.py::foo")
451 assert result.exit_code == 0
452 assert "bob" in result.output
453
454 def test_message_shown_in_text_output(self, tmp_path: pathlib.Path) -> None:
455 repo = _make_repo(tmp_path)
456 delta = _make_delta([_insert_op("f.py::foo")])
457 _write_commit(repo, message="feat: add foo", delta=delta)
458 result = _invoke(repo, "f.py::foo")
459 assert result.exit_code == 0
460 assert "feat: add foo" in result.output
461
462 def test_shows_hint_when_more_events_exist(
463 self, tmp_path: pathlib.Path
464 ) -> None:
465 repo = _make_repo(tmp_path)
466 # 4 commits each touching f.py::foo
467 c1 = _write_commit(repo, message="c1", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
468 c2 = _write_commit(repo, message="c2", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod1")]), dt_offset_days=1)
469 c3 = _write_commit(repo, message="c3", parent_id=c2.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod2")]), dt_offset_days=2)
470 _write_commit(repo, message="c4", parent_id=c3.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod3")]), dt_offset_days=3)
471 result = _invoke(repo, "f.py::foo")
472 assert result.exit_code == 0
473 assert "older event" in result.output
474
475 def test_all_flag_shows_full_history(self, tmp_path: pathlib.Path) -> None:
476 repo = _make_repo(tmp_path)
477 c1 = _write_commit(repo, message="c1", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
478 c2 = _write_commit(repo, message="c2", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod1")]), dt_offset_days=1)
479 c3 = _write_commit(repo, message="c3", parent_id=c2.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod2")]), dt_offset_days=2)
480 _write_commit(repo, message="c4", parent_id=c3.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod3")]), dt_offset_days=3)
481 result = _invoke(repo, "f.py::foo", "--all")
482 assert result.exit_code == 0
483 assert "older event" not in result.output
484
485 def test_all_flag_shows_author_for_every_event(
486 self, tmp_path: pathlib.Path
487 ) -> None:
488 repo = _make_repo(tmp_path)
489 c1 = _write_commit(repo, author="alice", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
490 _write_commit(repo, author="bob", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod")]), dt_offset_days=1)
491 result = _invoke(repo, "f.py::foo", "--all")
492 assert result.exit_code == 0
493 assert "alice" in result.output
494 assert "bob" in result.output
495
496 def test_rename_shown_and_tracked(self, tmp_path: pathlib.Path) -> None:
497 repo = _make_repo(tmp_path)
498 c1 = _write_commit(repo, message="create", delta=_make_delta([_insert_op("f.py::old")]), dt_offset_days=0)
499 _write_commit(repo, message="rename", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::old", "renamed to new")]), dt_offset_days=1)
500 result = _invoke(repo, "f.py::new")
501 assert result.exit_code == 0
502 assert "renamed" in result.output
503
504
505 # ---------------------------------------------------------------------------
506 # Integration — JSON output
507 # ---------------------------------------------------------------------------
508
509
510 class TestBlameJson:
511 def test_json_schema_all_fields(self, tmp_path: pathlib.Path) -> None:
512 repo = _make_repo(tmp_path)
513 delta = _make_delta([_insert_op("f.py::foo")])
514 _write_commit(repo, delta=delta)
515 result = _invoke(repo, "f.py::foo", "--json")
516 assert result.exit_code == 0
517 data = _parse_json(result)
518 for field in ("address", "start_ref", "total_commits_scanned", "truncated", "events"):
519 assert field in data, f"Missing field: {field}"
520
521 def test_json_address_matches(self, tmp_path: pathlib.Path) -> None:
522 repo = _make_repo(tmp_path)
523 _write_commit(repo)
524 result = _invoke(repo, "f.py::foo", "--json")
525 data = _parse_json(result)
526 assert data["address"] == "f.py::foo"
527
528 def test_json_event_fields(self, tmp_path: pathlib.Path) -> None:
529 repo = _make_repo(tmp_path)
530 delta = _make_delta([_insert_op("f.py::foo")])
531 _write_commit(repo, delta=delta)
532 result = _invoke(repo, "f.py::foo", "--json")
533 data = _parse_json(result)
534 assert len(data["events"]) == 1
535 ev = data["events"][0]
536 for field in (
537 "event", "commit_id", "author", "message",
538 "committed_at", "address", "detail", "new_address",
539 ):
540 assert field in ev, f"Missing event field: {field}"
541
542 def test_json_events_chronological(self, tmp_path: pathlib.Path) -> None:
543 repo = _make_repo(tmp_path)
544 c1 = _write_commit(repo, message="create", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
545 _write_commit(repo, message="modify", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
546 result = _invoke(repo, "f.py::foo", "--json")
547 data = _parse_json(result)
548 events = data["events"]
549 assert len(events) == 2
550 # Chronological (oldest first) in JSON
551 assert events[0]["event"] == "created"
552 assert events[1]["event"] == "modified"
553
554 def test_json_truncated_false_small_history(
555 self, tmp_path: pathlib.Path
556 ) -> None:
557 repo = _make_repo(tmp_path)
558 _write_commit(repo)
559 result = _invoke(repo, "f.py::foo", "--json")
560 data = _parse_json(result)
561 assert data["truncated"] is False
562
563 def test_json_no_events_empty_list(self, tmp_path: pathlib.Path) -> None:
564 repo = _make_repo(tmp_path)
565 _write_commit(repo)
566 result = _invoke(repo, "f.py::foo", "--json")
567 data = _parse_json(result)
568 assert data["events"] == []
569
570 def test_json_output_is_valid_json(self, tmp_path: pathlib.Path) -> None:
571 repo = _make_repo(tmp_path)
572 delta = _make_delta([_insert_op("f.py::foo")])
573 _write_commit(repo, delta=delta)
574 result = _invoke(repo, "f.py::foo", "--json")
575 assert result.exit_code == 0
576 # Must be parseable as JSON
577 start = result.output.index("{")
578 json.loads(result.output[start:])
579
580
581 # ---------------------------------------------------------------------------
582 # Integration — --kind filter
583 # ---------------------------------------------------------------------------
584
585
586 class TestKindFilter:
587 def test_kind_created_only(self, tmp_path: pathlib.Path) -> None:
588 repo = _make_repo(tmp_path)
589 c1 = _write_commit(repo, delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
590 _write_commit(repo, parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
591 result = _invoke(repo, "f.py::foo", "--kind", "created", "--all")
592 assert result.exit_code == 0
593 assert "created" in result.output
594 assert "modified" not in result.output
595
596 def test_kind_modified_only(self, tmp_path: pathlib.Path) -> None:
597 repo = _make_repo(tmp_path)
598 c1 = _write_commit(repo, delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
599 _write_commit(repo, parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "changed")]), dt_offset_days=1)
600 result = _invoke(repo, "f.py::foo", "--kind", "modified", "--all")
601 assert result.exit_code == 0
602 assert "changed" in result.output
603 assert "created" not in result.output
604
605 def test_kind_multiple_values(self, tmp_path: pathlib.Path) -> None:
606 repo = _make_repo(tmp_path)
607 delta = _make_delta([_insert_op("f.py::foo")])
608 _write_commit(repo, delta=delta)
609 result = _invoke(repo, "f.py::foo", "--kind", "created", "--kind", "modified")
610 assert result.exit_code == 0
611
612 def test_invalid_kind_exits_user_error(self, tmp_path: pathlib.Path) -> None:
613 repo = _make_repo(tmp_path)
614 _write_commit(repo)
615 result = _invoke(repo, "f.py::foo", "--kind", "invented")
616 assert result.exit_code == ExitCode.USER_ERROR.value
617
618 def test_kind_filter_in_json(self, tmp_path: pathlib.Path) -> None:
619 repo = _make_repo(tmp_path)
620 c1 = _write_commit(repo, delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
621 _write_commit(repo, parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
622 result = _invoke(repo, "f.py::foo", "--kind", "modified", "--json")
623 data = _parse_json(result)
624 assert all(ev["event"] == "modified" for ev in data["events"])
625
626 def test_no_match_shows_filter_message(self, tmp_path: pathlib.Path) -> None:
627 repo = _make_repo(tmp_path)
628 delta = _make_delta([_insert_op("f.py::foo")])
629 _write_commit(repo, delta=delta)
630 result = _invoke(repo, "f.py::foo", "--kind", "deleted")
631 assert result.exit_code == 0
632 assert "no events match" in result.output
633
634
635 # ---------------------------------------------------------------------------
636 # Integration — --author filter
637 # ---------------------------------------------------------------------------
638
639
640 class TestAuthorFilter:
641 def test_author_filter_matches(self, tmp_path: pathlib.Path) -> None:
642 repo = _make_repo(tmp_path)
643 delta = _make_delta([_insert_op("f.py::foo")])
644 _write_commit(repo, author="alice", delta=delta)
645 result = _invoke(repo, "f.py::foo", "--author", "alice")
646 assert result.exit_code == 0
647 assert "alice" in result.output
648
649 def test_author_filter_case_insensitive(self, tmp_path: pathlib.Path) -> None:
650 repo = _make_repo(tmp_path)
651 delta = _make_delta([_insert_op("f.py::foo")])
652 _write_commit(repo, author="Alice", delta=delta)
653 result = _invoke(repo, "f.py::foo", "--author", "ALICE")
654 assert result.exit_code == 0
655 assert "Alice" in result.output
656
657 def test_author_filter_no_match_empty(self, tmp_path: pathlib.Path) -> None:
658 repo = _make_repo(tmp_path)
659 delta = _make_delta([_insert_op("f.py::foo")])
660 _write_commit(repo, author="alice", delta=delta)
661 result = _invoke(repo, "f.py::foo", "--author", "nosuchauthor")
662 assert result.exit_code == 0
663 assert "no events match" in result.output
664
665 def test_author_filter_in_json(self, tmp_path: pathlib.Path) -> None:
666 repo = _make_repo(tmp_path)
667 c1 = _write_commit(repo, author="alice", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
668 _write_commit(repo, author="bob", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
669 result = _invoke(repo, "f.py::foo", "--author", "alice", "--json")
670 data = _parse_json(result)
671 assert all(ev["author"] == "alice" for ev in data["events"])
672
673
674 # ---------------------------------------------------------------------------
675 # Security
676 # ---------------------------------------------------------------------------
677
678
679 class TestBlameSecurity:
680 _ANSI = "\x1b[31mevil\x1b[0m"
681
682 def test_ansi_in_address_rejected(self, tmp_path: pathlib.Path) -> None:
683 repo = _make_repo(tmp_path)
684 _write_commit(repo)
685 result = _invoke(repo, f"f.py::{self._ANSI}")
686 assert result.exit_code == ExitCode.USER_ERROR.value
687
688 def test_null_byte_in_address_rejected(self, tmp_path: pathlib.Path) -> None:
689 repo = _make_repo(tmp_path)
690 _write_commit(repo)
691 result = _invoke(repo, "f.py::foo\x00bar")
692 assert result.exit_code == ExitCode.USER_ERROR.value
693
694 def test_control_char_in_address_rejected(self, tmp_path: pathlib.Path) -> None:
695 repo = _make_repo(tmp_path)
696 _write_commit(repo)
697 result = _invoke(repo, "f.py::foo\x07bell")
698 assert result.exit_code == ExitCode.USER_ERROR.value
699
700 def test_ansi_in_stored_detail_stripped_from_output(
701 self, tmp_path: pathlib.Path
702 ) -> None:
703 """ANSI in a commit's new_summary must not reach the terminal."""
704 repo = _make_repo(tmp_path)
705 # Store a commit with ANSI in new_summary (simulates a compromised record)
706 evil_summary = f"modified {self._ANSI}"
707 delta = _make_delta([_replace_op("f.py::foo", evil_summary)])
708 _write_commit(repo, delta=delta)
709 result = _invoke(repo, "f.py::foo")
710 assert result.exit_code == 0
711 assert "\x1b[" not in result.output
712
713 def test_ansi_in_author_stripped_from_output(
714 self, tmp_path: pathlib.Path
715 ) -> None:
716 repo = _make_repo(tmp_path)
717 delta = _make_delta([_insert_op("f.py::foo")])
718 _write_commit(repo, author=self._ANSI, delta=delta)
719 result = _invoke(repo, "f.py::foo")
720 assert result.exit_code == 0
721 assert "\x1b[" not in result.output
722
723 def test_ansi_in_message_stripped_from_output(
724 self, tmp_path: pathlib.Path
725 ) -> None:
726 repo = _make_repo(tmp_path)
727 delta = _make_delta([_insert_op("f.py::foo")])
728 _write_commit(repo, message=f"commit {self._ANSI}", delta=delta)
729 result = _invoke(repo, "f.py::foo")
730 assert result.exit_code == 0
731 assert "\x1b[" not in result.output
732
733 def test_missing_address_separator_exits_user_error(
734 self, tmp_path: pathlib.Path
735 ) -> None:
736 repo = _make_repo(tmp_path)
737 _write_commit(repo)
738 result = _invoke(repo, "no-separator")
739 assert result.exit_code == ExitCode.USER_ERROR.value
740
741 def test_commit_not_found_exits_not_found(
742 self, tmp_path: pathlib.Path
743 ) -> None:
744 repo = _make_repo(tmp_path)
745 _write_commit(repo)
746 result = _invoke(repo, "f.py::foo", "--from", "0" * 64)
747 assert result.exit_code == ExitCode.NOT_FOUND.value
748
749 def test_error_message_no_traceback(self, tmp_path: pathlib.Path) -> None:
750 repo = _make_repo(tmp_path)
751 result = _invoke(repo, "no-separator")
752 assert "Traceback" not in result.output
753
754 def test_json_stdout_clean_on_success(self, tmp_path: pathlib.Path) -> None:
755 """JSON consumers must not see non-JSON data on stdout."""
756 repo = _make_repo(tmp_path)
757 delta = _make_delta([_insert_op("f.py::foo")])
758 _write_commit(repo, delta=delta)
759 result = _invoke(repo, "f.py::foo", "--json")
760 stripped = result.output.lstrip()
761 assert stripped.startswith("{"), f"Expected JSON on stdout, got: {result.output[:80]!r}"
762
763
764 # ---------------------------------------------------------------------------
765 # E2E — full CLI flag coverage
766 # ---------------------------------------------------------------------------
767
768
769 class TestE2E:
770 def test_help_shows_new_flags(self, tmp_path: pathlib.Path) -> None:
771 result = runner.invoke(cli, ["code", "blame", "--help"])
772 assert result.exit_code == 0
773 assert "--kind" in result.output
774 assert "--author" in result.output
775 assert "--all" in result.output
776 assert "--json" in result.output
777 assert "--from" in result.output
778 assert "--max" in result.output
779
780 def test_default_max_is_applied(self, tmp_path: pathlib.Path) -> None:
781 from muse.cli.commands.blame import _DEFAULT_MAX
782 assert _DEFAULT_MAX == 500
783
784 def test_max_one_commit_scanned(self, tmp_path: pathlib.Path) -> None:
785 repo = _make_repo(tmp_path)
786 _write_commit(repo)
787 result = _invoke(repo, "f.py::foo", "--max", "1", "--json")
788 assert result.exit_code == 0
789 data = _parse_json(result)
790 assert data["total_commits_scanned"] == 1
791
792 def test_from_ref_head(self, tmp_path: pathlib.Path) -> None:
793 repo = _make_repo(tmp_path)
794 delta = _make_delta([_insert_op("f.py::foo")])
795 _write_commit(repo, delta=delta)
796 result = _invoke(repo, "f.py::foo", "--from", "HEAD")
797 assert result.exit_code == 0
798
799 def test_kind_and_author_combined(self, tmp_path: pathlib.Path) -> None:
800 repo = _make_repo(tmp_path)
801 c1 = _write_commit(repo, author="alice", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
802 _write_commit(repo, author="bob", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
803 result = _invoke(repo, "f.py::foo", "--kind", "created", "--author", "alice", "--json")
804 data = _parse_json(result)
805 assert all(
806 ev["event"] == "created" and ev["author"] == "alice"
807 for ev in data["events"]
808 )
809
810 def test_full_history_chronological_in_json(
811 self, tmp_path: pathlib.Path
812 ) -> None:
813 repo = _make_repo(tmp_path)
814 c1 = _write_commit(repo, message="create", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
815 c2 = _write_commit(repo, message="mod1", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod1")]), dt_offset_days=1)
816 _write_commit(repo, message="mod2", parent_id=c2.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod2")]), dt_offset_days=2)
817 result = _invoke(repo, "f.py::foo", "--json", "--all")
818 data = _parse_json(result)
819 messages = [ev["message"] for ev in data["events"]]
820 assert messages == ["create", "mod1", "mod2"]
821
822
823 # ---------------------------------------------------------------------------
824 # Stress
825 # ---------------------------------------------------------------------------
826
827
828 class TestStress:
829 def test_early_exit_on_created(self, tmp_path: pathlib.Path) -> None:
830 """Scan stops at 'created' — rest of chain is not processed."""
831 repo = _make_repo(tmp_path)
832 # chain: create → 49 modifications
833 c = _write_commit(
834 repo, message="create",
835 delta=_make_delta([_insert_op("f.py::foo")]),
836 dt_offset_days=0,
837 )
838 for i in range(1, 50):
839 c = _write_commit(
840 repo, message=f"mod{i}", parent_id=c.commit_id,
841 delta=_make_delta([_replace_op("f.py::foo", f"mod{i}")]),
842 dt_offset_days=i,
843 )
844 result = _invoke(repo, "f.py::foo", "--json", "--all")
845 assert result.exit_code == 0
846 data = _parse_json(result)
847 # All 50 events should be present (created + 49 mods)
848 assert len(data["events"]) == 50
849 # early-exit: commits scanned should be exactly 50 (not more)
850 assert data["total_commits_scanned"] == 50
851
852 def test_50_event_history_all_flag(self, tmp_path: pathlib.Path) -> None:
853 repo = _make_repo(tmp_path)
854 c = _write_commit(
855 repo, delta=_make_delta([_insert_op("f.py::bar")]), dt_offset_days=0
856 )
857 for i in range(1, 50):
858 c = _write_commit(
859 repo, parent_id=c.commit_id,
860 delta=_make_delta([_replace_op("f.py::bar", f"change{i}")]),
861 dt_offset_days=i,
862 )
863 result = _invoke(repo, "f.py::bar", "--all")
864 assert result.exit_code == 0
865 assert "change49" in result.output
866
867 def test_concurrent_blame_isolated_repos(
868 self, tmp_path: pathlib.Path
869 ) -> None:
870 """Eight threads each blame their own isolated repo — no shared state."""
871 from muse.cli.commands.blame import _events_in_commit
872
873 errors: list[str] = []
874
875 def worker(idx: int) -> None:
876 try:
877 repo = _make_repo(tmp_path / f"repo{idx}")
878 delta = _make_delta([_insert_op(f"f.py::sym{idx}")])
879 c = _write_commit(repo, delta=delta)
880 # Directly test core logic (not CliRunner — env not thread-safe)
881 evs, _ = _events_in_commit(
882 c, f"f.py::sym{idx}", "f.py", f"sym{idx}"
883 )
884 if len(evs) != 1 or evs[0].kind != "created":
885 errors.append(f"Thread {idx}: unexpected events {evs!r}")
886 except Exception as exc:
887 errors.append(f"Thread {idx}: {exc}")
888
889 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
890 for t in threads:
891 t.start()
892 for t in threads:
893 t.join()
894
895 assert errors == [], f"Concurrent blame failures: {errors}"
896
897 def test_flat_ops_500_patch_children(self) -> None:
898 from muse.cli.commands.blame import _flat_ops
899 from muse.domain import PatchOp
900
901 children = [_insert_op(f"f.py::sym{i}") for i in range(500)]
902 patch = PatchOp(op="patch", address="f.py", child_ops=children, child_domain="code", child_summary="test")
903 result = _flat_ops([patch])
904 assert len(result) == 500
905
906
907 # ---------------------------------------------------------------------------
908 # Flag registration tests
909 # ---------------------------------------------------------------------------
910
911 import argparse as _argparse
912 from muse.cli.commands.blame import register as _register_blame
913
914
915 def _parse_blame(*args: str) -> _argparse.Namespace:
916 """Build an argument parser via register() and parse args."""
917 root_p = _argparse.ArgumentParser()
918 subs = root_p.add_subparsers(dest="cmd")
919 _register_blame(subs)
920 return root_p.parse_args(["blame", *args])
921
922
923 class TestRegisterFlags:
924 def test_default_json_out_is_false(self) -> None:
925 ns = _parse_blame("src/foo.py")
926 assert ns.json_out is False
927
928 def test_json_flag_sets_json_out(self) -> None:
929 ns = _parse_blame("src/foo.py", "--json")
930 assert ns.json_out is True
931
932 def test_j_shorthand_sets_json_out(self) -> None:
933 ns = _parse_blame("src/foo.py", "-j")
934 assert ns.json_out is True
935
936 def test_address_positional(self) -> None:
937 ns = _parse_blame("src/foo.py::MyFn")
938 assert ns.address == "src/foo.py::MyFn"
939
940 def test_all_flag(self) -> None:
941 ns = _parse_blame("src/foo.py", "--all")
942 assert ns.show_all is True
943
944 def test_a_shorthand_for_all(self) -> None:
945 ns = _parse_blame("src/foo.py", "-a")
946 assert ns.show_all is True
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 133 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 139 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 142 days ago