gabriel / muse public
test_cmd_log.py python
806 lines 30.2 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 154 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 }
323 assert expected == set(d.keys())
324
325 def test_parent2_commit_id_is_none_for_linear(self) -> None:
326 from muse.cli.commands.log import _commit_to_json
327
328 c = self._make_commit()
329 d = _commit_to_json(c)
330 assert d["parent2_commit_id"] is None
331
332 def test_breaking_changes_is_always_list(self) -> None:
333 from muse.cli.commands.log import _commit_to_json
334
335 c = self._make_commit()
336 d = _commit_to_json(c)
337 assert isinstance(d["breaking_changes"], list)
338
339 def test_committed_at_is_iso_string(self) -> None:
340 from muse.cli.commands.log import _commit_to_json
341
342 c = self._make_commit()
343 d = _commit_to_json(c)
344 ts = d["committed_at"]
345 assert isinstance(ts, str)
346 assert "2025" in ts
347 assert "T" in ts or " " in ts
348
349
350 # ---------------------------------------------------------------------------
351 # Integration — JSON output schema
352 # ---------------------------------------------------------------------------
353
354
355 class TestJsonSchema:
356 _REQUIRED_COMMIT_KEYS = {
357 "commit_id", "branch", "message", "author", "committed_at",
358 "parent_commit_id", "parent2_commit_id", "snapshot_id",
359 "sem_ver_bump", "breaking_changes", "metadata",
360 }
361
362 def test_top_level_keys(self, tmp_path: pathlib.Path) -> None:
363 repo = _fresh_repo(tmp_path)
364 data = json.loads(_log(repo, "--json").output)
365 assert "commits" in data
366 assert "truncated" in data
367
368 def test_all_commit_keys_present(self, tmp_path: pathlib.Path) -> None:
369 repo = _fresh_repo(tmp_path, n_commits=2)
370 data = json.loads(_log(repo, "--json").output)
371 for c in data["commits"]:
372 missing = self._REQUIRED_COMMIT_KEYS - set(c.keys())
373 assert not missing, f"Missing keys: {missing}"
374
375 def test_parent2_commit_id_present(self, tmp_path: pathlib.Path) -> None:
376 repo = _fresh_repo(tmp_path)
377 data = json.loads(_log(repo, "--json").output)
378 assert "parent2_commit_id" in data["commits"][0]
379
380 def test_breaking_changes_is_list(self, tmp_path: pathlib.Path) -> None:
381 repo = _fresh_repo(tmp_path)
382 data = json.loads(_log(repo, "--json").output)
383 assert isinstance(data["commits"][0]["breaking_changes"], list)
384
385 def test_committed_at_is_iso(self, tmp_path: pathlib.Path) -> None:
386 repo = _fresh_repo(tmp_path)
387 data = json.loads(_log(repo, "--json").output)
388 ts = data["commits"][0]["committed_at"]
389 assert "T" in ts or "+" in ts
390
391 def test_truncated_false_by_default(self, tmp_path: pathlib.Path) -> None:
392 repo = _fresh_repo(tmp_path, n_commits=3)
393 data = json.loads(_log(repo, "--json").output)
394 assert data["truncated"] is False
395
396 def test_json_parseable_output(self, tmp_path: pathlib.Path) -> None:
397 repo = _fresh_repo(tmp_path, n_commits=5)
398 result = _log(repo, "--json")
399 data = json.loads(result.output)
400 assert isinstance(data["commits"], list)
401 assert len(data["commits"]) == 5
402
403 def test_empty_repo_json(self, tmp_path: pathlib.Path) -> None:
404 repo = tmp_path / "repo"
405 _init(repo)
406 result = _log(repo, "--json")
407 data = json.loads(result.output)
408 assert data["commits"] == []
409 assert data["truncated"] is False
410
411 def test_limit_n_json(self, tmp_path: pathlib.Path) -> None:
412 repo = _fresh_repo(tmp_path, n_commits=5)
413 data = json.loads(_log(repo, "--json", "-n", "2").output)
414 assert len(data["commits"]) == 2
415
416 def test_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
417 repo = _fresh_repo(tmp_path, n_commits=3)
418 data = json.loads(_log(repo, "--json").output)
419 timestamps = [c["committed_at"] for c in data["commits"]]
420 assert timestamps == sorted(timestamps, reverse=True)
421
422 def test_output_is_single_object(self, tmp_path: pathlib.Path) -> None:
423 """--json must produce one JSON object, not an array or newline-delimited."""
424 repo = _fresh_repo(tmp_path)
425 result = _log(repo, "--json")
426 # Must parse as a single dict
427 data = json.loads(result.output)
428 assert isinstance(data, dict)
429
430
431 # ---------------------------------------------------------------------------
432 # Integration — --oneline
433 # ---------------------------------------------------------------------------
434
435
436 class TestOneline:
437 def test_one_line_per_commit(self, tmp_path: pathlib.Path) -> None:
438 repo = _fresh_repo(tmp_path, n_commits=3)
439 result = _log(repo, "--oneline")
440 lines = [l for l in result.output.splitlines() if l.strip()]
441 assert len(lines) == 3
442
443 def test_short_hash_in_output(self, tmp_path: pathlib.Path) -> None:
444 repo = _fresh_repo(tmp_path)
445 data = json.loads(_log(repo, "--json").output)
446 commit_id = data["commits"][0]["commit_id"]
447 result = _log(repo, "--oneline")
448 assert commit_id[:8] in result.output
449
450 def test_message_on_same_line(self, tmp_path: pathlib.Path) -> None:
451 repo = _fresh_repo(tmp_path)
452 _commit(repo, "my special message", filename="z.py")
453 result = _log(repo, "--oneline", "-n", "1")
454 assert "my special message" in result.output
455 assert len(result.output.splitlines()) >= 1
456
457 def test_no_ansi_when_not_tty(self, tmp_path: pathlib.Path) -> None:
458 repo = _fresh_repo(tmp_path)
459 result = _log(repo, "--oneline")
460 # CLI runner is not a TTY — no escape sequences
461 assert "\x1b[" not in result.output
462
463
464 # ---------------------------------------------------------------------------
465 # Integration — --stat
466 # ---------------------------------------------------------------------------
467
468
469 class TestStat:
470 def test_stat_shows_added_files(self, tmp_path: pathlib.Path) -> None:
471 repo = _fresh_repo(tmp_path, n_commits=1)
472 result = _log(repo, "--stat")
473 assert "added" in result.output
474 assert "+" in result.output
475
476 def test_stat_shows_summary_line(self, tmp_path: pathlib.Path) -> None:
477 repo = _fresh_repo(tmp_path, n_commits=1)
478 result = _log(repo, "--stat")
479 assert "added" in result.output
480 assert "removed" in result.output
481
482 def test_stat_exit_zero(self, tmp_path: pathlib.Path) -> None:
483 repo = _fresh_repo(tmp_path)
484 result = _log(repo, "--stat")
485 assert result.exit_code == 0
486
487
488 # ---------------------------------------------------------------------------
489 # Integration — filters
490 # ---------------------------------------------------------------------------
491
492
493 class TestFilters:
494 def test_author_filter_matches(self, tmp_path: pathlib.Path) -> None:
495 repo = _fresh_repo(tmp_path, n_commits=2)
496 # The author will be whatever muse uses by default
497 # We just verify that filtering by nonexistent author returns none
498 result = _log(repo, "--author", "zzz_nobody_zzz")
499 assert "(no commits)" in result.output
500
501 def test_since_filters_old_commits(self, tmp_path: pathlib.Path) -> None:
502 repo = _fresh_repo(tmp_path, n_commits=2)
503 result = _log(repo, "--since", "2099-01-01")
504 # Future date — should return no commits
505 assert "(no commits)" in result.output
506
507 def test_until_filters_future_commits(self, tmp_path: pathlib.Path) -> None:
508 repo = _fresh_repo(tmp_path, n_commits=2)
509 # Past date — all commits should be excluded
510 result = _log(repo, "--until", "2000-01-01")
511 assert "(no commits)" in result.output
512
513 def test_limit_shorthand(self, tmp_path: pathlib.Path) -> None:
514 """muse log -2 must show at most 2 commits."""
515 repo = _fresh_repo(tmp_path, n_commits=5)
516 result = _log(repo, "--oneline", "-n", "2")
517 lines = [l for l in result.output.splitlines() if l.strip()]
518 assert len(lines) == 2
519
520 def test_json_since_filters(self, tmp_path: pathlib.Path) -> None:
521 repo = _fresh_repo(tmp_path, n_commits=2)
522 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
523 assert data["commits"] == []
524
525 def test_invalid_since_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
526 repo = _fresh_repo(tmp_path)
527 result = _log(repo, "--since", "not-a-date")
528 assert result.exit_code != 0
529
530 def test_invalid_until_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
531 repo = _fresh_repo(tmp_path)
532 result = _log(repo, "--until", "not-a-date")
533 assert result.exit_code != 0
534
535 def test_invalid_since_no_traceback(self, tmp_path: pathlib.Path) -> None:
536 repo = _fresh_repo(tmp_path)
537 result = _log(repo, "--since", "baddate")
538 assert "Traceback" not in result.output
539
540 def test_invalid_until_clean_error(self, tmp_path: pathlib.Path) -> None:
541 repo = _fresh_repo(tmp_path)
542 result = _log(repo, "--until", "foo")
543 assert "Cannot parse" in result.output or result.exit_code != 0
544
545
546 # ---------------------------------------------------------------------------
547 # Integration — format validation
548 # ---------------------------------------------------------------------------
549
550
551 class TestFormatValidation:
552 def test_invalid_format_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
553 repo = _fresh_repo(tmp_path)
554 result = _log(repo, "--format", "xml")
555 assert result.exit_code != 0
556
557 def test_invalid_format_no_traceback(self, tmp_path: pathlib.Path) -> None:
558 repo = _fresh_repo(tmp_path)
559 result = _log(repo, "--format", "yaml")
560 assert "Traceback" not in result.output
561
562 def test_json_format_alias(self, tmp_path: pathlib.Path) -> None:
563 repo = _fresh_repo(tmp_path, n_commits=2)
564 r1 = _log(repo, "--json")
565 r2 = _log(repo, "--format", "json")
566 assert json.loads(r1.output) == json.loads(r2.output)
567
568 def test_invalid_max_count_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
569 repo = _fresh_repo(tmp_path)
570 result = _log(repo, "-n", "0")
571 assert result.exit_code != 0
572
573
574 # ---------------------------------------------------------------------------
575 # Security — ANSI injection
576 # ---------------------------------------------------------------------------
577
578
579 class TestSecurity:
580 def test_ansi_in_commit_message_sanitized_oneline(self, tmp_path: pathlib.Path) -> None:
581 repo = tmp_path / "repo"
582 _init(repo)
583 # Commit a message with ANSI in it
584 _commit(repo, "\x1b[31mevil\x1b[0m", filename="evil.py")
585 result = _log(repo, "--oneline")
586 # The runner is not a tty — any escape from the message must be sanitized
587 assert "\x1b[31m" not in result.output
588
589 def test_ansi_in_commit_message_sanitized_long(self, tmp_path: pathlib.Path) -> None:
590 repo = tmp_path / "repo"
591 _init(repo)
592 _commit(repo, "\x1b[31mhacked\x1b[0m", filename="h.py")
593 result = _log(repo)
594 assert "\x1b[31m" not in result.output
595
596 def test_ansi_in_author_sanitized(self, tmp_path: pathlib.Path) -> None:
597 """Author names from CommitRecord must be sanitized in output."""
598 repo = _fresh_repo(tmp_path)
599 result = _log(repo)
600 # No raw escape from author field in text output (we can't control
601 # author easily, but ensure output is escape-free when not tty)
602 assert "\x1b[31m" not in result.output
603
604 def test_multiline_message_all_lines_indented(self, tmp_path: pathlib.Path) -> None:
605 """Every line of a multiline message must start with 4-space indent."""
606 repo = tmp_path / "repo"
607 _init(repo)
608 _commit(repo, "Line1\nLine2\nLine3", filename="f.py")
609 result = _log(repo)
610 body_lines = [l for l in result.output.splitlines() if l.strip() in ("Line1", "Line2", "Line3")]
611 assert body_lines, f"Body lines not found in: {result.output}"
612 for line in body_lines:
613 assert line.startswith(" "), f"Not indented: {repr(line)}"
614
615 def test_invalid_fmt_sanitized_in_error(self, tmp_path: pathlib.Path) -> None:
616 repo = _fresh_repo(tmp_path)
617 evil_fmt = "\x1b[31mevil\x1b[0m"
618 result = _log(repo, "--format", evil_fmt)
619 assert result.exit_code != 0
620 assert "\x1b[31m" not in result.output
621
622 def test_no_repo_id_in_json_output(self, tmp_path: pathlib.Path) -> None:
623 repo = _fresh_repo(tmp_path)
624 stored = json.loads((repo / ".muse" / "repo.json").read_text())["repo_id"]
625 result = _log(repo, "--json")
626 assert stored not in result.output
627
628
629 # ---------------------------------------------------------------------------
630 # Integration — nonexistent branch
631 # ---------------------------------------------------------------------------
632
633
634 class TestNonexistentBranch:
635 def test_nonexistent_branch_contextual_message(self, tmp_path: pathlib.Path) -> None:
636 repo = _fresh_repo(tmp_path)
637 result = _log(repo, "bogus-branch")
638 assert "bogus-branch" in result.output
639
640 def test_nonexistent_branch_exits_zero(self, tmp_path: pathlib.Path) -> None:
641 """log on a nonexistent branch is not a fatal error."""
642 repo = _fresh_repo(tmp_path)
643 result = _log(repo, "bogus-branch")
644 assert result.exit_code == 0
645
646 def test_nonexistent_branch_json_empty_commits(self, tmp_path: pathlib.Path) -> None:
647 repo = _fresh_repo(tmp_path)
648 data = json.loads(_log(repo, "--json", "bogus-branch").output)
649 assert data["commits"] == []
650
651 def test_empty_repo_shows_no_commits(self, tmp_path: pathlib.Path) -> None:
652 repo = tmp_path / "repo"
653 _init(repo)
654 result = _log(repo)
655 assert "no commits" in result.output.lower()
656
657
658 # ---------------------------------------------------------------------------
659 # End-to-end — complete workflows
660 # ---------------------------------------------------------------------------
661
662
663 class TestEndToEnd:
664 def test_single_commit_log(self, tmp_path: pathlib.Path) -> None:
665 repo = _fresh_repo(tmp_path, n_commits=1)
666 result = _log(repo)
667 assert result.exit_code == 0
668 assert "commit" in result.output.lower()
669
670 def test_multiple_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
671 repo = _fresh_repo(tmp_path, n_commits=3)
672 result = _log(repo, "--oneline")
673 lines = [l for l in result.output.strip().splitlines() if l]
674 assert len(lines) == 3
675
676 def test_head_decoration_on_latest(self, tmp_path: pathlib.Path) -> None:
677 repo = _fresh_repo(tmp_path, n_commits=2)
678 result = _log(repo)
679 lines = result.output.strip().splitlines()
680 # First commit line should have HEAD
681 first = next((l for l in lines if "commit" in l.lower()), "")
682 assert "HEAD" in first
683
684 def test_subprocess_call_works(self, tmp_path: pathlib.Path) -> None:
685 repo = _fresh_repo(tmp_path, n_commits=2)
686 r = subprocess.run(
687 ["muse", "log", "--json"],
688 capture_output=True, text=True, cwd=str(repo),
689 )
690 assert r.returncode == 0
691 data = json.loads(r.stdout)
692 assert len(data["commits"]) == 2
693
694 def test_log_after_branch_switch(self, tmp_path: pathlib.Path) -> None:
695 from muse.cli.app import main as cli
696
697 repo = _fresh_repo(tmp_path, n_commits=2)
698 saved = os.getcwd()
699 os.chdir(repo)
700 try:
701 runner.invoke(cli, ["branch", "feat/x"])
702 runner.invoke(cli, ["checkout", "feat/x"])
703 finally:
704 os.chdir(saved)
705 _commit(repo, "feat commit", filename="feat.py")
706 data = json.loads(_log(repo, "--json").output)
707 # feat branch should have 3 commits (2 from main + 1 new)
708 assert len(data["commits"]) == 3
709
710 def test_log_on_explicit_branch(self, tmp_path: pathlib.Path) -> None:
711 from muse.cli.app import main as cli
712
713 repo = _fresh_repo(tmp_path, n_commits=2)
714 saved = os.getcwd()
715 os.chdir(repo)
716 try:
717 runner.invoke(cli, ["branch", "feat/y"])
718 runner.invoke(cli, ["checkout", "feat/y"])
719 finally:
720 os.chdir(saved)
721 _commit(repo, "only on feat", filename="feat_y.py")
722 # Log main explicitly — should not include feat commit
723 data_main = json.loads(_log(repo, "--json", "main").output)
724 messages = [c["message"] for c in data_main["commits"]]
725 assert "only on feat" not in messages
726
727 def test_merge_commit_has_parent2(self, tmp_path: pathlib.Path) -> None:
728 from muse.cli.app import main as cli
729
730 repo = _fresh_repo(tmp_path, n_commits=1)
731 saved = os.getcwd()
732 os.chdir(repo)
733 try:
734 runner.invoke(cli, ["branch", "feat/merge-test"])
735 runner.invoke(cli, ["checkout", "feat/merge-test"])
736 (repo / "feat_file.py").write_text("f=1\n")
737 runner.invoke(cli, ["commit", "-m", "feat commit"])
738 runner.invoke(cli, ["checkout", "main"])
739 (repo / "main_file.py").write_text("m=1\n")
740 runner.invoke(cli, ["commit", "-m", "main diverge"])
741 runner.invoke(cli, ["merge", "feat/merge-test"])
742 finally:
743 os.chdir(saved)
744
745 data = json.loads(_log(repo, "--json", "-n", "1").output)
746 merge_commit = data["commits"][0]
747 # A merge commit must have parent2_commit_id set
748 assert merge_commit["parent2_commit_id"] is not None
749
750
751 # ---------------------------------------------------------------------------
752 # Stress — large history and rapid calls
753 # ---------------------------------------------------------------------------
754
755
756 class TestStress:
757 @pytest.mark.slow
758 def test_log_200_commits_json(self, tmp_path: pathlib.Path) -> None:
759 """log --json on 200 commits must exit 0 with correct count."""
760 repo = _fresh_repo(tmp_path, n_commits=200)
761 result = _log(repo, "--json")
762 assert result.exit_code == 0
763 data = json.loads(result.output)
764 assert len(data["commits"]) == 200
765
766 @pytest.mark.slow
767 def test_log_200_commits_oneline(self, tmp_path: pathlib.Path) -> None:
768 repo = _fresh_repo(tmp_path, n_commits=200)
769 result = _log(repo, "--oneline")
770 assert result.exit_code == 0
771 lines = [l for l in result.output.splitlines() if l.strip()]
772 assert len(lines) == 200
773
774 @pytest.mark.slow
775 def test_rapid_sequential_calls(self, tmp_path: pathlib.Path) -> None:
776 """20 sequential muse log calls must all succeed."""
777 repo = _fresh_repo(tmp_path, n_commits=10)
778 for i in range(20):
779 result = _log(repo, "--json")
780 assert result.exit_code == 0, f"Call {i} failed"
781
782 def test_limit_n_large(self, tmp_path: pathlib.Path) -> None:
783 repo = _fresh_repo(tmp_path, n_commits=10)
784 data = json.loads(_log(repo, "--json", "-n", "5").output)
785 assert len(data["commits"]) == 5
786
787 def test_filter_returns_subset(self, tmp_path: pathlib.Path) -> None:
788 """Limiting to 5 commits from a 20-commit repo returns exactly 5."""
789 repo = _fresh_repo(tmp_path, n_commits=20)
790 data = json.loads(_log(repo, "--json", "-n", "5").output)
791 assert len(data["commits"]) == 5
792
793 def test_truncated_true_when_filter_skips_commits(self, tmp_path: pathlib.Path) -> None:
794 """With active filter + large walk cap, walk_truncated can be True.
795
796 Use --since=future so the filter skips all commits, but the walk still
797 fetches them all up to walk_cap. We exercise the truncated-when-filter
798 path by creating more commits than the walk ceiling.
799 """
800 repo = _fresh_repo(tmp_path, n_commits=10)
801 # Verify that --since=2099 returns an empty but valid JSON object.
802 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
803 assert data["commits"] == []
804 # truncated may or may not be True here depending on walk_cap;
805 # the key invariant is that the output is well-formed JSON.
806 assert isinstance(data["truncated"], bool)
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 154 days ago