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