gabriel / muse public
test_cmd_log.py python
1,025 lines 39.7 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 149 days ago
1 """Comprehensive tests for ``muse log``.
2
3 Coverage tiers:
4 - Unit: _parse_date, _apply_filters, _commit_to_json, _format_date,
5 _file_diff, _branch_tips, _collect_all_commits, _topo_sort
6 - Integration: all flags (--json, --oneline, --stat, --graph, --all,
7 --since, --until, --author, --section, --track, --emotion, -n)
8 - End-to-end: full workflows (init→commit(s)→log, branch→merge→log --all)
9 - Security: ANSI injection via commit messages/authors, invalid date formats,
10 bad --format value, multiline message sanitization
11 - Stress: 500-commit repos, rapid sequential calls, filter on large history
12 """
13 from __future__ import annotations
14
15 import json
16 import os
17 import pathlib
18 import subprocess
19 from datetime import datetime, timezone
20
21 import pytest
22
23 from muse.core.store import CommitRecord
24 from tests.cli_test_helper import CliRunner, InvokeResult
25
26 runner = CliRunner()
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32
33 def _init(repo: pathlib.Path) -> InvokeResult:
34 from muse.cli.app import main as cli
35
36 repo.mkdir(parents=True, exist_ok=True)
37 saved = os.getcwd()
38 try:
39 os.chdir(repo)
40 return runner.invoke(cli, ["init"])
41 finally:
42 os.chdir(saved)
43
44
45 def _log(repo: pathlib.Path, *extra: str) -> InvokeResult:
46 from muse.cli.app import main as cli
47
48 saved = os.getcwd()
49 try:
50 os.chdir(repo)
51 return runner.invoke(cli, ["log", *extra])
52 finally:
53 os.chdir(saved)
54
55
56 def _commit(repo: pathlib.Path, msg: str = "commit", filename: str | None = None) -> None:
57 from muse.cli.app import main as cli
58
59 fname = filename or f"file_{abs(hash(msg))}.py"
60 (repo / fname).write_text(f"# {msg}\n")
61 saved = os.getcwd()
62 try:
63 os.chdir(repo)
64 runner.invoke(cli, ["commit", "-m", msg])
65 finally:
66 os.chdir(saved)
67
68
69 def _fresh_repo(tmp: pathlib.Path, n_commits: int = 1) -> pathlib.Path:
70 repo = tmp / "repo"
71 _init(repo)
72 for i in range(n_commits):
73 _commit(repo, f"commit {i}", filename=f"file_{i}.py")
74 return repo
75
76
77 # ---------------------------------------------------------------------------
78 # Unit — _parse_date
79 # ---------------------------------------------------------------------------
80
81
82 class TestParseDate:
83 def test_today(self) -> None:
84 from muse.cli.commands.log import _parse_date
85
86 dt = _parse_date("today")
87 now = datetime.now(timezone.utc)
88 assert dt.date() == now.date()
89 assert dt.tzinfo is not None
90
91 def test_yesterday(self) -> None:
92 from muse.cli.commands.log import _parse_date
93 from datetime import timedelta
94
95 dt = _parse_date("yesterday")
96 now = datetime.now(timezone.utc)
97 assert dt.date() == (now - timedelta(days=1)).date()
98
99 def test_n_days_ago(self) -> None:
100 from muse.cli.commands.log import _parse_date
101 from datetime import timedelta
102
103 dt = _parse_date("7 days ago")
104 now = datetime.now(timezone.utc)
105 diff = now - dt
106 assert abs(diff.total_seconds() - 7 * 86400) < 5
107
108 def test_n_weeks_ago(self) -> None:
109 from muse.cli.commands.log import _parse_date
110 from datetime import timedelta
111
112 dt = _parse_date("2 weeks ago")
113 now = datetime.now(timezone.utc)
114 diff = now - dt
115 assert abs(diff.total_seconds() - 14 * 86400) < 5
116
117 def test_iso_date(self) -> None:
118 from muse.cli.commands.log import _parse_date
119
120 dt = _parse_date("2025-01-15")
121 assert dt.year == 2025
122 assert dt.month == 1
123 assert dt.day == 15
124 assert dt.tzinfo is not None
125
126 def test_iso_datetime(self) -> None:
127 from muse.cli.commands.log import _parse_date
128
129 dt = _parse_date("2025-01-15T12:30:00")
130 assert dt.hour == 12
131 assert dt.minute == 30
132
133 def test_space_datetime(self) -> None:
134 from muse.cli.commands.log import _parse_date
135
136 dt = _parse_date("2025-06-01 09:00:00")
137 assert dt.year == 2025
138 assert dt.hour == 9
139
140 def test_invalid_raises_value_error(self) -> None:
141 from muse.cli.commands.log import _parse_date
142
143 with pytest.raises(ValueError, match="Cannot parse date"):
144 _parse_date("not-a-date")
145
146 def test_empty_string_raises(self) -> None:
147 from muse.cli.commands.log import _parse_date
148
149 with pytest.raises(ValueError):
150 _parse_date("")
151
152 def test_case_insensitive(self) -> None:
153 from muse.cli.commands.log import _parse_date
154
155 dt1 = _parse_date("TODAY")
156 dt2 = _parse_date("today")
157 assert dt1.date() == dt2.date()
158
159 def test_plural_days(self) -> None:
160 from muse.cli.commands.log import _parse_date
161
162 dt1 = _parse_date("1 day ago")
163 dt2 = _parse_date("1 days ago")
164 assert abs((dt1 - dt2).total_seconds()) < 2
165
166
167 # ---------------------------------------------------------------------------
168 # Unit — _apply_filters
169 # ---------------------------------------------------------------------------
170
171
172 class TestApplyFilters:
173 def _make_commits(self, n: int, author: str = "alice") -> list[CommitRecord]:
174 return [
175 CommitRecord(
176 commit_id=f"{'a' * 63}{i:x}"[:64],
177 repo_id="r" * 36,
178 branch="main",
179 message=f"msg {i}",
180 author=author,
181 committed_at=datetime(2025, 6, i % 28 + 1, tzinfo=timezone.utc),
182 parent_commit_id=None,
183 snapshot_id="b" * 64,
184 )
185 for i in range(n)
186 ]
187
188 def test_no_filters_returns_all(self) -> None:
189 from muse.cli.commands.log import _apply_filters
190
191 commits = self._make_commits(5)
192 result, truncated = _apply_filters(
193 commits,
194 since_dt=None, until_dt=None, author=None,
195 section=None, track=None, emotion=None, limit=100,
196 )
197 assert len(result) == 5
198 assert not truncated
199
200 def test_limit_enforced(self) -> None:
201 from muse.cli.commands.log import _apply_filters
202
203 commits = self._make_commits(10)
204 result, truncated = _apply_filters(
205 commits,
206 since_dt=None, until_dt=None, author=None,
207 section=None, track=None, emotion=None, limit=3,
208 )
209 assert len(result) == 3
210 assert truncated
211
212 def test_author_filter_case_insensitive(self) -> None:
213 from muse.cli.commands.log import _apply_filters
214
215 alice = CommitRecord(
216 commit_id="a" * 64, repo_id="r" * 36, branch="main", message="m",
217 author="Alice",
218 committed_at=datetime(2025, 1, 1, tzinfo=timezone.utc),
219 parent_commit_id=None, snapshot_id="b" * 64,
220 )
221 bob = CommitRecord(
222 commit_id="b" * 64, repo_id="r" * 36, branch="main", message="m",
223 author="Bob",
224 committed_at=datetime(2025, 1, 2, tzinfo=timezone.utc),
225 parent_commit_id=None, snapshot_id="c" * 64,
226 )
227 result, _ = _apply_filters(
228 [alice, bob],
229 since_dt=None, until_dt=None, author="alice",
230 section=None, track=None, emotion=None, limit=100,
231 )
232 assert len(result) == 1
233 assert result[0].author == "Alice"
234
235 def test_since_filter(self) -> None:
236 from muse.cli.commands.log import _apply_filters
237
238 old = CommitRecord(
239 commit_id="a" * 64, repo_id="r" * 36, branch="main", message="old",
240 author="x",
241 committed_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
242 parent_commit_id=None, snapshot_id="b" * 64,
243 )
244 new_commit = CommitRecord(
245 commit_id="b" * 64, repo_id="r" * 36, branch="main", message="new",
246 author="x",
247 committed_at=datetime(2025, 6, 1, tzinfo=timezone.utc),
248 parent_commit_id=None, snapshot_id="c" * 64,
249 )
250 since = datetime(2025, 1, 1, tzinfo=timezone.utc)
251 result, _ = _apply_filters(
252 [old, new_commit],
253 since_dt=since, until_dt=None, author=None,
254 section=None, track=None, emotion=None, limit=100,
255 )
256 assert len(result) == 1
257 assert result[0].message == "new"
258
259 def test_until_filter(self) -> None:
260 from muse.cli.commands.log import _apply_filters
261
262 early = CommitRecord(
263 commit_id="a" * 64, repo_id="r" * 36, branch="main", message="early",
264 author="x",
265 committed_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
266 parent_commit_id=None, snapshot_id="b" * 64,
267 )
268 late = CommitRecord(
269 commit_id="b" * 64, repo_id="r" * 36, branch="main", message="late",
270 author="x",
271 committed_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
272 parent_commit_id=None, snapshot_id="c" * 64,
273 )
274 until = datetime(2025, 1, 1, tzinfo=timezone.utc)
275 result, _ = _apply_filters(
276 [early, late],
277 since_dt=None, until_dt=until, author=None,
278 section=None, track=None, emotion=None, limit=100,
279 )
280 assert len(result) == 1
281 assert result[0].message == "early"
282
283 def test_empty_input_returns_empty(self) -> None:
284 from muse.cli.commands.log import _apply_filters
285
286 result, truncated = _apply_filters(
287 [],
288 since_dt=None, until_dt=None, author=None,
289 section=None, track=None, emotion=None, limit=10,
290 )
291 assert result == []
292 assert not truncated
293
294
295 # ---------------------------------------------------------------------------
296 # Unit — _commit_to_json
297 # ---------------------------------------------------------------------------
298
299
300 class TestCommitToJson:
301 def _make_commit(self) -> CommitRecord:
302 return CommitRecord(
303 commit_id="a" * 64,
304 repo_id="r" * 36,
305 branch="main",
306 message="hello",
307 author="alice",
308 committed_at=datetime(2025, 6, 1, tzinfo=timezone.utc),
309 parent_commit_id=None,
310 snapshot_id="b" * 64,
311 )
312
313 def test_all_keys_present(self) -> None:
314 from muse.cli.commands.log import _commit_to_json
315
316 c = self._make_commit()
317 d = _commit_to_json(c)
318 expected = {
319 "commit_id", "branch", "message", "author", "committed_at",
320 "parent_commit_id", "parent2_commit_id", "snapshot_id",
321 "sem_ver_bump", "breaking_changes", "metadata",
322 "files_added", "files_removed", "files_modified",
323 }
324 assert expected == set(d.keys())
325
326 def test_file_lists_empty_without_stat(self) -> None:
327 from muse.cli.commands.log import _commit_to_json
328
329 c = self._make_commit()
330 d = _commit_to_json(c)
331 assert d["files_added"] == []
332 assert d["files_removed"] == []
333 assert d["files_modified"] == []
334
335 def test_parent2_commit_id_is_none_for_linear(self) -> None:
336 from muse.cli.commands.log import _commit_to_json
337
338 c = self._make_commit()
339 d = _commit_to_json(c)
340 assert d["parent2_commit_id"] is None
341
342 def test_breaking_changes_is_always_list(self) -> None:
343 from muse.cli.commands.log import _commit_to_json
344
345 c = self._make_commit()
346 d = _commit_to_json(c)
347 assert isinstance(d["breaking_changes"], list)
348
349 def test_committed_at_is_iso_string(self) -> None:
350 from muse.cli.commands.log import _commit_to_json
351
352 c = self._make_commit()
353 d = _commit_to_json(c)
354 ts = d["committed_at"]
355 assert isinstance(ts, str)
356 assert "2025" in ts
357 assert "T" in ts or " " in ts
358
359
360 # ---------------------------------------------------------------------------
361 # Integration — JSON output schema
362 # ---------------------------------------------------------------------------
363
364
365 class TestJsonSchema:
366 _REQUIRED_COMMIT_KEYS = {
367 "commit_id", "branch", "message", "author", "committed_at",
368 "parent_commit_id", "parent2_commit_id", "snapshot_id",
369 "sem_ver_bump", "breaking_changes", "metadata",
370 "files_added", "files_removed", "files_modified",
371 }
372
373 def test_top_level_keys(self, tmp_path: pathlib.Path) -> None:
374 repo = _fresh_repo(tmp_path)
375 data = json.loads(_log(repo, "--json").output)
376 assert "commits" in data
377 assert "truncated" in data
378
379 def test_all_commit_keys_present(self, tmp_path: pathlib.Path) -> None:
380 repo = _fresh_repo(tmp_path, n_commits=2)
381 data = json.loads(_log(repo, "--json").output)
382 for c in data["commits"]:
383 missing = self._REQUIRED_COMMIT_KEYS - set(c.keys())
384 assert not missing, f"Missing keys: {missing}"
385
386 def test_parent2_commit_id_present(self, tmp_path: pathlib.Path) -> None:
387 repo = _fresh_repo(tmp_path)
388 data = json.loads(_log(repo, "--json").output)
389 assert "parent2_commit_id" in data["commits"][0]
390
391 def test_breaking_changes_is_list(self, tmp_path: pathlib.Path) -> None:
392 repo = _fresh_repo(tmp_path)
393 data = json.loads(_log(repo, "--json").output)
394 assert isinstance(data["commits"][0]["breaking_changes"], list)
395
396 def test_committed_at_is_iso(self, tmp_path: pathlib.Path) -> None:
397 repo = _fresh_repo(tmp_path)
398 data = json.loads(_log(repo, "--json").output)
399 ts = data["commits"][0]["committed_at"]
400 assert "T" in ts or "+" in ts
401
402 def test_truncated_false_by_default(self, tmp_path: pathlib.Path) -> None:
403 repo = _fresh_repo(tmp_path, n_commits=3)
404 data = json.loads(_log(repo, "--json").output)
405 assert data["truncated"] is False
406
407 def test_json_parseable_output(self, tmp_path: pathlib.Path) -> None:
408 repo = _fresh_repo(tmp_path, n_commits=5)
409 result = _log(repo, "--json")
410 data = json.loads(result.output)
411 assert isinstance(data["commits"], list)
412 assert len(data["commits"]) == 5
413
414 def test_empty_repo_json(self, tmp_path: pathlib.Path) -> None:
415 repo = tmp_path / "repo"
416 _init(repo)
417 result = _log(repo, "--json")
418 data = json.loads(result.output)
419 assert data["commits"] == []
420 assert data["truncated"] is False
421
422 def test_limit_n_json(self, tmp_path: pathlib.Path) -> None:
423 repo = _fresh_repo(tmp_path, n_commits=5)
424 data = json.loads(_log(repo, "--json", "-n", "2").output)
425 assert len(data["commits"]) == 2
426
427 def test_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
428 repo = _fresh_repo(tmp_path, n_commits=3)
429 data = json.loads(_log(repo, "--json").output)
430 timestamps = [c["committed_at"] for c in data["commits"]]
431 assert timestamps == sorted(timestamps, reverse=True)
432
433 def test_output_is_single_object(self, tmp_path: pathlib.Path) -> None:
434 """--json must produce one JSON object, not an array or newline-delimited."""
435 repo = _fresh_repo(tmp_path)
436 result = _log(repo, "--json")
437 # Must parse as a single dict
438 data = json.loads(result.output)
439 assert isinstance(data, dict)
440
441
442 # ---------------------------------------------------------------------------
443 # Integration — --oneline
444 # ---------------------------------------------------------------------------
445
446
447 class TestOneline:
448 def test_one_line_per_commit(self, tmp_path: pathlib.Path) -> None:
449 repo = _fresh_repo(tmp_path, n_commits=3)
450 result = _log(repo, "--oneline")
451 lines = [l for l in result.output.splitlines() if l.strip()]
452 assert len(lines) == 3
453
454 def test_short_hash_in_output(self, tmp_path: pathlib.Path) -> None:
455 repo = _fresh_repo(tmp_path)
456 data = json.loads(_log(repo, "--json").output)
457 commit_id = data["commits"][0]["commit_id"]
458 result = _log(repo, "--oneline")
459 assert commit_id[:8] in result.output
460
461 def test_message_on_same_line(self, tmp_path: pathlib.Path) -> None:
462 repo = _fresh_repo(tmp_path)
463 _commit(repo, "my special message", filename="z.py")
464 result = _log(repo, "--oneline", "-n", "1")
465 assert "my special message" in result.output
466 assert len(result.output.splitlines()) >= 1
467
468 def test_no_ansi_when_not_tty(self, tmp_path: pathlib.Path) -> None:
469 repo = _fresh_repo(tmp_path)
470 result = _log(repo, "--oneline")
471 # CLI runner is not a TTY — no escape sequences
472 assert "\x1b[" not in result.output
473
474
475 # ---------------------------------------------------------------------------
476 # Integration — --stat
477 # ---------------------------------------------------------------------------
478
479
480 class TestStat:
481 def test_stat_shows_added_files(self, tmp_path: pathlib.Path) -> None:
482 repo = _fresh_repo(tmp_path, n_commits=1)
483 result = _log(repo, "--stat")
484 assert "added" in result.output
485 assert "+" in result.output
486
487 def test_stat_shows_summary_line(self, tmp_path: pathlib.Path) -> None:
488 repo = _fresh_repo(tmp_path, n_commits=1)
489 result = _log(repo, "--stat")
490 assert "added" in result.output
491 assert "removed" in result.output
492
493 def test_stat_shows_modified_marker(self, tmp_path: pathlib.Path) -> None:
494 repo = _fresh_repo(tmp_path, n_commits=1)
495 # Modify the same file in a second commit so "modified" fires.
496 (repo / "file_0.py").write_text("# changed\n")
497 _commit(repo, "modify existing")
498 result = _log(repo, "--stat", "-n", "1")
499 assert "~" in result.output
500 assert "modified" in result.output
501
502 def test_stat_exit_zero(self, tmp_path: pathlib.Path) -> None:
503 repo = _fresh_repo(tmp_path)
504 result = _log(repo, "--stat")
505 assert result.exit_code == 0
506
507 def test_stat_json_file_lists_populated(self, tmp_path: pathlib.Path) -> None:
508 repo = _fresh_repo(tmp_path, n_commits=1)
509 data = json.loads(_log(repo, "--stat", "--json").output)
510 commit = data["commits"][0]
511 # The initial commit adds at least one file.
512 assert isinstance(commit["files_added"], list)
513 assert isinstance(commit["files_removed"], list)
514 assert isinstance(commit["files_modified"], list)
515 assert len(commit["files_added"]) > 0
516
517 def test_stat_json_modified_populated(self, tmp_path: pathlib.Path) -> None:
518 repo = _fresh_repo(tmp_path, n_commits=1)
519 # Overwrite the existing file so the second commit shows a modification.
520 (repo / "file_0.py").write_text("# changed\n")
521 _commit(repo, "modify existing")
522 data = json.loads(_log(repo, "--stat", "--json", "-n", "1").output)
523 commit = data["commits"][0]
524 assert "file_0.py" in commit["files_modified"]
525
526 def test_json_file_lists_populated_without_stat_flag(self, tmp_path: pathlib.Path) -> None:
527 """--json always populates file lists — agents must not need --stat."""
528 repo = _fresh_repo(tmp_path, n_commits=1)
529 data = json.loads(_log(repo, "--json").output)
530 commit = data["commits"][0]
531 # The initial commit adds at least one file; file lists must be
532 # populated even without the --stat flag.
533 assert isinstance(commit["files_added"], list)
534 assert isinstance(commit["files_removed"], list)
535 assert isinstance(commit["files_modified"], list)
536 assert len(commit["files_added"]) > 0
537
538
539 # ---------------------------------------------------------------------------
540 # Integration — filters
541 # ---------------------------------------------------------------------------
542
543
544 def _commit_as(repo: pathlib.Path, msg: str, author: str, filename: str | None = None) -> None:
545 """Invoke muse commit with an explicit --author flag."""
546 from muse.cli.app import main as cli
547 fname = filename or f"file_{abs(hash(msg))}.py"
548 (repo / fname).write_text(f"# {msg}\n")
549 saved = os.getcwd()
550 try:
551 os.chdir(repo)
552 runner.invoke(cli, ["commit", "-m", msg, "--author", author])
553 finally:
554 os.chdir(saved)
555
556
557 def _commit_with_config_author(repo: pathlib.Path, msg: str, author: str, filename: str | None = None) -> None:
558 """Write user.name to repo config, then invoke muse commit without --author."""
559 from muse.cli.app import main as cli
560 from muse.cli.config import set_user_field
561 set_user_field("name", author, repo)
562 fname = filename or f"file_{abs(hash(msg))}.py"
563 (repo / fname).write_text(f"# {msg}\n")
564 saved = os.getcwd()
565 try:
566 os.chdir(repo)
567 runner.invoke(cli, ["commit", "-m", msg])
568 finally:
569 os.chdir(saved)
570
571
572 class TestAuthorField:
573 """Author field in log JSON must come from user.name config when --author not given."""
574
575 def test_commit_with_explicit_author_appears_in_log(self, tmp_path: pathlib.Path) -> None:
576 """--author flag sets author field that muse log --json exposes."""
577 repo = _fresh_repo(tmp_path, n_commits=0)
578 _commit_as(repo, "my commit", "charlie")
579 result = _log(repo, "--json")
580 data = json.loads(result.output)
581 assert data["commits"][0]["author"] == "charlie"
582
583 def test_commit_reads_user_name_from_config(self, tmp_path: pathlib.Path) -> None:
584 """muse commit without --author reads user.name from repo config."""
585 repo = _fresh_repo(tmp_path, n_commits=0)
586 _commit_with_config_author(repo, "config commit", "diana")
587 result = _log(repo, "--json")
588 data = json.loads(result.output)
589 assert data["commits"][0]["author"] == "diana"
590
591 def test_author_filter_returns_matching_commits(self, tmp_path: pathlib.Path) -> None:
592 """--author filter must return commits whose author matches the substring."""
593 repo = _fresh_repo(tmp_path, n_commits=0)
594 _commit_as(repo, "alice commit", "alice", filename="a.py")
595 _commit_as(repo, "bob commit", "bob", filename="b.py")
596 result = _log(repo, "--author", "alice", "--json")
597 data = json.loads(result.output)
598 assert len(data["commits"]) == 1
599 assert data["commits"][0]["author"] == "alice"
600
601 def test_author_filter_nonexistent_returns_no_commits(self, tmp_path: pathlib.Path) -> None:
602 """--author filter with no match must return empty list."""
603 repo = _fresh_repo(tmp_path, n_commits=0)
604 _commit_as(repo, "some commit", "alice", filename="a.py")
605 result = _log(repo, "--author", "zzz_nobody_zzz", "--json")
606 data = json.loads(result.output)
607 assert data["commits"] == []
608
609
610 class TestFilters:
611 def test_author_filter_matches(self, tmp_path: pathlib.Path) -> None:
612 repo = _fresh_repo(tmp_path, n_commits=2)
613 # The author will be whatever muse uses by default
614 # We just verify that filtering by nonexistent author returns none
615 result = _log(repo, "--author", "zzz_nobody_zzz")
616 assert "(no commits)" in result.output
617
618 def test_since_filters_old_commits(self, tmp_path: pathlib.Path) -> None:
619 repo = _fresh_repo(tmp_path, n_commits=2)
620 result = _log(repo, "--since", "2099-01-01")
621 # Future date — should return no commits
622 assert "(no commits)" in result.output
623
624 def test_until_filters_future_commits(self, tmp_path: pathlib.Path) -> None:
625 repo = _fresh_repo(tmp_path, n_commits=2)
626 # Past date — all commits should be excluded
627 result = _log(repo, "--until", "2000-01-01")
628 assert "(no commits)" in result.output
629
630 def test_limit_shorthand(self, tmp_path: pathlib.Path) -> None:
631 """muse log -2 must show at most 2 commits."""
632 repo = _fresh_repo(tmp_path, n_commits=5)
633 result = _log(repo, "--oneline", "-n", "2")
634 lines = [l for l in result.output.splitlines() if l.strip()]
635 assert len(lines) == 2
636
637 def test_limit_flag_alias(self, tmp_path: pathlib.Path) -> None:
638 """--limit is an alias for -n/--max-count."""
639 repo = _fresh_repo(tmp_path, n_commits=5)
640 result = _log(repo, "--oneline", "--limit", "3")
641 lines = [l for l in result.output.splitlines() if l.strip()]
642 assert len(lines) == 3
643
644 def test_limit_flag_json(self, tmp_path: pathlib.Path) -> None:
645 """--limit works with --json output."""
646 repo = _fresh_repo(tmp_path, n_commits=5)
647 data = json.loads(_log(repo, "--json", "--limit", "2").output)
648 assert len(data["commits"]) == 2
649
650 def test_json_since_filters(self, tmp_path: pathlib.Path) -> None:
651 repo = _fresh_repo(tmp_path, n_commits=2)
652 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
653 assert data["commits"] == []
654
655 def test_invalid_since_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
656 repo = _fresh_repo(tmp_path)
657 result = _log(repo, "--since", "not-a-date")
658 assert result.exit_code != 0
659
660 def test_invalid_until_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
661 repo = _fresh_repo(tmp_path)
662 result = _log(repo, "--until", "not-a-date")
663 assert result.exit_code != 0
664
665 def test_invalid_since_no_traceback(self, tmp_path: pathlib.Path) -> None:
666 repo = _fresh_repo(tmp_path)
667 result = _log(repo, "--since", "baddate")
668 assert "Traceback" not in result.output
669
670 def test_invalid_until_clean_error(self, tmp_path: pathlib.Path) -> None:
671 repo = _fresh_repo(tmp_path)
672 result = _log(repo, "--until", "foo")
673 assert "Cannot parse" in result.output or result.exit_code != 0
674
675
676 # ---------------------------------------------------------------------------
677 # Integration — format validation
678 # ---------------------------------------------------------------------------
679
680
681 class TestFormatValidation:
682 def test_invalid_format_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
683 repo = _fresh_repo(tmp_path)
684 result = _log(repo, "--format", "xml")
685 assert result.exit_code != 0
686
687 def test_invalid_format_no_traceback(self, tmp_path: pathlib.Path) -> None:
688 repo = _fresh_repo(tmp_path)
689 result = _log(repo, "--format", "yaml")
690 assert "Traceback" not in result.output
691
692 def test_json_format_alias(self, tmp_path: pathlib.Path) -> None:
693 repo = _fresh_repo(tmp_path, n_commits=2)
694 r1 = _log(repo, "--json")
695 r2 = _log(repo, "--format", "json")
696 assert json.loads(r1.output) == json.loads(r2.output)
697
698 def test_invalid_max_count_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
699 repo = _fresh_repo(tmp_path)
700 result = _log(repo, "-n", "0")
701 assert result.exit_code != 0
702
703
704 # ---------------------------------------------------------------------------
705 # Security — ANSI injection
706 # ---------------------------------------------------------------------------
707
708
709 class TestSecurity:
710 def test_ansi_in_commit_message_sanitized_oneline(self, tmp_path: pathlib.Path) -> None:
711 repo = tmp_path / "repo"
712 _init(repo)
713 # Commit a message with ANSI in it
714 _commit(repo, "\x1b[31mevil\x1b[0m", filename="evil.py")
715 result = _log(repo, "--oneline")
716 # The runner is not a tty — any escape from the message must be sanitized
717 assert "\x1b[31m" not in result.output
718
719 def test_ansi_in_commit_message_sanitized_long(self, tmp_path: pathlib.Path) -> None:
720 repo = tmp_path / "repo"
721 _init(repo)
722 _commit(repo, "\x1b[31mhacked\x1b[0m", filename="h.py")
723 result = _log(repo)
724 assert "\x1b[31m" not in result.output
725
726 def test_ansi_in_author_sanitized(self, tmp_path: pathlib.Path) -> None:
727 """Author names from CommitRecord must be sanitized in output."""
728 repo = _fresh_repo(tmp_path)
729 result = _log(repo)
730 # No raw escape from author field in text output (we can't control
731 # author easily, but ensure output is escape-free when not tty)
732 assert "\x1b[31m" not in result.output
733
734 def test_multiline_message_all_lines_indented(self, tmp_path: pathlib.Path) -> None:
735 """Every line of a multiline message must start with 4-space indent."""
736 repo = tmp_path / "repo"
737 _init(repo)
738 _commit(repo, "Line1\nLine2\nLine3", filename="f.py")
739 result = _log(repo)
740 body_lines = [l for l in result.output.splitlines() if l.strip() in ("Line1", "Line2", "Line3")]
741 assert body_lines, f"Body lines not found in: {result.output}"
742 for line in body_lines:
743 assert line.startswith(" "), f"Not indented: {repr(line)}"
744
745 def test_invalid_fmt_sanitized_in_error(self, tmp_path: pathlib.Path) -> None:
746 repo = _fresh_repo(tmp_path)
747 evil_fmt = "\x1b[31mevil\x1b[0m"
748 result = _log(repo, "--format", evil_fmt)
749 assert result.exit_code != 0
750 assert "\x1b[31m" not in result.output
751
752 def test_no_repo_id_in_json_output(self, tmp_path: pathlib.Path) -> None:
753 repo = _fresh_repo(tmp_path)
754 stored = json.loads((repo / ".muse" / "repo.json").read_text())["repo_id"]
755 result = _log(repo, "--json")
756 assert stored not in result.output
757
758
759 # ---------------------------------------------------------------------------
760 # Integration — nonexistent branch
761 # ---------------------------------------------------------------------------
762
763
764 class TestNonexistentBranch:
765 def test_nonexistent_branch_contextual_message(self, tmp_path: pathlib.Path) -> None:
766 repo = _fresh_repo(tmp_path)
767 result = _log(repo, "bogus-branch")
768 assert "bogus-branch" in result.output
769
770 def test_nonexistent_branch_exits_zero(self, tmp_path: pathlib.Path) -> None:
771 """log on a nonexistent branch is not a fatal error."""
772 repo = _fresh_repo(tmp_path)
773 result = _log(repo, "bogus-branch")
774 assert result.exit_code == 0
775
776 def test_nonexistent_branch_json_empty_commits(self, tmp_path: pathlib.Path) -> None:
777 repo = _fresh_repo(tmp_path)
778 data = json.loads(_log(repo, "--json", "bogus-branch").output)
779 assert data["commits"] == []
780
781 def test_empty_repo_shows_no_commits(self, tmp_path: pathlib.Path) -> None:
782 repo = tmp_path / "repo"
783 _init(repo)
784 result = _log(repo)
785 assert "no commits" in result.output.lower()
786
787
788 # ---------------------------------------------------------------------------
789 # End-to-end — complete workflows
790 # ---------------------------------------------------------------------------
791
792
793 class TestEndToEnd:
794 def test_single_commit_log(self, tmp_path: pathlib.Path) -> None:
795 repo = _fresh_repo(tmp_path, n_commits=1)
796 result = _log(repo)
797 assert result.exit_code == 0
798 assert "commit" in result.output.lower()
799
800 def test_multiple_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
801 repo = _fresh_repo(tmp_path, n_commits=3)
802 result = _log(repo, "--oneline")
803 lines = [l for l in result.output.strip().splitlines() if l]
804 assert len(lines) == 3
805
806 def test_head_decoration_on_latest(self, tmp_path: pathlib.Path) -> None:
807 repo = _fresh_repo(tmp_path, n_commits=2)
808 result = _log(repo)
809 lines = result.output.strip().splitlines()
810 # First commit line should have HEAD
811 first = next((l for l in lines if "commit" in l.lower()), "")
812 assert "HEAD" in first
813
814 def test_subprocess_call_works(self, tmp_path: pathlib.Path) -> None:
815 repo = _fresh_repo(tmp_path, n_commits=2)
816 r = subprocess.run(
817 ["muse", "log", "--json"],
818 capture_output=True, text=True, cwd=str(repo),
819 )
820 assert r.returncode == 0
821 data = json.loads(r.stdout)
822 assert len(data["commits"]) == 2
823
824 def test_log_after_branch_switch(self, tmp_path: pathlib.Path) -> None:
825 from muse.cli.app import main as cli
826
827 repo = _fresh_repo(tmp_path, n_commits=2)
828 saved = os.getcwd()
829 os.chdir(repo)
830 try:
831 runner.invoke(cli, ["branch", "feat/x"])
832 runner.invoke(cli, ["checkout", "feat/x"])
833 finally:
834 os.chdir(saved)
835 _commit(repo, "feat commit", filename="feat.py")
836 data = json.loads(_log(repo, "--json").output)
837 # feat branch should have 3 commits (2 from main + 1 new)
838 assert len(data["commits"]) == 3
839
840 def test_log_on_explicit_branch(self, tmp_path: pathlib.Path) -> None:
841 from muse.cli.app import main as cli
842
843 repo = _fresh_repo(tmp_path, n_commits=2)
844 saved = os.getcwd()
845 os.chdir(repo)
846 try:
847 runner.invoke(cli, ["branch", "feat/y"])
848 runner.invoke(cli, ["checkout", "feat/y"])
849 finally:
850 os.chdir(saved)
851 _commit(repo, "only on feat", filename="feat_y.py")
852 # Log main explicitly — should not include feat commit
853 data_main = json.loads(_log(repo, "--json", "main").output)
854 messages = [c["message"] for c in data_main["commits"]]
855 assert "only on feat" not in messages
856
857 def test_merge_commit_has_parent2(self, tmp_path: pathlib.Path) -> None:
858 from muse.cli.app import main as cli
859
860 repo = _fresh_repo(tmp_path, n_commits=1)
861 saved = os.getcwd()
862 os.chdir(repo)
863 try:
864 runner.invoke(cli, ["branch", "feat/merge-test"])
865 runner.invoke(cli, ["checkout", "feat/merge-test"])
866 (repo / "feat_file.py").write_text("f=1\n")
867 runner.invoke(cli, ["commit", "-m", "feat commit"])
868 runner.invoke(cli, ["checkout", "main"])
869 (repo / "main_file.py").write_text("m=1\n")
870 runner.invoke(cli, ["commit", "-m", "main diverge"])
871 runner.invoke(cli, ["merge", "feat/merge-test"])
872 finally:
873 os.chdir(saved)
874
875 data = json.loads(_log(repo, "--json", "-n", "1").output)
876 merge_commit = data["commits"][0]
877 # A merge commit must have parent2_commit_id set
878 assert merge_commit["parent2_commit_id"] is not None
879
880
881 # ---------------------------------------------------------------------------
882 # Stress — large history and rapid calls
883 # ---------------------------------------------------------------------------
884
885
886 class TestStress:
887 @pytest.mark.slow
888 def test_log_200_commits_json(self, tmp_path: pathlib.Path) -> None:
889 """log --json on 200 commits must exit 0 with correct count."""
890 repo = _fresh_repo(tmp_path, n_commits=200)
891 result = _log(repo, "--json")
892 assert result.exit_code == 0
893 data = json.loads(result.output)
894 assert len(data["commits"]) == 200
895
896 @pytest.mark.slow
897 def test_log_200_commits_oneline(self, tmp_path: pathlib.Path) -> None:
898 repo = _fresh_repo(tmp_path, n_commits=200)
899 result = _log(repo, "--oneline")
900 assert result.exit_code == 0
901 lines = [l for l in result.output.splitlines() if l.strip()]
902 assert len(lines) == 200
903
904 @pytest.mark.slow
905 def test_rapid_sequential_calls(self, tmp_path: pathlib.Path) -> None:
906 """20 sequential muse log calls must all succeed."""
907 repo = _fresh_repo(tmp_path, n_commits=10)
908 for i in range(20):
909 result = _log(repo, "--json")
910 assert result.exit_code == 0, f"Call {i} failed"
911
912 def test_limit_n_large(self, tmp_path: pathlib.Path) -> None:
913 repo = _fresh_repo(tmp_path, n_commits=10)
914 data = json.loads(_log(repo, "--json", "-n", "5").output)
915 assert len(data["commits"]) == 5
916
917 def test_filter_returns_subset(self, tmp_path: pathlib.Path) -> None:
918 """Limiting to 5 commits from a 20-commit repo returns exactly 5."""
919 repo = _fresh_repo(tmp_path, n_commits=20)
920 data = json.loads(_log(repo, "--json", "-n", "5").output)
921 assert len(data["commits"]) == 5
922
923 def test_truncated_true_when_filter_skips_commits(self, tmp_path: pathlib.Path) -> None:
924 """With active filter + large walk cap, walk_truncated can be True.
925
926 Use --since=future so the filter skips all commits, but the walk still
927 fetches them all up to walk_cap. We exercise the truncated-when-filter
928 path by creating more commits than the walk ceiling.
929 """
930 repo = _fresh_repo(tmp_path, n_commits=10)
931 # Verify that --since=2099 returns an empty but valid JSON object.
932 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
933 assert data["commits"] == []
934 # truncated may or may not be True here depending on walk_cap;
935 # the key invariant is that the output is well-formed JSON.
936 assert isinstance(data["truncated"], bool)
937
938
939 # ===========================================================================
940 # Manifest cache — each commit's snapshot must be read at most once per run
941 # ===========================================================================
942
943
944 class TestManifestCache:
945 """get_commit_snapshot_manifest must not be called more than once per commit_id.
946
947 Before the fix, _commit_touches_path and _file_diff each called
948 get_commit_snapshot_manifest independently. With a pathspec filter plus
949 JSON output (which always runs _file_diff), the same commit_id was read 4×
950 per commit (current + parent in each function).
951
952 After the fix, a shared manifest_cache dict deduplicates reads so each
953 commit_id is read at most once regardless of how many callers need it.
954 """
955
956 def test_manifest_cache_used_structurally(self) -> None:
957 """manifest_cache dict must be threaded through the log pipeline."""
958 import inspect
959 from muse.cli.commands import log as log_module
960
961 source = inspect.getsource(log_module)
962 assert "manifest_cache" in source, (
963 "log.py must use a manifest_cache dict to deduplicate snapshot reads"
964 )
965
966 def test_each_commit_id_read_at_most_once(self, tmp_path: pathlib.Path) -> None:
967 """With pathspec + JSON mode, each commit's snapshot read ≤ 1×.
968
969 JSON mode always calls _file_diff (stat=True).
970 Pathspec filter calls _commit_touches_path.
971 Without a shared cache, the same manifest is loaded 4× per commit.
972 With a shared cache it is loaded exactly once.
973 """
974 from unittest.mock import patch, call
975 from muse.core import store as store_module
976
977 repo = tmp_path / "r"
978 _init(repo)
979 (repo / "src").mkdir(exist_ok=True)
980 # Create 3 commits each touching a distinct file.
981 for i in range(3):
982 _commit(repo, f"msg{i}", filename=f"src/file{i}.py")
983
984 seen_ids: list[str] = []
985
986 original_fn = store_module.get_commit_snapshot_manifest
987
988 def tracking_fn(root, commit_id):
989 seen_ids.append(commit_id)
990 return original_fn(root, commit_id)
991
992 with patch.object(store_module, "get_commit_snapshot_manifest", side_effect=tracking_fn):
993 result = _log(repo, "--json", "--", "src/")
994
995 assert result.exit_code == 0
996 data = json.loads(result.output)
997 assert len(data["commits"]) > 0
998
999 # Each commit_id must appear at most once in the call log.
1000 from collections import Counter
1001 counts = Counter(seen_ids)
1002 duplicates = {cid: n for cid, n in counts.items() if n > 1}
1003 assert not duplicates, (
1004 f"get_commit_snapshot_manifest called >1× for commit IDs: {duplicates}. "
1005 "Manifest cache not working."
1006 )
1007
1008 def test_pathspec_filter_correct_with_cache(self, tmp_path: pathlib.Path) -> None:
1009 """Pathspec filter returns correct commits when manifest cache is active."""
1010 repo = tmp_path / "r"
1011 _init(repo)
1012 _commit(repo, "add alpha", filename="alpha.py")
1013 _commit(repo, "add beta", filename="beta.py")
1014 _commit(repo, "add gamma", filename="gamma.py")
1015
1016 result = _log(repo, "--json", "--", "alpha.py")
1017 assert result.exit_code == 0
1018 data = json.loads(result.output)
1019 messages = [c["message"] for c in data["commits"]]
1020 assert any("alpha" in m for m in messages), (
1021 "alpha.py pathspec should include the 'add alpha' commit"
1022 )
1023 assert not any("beta" in m for m in messages), (
1024 "beta.py pathspec should NOT include the 'add beta' commit"
1025 )
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 149 days ago