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