gabriel / muse public
test_cmd_log.py python
1,076 lines 41.5 KB
Raw
sha256:144495ae90498f9d51aeac4bf7f9ee9e9502d14474745a856b96921f7349c0ce next round of fixing tests Human 105 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 "signer_public_key",
354 }
355 assert expected == set(d.keys())
356
357 def test_file_lists_empty_without_stat(self) -> None:
358 from muse.cli.commands.log import _commit_to_json
359
360 c = self._make_commit()
361 d = _commit_to_json(c)
362 assert d["files_added"] == []
363 assert d["files_removed"] == []
364 assert d["files_modified"] == []
365
366 def test_parent2_commit_id_is_none_for_linear(self) -> None:
367 from muse.cli.commands.log import _commit_to_json
368
369 c = self._make_commit()
370 d = _commit_to_json(c)
371 assert d["parent2_commit_id"] is None
372
373 def test_breaking_changes_is_always_list(self) -> None:
374 from muse.cli.commands.log import _commit_to_json
375
376 c = self._make_commit()
377 d = _commit_to_json(c)
378 assert isinstance(d["breaking_changes"], list)
379
380 def test_committed_at_is_iso_string(self) -> None:
381 from muse.cli.commands.log import _commit_to_json
382
383 c = self._make_commit()
384 d = _commit_to_json(c)
385 ts = d["committed_at"]
386 assert isinstance(ts, str)
387 assert "2025" in ts
388 assert "T" in ts or " " in ts
389
390
391 # ---------------------------------------------------------------------------
392 # Integration — JSON output schema
393 # ---------------------------------------------------------------------------
394
395
396 class TestJsonSchema:
397 _REQUIRED_COMMIT_KEYS = {
398 "commit_id", "branch", "message", "author", "committed_at",
399 "parent_commit_id", "parent2_commit_id", "snapshot_id",
400 "sem_ver_bump", "breaking_changes", "metadata",
401 "files_added", "files_removed", "files_modified",
402 }
403
404 def test_top_level_keys(self, tmp_path: pathlib.Path) -> None:
405 repo = _fresh_repo(tmp_path)
406 data = json.loads(_log(repo, "--json").output)
407 assert "commits" in data
408 assert "truncated" in data
409
410 def test_all_commit_keys_present(self, tmp_path: pathlib.Path) -> None:
411 repo = _fresh_repo(tmp_path, n_commits=2)
412 data = json.loads(_log(repo, "--json").output)
413 for c in data["commits"]:
414 missing = self._REQUIRED_COMMIT_KEYS - set(c.keys())
415 assert not missing, f"Missing keys: {missing}"
416
417 def test_parent2_commit_id_present(self, tmp_path: pathlib.Path) -> None:
418 repo = _fresh_repo(tmp_path)
419 data = json.loads(_log(repo, "--json").output)
420 assert "parent2_commit_id" in data["commits"][0]
421
422 def test_breaking_changes_is_list(self, tmp_path: pathlib.Path) -> None:
423 repo = _fresh_repo(tmp_path)
424 data = json.loads(_log(repo, "--json").output)
425 assert isinstance(data["commits"][0]["breaking_changes"], list)
426
427 def test_committed_at_is_iso(self, tmp_path: pathlib.Path) -> None:
428 repo = _fresh_repo(tmp_path)
429 data = json.loads(_log(repo, "--json").output)
430 ts = data["commits"][0]["committed_at"]
431 assert "T" in ts or "+" in ts
432
433 def test_truncated_false_by_default(self, tmp_path: pathlib.Path) -> None:
434 repo = _fresh_repo(tmp_path, n_commits=3)
435 data = json.loads(_log(repo, "--json").output)
436 assert data["truncated"] is False
437
438 def test_json_parseable_output(self, tmp_path: pathlib.Path) -> None:
439 repo = _fresh_repo(tmp_path, n_commits=5)
440 result = _log(repo, "--json")
441 data = json.loads(result.output)
442 assert isinstance(data["commits"], list)
443 assert len(data["commits"]) == 5
444
445 def test_empty_repo_json(self, tmp_path: pathlib.Path) -> None:
446 repo = tmp_path / "repo"
447 _init(repo)
448 result = _log(repo, "--json")
449 data = json.loads(result.output)
450 assert data["commits"] == []
451 assert data["truncated"] is False
452
453 def test_limit_n_json(self, tmp_path: pathlib.Path) -> None:
454 repo = _fresh_repo(tmp_path, n_commits=5)
455 data = json.loads(_log(repo, "--json", "--limit", "2").output)
456 assert len(data["commits"]) == 2
457
458 def test_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
459 repo = _fresh_repo(tmp_path, n_commits=3)
460 data = json.loads(_log(repo, "--json").output)
461 timestamps = [c["committed_at"] for c in data["commits"]]
462 assert timestamps == sorted(timestamps, reverse=True)
463
464 def test_output_is_single_object(self, tmp_path: pathlib.Path) -> None:
465 """--json must produce one JSON object, not an array or newline-delimited."""
466 repo = _fresh_repo(tmp_path)
467 result = _log(repo, "--json")
468 # Must parse as a single dict
469 data = json.loads(result.output)
470 assert isinstance(data, dict)
471
472
473 # ---------------------------------------------------------------------------
474 # Integration — --oneline
475 # ---------------------------------------------------------------------------
476
477
478 class TestOneline:
479 def test_one_line_per_commit(self, tmp_path: pathlib.Path) -> None:
480 repo = _fresh_repo(tmp_path, n_commits=3)
481 result = _log(repo, "--oneline")
482 lines = [l for l in result.output.splitlines() if l.strip()]
483 assert len(lines) == 3
484
485 def test_short_hash_in_output(self, tmp_path: pathlib.Path) -> None:
486 repo = _fresh_repo(tmp_path)
487 data = json.loads(_log(repo, "--json").output)
488 commit_id = data["commits"][0]["commit_id"]
489 result = _log(repo, "--oneline")
490 assert commit_id[:8] in result.output
491
492 def test_message_on_same_line(self, tmp_path: pathlib.Path) -> None:
493 repo = _fresh_repo(tmp_path)
494 _commit(repo, "my special message", filename="z.py")
495 result = _log(repo, "--oneline", "--limit", "1")
496 assert "my special message" in result.output
497 assert len(result.output.splitlines()) >= 1
498
499 def test_no_ansi_when_not_tty(self, tmp_path: pathlib.Path) -> None:
500 repo = _fresh_repo(tmp_path)
501 result = _log(repo, "--oneline")
502 # CLI runner is not a TTY — no escape sequences
503 assert "\x1b[" not in result.output
504
505
506 # ---------------------------------------------------------------------------
507 # Integration — --stat
508 # ---------------------------------------------------------------------------
509
510
511 class TestStat:
512 def test_stat_shows_added_files(self, tmp_path: pathlib.Path) -> None:
513 repo = _fresh_repo(tmp_path, n_commits=1)
514 result = _log(repo, "--stat")
515 assert "added" in result.output
516 assert "+" in result.output
517
518 def test_stat_shows_summary_line(self, tmp_path: pathlib.Path) -> None:
519 repo = _fresh_repo(tmp_path, n_commits=1)
520 result = _log(repo, "--stat")
521 assert "added" in result.output
522 assert "removed" in result.output
523
524 def test_stat_shows_modified_marker(self, tmp_path: pathlib.Path) -> None:
525 repo = _fresh_repo(tmp_path, n_commits=1)
526 # Modify the same file in a second commit so "modified" fires.
527 (repo / "file_0.py").write_text("# changed\n")
528 _commit(repo, "modify existing")
529 result = _log(repo, "--stat", "--limit", "1")
530 assert "~" in result.output
531 assert "modified" in result.output
532
533 def test_stat_exit_zero(self, tmp_path: pathlib.Path) -> None:
534 repo = _fresh_repo(tmp_path)
535 result = _log(repo, "--stat")
536 assert result.exit_code == 0
537
538 def test_stat_json_file_lists_populated(self, tmp_path: pathlib.Path) -> None:
539 repo = _fresh_repo(tmp_path, n_commits=1)
540 data = json.loads(_log(repo, "--stat", "--json").output)
541 commit = data["commits"][0]
542 # The initial commit adds at least one file.
543 assert isinstance(commit["files_added"], list)
544 assert isinstance(commit["files_removed"], list)
545 assert isinstance(commit["files_modified"], list)
546 assert len(commit["files_added"]) > 0
547
548 def test_stat_json_modified_populated(self, tmp_path: pathlib.Path) -> None:
549 repo = _fresh_repo(tmp_path, n_commits=1)
550 # Overwrite the existing file so the second commit shows a modification.
551 (repo / "file_0.py").write_text("# changed\n")
552 _commit(repo, "modify existing")
553 data = json.loads(_log(repo, "--stat", "--json", "--limit", "1").output)
554 commit = data["commits"][0]
555 assert "file_0.py" in commit["files_modified"]
556
557 def test_json_file_lists_populated_without_stat_flag(self, tmp_path: pathlib.Path) -> None:
558 """--json always populates file lists — agents must not need --stat."""
559 repo = _fresh_repo(tmp_path, n_commits=1)
560 data = json.loads(_log(repo, "--json").output)
561 commit = data["commits"][0]
562 # The initial commit adds at least one file; file lists must be
563 # populated even without the --stat flag.
564 assert isinstance(commit["files_added"], list)
565 assert isinstance(commit["files_removed"], list)
566 assert isinstance(commit["files_modified"], list)
567 assert len(commit["files_added"]) > 0
568
569
570 # ---------------------------------------------------------------------------
571 # Integration — filters
572 # ---------------------------------------------------------------------------
573
574
575 def _commit_as(repo: pathlib.Path, msg: str, author: str, filename: str | None = None) -> None:
576 """Invoke muse commit with an explicit --author flag."""
577 from muse.cli.app import main as cli
578 fname = filename or f"file_{abs(hash(msg))}.py"
579 (repo / fname).write_text(f"# {msg}\n")
580 saved = os.getcwd()
581 try:
582 os.chdir(repo)
583 runner.invoke(cli, ["commit", "-m", msg, "--author", author])
584 finally:
585 os.chdir(saved)
586
587
588 def _commit_with_identity_author(repo: pathlib.Path, msg: str, author: str, hub_url: str, filename: str | None = None) -> None:
589 """Seed identity.toml with a handle, then invoke muse commit without --author."""
590 from muse.cli.app import main as cli
591 from unittest.mock import patch
592 import pathlib
593 identity_file = repo / "identity.toml"
594 hostname = hub_url.split("://", 1)[-1].rstrip("/")
595 identity_file.write_text(
596 f'["{hostname}"]\ntype = "human"\nhandle = "{author}"\n'
597 f'algorithm = "ed25519"\nfingerprint = "sha256:abc"\nhd_path = "m/0\'"\n',
598 encoding="utf-8",
599 )
600 from muse.cli.config import set_hub_url
601 set_hub_url(hub_url, repo)
602 fname = filename or f"file_{abs(hash(msg))}.py"
603 (repo / fname).write_text(f"# {msg}\n")
604 saved = os.getcwd()
605 try:
606 os.chdir(repo)
607 with patch("muse.core.identity._IDENTITY_FILE", identity_file):
608 runner.invoke(cli, ["commit", "-m", msg])
609 finally:
610 os.chdir(saved)
611
612
613 class TestAuthorField:
614 """Author field in log JSON must come from identity.toml when --author not given."""
615
616 def test_commit_with_explicit_author_appears_in_log(self, tmp_path: pathlib.Path) -> None:
617 """--author flag sets author field that muse log --json exposes."""
618 repo = _fresh_repo(tmp_path, n_commits=0)
619 _commit_as(repo, "my commit", "charlie")
620 result = _log(repo, "--json")
621 data = json.loads(result.output)
622 assert data["commits"][0]["author"] == "charlie"
623
624 def test_commit_reads_user_name_from_identity(self, tmp_path: pathlib.Path) -> None:
625 """muse commit without --author reads user.handle from identity.toml."""
626 repo = _fresh_repo(tmp_path, n_commits=0)
627 _commit_with_identity_author(repo, "identity commit", "diana", "https://localhost:1337")
628 result = _log(repo, "--json")
629 data = json.loads(result.output)
630 assert data["commits"][0]["author"] == "diana"
631
632 def test_author_filter_returns_matching_commits(self, tmp_path: pathlib.Path) -> None:
633 """--author filter must return commits whose author matches the substring."""
634 repo = _fresh_repo(tmp_path, n_commits=0)
635 _commit_as(repo, "alice commit", "alice", filename="a.py")
636 _commit_as(repo, "bob commit", "bob", filename="b.py")
637 result = _log(repo, "--author", "alice", "--json")
638 data = json.loads(result.output)
639 assert len(data["commits"]) == 1
640 assert data["commits"][0]["author"] == "alice"
641
642 def test_author_filter_nonexistent_returns_no_commits(self, tmp_path: pathlib.Path) -> None:
643 """--author filter with no match must return empty list."""
644 repo = _fresh_repo(tmp_path, n_commits=0)
645 _commit_as(repo, "some commit", "alice", filename="a.py")
646 result = _log(repo, "--author", "zzz_nobody_zzz", "--json")
647 data = json.loads(result.output)
648 assert data["commits"] == []
649
650
651 class TestFilters:
652 def test_author_filter_matches(self, tmp_path: pathlib.Path) -> None:
653 repo = _fresh_repo(tmp_path, n_commits=2)
654 # The author will be whatever muse uses by default
655 # We just verify that filtering by nonexistent author returns none
656 result = _log(repo, "--author", "zzz_nobody_zzz")
657 assert "(no commits)" in result.output
658
659 def test_since_filters_old_commits(self, tmp_path: pathlib.Path) -> None:
660 repo = _fresh_repo(tmp_path, n_commits=2)
661 result = _log(repo, "--since", "2099-01-01")
662 # Future date — should return no commits
663 assert "(no commits)" in result.output
664
665 def test_until_filters_future_commits(self, tmp_path: pathlib.Path) -> None:
666 repo = _fresh_repo(tmp_path, n_commits=2)
667 # Past date — all commits should be excluded
668 result = _log(repo, "--until", "2000-01-01")
669 assert "(no commits)" in result.output
670
671 def test_limit_shorthand(self, tmp_path: pathlib.Path) -> None:
672 """muse log -2 must show at most 2 commits."""
673 repo = _fresh_repo(tmp_path, n_commits=5)
674 result = _log(repo, "--oneline", "--limit", "2")
675 lines = [l for l in result.output.splitlines() if l.strip()]
676 assert len(lines) == 2
677
678 def test_limit_flag_alias(self, tmp_path: pathlib.Path) -> None:
679 """--limit is an alias for -n/--max-count."""
680 repo = _fresh_repo(tmp_path, n_commits=5)
681 result = _log(repo, "--oneline", "--limit", "3")
682 lines = [l for l in result.output.splitlines() if l.strip()]
683 assert len(lines) == 3
684
685 def test_limit_flag_json(self, tmp_path: pathlib.Path) -> None:
686 """--limit works with --json output."""
687 repo = _fresh_repo(tmp_path, n_commits=5)
688 data = json.loads(_log(repo, "--json", "--limit", "2").output)
689 assert len(data["commits"]) == 2
690
691 def test_json_since_filters(self, tmp_path: pathlib.Path) -> None:
692 repo = _fresh_repo(tmp_path, n_commits=2)
693 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
694 assert data["commits"] == []
695
696 def test_invalid_since_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
697 repo = _fresh_repo(tmp_path)
698 result = _log(repo, "--since", "not-a-date")
699 assert result.exit_code != 0
700
701 def test_invalid_until_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
702 repo = _fresh_repo(tmp_path)
703 result = _log(repo, "--until", "not-a-date")
704 assert result.exit_code != 0
705
706 def test_invalid_since_no_traceback(self, tmp_path: pathlib.Path) -> None:
707 repo = _fresh_repo(tmp_path)
708 result = _log(repo, "--since", "baddate")
709 assert "Traceback" not in result.output
710
711 def test_invalid_until_clean_error(self, tmp_path: pathlib.Path) -> None:
712 repo = _fresh_repo(tmp_path)
713 result = _log(repo, "--until", "foo")
714 assert "Cannot parse" in result.output or result.exit_code != 0
715
716
717 # ---------------------------------------------------------------------------
718 # Integration — format validation
719 # ---------------------------------------------------------------------------
720
721
722 class TestFormatValidation:
723 def test_unknown_flag_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
724 repo = _fresh_repo(tmp_path)
725 result = _log(repo, "--format", "xml")
726 assert result.exit_code != 0
727
728 def test_unknown_flag_no_traceback(self, tmp_path: pathlib.Path) -> None:
729 repo = _fresh_repo(tmp_path)
730 result = _log(repo, "--format", "yaml")
731 assert "Traceback" not in result.output
732
733 def test_j_shorthand_same_as_json_flag(self, tmp_path: pathlib.Path) -> None:
734 repo = _fresh_repo(tmp_path, n_commits=2)
735 r1 = _log(repo, "--json")
736 r2 = _log(repo, "-j")
737 d1 = json.loads(r1.output)
738 d2 = json.loads(r2.output)
739 # duration_ms and timestamp are wall-clock values — exclude them
740 for d in (d1, d2):
741 d.pop("duration_ms", None)
742 d.pop("timestamp", None)
743 for c in d.get("commits", []):
744 c.pop("duration_ms", None)
745 assert d1 == d2
746
747 def test_invalid_max_count_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
748 repo = _fresh_repo(tmp_path)
749 result = _log(repo, "--limit", "0")
750 assert result.exit_code != 0
751
752
753 # ---------------------------------------------------------------------------
754 # Security — ANSI injection
755 # ---------------------------------------------------------------------------
756
757
758 class TestSecurity:
759 def test_ansi_in_commit_message_sanitized_oneline(self, tmp_path: pathlib.Path) -> None:
760 repo = tmp_path / "repo"
761 _init(repo)
762 # Commit a message with ANSI in it
763 _commit(repo, "\x1b[31mmalicious\x1b[0m", filename="malicious.py")
764 result = _log(repo, "--oneline")
765 # The runner is not a tty — any escape from the message must be sanitized
766 assert "\x1b[31m" not in result.output
767
768 def test_ansi_in_commit_message_sanitized_long(self, tmp_path: pathlib.Path) -> None:
769 repo = tmp_path / "repo"
770 _init(repo)
771 _commit(repo, "\x1b[31mhacked\x1b[0m", filename="h.py")
772 result = _log(repo)
773 assert "\x1b[31m" not in result.output
774
775 def test_ansi_in_author_sanitized(self, tmp_path: pathlib.Path) -> None:
776 """Author names from CommitRecord must be sanitized in output."""
777 repo = _fresh_repo(tmp_path)
778 result = _log(repo)
779 # No raw escape from author field in text output (we can't control
780 # author easily, but ensure output is escape-free when not tty)
781 assert "\x1b[31m" not in result.output
782
783 def test_multiline_message_all_lines_indented(self, tmp_path: pathlib.Path) -> None:
784 """Every line of a multiline message must start with 4-space indent."""
785 repo = tmp_path / "repo"
786 _init(repo)
787 _commit(repo, "Line1\nLine2\nLine3", filename="f.py")
788 result = _log(repo)
789 body_lines = [l for l in result.output.splitlines() if l.strip() in ("Line1", "Line2", "Line3")]
790 assert body_lines, f"Body lines not found in: {result.output}"
791 for line in body_lines:
792 assert line.startswith(" "), f"Not indented: {repr(line)}"
793
794 def test_unknown_flag_exits_nonzero_ansi(self, tmp_path: pathlib.Path) -> None:
795 repo = _fresh_repo(tmp_path)
796 malicious_fmt = "\x1b[31mmalicious\x1b[0m"
797 result = _log(repo, "--format", malicious_fmt)
798 assert result.exit_code != 0
799
800 def test_repo_id_in_json_envelope(self, tmp_path: pathlib.Path) -> None:
801 """repo_id is included in the JSON envelope for agent cross-referencing."""
802 repo = _fresh_repo(tmp_path)
803 stored = json.loads((repo_json_path(repo)).read_text())["repo_id"]
804 result = _log(repo, "--json")
805 data = json.loads(result.output)
806 assert data["repo_id"] == stored
807
808
809 # ---------------------------------------------------------------------------
810 # Integration — nonexistent branch
811 # ---------------------------------------------------------------------------
812
813
814 class TestNonexistentBranch:
815 def test_nonexistent_branch_contextual_message(self, tmp_path: pathlib.Path) -> None:
816 repo = _fresh_repo(tmp_path)
817 result = _log(repo, "bogus-branch")
818 assert "bogus-branch" in result.output
819
820 def test_nonexistent_branch_exits_zero(self, tmp_path: pathlib.Path) -> None:
821 """log on a nonexistent branch is not a fatal error."""
822 repo = _fresh_repo(tmp_path)
823 result = _log(repo, "bogus-branch")
824 assert result.exit_code == 0
825
826 def test_nonexistent_branch_json_empty_commits(self, tmp_path: pathlib.Path) -> None:
827 repo = _fresh_repo(tmp_path)
828 data = json.loads(_log(repo, "--json", "bogus-branch").output)
829 assert data["commits"] == []
830
831 def test_empty_repo_shows_no_commits(self, tmp_path: pathlib.Path) -> None:
832 repo = tmp_path / "repo"
833 _init(repo)
834 result = _log(repo)
835 assert "no commits" in result.output.lower()
836
837
838 # ---------------------------------------------------------------------------
839 # End-to-end — complete workflows
840 # ---------------------------------------------------------------------------
841
842
843 class TestEndToEnd:
844 def test_single_commit_log(self, tmp_path: pathlib.Path) -> None:
845 repo = _fresh_repo(tmp_path, n_commits=1)
846 result = _log(repo)
847 assert result.exit_code == 0
848 assert "commit" in result.output.lower()
849
850 def test_multiple_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
851 repo = _fresh_repo(tmp_path, n_commits=3)
852 result = _log(repo, "--oneline")
853 lines = [l for l in result.output.strip().splitlines() if l]
854 assert len(lines) == 3
855
856 def test_head_decoration_on_latest(self, tmp_path: pathlib.Path) -> None:
857 repo = _fresh_repo(tmp_path, n_commits=2)
858 result = _log(repo)
859 lines = result.output.strip().splitlines()
860 # First commit line should have HEAD
861 first = next((l for l in lines if "commit" in l.lower()), "")
862 assert "HEAD" in first
863
864 def test_subprocess_call_works(self, tmp_path: pathlib.Path) -> None:
865 repo = _fresh_repo(tmp_path, n_commits=2)
866 r = subprocess.run(
867 ["muse", "log", "--json"],
868 capture_output=True, text=True, cwd=str(repo),
869 )
870 assert r.returncode == 0
871 data = json.loads(r.stdout)
872 assert len(data["commits"]) == 2
873
874 def test_log_after_branch_switch(self, tmp_path: pathlib.Path) -> None:
875 from muse.cli.app import main as cli
876
877 repo = _fresh_repo(tmp_path, n_commits=2)
878 saved = os.getcwd()
879 os.chdir(repo)
880 try:
881 runner.invoke(cli, ["branch", "feat/x"])
882 runner.invoke(cli, ["checkout", "feat/x"])
883 finally:
884 os.chdir(saved)
885 _commit(repo, "feat commit", filename="feat.py")
886 data = json.loads(_log(repo, "--json").output)
887 # feat branch should have 3 commits (2 from main + 1 new)
888 assert len(data["commits"]) == 3
889
890 def test_log_on_explicit_branch(self, tmp_path: pathlib.Path) -> None:
891 from muse.cli.app import main as cli
892
893 repo = _fresh_repo(tmp_path, n_commits=2)
894 saved = os.getcwd()
895 os.chdir(repo)
896 try:
897 runner.invoke(cli, ["branch", "feat/y"])
898 runner.invoke(cli, ["checkout", "feat/y"])
899 finally:
900 os.chdir(saved)
901 _commit(repo, "only on feat", filename="feat_y.py")
902 # Log main explicitly — should not include feat commit
903 data_main = json.loads(_log(repo, "--json", "main").output)
904 messages = [c["message"] for c in data_main["commits"]]
905 assert "only on feat" not in messages
906
907 def test_merge_commit_has_parent2(self, tmp_path: pathlib.Path) -> None:
908 from muse.cli.app import main as cli
909
910 repo = _fresh_repo(tmp_path, n_commits=1)
911 saved = os.getcwd()
912 os.chdir(repo)
913 try:
914 runner.invoke(cli, ["branch", "feat/merge-test"])
915 runner.invoke(cli, ["checkout", "feat/merge-test"])
916 (repo / "feat_file.py").write_text("f=1\n")
917 runner.invoke(cli, ["commit", "-m", "feat commit"])
918 runner.invoke(cli, ["checkout", "main"])
919 (repo / "main_file.py").write_text("m=1\n")
920 runner.invoke(cli, ["commit", "-m", "main diverge"])
921 runner.invoke(cli, ["merge", "feat/merge-test"])
922 finally:
923 os.chdir(saved)
924
925 data = json.loads(_log(repo, "--json", "--limit", "1").output)
926 merge_commit = data["commits"][0]
927 # A merge commit must have parent2_commit_id set
928 assert merge_commit["parent2_commit_id"] is not None
929
930
931 # ---------------------------------------------------------------------------
932 # Stress — large history and rapid calls
933 # ---------------------------------------------------------------------------
934
935
936 class TestStress:
937 @pytest.mark.slow
938 def test_log_200_commits_json(self, tmp_path: pathlib.Path) -> None:
939 """log --json on 200 commits must exit 0 with correct count."""
940 repo = _fresh_repo(tmp_path, n_commits=200)
941 result = _log(repo, "--json")
942 assert result.exit_code == 0
943 data = json.loads(result.output)
944 assert len(data["commits"]) == 200
945
946 @pytest.mark.slow
947 def test_log_200_commits_oneline(self, tmp_path: pathlib.Path) -> None:
948 repo = _fresh_repo(tmp_path, n_commits=200)
949 result = _log(repo, "--oneline")
950 assert result.exit_code == 0
951 lines = [l for l in result.output.splitlines() if l.strip()]
952 assert len(lines) == 200
953
954 @pytest.mark.slow
955 def test_rapid_sequential_calls(self, tmp_path: pathlib.Path) -> None:
956 """20 sequential muse log calls must all succeed."""
957 repo = _fresh_repo(tmp_path, n_commits=10)
958 for i in range(20):
959 result = _log(repo, "--json")
960 assert result.exit_code == 0, f"Call {i} failed"
961
962 def test_limit_n_large(self, tmp_path: pathlib.Path) -> None:
963 repo = _fresh_repo(tmp_path, n_commits=10)
964 data = json.loads(_log(repo, "--json", "--limit", "5").output)
965 assert len(data["commits"]) == 5
966
967 def test_filter_returns_subset(self, tmp_path: pathlib.Path) -> None:
968 """Limiting to 5 commits from a 20-commit repo returns exactly 5."""
969 repo = _fresh_repo(tmp_path, n_commits=20)
970 data = json.loads(_log(repo, "--json", "--limit", "5").output)
971 assert len(data["commits"]) == 5
972
973 def test_truncated_true_when_filter_skips_commits(self, tmp_path: pathlib.Path) -> None:
974 """With active filter + large walk cap, walk_truncated can be True.
975
976 Use --since=future so the filter skips all commits, but the walk still
977 fetches them all up to walk_cap. We exercise the truncated-when-filter
978 path by creating more commits than the walk ceiling.
979 """
980 repo = _fresh_repo(tmp_path, n_commits=10)
981 # Verify that --since=2099 returns an empty but valid JSON object.
982 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
983 assert data["commits"] == []
984 # truncated may or may not be True here depending on walk_cap;
985 # the key invariant is that the output is well-formed JSON.
986 assert isinstance(data["truncated"], bool)
987
988
989 # ===========================================================================
990 # Manifest cache — each commit's snapshot must be read at most once per run
991 # ===========================================================================
992
993
994 class TestManifestCache:
995 """get_commit_snapshot_manifest must not be called more than once per commit_id.
996
997 Before the fix, _commit_touches_path and _file_diff each called
998 get_commit_snapshot_manifest independently. With a pathspec filter plus
999 JSON output (which always runs _file_diff), the same commit_id was read 4×
1000 per commit (current + parent in each function).
1001
1002 After the fix, a shared manifest_cache dict deduplicates reads so each
1003 commit_id is read at most once regardless of how many callers need it.
1004 """
1005
1006 def test_manifest_cache_used_structurally(self) -> None:
1007 """manifest_cache dict must be threaded through the log pipeline."""
1008 import inspect
1009 from muse.cli.commands import log as log_module
1010
1011 source = inspect.getsource(log_module)
1012 assert "manifest_cache" in source, (
1013 "log.py must use a manifest_cache dict to deduplicate snapshot reads"
1014 )
1015
1016 def test_each_commit_id_read_at_most_once(self, tmp_path: pathlib.Path) -> None:
1017 """With pathspec + JSON mode, each commit's snapshot read ≤ 1×.
1018
1019 JSON mode always calls _file_diff (stat=True).
1020 Pathspec filter calls _commit_touches_path.
1021 Without a shared cache, the same manifest is loaded 4× per commit.
1022 With a shared cache it is loaded exactly once.
1023 """
1024 from unittest.mock import patch, call
1025 import muse.cli.commands.log as log_mod
1026 from muse.core.snapshots import get_commit_snapshot_manifest
1027
1028 repo = tmp_path / "r"
1029 _init(repo)
1030 (repo / "src").mkdir(exist_ok=True)
1031 # Create 3 commits each touching a distinct file.
1032 for i in range(3):
1033 _commit(repo, f"msg{i}", filename=f"src/file{i}.py")
1034
1035 seen_ids: list[str] = []
1036
1037 original_fn = get_commit_snapshot_manifest
1038
1039 def tracking_fn(root: pathlib.Path, commit_id: str) -> Mapping[str, str]:
1040 seen_ids.append(commit_id)
1041 return original_fn(root, commit_id)
1042
1043 with patch.object(log_mod, "get_commit_snapshot_manifest", side_effect=tracking_fn):
1044 result = _log(repo, "--json", "--", "src/")
1045
1046 assert result.exit_code == 0
1047 data = json.loads(result.output)
1048 assert len(data["commits"]) > 0
1049
1050 # Each commit_id must appear at most once in the call log.
1051 from collections import Counter
1052 counts = Counter(seen_ids)
1053 duplicates = {cid: n for cid, n in counts.items() if n > 1}
1054 assert not duplicates, (
1055 f"get_commit_snapshot_manifest called >1× for commit IDs: {duplicates}. "
1056 "Manifest cache not working."
1057 )
1058
1059 def test_pathspec_filter_correct_with_cache(self, tmp_path: pathlib.Path) -> None:
1060 """Pathspec filter returns correct commits when manifest cache is active."""
1061 repo = tmp_path / "r"
1062 _init(repo)
1063 _commit(repo, "add alpha", filename="alpha.py")
1064 _commit(repo, "add beta", filename="beta.py")
1065 _commit(repo, "add gamma", filename="gamma.py")
1066
1067 result = _log(repo, "--json", "--", "alpha.py")
1068 assert result.exit_code == 0
1069 data = json.loads(result.output)
1070 messages = [c["message"] for c in data["commits"]]
1071 assert any("alpha" in m for m in messages), (
1072 "alpha.py pathspec should include the 'add alpha' commit"
1073 )
1074 assert not any("beta" in m for m in messages), (
1075 "beta.py pathspec should NOT include the 'add beta' commit"
1076 )
File History 1 commit
sha256:144495ae90498f9d51aeac4bf7f9ee9e9502d14474745a856b96921f7349c0ce next round of fixing tests Human 105 days ago