gabriel / muse public
test_cmd_log.py python
1,076 lines 41.6 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 123 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.store 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 repo_id="r" * 36,
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, repo_id="r" * 36, 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, repo_id="r" * 36, 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, repo_id="r" * 36, 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, repo_id="r" * 36, 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, repo_id="r" * 36, 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, repo_id="r" * 36, 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 repo_id="r" * 36,
334 branch="main",
335 message="hello",
336 author="alice",
337 committed_at=datetime(2025, 6, 1, tzinfo=timezone.utc),
338 parent_commit_id=None,
339 snapshot_id="b" * 64,
340 )
341
342 def test_all_keys_present(self) -> None:
343 from muse.cli.commands.log import _commit_to_json
344
345 c = self._make_commit()
346 d = _commit_to_json(c)
347 expected = {
348 "commit_id", "branch", "message", "author",
349 "agent_id", "model_id",
350 "committed_at",
351 "parent_commit_id", "parent2_commit_id", "snapshot_id",
352 "sem_ver_bump", "breaking_changes", "metadata",
353 "files_added", "files_removed", "files_modified",
354 "structured_delta",
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, ["commit", "-m", msg, "--author", author])
585 finally:
586 os.chdir(saved)
587
588
589 def _commit_with_identity_author(repo: pathlib.Path, msg: str, author: str, hub_url: str, filename: str | None = None) -> None:
590 """Seed identity.toml with a handle, then invoke muse commit without --author."""
591 from muse.cli.app import main as cli
592 from unittest.mock import patch
593 import pathlib
594 identity_file = repo / "identity.toml"
595 hostname = hub_url.split("://", 1)[-1].rstrip("/")
596 identity_file.write_text(
597 f'["{hostname}"]\ntype = "human"\nhandle = "{author}"\n'
598 f'algorithm = "ed25519"\nfingerprint = "sha256:abc"\nhd_path = "m/0\'"\n',
599 encoding="utf-8",
600 )
601 from muse.cli.config import set_hub_url
602 set_hub_url(hub_url, repo)
603 fname = filename or f"file_{abs(hash(msg))}.py"
604 (repo / fname).write_text(f"# {msg}\n")
605 saved = os.getcwd()
606 try:
607 os.chdir(repo)
608 with patch("muse.core.identity._IDENTITY_FILE", identity_file):
609 runner.invoke(cli, ["commit", "-m", msg])
610 finally:
611 os.chdir(saved)
612
613
614 class TestAuthorField:
615 """Author field in log JSON must come from identity.toml when --author not given."""
616
617 def test_commit_with_explicit_author_appears_in_log(self, tmp_path: pathlib.Path) -> None:
618 """--author flag sets author field that muse log --json exposes."""
619 repo = _fresh_repo(tmp_path, n_commits=0)
620 _commit_as(repo, "my commit", "charlie")
621 result = _log(repo, "--json")
622 data = json.loads(result.output)
623 assert data["commits"][0]["author"] == "charlie"
624
625 def test_commit_reads_user_name_from_identity(self, tmp_path: pathlib.Path) -> None:
626 """muse commit without --author reads user.handle from identity.toml."""
627 repo = _fresh_repo(tmp_path, n_commits=0)
628 _commit_with_identity_author(repo, "identity commit", "diana", "https://localhost:1337")
629 result = _log(repo, "--json")
630 data = json.loads(result.output)
631 assert data["commits"][0]["author"] == "diana"
632
633 def test_author_filter_returns_matching_commits(self, tmp_path: pathlib.Path) -> None:
634 """--author filter must return commits whose author matches the substring."""
635 repo = _fresh_repo(tmp_path, n_commits=0)
636 _commit_as(repo, "alice commit", "alice", filename="a.py")
637 _commit_as(repo, "bob commit", "bob", filename="b.py")
638 result = _log(repo, "--author", "alice", "--json")
639 data = json.loads(result.output)
640 assert len(data["commits"]) == 1
641 assert data["commits"][0]["author"] == "alice"
642
643 def test_author_filter_nonexistent_returns_no_commits(self, tmp_path: pathlib.Path) -> None:
644 """--author filter with no match must return empty list."""
645 repo = _fresh_repo(tmp_path, n_commits=0)
646 _commit_as(repo, "some commit", "alice", filename="a.py")
647 result = _log(repo, "--author", "zzz_nobody_zzz", "--json")
648 data = json.loads(result.output)
649 assert data["commits"] == []
650
651
652 class TestFilters:
653 def test_author_filter_matches(self, tmp_path: pathlib.Path) -> None:
654 repo = _fresh_repo(tmp_path, n_commits=2)
655 # The author will be whatever muse uses by default
656 # We just verify that filtering by nonexistent author returns none
657 result = _log(repo, "--author", "zzz_nobody_zzz")
658 assert "(no commits)" in result.output
659
660 def test_since_filters_old_commits(self, tmp_path: pathlib.Path) -> None:
661 repo = _fresh_repo(tmp_path, n_commits=2)
662 result = _log(repo, "--since", "2099-01-01")
663 # Future date — should return no commits
664 assert "(no commits)" in result.output
665
666 def test_until_filters_future_commits(self, tmp_path: pathlib.Path) -> None:
667 repo = _fresh_repo(tmp_path, n_commits=2)
668 # Past date — all commits should be excluded
669 result = _log(repo, "--until", "2000-01-01")
670 assert "(no commits)" in result.output
671
672 def test_limit_shorthand(self, tmp_path: pathlib.Path) -> None:
673 """muse log -2 must show at most 2 commits."""
674 repo = _fresh_repo(tmp_path, n_commits=5)
675 result = _log(repo, "--oneline", "--limit", "2")
676 lines = [l for l in result.output.splitlines() if l.strip()]
677 assert len(lines) == 2
678
679 def test_limit_flag_alias(self, tmp_path: pathlib.Path) -> None:
680 """--limit is an alias for -n/--max-count."""
681 repo = _fresh_repo(tmp_path, n_commits=5)
682 result = _log(repo, "--oneline", "--limit", "3")
683 lines = [l for l in result.output.splitlines() if l.strip()]
684 assert len(lines) == 3
685
686 def test_limit_flag_json(self, tmp_path: pathlib.Path) -> None:
687 """--limit works with --json output."""
688 repo = _fresh_repo(tmp_path, n_commits=5)
689 data = json.loads(_log(repo, "--json", "--limit", "2").output)
690 assert len(data["commits"]) == 2
691
692 def test_json_since_filters(self, tmp_path: pathlib.Path) -> None:
693 repo = _fresh_repo(tmp_path, n_commits=2)
694 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
695 assert data["commits"] == []
696
697 def test_invalid_since_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
698 repo = _fresh_repo(tmp_path)
699 result = _log(repo, "--since", "not-a-date")
700 assert result.exit_code != 0
701
702 def test_invalid_until_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
703 repo = _fresh_repo(tmp_path)
704 result = _log(repo, "--until", "not-a-date")
705 assert result.exit_code != 0
706
707 def test_invalid_since_no_traceback(self, tmp_path: pathlib.Path) -> None:
708 repo = _fresh_repo(tmp_path)
709 result = _log(repo, "--since", "baddate")
710 assert "Traceback" not in result.output
711
712 def test_invalid_until_clean_error(self, tmp_path: pathlib.Path) -> None:
713 repo = _fresh_repo(tmp_path)
714 result = _log(repo, "--until", "foo")
715 assert "Cannot parse" in result.output or result.exit_code != 0
716
717
718 # ---------------------------------------------------------------------------
719 # Integration — format validation
720 # ---------------------------------------------------------------------------
721
722
723 class TestFormatValidation:
724 def test_unknown_flag_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
725 repo = _fresh_repo(tmp_path)
726 result = _log(repo, "--format", "xml")
727 assert result.exit_code != 0
728
729 def test_unknown_flag_no_traceback(self, tmp_path: pathlib.Path) -> None:
730 repo = _fresh_repo(tmp_path)
731 result = _log(repo, "--format", "yaml")
732 assert "Traceback" not in result.output
733
734 def test_j_shorthand_same_as_json_flag(self, tmp_path: pathlib.Path) -> None:
735 repo = _fresh_repo(tmp_path, n_commits=2)
736 r1 = _log(repo, "--json")
737 r2 = _log(repo, "-j")
738 d1 = json.loads(r1.output)
739 d2 = json.loads(r2.output)
740 # duration_ms and timestamp are wall-clock values — exclude them
741 for d in (d1, d2):
742 d.pop("duration_ms", None)
743 d.pop("timestamp", None)
744 for c in d.get("commits", []):
745 c.pop("duration_ms", None)
746 assert d1 == d2
747
748 def test_invalid_max_count_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
749 repo = _fresh_repo(tmp_path)
750 result = _log(repo, "--limit", "0")
751 assert result.exit_code != 0
752
753
754 # ---------------------------------------------------------------------------
755 # Security — ANSI injection
756 # ---------------------------------------------------------------------------
757
758
759 class TestSecurity:
760 def test_ansi_in_commit_message_sanitized_oneline(self, tmp_path: pathlib.Path) -> None:
761 repo = tmp_path / "repo"
762 _init(repo)
763 # Commit a message with ANSI in it
764 _commit(repo, "\x1b[31mmalicious\x1b[0m", filename="malicious.py")
765 result = _log(repo, "--oneline")
766 # The runner is not a tty — any escape from the message must be sanitized
767 assert "\x1b[31m" not in result.output
768
769 def test_ansi_in_commit_message_sanitized_long(self, tmp_path: pathlib.Path) -> None:
770 repo = tmp_path / "repo"
771 _init(repo)
772 _commit(repo, "\x1b[31mhacked\x1b[0m", filename="h.py")
773 result = _log(repo)
774 assert "\x1b[31m" not in result.output
775
776 def test_ansi_in_author_sanitized(self, tmp_path: pathlib.Path) -> None:
777 """Author names from CommitRecord must be sanitized in output."""
778 repo = _fresh_repo(tmp_path)
779 result = _log(repo)
780 # No raw escape from author field in text output (we can't control
781 # author easily, but ensure output is escape-free when not tty)
782 assert "\x1b[31m" not in result.output
783
784 def test_multiline_message_all_lines_indented(self, tmp_path: pathlib.Path) -> None:
785 """Every line of a multiline message must start with 4-space indent."""
786 repo = tmp_path / "repo"
787 _init(repo)
788 _commit(repo, "Line1\nLine2\nLine3", filename="f.py")
789 result = _log(repo)
790 body_lines = [l for l in result.output.splitlines() if l.strip() in ("Line1", "Line2", "Line3")]
791 assert body_lines, f"Body lines not found in: {result.output}"
792 for line in body_lines:
793 assert line.startswith(" "), f"Not indented: {repr(line)}"
794
795 def test_unknown_flag_exits_nonzero_ansi(self, tmp_path: pathlib.Path) -> None:
796 repo = _fresh_repo(tmp_path)
797 malicious_fmt = "\x1b[31mmalicious\x1b[0m"
798 result = _log(repo, "--format", malicious_fmt)
799 assert result.exit_code != 0
800
801 def test_repo_id_in_json_envelope(self, tmp_path: pathlib.Path) -> None:
802 """repo_id is included in the JSON envelope for agent cross-referencing."""
803 repo = _fresh_repo(tmp_path)
804 stored = json.loads((repo_json_path(repo)).read_text())["repo_id"]
805 result = _log(repo, "--json")
806 data = json.loads(result.output)
807 assert data["repo_id"] == stored
808
809
810 # ---------------------------------------------------------------------------
811 # Integration — nonexistent branch
812 # ---------------------------------------------------------------------------
813
814
815 class TestNonexistentBranch:
816 def test_nonexistent_branch_contextual_message(self, tmp_path: pathlib.Path) -> None:
817 repo = _fresh_repo(tmp_path)
818 result = _log(repo, "bogus-branch")
819 assert "bogus-branch" in result.output
820
821 def test_nonexistent_branch_exits_zero(self, tmp_path: pathlib.Path) -> None:
822 """log on a nonexistent branch is not a fatal error."""
823 repo = _fresh_repo(tmp_path)
824 result = _log(repo, "bogus-branch")
825 assert result.exit_code == 0
826
827 def test_nonexistent_branch_json_empty_commits(self, tmp_path: pathlib.Path) -> None:
828 repo = _fresh_repo(tmp_path)
829 data = json.loads(_log(repo, "--json", "bogus-branch").output)
830 assert data["commits"] == []
831
832 def test_empty_repo_shows_no_commits(self, tmp_path: pathlib.Path) -> None:
833 repo = tmp_path / "repo"
834 _init(repo)
835 result = _log(repo)
836 assert "no commits" in result.output.lower()
837
838
839 # ---------------------------------------------------------------------------
840 # End-to-end — complete workflows
841 # ---------------------------------------------------------------------------
842
843
844 class TestEndToEnd:
845 def test_single_commit_log(self, tmp_path: pathlib.Path) -> None:
846 repo = _fresh_repo(tmp_path, n_commits=1)
847 result = _log(repo)
848 assert result.exit_code == 0
849 assert "commit" in result.output.lower()
850
851 def test_multiple_commits_ordered_newest_first(self, tmp_path: pathlib.Path) -> None:
852 repo = _fresh_repo(tmp_path, n_commits=3)
853 result = _log(repo, "--oneline")
854 lines = [l for l in result.output.strip().splitlines() if l]
855 assert len(lines) == 3
856
857 def test_head_decoration_on_latest(self, tmp_path: pathlib.Path) -> None:
858 repo = _fresh_repo(tmp_path, n_commits=2)
859 result = _log(repo)
860 lines = result.output.strip().splitlines()
861 # First commit line should have HEAD
862 first = next((l for l in lines if "commit" in l.lower()), "")
863 assert "HEAD" in first
864
865 def test_subprocess_call_works(self, tmp_path: pathlib.Path) -> None:
866 repo = _fresh_repo(tmp_path, n_commits=2)
867 r = subprocess.run(
868 ["muse", "log", "--json"],
869 capture_output=True, text=True, cwd=str(repo),
870 )
871 assert r.returncode == 0
872 data = json.loads(r.stdout)
873 assert len(data["commits"]) == 2
874
875 def test_log_after_branch_switch(self, tmp_path: pathlib.Path) -> None:
876 from muse.cli.app import main as cli
877
878 repo = _fresh_repo(tmp_path, n_commits=2)
879 saved = os.getcwd()
880 os.chdir(repo)
881 try:
882 runner.invoke(cli, ["branch", "feat/x"])
883 runner.invoke(cli, ["checkout", "feat/x"])
884 finally:
885 os.chdir(saved)
886 _commit(repo, "feat commit", filename="feat.py")
887 data = json.loads(_log(repo, "--json").output)
888 # feat branch should have 3 commits (2 from main + 1 new)
889 assert len(data["commits"]) == 3
890
891 def test_log_on_explicit_branch(self, tmp_path: pathlib.Path) -> None:
892 from muse.cli.app import main as cli
893
894 repo = _fresh_repo(tmp_path, n_commits=2)
895 saved = os.getcwd()
896 os.chdir(repo)
897 try:
898 runner.invoke(cli, ["branch", "feat/y"])
899 runner.invoke(cli, ["checkout", "feat/y"])
900 finally:
901 os.chdir(saved)
902 _commit(repo, "only on feat", filename="feat_y.py")
903 # Log main explicitly — should not include feat commit
904 data_main = json.loads(_log(repo, "--json", "main").output)
905 messages = [c["message"] for c in data_main["commits"]]
906 assert "only on feat" not in messages
907
908 def test_merge_commit_has_parent2(self, tmp_path: pathlib.Path) -> None:
909 from muse.cli.app import main as cli
910
911 repo = _fresh_repo(tmp_path, n_commits=1)
912 saved = os.getcwd()
913 os.chdir(repo)
914 try:
915 runner.invoke(cli, ["branch", "feat/merge-test"])
916 runner.invoke(cli, ["checkout", "feat/merge-test"])
917 (repo / "feat_file.py").write_text("f=1\n")
918 runner.invoke(cli, ["commit", "-m", "feat commit"])
919 runner.invoke(cli, ["checkout", "main"])
920 (repo / "main_file.py").write_text("m=1\n")
921 runner.invoke(cli, ["commit", "-m", "main diverge"])
922 runner.invoke(cli, ["merge", "feat/merge-test"])
923 finally:
924 os.chdir(saved)
925
926 data = json.loads(_log(repo, "--json", "--limit", "1").output)
927 merge_commit = data["commits"][0]
928 # A merge commit must have parent2_commit_id set
929 assert merge_commit["parent2_commit_id"] is not None
930
931
932 # ---------------------------------------------------------------------------
933 # Stress — large history and rapid calls
934 # ---------------------------------------------------------------------------
935
936
937 class TestStress:
938 @pytest.mark.slow
939 def test_log_200_commits_json(self, tmp_path: pathlib.Path) -> None:
940 """log --json on 200 commits must exit 0 with correct count."""
941 repo = _fresh_repo(tmp_path, n_commits=200)
942 result = _log(repo, "--json")
943 assert result.exit_code == 0
944 data = json.loads(result.output)
945 assert len(data["commits"]) == 200
946
947 @pytest.mark.slow
948 def test_log_200_commits_oneline(self, tmp_path: pathlib.Path) -> None:
949 repo = _fresh_repo(tmp_path, n_commits=200)
950 result = _log(repo, "--oneline")
951 assert result.exit_code == 0
952 lines = [l for l in result.output.splitlines() if l.strip()]
953 assert len(lines) == 200
954
955 @pytest.mark.slow
956 def test_rapid_sequential_calls(self, tmp_path: pathlib.Path) -> None:
957 """20 sequential muse log calls must all succeed."""
958 repo = _fresh_repo(tmp_path, n_commits=10)
959 for i in range(20):
960 result = _log(repo, "--json")
961 assert result.exit_code == 0, f"Call {i} failed"
962
963 def test_limit_n_large(self, tmp_path: pathlib.Path) -> None:
964 repo = _fresh_repo(tmp_path, n_commits=10)
965 data = json.loads(_log(repo, "--json", "--limit", "5").output)
966 assert len(data["commits"]) == 5
967
968 def test_filter_returns_subset(self, tmp_path: pathlib.Path) -> None:
969 """Limiting to 5 commits from a 20-commit repo returns exactly 5."""
970 repo = _fresh_repo(tmp_path, n_commits=20)
971 data = json.loads(_log(repo, "--json", "--limit", "5").output)
972 assert len(data["commits"]) == 5
973
974 def test_truncated_true_when_filter_skips_commits(self, tmp_path: pathlib.Path) -> None:
975 """With active filter + large walk cap, walk_truncated can be True.
976
977 Use --since=future so the filter skips all commits, but the walk still
978 fetches them all up to walk_cap. We exercise the truncated-when-filter
979 path by creating more commits than the walk ceiling.
980 """
981 repo = _fresh_repo(tmp_path, n_commits=10)
982 # Verify that --since=2099 returns an empty but valid JSON object.
983 data = json.loads(_log(repo, "--json", "--since", "2099-01-01").output)
984 assert data["commits"] == []
985 # truncated may or may not be True here depending on walk_cap;
986 # the key invariant is that the output is well-formed JSON.
987 assert isinstance(data["truncated"], bool)
988
989
990 # ===========================================================================
991 # Manifest cache — each commit's snapshot must be read at most once per run
992 # ===========================================================================
993
994
995 class TestManifestCache:
996 """get_commit_snapshot_manifest must not be called more than once per commit_id.
997
998 Before the fix, _commit_touches_path and _file_diff each called
999 get_commit_snapshot_manifest independently. With a pathspec filter plus
1000 JSON output (which always runs _file_diff), the same commit_id was read 4×
1001 per commit (current + parent in each function).
1002
1003 After the fix, a shared manifest_cache dict deduplicates reads so each
1004 commit_id is read at most once regardless of how many callers need it.
1005 """
1006
1007 def test_manifest_cache_used_structurally(self) -> None:
1008 """manifest_cache dict must be threaded through the log pipeline."""
1009 import inspect
1010 from muse.cli.commands import log as log_module
1011
1012 source = inspect.getsource(log_module)
1013 assert "manifest_cache" in source, (
1014 "log.py must use a manifest_cache dict to deduplicate snapshot reads"
1015 )
1016
1017 def test_each_commit_id_read_at_most_once(self, tmp_path: pathlib.Path) -> None:
1018 """With pathspec + JSON mode, each commit's snapshot read ≤ 1×.
1019
1020 JSON mode always calls _file_diff (stat=True).
1021 Pathspec filter calls _commit_touches_path.
1022 Without a shared cache, the same manifest is loaded 4× per commit.
1023 With a shared cache it is loaded exactly once.
1024 """
1025 from unittest.mock import patch, call
1026 from muse.core import store as store_module
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 = store_module.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(store_module, "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:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 123 days ago