gabriel / muse public
test_cmd_blame_hardening.py python
897 lines 36.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 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(parents, snap_id, message, committed_at.isoformat())
101 record = CommitRecord(
102 commit_id=cid,
103 repo_id="test-repo",
104 branch=branch,
105 snapshot_id=snap_id,
106 message=message,
107 committed_at=committed_at,
108 author=author,
109 parent_commit_id=parent_id,
110 structured_delta=delta,
111 )
112 write_commit(root, record)
113 (root / ".muse" / "refs" / "heads" / branch).write_text(cid, encoding="utf-8")
114 return record
115
116
117 def _make_delta(ops: list[DomainOp]) -> StructuredDelta:
118 return StructuredDelta(domain="code", ops=ops, summary="")
119
120
121 _FAKE_HASH_A = "a" * 64
122 _FAKE_HASH_B = "b" * 64
123
124
125 def _insert_op(address: str, summary: str = "created") -> DomainOp:
126 from muse.domain import InsertOp
127 return InsertOp(
128 op="insert", address=address,
129 position=None, content_id=_FAKE_HASH_A,
130 content_summary=summary,
131 )
132
133
134 def _delete_op(address: str, summary: str = "deleted") -> DomainOp:
135 from muse.domain import DeleteOp
136 return DeleteOp(
137 op="delete", address=address,
138 position=None, content_id=_FAKE_HASH_A,
139 content_summary=summary,
140 )
141
142
143 def _replace_op(address: str, new_summary: str = "modified") -> DomainOp:
144 from muse.domain import ReplaceOp
145 return ReplaceOp(
146 op="replace", address=address,
147 position=None,
148 old_content_id=_FAKE_HASH_A, new_content_id=_FAKE_HASH_B,
149 old_summary="old", new_summary=new_summary,
150 )
151
152
153 def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
154 return runner.invoke(
155 cli,
156 ["code", "blame", *args],
157 env={"MUSE_REPO_ROOT": str(root)},
158 )
159
160
161 def _parse_json(result: InvokeResult) -> "_BlameResultJson":
162 from muse.cli.commands.blame import _BlameResultJson, _BlameEventJson
163
164 start = result.output.index("{")
165 blob = result.output[start:]
166 depth = 0
167 end = 0
168 for i, ch in enumerate(blob):
169 if ch == "{":
170 depth += 1
171 elif ch == "}":
172 depth -= 1
173 if depth == 0:
174 end = i + 1
175 break
176 raw = json.loads(blob[:end])
177 assert isinstance(raw, dict)
178 raw_events = raw.get("events", [])
179 assert isinstance(raw_events, list)
180 _valid_kinds = frozenset(("created", "modified", "renamed", "moved", "deleted", "signature"))
181 events: list[_BlameEventJson] = []
182 for e in raw_events:
183 assert isinstance(e, dict)
184 raw_kind = e.get("event", "modified")
185 kind: SymbolEventKind = raw_kind if raw_kind in _valid_kinds else "modified"
186 events.append(_BlameEventJson(
187 event=kind,
188 commit_id=str(e.get("commit_id", "")),
189 author=str(e.get("author", "")),
190 message=str(e.get("message", "")),
191 committed_at=str(e.get("committed_at", "")),
192 address=str(e.get("address", "")),
193 detail=str(e.get("detail", "")),
194 new_address=e.get("new_address"),
195 ))
196 return _BlameResultJson(
197 address=str(raw.get("address", "")),
198 start_ref=str(raw.get("start_ref", "")),
199 total_commits_scanned=int(raw.get("total_commits_scanned", 0)),
200 truncated=bool(raw.get("truncated", False)),
201 events=events,
202 )
203
204
205 # ---------------------------------------------------------------------------
206 # Unit — _flat_ops
207 # ---------------------------------------------------------------------------
208
209
210 class TestFlatOps:
211 def test_passthrough_non_patch_ops(self) -> None:
212 from muse.cli.commands.blame import _flat_ops
213
214 op = _insert_op("f.py::foo")
215 assert _flat_ops([op]) == [op]
216
217 def test_flattens_patch_children(self) -> None:
218 from muse.cli.commands.blame import _flat_ops
219 from muse.domain import PatchOp
220
221 child1 = _insert_op("f.py::foo")
222 child2 = _replace_op("f.py::bar")
223 patch = PatchOp(op="patch", address="f.py", child_ops=[child1, child2], child_domain="code", child_summary="test")
224 result = _flat_ops([patch])
225 assert result == [child1, child2]
226
227 def test_empty_ops(self) -> None:
228 from muse.cli.commands.blame import _flat_ops
229
230 assert _flat_ops([]) == []
231
232 def test_mixed_patch_and_leaf(self) -> None:
233 from muse.cli.commands.blame import _flat_ops
234 from muse.domain import PatchOp
235
236 child = _insert_op("f.py::child")
237 patch = PatchOp(op="patch", address="f.py", child_ops=[child], child_domain="code", child_summary="test")
238 leaf = _delete_op("g.py::gone")
239 result = _flat_ops([patch, leaf])
240 assert result == [child, leaf]
241
242
243 # ---------------------------------------------------------------------------
244 # Unit — _events_in_commit
245 # ---------------------------------------------------------------------------
246
247
248 class TestEventsInCommit:
249 def _commit(
250 self, root: pathlib.Path, delta: StructuredDelta | None = None
251 ) -> CommitRecord:
252 return _write_commit(root, delta=delta)
253
254 def test_insert_yields_created(self, tmp_path: pathlib.Path) -> None:
255 from muse.cli.commands.blame import _events_in_commit
256
257 repo = _make_repo(tmp_path)
258 delta = _make_delta([_insert_op("f.py::foo", "initial")])
259 c = self._commit(repo, delta)
260 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
261 assert len(evs) == 1
262 assert evs[0].kind == "created"
263 assert next_addr == "f.py::foo"
264
265 def test_replace_yields_modified(self, tmp_path: pathlib.Path) -> None:
266 from muse.cli.commands.blame import _events_in_commit
267
268 repo = _make_repo(tmp_path)
269 delta = _make_delta([_replace_op("f.py::foo", "refactored")])
270 c = self._commit(repo, delta)
271 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
272 assert len(evs) == 1
273 assert evs[0].kind == "modified"
274
275 def test_replace_rename_yields_renamed(self, tmp_path: pathlib.Path) -> None:
276 from muse.cli.commands.blame import _events_in_commit
277
278 repo = _make_repo(tmp_path)
279 delta = _make_delta([_replace_op("f.py::foo", "renamed to bar")])
280 c = self._commit(repo, delta)
281 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
282 assert len(evs) == 1
283 assert evs[0].kind == "renamed"
284 assert evs[0].new_address == "f.py::bar"
285 assert next_addr == "f.py::foo" # old name — unchanged when walking backward
286
287 def test_delete_yields_deleted(self, tmp_path: pathlib.Path) -> None:
288 from muse.cli.commands.blame import _events_in_commit
289
290 repo = _make_repo(tmp_path)
291 delta = _make_delta([_delete_op("f.py::foo", "removed")])
292 c = self._commit(repo, delta)
293 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
294 assert len(evs) == 1
295 assert evs[0].kind == "deleted"
296
297 def test_delete_moved_to_yields_moved(self, tmp_path: pathlib.Path) -> None:
298 from muse.cli.commands.blame import _events_in_commit
299
300 repo = _make_repo(tmp_path)
301 delta = _make_delta([_delete_op("f.py::foo", "moved to g.py")])
302 c = self._commit(repo, delta)
303 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
304 assert evs[0].kind == "moved"
305
306 def test_replace_signature_yields_signature(
307 self, tmp_path: pathlib.Path
308 ) -> None:
309 from muse.cli.commands.blame import _events_in_commit
310
311 repo = _make_repo(tmp_path)
312 delta = _make_delta([_replace_op("f.py::foo", "signature changed")])
313 c = self._commit(repo, delta)
314 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
315 assert evs[0].kind == "signature"
316
317 def test_no_delta_returns_empty(self, tmp_path: pathlib.Path) -> None:
318 from muse.cli.commands.blame import _events_in_commit
319
320 repo = _make_repo(tmp_path)
321 c = self._commit(repo, delta=None)
322 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
323 assert evs == []
324 assert next_addr == "f.py::foo"
325
326 def test_unrelated_op_not_matched(self, tmp_path: pathlib.Path) -> None:
327 from muse.cli.commands.blame import _events_in_commit
328
329 repo = _make_repo(tmp_path)
330 delta = _make_delta([_insert_op("f.py::other")])
331 c = self._commit(repo, delta)
332 evs, _ = _events_in_commit(c, "f.py::foo", "f.py", "foo")
333 assert evs == []
334
335 def test_reverse_rename_switches_next_address(
336 self, tmp_path: pathlib.Path
337 ) -> None:
338 from muse.cli.commands.blame import _events_in_commit
339
340 repo = _make_repo(tmp_path)
341 # op: old name "f.py::old" was renamed to "foo"
342 delta = _make_delta([_replace_op("f.py::old", "renamed to foo")])
343 c = self._commit(repo, delta)
344 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
345 assert len(evs) == 1
346 assert evs[0].kind == "renamed"
347 assert next_addr == "f.py::old"
348
349 def test_malformed_op_address_without_colons_skipped(
350 self, tmp_path: pathlib.Path
351 ) -> None:
352 from muse.cli.commands.blame import _events_in_commit
353
354 repo = _make_repo(tmp_path)
355 # op_address without '::' — should not cause crash or incorrect match
356 delta = _make_delta([_replace_op("nofile", "renamed to foo")])
357 c = self._commit(repo, delta)
358 # Should not raise and should not produce events for "f.py::foo"
359 evs, next_addr = _events_in_commit(c, "f.py::foo", "f.py", "foo")
360 assert evs == []
361 assert next_addr == "f.py::foo"
362
363
364 # ---------------------------------------------------------------------------
365 # Unit — _BlameEvent.to_dict
366 # ---------------------------------------------------------------------------
367
368
369 class TestBlameEventToDict:
370 def test_all_fields_present(self, tmp_path: pathlib.Path) -> None:
371 from muse.cli.commands.blame import _BlameEvent
372
373 repo = _make_repo(tmp_path)
374 c = _write_commit(repo)
375 ev = _BlameEvent("created", c, "f.py::foo", "initial", None)
376 d = ev.to_dict()
377 for field in (
378 "event", "commit_id", "author", "message",
379 "committed_at", "address", "detail", "new_address",
380 ):
381 assert field in d, f"Missing field: {field}"
382
383 def test_event_kind_preserved(self, tmp_path: pathlib.Path) -> None:
384 from muse.cli.commands.blame import _BlameEvent
385
386 repo = _make_repo(tmp_path)
387 c = _write_commit(repo)
388 for kind in ("created", "modified", "renamed", "moved", "deleted", "signature"):
389 typed_kind: SymbolEventKind = kind
390 ev = _BlameEvent(typed_kind, c, "f.py::foo", "detail", None)
391 assert ev.to_dict()["event"] == kind
392
393 def test_new_address_none_when_absent(self, tmp_path: pathlib.Path) -> None:
394 from muse.cli.commands.blame import _BlameEvent
395
396 repo = _make_repo(tmp_path)
397 c = _write_commit(repo)
398 ev = _BlameEvent("modified", c, "f.py::foo", "mod", None)
399 assert ev.to_dict()["new_address"] is None
400
401 def test_new_address_string_when_set(self, tmp_path: pathlib.Path) -> None:
402 from muse.cli.commands.blame import _BlameEvent
403
404 repo = _make_repo(tmp_path)
405 c = _write_commit(repo)
406 ev = _BlameEvent("renamed", c, "f.py::old", "renamed to new", "f.py::new")
407 assert ev.to_dict()["new_address"] == "f.py::new"
408
409
410 # ---------------------------------------------------------------------------
411 # Integration — basic blame scenarios
412 # ---------------------------------------------------------------------------
413
414
415 class TestBlameShow:
416 def test_no_events_shows_message(self, tmp_path: pathlib.Path) -> None:
417 repo = _make_repo(tmp_path)
418 _write_commit(repo)
419 result = _invoke(repo, "f.py::foo")
420 assert result.exit_code == 0
421 assert "no events found" in result.output
422
423 def test_created_event_shown(self, tmp_path: pathlib.Path) -> None:
424 repo = _make_repo(tmp_path)
425 delta = _make_delta([_insert_op("f.py::foo")])
426 _write_commit(repo, delta=delta)
427 result = _invoke(repo, "f.py::foo")
428 assert result.exit_code == 0
429 assert "created" in result.output
430
431 def test_modified_event_shown(self, tmp_path: pathlib.Path) -> None:
432 repo = _make_repo(tmp_path)
433 delta = _make_delta([_replace_op("f.py::foo", "big refactor")])
434 _write_commit(repo, delta=delta)
435 result = _invoke(repo, "f.py::foo")
436 assert result.exit_code == 0
437 assert "big refactor" in result.output
438
439 def test_author_shown_in_text_output(self, tmp_path: pathlib.Path) -> None:
440 repo = _make_repo(tmp_path)
441 delta = _make_delta([_insert_op("f.py::foo")])
442 _write_commit(repo, author="bob", delta=delta)
443 result = _invoke(repo, "f.py::foo")
444 assert result.exit_code == 0
445 assert "bob" in result.output
446
447 def test_message_shown_in_text_output(self, tmp_path: pathlib.Path) -> None:
448 repo = _make_repo(tmp_path)
449 delta = _make_delta([_insert_op("f.py::foo")])
450 _write_commit(repo, message="feat: add foo", delta=delta)
451 result = _invoke(repo, "f.py::foo")
452 assert result.exit_code == 0
453 assert "feat: add foo" in result.output
454
455 def test_shows_hint_when_more_events_exist(
456 self, tmp_path: pathlib.Path
457 ) -> None:
458 repo = _make_repo(tmp_path)
459 # 4 commits each touching f.py::foo
460 c1 = _write_commit(repo, message="c1", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
461 c2 = _write_commit(repo, message="c2", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod1")]), dt_offset_days=1)
462 c3 = _write_commit(repo, message="c3", parent_id=c2.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod2")]), dt_offset_days=2)
463 _write_commit(repo, message="c4", parent_id=c3.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod3")]), dt_offset_days=3)
464 result = _invoke(repo, "f.py::foo")
465 assert result.exit_code == 0
466 assert "older event" in result.output
467
468 def test_all_flag_shows_full_history(self, tmp_path: pathlib.Path) -> None:
469 repo = _make_repo(tmp_path)
470 c1 = _write_commit(repo, message="c1", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
471 c2 = _write_commit(repo, message="c2", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod1")]), dt_offset_days=1)
472 c3 = _write_commit(repo, message="c3", parent_id=c2.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod2")]), dt_offset_days=2)
473 _write_commit(repo, message="c4", parent_id=c3.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod3")]), dt_offset_days=3)
474 result = _invoke(repo, "f.py::foo", "--all")
475 assert result.exit_code == 0
476 assert "older event" not in result.output
477
478 def test_all_flag_shows_author_for_every_event(
479 self, tmp_path: pathlib.Path
480 ) -> None:
481 repo = _make_repo(tmp_path)
482 c1 = _write_commit(repo, author="alice", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
483 _write_commit(repo, author="bob", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod")]), dt_offset_days=1)
484 result = _invoke(repo, "f.py::foo", "--all")
485 assert result.exit_code == 0
486 assert "alice" in result.output
487 assert "bob" in result.output
488
489 def test_rename_shown_and_tracked(self, tmp_path: pathlib.Path) -> None:
490 repo = _make_repo(tmp_path)
491 c1 = _write_commit(repo, message="create", delta=_make_delta([_insert_op("f.py::old")]), dt_offset_days=0)
492 _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)
493 result = _invoke(repo, "f.py::new")
494 assert result.exit_code == 0
495 assert "renamed" in result.output
496
497
498 # ---------------------------------------------------------------------------
499 # Integration — JSON output
500 # ---------------------------------------------------------------------------
501
502
503 class TestBlameJson:
504 def test_json_schema_all_fields(self, tmp_path: pathlib.Path) -> None:
505 repo = _make_repo(tmp_path)
506 delta = _make_delta([_insert_op("f.py::foo")])
507 _write_commit(repo, delta=delta)
508 result = _invoke(repo, "f.py::foo", "--json")
509 assert result.exit_code == 0
510 data = _parse_json(result)
511 for field in ("address", "start_ref", "total_commits_scanned", "truncated", "events"):
512 assert field in data, f"Missing field: {field}"
513
514 def test_json_address_matches(self, tmp_path: pathlib.Path) -> None:
515 repo = _make_repo(tmp_path)
516 _write_commit(repo)
517 result = _invoke(repo, "f.py::foo", "--json")
518 data = _parse_json(result)
519 assert data["address"] == "f.py::foo"
520
521 def test_json_event_fields(self, tmp_path: pathlib.Path) -> None:
522 repo = _make_repo(tmp_path)
523 delta = _make_delta([_insert_op("f.py::foo")])
524 _write_commit(repo, delta=delta)
525 result = _invoke(repo, "f.py::foo", "--json")
526 data = _parse_json(result)
527 assert len(data["events"]) == 1
528 ev = data["events"][0]
529 for field in (
530 "event", "commit_id", "author", "message",
531 "committed_at", "address", "detail", "new_address",
532 ):
533 assert field in ev, f"Missing event field: {field}"
534
535 def test_json_events_chronological(self, tmp_path: pathlib.Path) -> None:
536 repo = _make_repo(tmp_path)
537 c1 = _write_commit(repo, message="create", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
538 _write_commit(repo, message="modify", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
539 result = _invoke(repo, "f.py::foo", "--json")
540 data = _parse_json(result)
541 events = data["events"]
542 assert len(events) == 2
543 # Chronological (oldest first) in JSON
544 assert events[0]["event"] == "created"
545 assert events[1]["event"] == "modified"
546
547 def test_json_truncated_false_small_history(
548 self, tmp_path: pathlib.Path
549 ) -> None:
550 repo = _make_repo(tmp_path)
551 _write_commit(repo)
552 result = _invoke(repo, "f.py::foo", "--json")
553 data = _parse_json(result)
554 assert data["truncated"] is False
555
556 def test_json_no_events_empty_list(self, tmp_path: pathlib.Path) -> 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["events"] == []
562
563 def test_json_output_is_valid_json(self, tmp_path: pathlib.Path) -> None:
564 repo = _make_repo(tmp_path)
565 delta = _make_delta([_insert_op("f.py::foo")])
566 _write_commit(repo, delta=delta)
567 result = _invoke(repo, "f.py::foo", "--json")
568 assert result.exit_code == 0
569 # Must be parseable as JSON
570 start = result.output.index("{")
571 json.loads(result.output[start:])
572
573
574 # ---------------------------------------------------------------------------
575 # Integration — --kind filter
576 # ---------------------------------------------------------------------------
577
578
579 class TestKindFilter:
580 def test_kind_created_only(self, tmp_path: pathlib.Path) -> None:
581 repo = _make_repo(tmp_path)
582 c1 = _write_commit(repo, delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
583 _write_commit(repo, parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
584 result = _invoke(repo, "f.py::foo", "--kind", "created", "--all")
585 assert result.exit_code == 0
586 assert "created" in result.output
587 assert "modified" not in result.output
588
589 def test_kind_modified_only(self, tmp_path: pathlib.Path) -> None:
590 repo = _make_repo(tmp_path)
591 c1 = _write_commit(repo, delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
592 _write_commit(repo, parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "changed")]), dt_offset_days=1)
593 result = _invoke(repo, "f.py::foo", "--kind", "modified", "--all")
594 assert result.exit_code == 0
595 assert "changed" in result.output
596 assert "created" not in result.output
597
598 def test_kind_multiple_values(self, tmp_path: pathlib.Path) -> None:
599 repo = _make_repo(tmp_path)
600 delta = _make_delta([_insert_op("f.py::foo")])
601 _write_commit(repo, delta=delta)
602 result = _invoke(repo, "f.py::foo", "--kind", "created", "--kind", "modified")
603 assert result.exit_code == 0
604
605 def test_invalid_kind_exits_user_error(self, tmp_path: pathlib.Path) -> None:
606 repo = _make_repo(tmp_path)
607 _write_commit(repo)
608 result = _invoke(repo, "f.py::foo", "--kind", "invented")
609 assert result.exit_code == ExitCode.USER_ERROR.value
610
611 def test_kind_filter_in_json(self, tmp_path: pathlib.Path) -> None:
612 repo = _make_repo(tmp_path)
613 c1 = _write_commit(repo, delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
614 _write_commit(repo, parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
615 result = _invoke(repo, "f.py::foo", "--kind", "modified", "--json")
616 data = _parse_json(result)
617 assert all(ev["event"] == "modified" for ev in data["events"])
618
619 def test_no_match_shows_filter_message(self, tmp_path: pathlib.Path) -> None:
620 repo = _make_repo(tmp_path)
621 delta = _make_delta([_insert_op("f.py::foo")])
622 _write_commit(repo, delta=delta)
623 result = _invoke(repo, "f.py::foo", "--kind", "deleted")
624 assert result.exit_code == 0
625 assert "no events match" in result.output
626
627
628 # ---------------------------------------------------------------------------
629 # Integration — --author filter
630 # ---------------------------------------------------------------------------
631
632
633 class TestAuthorFilter:
634 def test_author_filter_matches(self, tmp_path: pathlib.Path) -> None:
635 repo = _make_repo(tmp_path)
636 delta = _make_delta([_insert_op("f.py::foo")])
637 _write_commit(repo, author="alice", delta=delta)
638 result = _invoke(repo, "f.py::foo", "--author", "alice")
639 assert result.exit_code == 0
640 assert "alice" in result.output
641
642 def test_author_filter_case_insensitive(self, tmp_path: pathlib.Path) -> None:
643 repo = _make_repo(tmp_path)
644 delta = _make_delta([_insert_op("f.py::foo")])
645 _write_commit(repo, author="Alice", delta=delta)
646 result = _invoke(repo, "f.py::foo", "--author", "ALICE")
647 assert result.exit_code == 0
648 assert "Alice" in result.output
649
650 def test_author_filter_no_match_empty(self, tmp_path: pathlib.Path) -> None:
651 repo = _make_repo(tmp_path)
652 delta = _make_delta([_insert_op("f.py::foo")])
653 _write_commit(repo, author="alice", delta=delta)
654 result = _invoke(repo, "f.py::foo", "--author", "nosuchauthor")
655 assert result.exit_code == 0
656 assert "no events match" in result.output
657
658 def test_author_filter_in_json(self, tmp_path: pathlib.Path) -> None:
659 repo = _make_repo(tmp_path)
660 c1 = _write_commit(repo, author="alice", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
661 _write_commit(repo, author="bob", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
662 result = _invoke(repo, "f.py::foo", "--author", "alice", "--json")
663 data = _parse_json(result)
664 assert all(ev["author"] == "alice" for ev in data["events"])
665
666
667 # ---------------------------------------------------------------------------
668 # Security
669 # ---------------------------------------------------------------------------
670
671
672 class TestBlameSecurity:
673 _ANSI = "\x1b[31mevil\x1b[0m"
674
675 def test_ansi_in_address_rejected(self, tmp_path: pathlib.Path) -> None:
676 repo = _make_repo(tmp_path)
677 _write_commit(repo)
678 result = _invoke(repo, f"f.py::{self._ANSI}")
679 assert result.exit_code == ExitCode.USER_ERROR.value
680
681 def test_null_byte_in_address_rejected(self, tmp_path: pathlib.Path) -> None:
682 repo = _make_repo(tmp_path)
683 _write_commit(repo)
684 result = _invoke(repo, "f.py::foo\x00bar")
685 assert result.exit_code == ExitCode.USER_ERROR.value
686
687 def test_control_char_in_address_rejected(self, tmp_path: pathlib.Path) -> None:
688 repo = _make_repo(tmp_path)
689 _write_commit(repo)
690 result = _invoke(repo, "f.py::foo\x07bell")
691 assert result.exit_code == ExitCode.USER_ERROR.value
692
693 def test_ansi_in_stored_detail_stripped_from_output(
694 self, tmp_path: pathlib.Path
695 ) -> None:
696 """ANSI in a commit's new_summary must not reach the terminal."""
697 repo = _make_repo(tmp_path)
698 # Store a commit with ANSI in new_summary (simulates a compromised record)
699 evil_summary = f"modified {self._ANSI}"
700 delta = _make_delta([_replace_op("f.py::foo", evil_summary)])
701 _write_commit(repo, delta=delta)
702 result = _invoke(repo, "f.py::foo")
703 assert result.exit_code == 0
704 assert "\x1b[" not in result.output
705
706 def test_ansi_in_author_stripped_from_output(
707 self, tmp_path: pathlib.Path
708 ) -> None:
709 repo = _make_repo(tmp_path)
710 delta = _make_delta([_insert_op("f.py::foo")])
711 _write_commit(repo, author=self._ANSI, delta=delta)
712 result = _invoke(repo, "f.py::foo")
713 assert result.exit_code == 0
714 assert "\x1b[" not in result.output
715
716 def test_ansi_in_message_stripped_from_output(
717 self, tmp_path: pathlib.Path
718 ) -> None:
719 repo = _make_repo(tmp_path)
720 delta = _make_delta([_insert_op("f.py::foo")])
721 _write_commit(repo, message=f"commit {self._ANSI}", delta=delta)
722 result = _invoke(repo, "f.py::foo")
723 assert result.exit_code == 0
724 assert "\x1b[" not in result.output
725
726 def test_missing_address_separator_exits_user_error(
727 self, tmp_path: pathlib.Path
728 ) -> None:
729 repo = _make_repo(tmp_path)
730 _write_commit(repo)
731 result = _invoke(repo, "no-separator")
732 assert result.exit_code == ExitCode.USER_ERROR.value
733
734 def test_commit_not_found_exits_not_found(
735 self, tmp_path: pathlib.Path
736 ) -> None:
737 repo = _make_repo(tmp_path)
738 _write_commit(repo)
739 result = _invoke(repo, "f.py::foo", "--from", "0" * 64)
740 assert result.exit_code == ExitCode.NOT_FOUND.value
741
742 def test_error_message_no_traceback(self, tmp_path: pathlib.Path) -> None:
743 repo = _make_repo(tmp_path)
744 result = _invoke(repo, "no-separator")
745 assert "Traceback" not in result.output
746
747 def test_json_stdout_clean_on_success(self, tmp_path: pathlib.Path) -> None:
748 """JSON consumers must not see non-JSON data on stdout."""
749 repo = _make_repo(tmp_path)
750 delta = _make_delta([_insert_op("f.py::foo")])
751 _write_commit(repo, delta=delta)
752 result = _invoke(repo, "f.py::foo", "--json")
753 stripped = result.output.lstrip()
754 assert stripped.startswith("{"), f"Expected JSON on stdout, got: {result.output[:80]!r}"
755
756
757 # ---------------------------------------------------------------------------
758 # E2E — full CLI flag coverage
759 # ---------------------------------------------------------------------------
760
761
762 class TestE2E:
763 def test_help_shows_new_flags(self, tmp_path: pathlib.Path) -> None:
764 result = runner.invoke(cli, ["code", "blame", "--help"])
765 assert result.exit_code == 0
766 assert "--kind" in result.output
767 assert "--author" in result.output
768 assert "--all" in result.output
769 assert "--json" in result.output
770 assert "--from" in result.output
771 assert "--max" in result.output
772
773 def test_default_max_is_applied(self, tmp_path: pathlib.Path) -> None:
774 from muse.cli.commands.blame import _DEFAULT_MAX
775 assert _DEFAULT_MAX == 500
776
777 def test_max_one_commit_scanned(self, tmp_path: pathlib.Path) -> None:
778 repo = _make_repo(tmp_path)
779 _write_commit(repo)
780 result = _invoke(repo, "f.py::foo", "--max", "1", "--json")
781 assert result.exit_code == 0
782 data = _parse_json(result)
783 assert data["total_commits_scanned"] == 1
784
785 def test_from_ref_head(self, tmp_path: pathlib.Path) -> None:
786 repo = _make_repo(tmp_path)
787 delta = _make_delta([_insert_op("f.py::foo")])
788 _write_commit(repo, delta=delta)
789 result = _invoke(repo, "f.py::foo", "--from", "HEAD")
790 assert result.exit_code == 0
791
792 def test_kind_and_author_combined(self, tmp_path: pathlib.Path) -> None:
793 repo = _make_repo(tmp_path)
794 c1 = _write_commit(repo, author="alice", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
795 _write_commit(repo, author="bob", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo")]), dt_offset_days=1)
796 result = _invoke(repo, "f.py::foo", "--kind", "created", "--author", "alice", "--json")
797 data = _parse_json(result)
798 assert all(
799 ev["event"] == "created" and ev["author"] == "alice"
800 for ev in data["events"]
801 )
802
803 def test_full_history_chronological_in_json(
804 self, tmp_path: pathlib.Path
805 ) -> None:
806 repo = _make_repo(tmp_path)
807 c1 = _write_commit(repo, message="create", delta=_make_delta([_insert_op("f.py::foo")]), dt_offset_days=0)
808 c2 = _write_commit(repo, message="mod1", parent_id=c1.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod1")]), dt_offset_days=1)
809 _write_commit(repo, message="mod2", parent_id=c2.commit_id, delta=_make_delta([_replace_op("f.py::foo", "mod2")]), dt_offset_days=2)
810 result = _invoke(repo, "f.py::foo", "--json", "--all")
811 data = _parse_json(result)
812 messages = [ev["message"] for ev in data["events"]]
813 assert messages == ["create", "mod1", "mod2"]
814
815
816 # ---------------------------------------------------------------------------
817 # Stress
818 # ---------------------------------------------------------------------------
819
820
821 class TestStress:
822 def test_early_exit_on_created(self, tmp_path: pathlib.Path) -> None:
823 """Scan stops at 'created' — rest of chain is not processed."""
824 repo = _make_repo(tmp_path)
825 # chain: create → 49 modifications
826 c = _write_commit(
827 repo, message="create",
828 delta=_make_delta([_insert_op("f.py::foo")]),
829 dt_offset_days=0,
830 )
831 for i in range(1, 50):
832 c = _write_commit(
833 repo, message=f"mod{i}", parent_id=c.commit_id,
834 delta=_make_delta([_replace_op("f.py::foo", f"mod{i}")]),
835 dt_offset_days=i,
836 )
837 result = _invoke(repo, "f.py::foo", "--json", "--all")
838 assert result.exit_code == 0
839 data = _parse_json(result)
840 # All 50 events should be present (created + 49 mods)
841 assert len(data["events"]) == 50
842 # early-exit: commits scanned should be exactly 50 (not more)
843 assert data["total_commits_scanned"] == 50
844
845 def test_50_event_history_all_flag(self, tmp_path: pathlib.Path) -> None:
846 repo = _make_repo(tmp_path)
847 c = _write_commit(
848 repo, delta=_make_delta([_insert_op("f.py::bar")]), dt_offset_days=0
849 )
850 for i in range(1, 50):
851 c = _write_commit(
852 repo, parent_id=c.commit_id,
853 delta=_make_delta([_replace_op("f.py::bar", f"change{i}")]),
854 dt_offset_days=i,
855 )
856 result = _invoke(repo, "f.py::bar", "--all")
857 assert result.exit_code == 0
858 assert "change49" in result.output
859
860 def test_concurrent_blame_isolated_repos(
861 self, tmp_path: pathlib.Path
862 ) -> None:
863 """Eight threads each blame their own isolated repo — no shared state."""
864 from muse.cli.commands.blame import _events_in_commit
865
866 errors: list[str] = []
867
868 def worker(idx: int) -> None:
869 try:
870 repo = _make_repo(tmp_path / f"repo{idx}")
871 delta = _make_delta([_insert_op(f"f.py::sym{idx}")])
872 c = _write_commit(repo, delta=delta)
873 # Directly test core logic (not CliRunner — env not thread-safe)
874 evs, _ = _events_in_commit(
875 c, f"f.py::sym{idx}", "f.py", f"sym{idx}"
876 )
877 if len(evs) != 1 or evs[0].kind != "created":
878 errors.append(f"Thread {idx}: unexpected events {evs!r}")
879 except Exception as exc:
880 errors.append(f"Thread {idx}: {exc}")
881
882 threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
883 for t in threads:
884 t.start()
885 for t in threads:
886 t.join()
887
888 assert errors == [], f"Concurrent blame failures: {errors}"
889
890 def test_flat_ops_500_patch_children(self) -> None:
891 from muse.cli.commands.blame import _flat_ops
892 from muse.domain import PatchOp
893
894 children = [_insert_op(f"f.py::sym{i}") for i in range(500)]
895 patch = PatchOp(op="patch", address="f.py", child_ops=children, child_domain="code", child_summary="test")
896 result = _flat_ops([patch])
897 assert len(result) == 500
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