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