gabriel / muse public
test_cmd_shortlog_hardening.py python
861 lines 30.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Hardening test suite for ``muse shortlog``.
2
3 Coverage:
4 - Unit: _branch_names (symlink guard), _group_key (all four modes),
5 _build_groups (email flag, dedup), _parse_date (valid + invalid)
6 - Security: ANSI in author/message sanitized in text, raw in JSON;
7 symlink inside refs/heads is skipped
8 - Error routing: all user errors go to stderr
9 - JSON schema: _ShortlogJson shape (repo_id, branch, groups), all fields
10 - New flags: --group-by (agent, model, branch), --summary, --no-merges,
11 --since, --until, combined filters
12 - --json: empty, single group, multi-group, provenance fields
13 - Integration: --all branches with dedup, --limit early-exit, date range
14 - E2E: help output, combined flags
15 - Stress: 500 commits × 5 authors, 50-branch repo, concurrent reads
16 """
17
18 from __future__ import annotations
19 from collections.abc import Mapping
20
21 import datetime
22 import json
23 import os
24 import pathlib
25 from typing import TypedDict
26 from unittest.mock import patch
27
28 import pytest
29 from tests.cli_test_helper import CliRunner, InvokeResult
30
31 from muse.cli.commands.shortlog import _branch_names, _build_groups, _group_key, _parse_date
32 from muse.core.object_store import write_object
33 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
34 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
35 from muse.core.types import Manifest, blob_id
36 from muse.core.paths import heads_dir, muse_dir
37
38 runner = CliRunner()
39 _REPO_ID = "shortlog-hard-test"
40
41 # Tracks the latest commit_id per (str(root), branch) so _make_commit
42 # can auto-chain without callers needing to pass parent_id explicitly.
43 _branch_heads_map: Manifest = {}
44
45
46 # ---------------------------------------------------------------------------
47 # Helpers
48 # ---------------------------------------------------------------------------
49
50
51
52
53 def _init_repo(path: pathlib.Path, *, domain: str = "code") -> pathlib.Path:
54 muse = muse_dir(path)
55 for sub in ("commits", "snapshots", "objects", "refs/heads"):
56 (muse / sub).mkdir(parents=True, exist_ok=True)
57 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
58 (muse / "repo.json").write_text(
59 json.dumps({"repo_id": _REPO_ID, "domain": domain}),
60 encoding="utf-8",
61 )
62 return path
63
64
65 _commit_counter = 0
66
67
68 def _make_commit(
69 root: pathlib.Path,
70 *,
71 author: str = "Alice",
72 agent_id: str | None = None,
73 model_id: str | None = None,
74 branch: str = "main",
75 parent_id: str | None = None,
76 parent2_id: str | None = None,
77 committed_at: datetime.datetime | None = None,
78 ) -> str:
79 """Create and store a commit, auto-chaining to the previous on the same branch."""
80 global _commit_counter
81 _commit_counter += 1
82 content = f"c{_commit_counter}".encode()
83 obj_id = blob_id(content)
84 write_object(root, obj_id, content)
85 manifest = {f"f{_commit_counter}.txt": obj_id}
86 snap_id = compute_snapshot_id(manifest)
87 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
88 ts = committed_at or datetime.datetime.now(datetime.timezone.utc)
89
90 # Auto-chain: if caller didn't provide parent_id, use the last known head.
91 effective_parent = parent_id
92 if effective_parent is None:
93 effective_parent = _branch_heads_map.get(f"{root}:{branch}")
94
95 pids = [pid for pid in (effective_parent, parent2_id) if pid is not None]
96 commit_id = compute_commit_id(
97 parent_ids=pids,
98 snapshot_id=snap_id,
99 message=f"msg {_commit_counter}",
100 committed_at_iso=ts.isoformat(),
101 author=author,
102 )
103 rec = CommitRecord(
104 repo_id=_REPO_ID,
105 commit_id=commit_id,
106 branch=branch,
107 snapshot_id=snap_id,
108 message=f"msg {_commit_counter}",
109 committed_at=ts,
110 parent_commit_id=effective_parent,
111 parent2_commit_id=parent2_id,
112 author=author,
113 agent_id=agent_id or "",
114 model_id=model_id or "",
115 )
116 write_commit(root, rec)
117 ref_dir = heads_dir(root)
118 ref_file = ref_dir / branch
119 ref_file.parent.mkdir(parents=True, exist_ok=True)
120 ref_file.write_text(commit_id, encoding="utf-8")
121 _branch_heads_map[f"{root}:{branch}"] = commit_id
122 return commit_id
123
124
125 def _env(repo: pathlib.Path) -> Manifest:
126 return {"MUSE_REPO_ROOT": str(repo)}
127
128
129 def _invoke(args: list[str], env: Manifest) -> InvokeResult:
130 return runner.invoke(None, args, env=env)
131
132
133 class _GroupOut(TypedDict):
134 key: str
135 count: int
136 commits: list[Mapping[str, str | None]]
137
138
139 class _ShortlogOut(TypedDict):
140 repo_id: str
141 branch: str
142 groups: list[_GroupOut]
143
144
145 def _parse_json(result: InvokeResult) -> _ShortlogOut:
146 raw = json.loads(result.output.strip())
147 groups: list[_GroupOut] = [
148 _GroupOut(
149 key=g["key"],
150 count=g["count"],
151 commits=g["commits"],
152 )
153 for g in raw["groups"]
154 ]
155 return _ShortlogOut(
156 repo_id=raw["repo_id"],
157 branch=raw["branch"],
158 groups=groups,
159 )
160
161
162 # ---------------------------------------------------------------------------
163 # Unit: _branch_names — symlink guard
164 # ---------------------------------------------------------------------------
165
166
167 def test_branch_names_returns_normal_branches(tmp_path: pathlib.Path) -> None:
168 _init_repo(tmp_path)
169 _make_commit(tmp_path, branch="main")
170 _make_commit(tmp_path, branch="dev")
171 names = _branch_names(tmp_path)
172 assert "main" in names
173 assert "dev" in names
174
175
176 def test_branch_names_skips_symlinks(tmp_path: pathlib.Path) -> None:
177 _init_repo(tmp_path)
178 _make_commit(tmp_path, branch="main")
179 h_dir = heads_dir(tmp_path)
180 malicious = h_dir / "malicious-branch"
181 try:
182 malicious.symlink_to(tmp_path / "some_other_file")
183 except OSError:
184 pytest.skip("filesystem does not support symlinks")
185 names = _branch_names(tmp_path)
186 assert "malicious-branch" not in names
187 assert "main" in names
188
189
190 def test_branch_names_missing_heads_dir(tmp_path: pathlib.Path) -> None:
191 _init_repo(tmp_path)
192 import shutil
193 shutil.rmtree(heads_dir(tmp_path))
194 assert _branch_names(tmp_path) == []
195
196
197 # ---------------------------------------------------------------------------
198 # Unit: _group_key
199 # ---------------------------------------------------------------------------
200
201
202 def _make_rec(
203 *,
204 author: str = "",
205 agent_id: str = "",
206 model_id: str = "",
207 branch: str = "main",
208 ) -> CommitRecord:
209 return CommitRecord(
210 repo_id=_REPO_ID,
211 commit_id="aaa",
212 branch=branch,
213 snapshot_id="snap",
214 message="x",
215 committed_at=datetime.datetime.now(datetime.timezone.utc),
216 author=author,
217 agent_id=agent_id,
218 model_id=model_id,
219 )
220
221
222 def test_group_key_author_with_author() -> None:
223 rec = _make_rec(author="Alice")
224 assert _group_key(rec, "author") == "Alice"
225
226
227 def test_group_key_author_fallback_to_agent() -> None:
228 rec = _make_rec(agent_id="bot-1")
229 assert _group_key(rec, "author") == "bot-1 (agent)"
230
231
232 def test_group_key_author_unknown() -> None:
233 rec = _make_rec()
234 assert _group_key(rec, "author") == "(unknown)"
235
236
237 def test_group_key_agent() -> None:
238 rec = _make_rec(agent_id="gpt-agent")
239 assert _group_key(rec, "agent") == "gpt-agent"
240
241
242 def test_group_key_agent_no_agent() -> None:
243 rec = _make_rec()
244 assert _group_key(rec, "agent") == "(no agent)"
245
246
247 def test_group_key_model() -> None:
248 rec = _make_rec(model_id="gpt-4o")
249 assert _group_key(rec, "model") == "gpt-4o"
250
251
252 def test_group_key_model_no_model() -> None:
253 rec = _make_rec()
254 assert _group_key(rec, "model") == "(no model)"
255
256
257 def test_group_key_branch() -> None:
258 rec = _make_rec(branch="feat/my-thing")
259 assert _group_key(rec, "branch") == "feat/my-thing"
260
261
262 # ---------------------------------------------------------------------------
263 # Unit: _parse_date
264 # ---------------------------------------------------------------------------
265
266
267 def test_parse_date_valid() -> None:
268 dt = _parse_date("2025-03-15", "--since")
269 assert dt.year == 2025
270 assert dt.month == 3
271 assert dt.day == 15
272 assert dt.tzinfo == datetime.timezone.utc
273
274
275 def test_parse_date_invalid_exits() -> None:
276 with pytest.raises(ValueError):
277 _parse_date("not-a-date", "--since")
278
279
280 def test_parse_date_wrong_format_exits() -> None:
281 with pytest.raises(ValueError):
282 _parse_date("15/03/2025", "--since")
283
284
285 # ---------------------------------------------------------------------------
286 # Security: ANSI injection
287 # ---------------------------------------------------------------------------
288
289
290 def test_ansi_in_author_name_stripped_text(tmp_path: pathlib.Path) -> None:
291 _init_repo(tmp_path)
292 _make_commit(tmp_path, author="Evil\x1b[31mRED\x1b[0m")
293 result = _invoke(["shortlog"], _env(tmp_path))
294 assert result.exit_code == 0
295 assert "\x1b[31m" not in result.output
296
297
298 def test_ansi_in_author_name_raw_in_json(tmp_path: pathlib.Path) -> None:
299 _init_repo(tmp_path)
300 _make_commit(tmp_path, author="Evil\x1b[31mRED\x1b[0m")
301 result = _invoke(["shortlog", "--json"], _env(tmp_path))
302 assert result.exit_code == 0
303 data = _parse_json(result)
304 assert data["groups"][0]["key"] == "Evil\x1b[31mRED\x1b[0m"
305
306
307 def test_ansi_in_message_stripped_text(tmp_path: pathlib.Path) -> None:
308 _init_repo(tmp_path)
309 commit_id = _make_commit(tmp_path)
310 # Directly overwrite message in stored commit to contain ANSI.
311 from muse.core.store import read_commit
312 original = read_commit(tmp_path, commit_id)
313 assert original is not None
314 from muse.core.snapshot import compute_commit_id
315 from muse.core.store import write_commit
316 malicious_msg = "fix: \x1b[1mBOLD\x1b[0m thing"
317 parent_ids = [original.parent_commit_id] if original.parent_commit_id else []
318 new_cid = compute_commit_id(
319 parent_ids=parent_ids,
320 snapshot_id=original.snapshot_id,
321 message=malicious_msg,
322 committed_at_iso=original.committed_at.isoformat(),
323 author=original.author,
324 )
325 patched = CommitRecord(
326 repo_id=_REPO_ID,
327 commit_id=new_cid,
328 branch=original.branch,
329 snapshot_id=original.snapshot_id,
330 message=malicious_msg,
331 committed_at=original.committed_at,
332 author=original.author,
333 )
334 write_commit(tmp_path, patched)
335 (heads_dir(tmp_path) / "main").write_text(new_cid)
336 result = _invoke(["shortlog"], _env(tmp_path))
337 assert "\x1b[1m" not in result.output
338
339
340 # ---------------------------------------------------------------------------
341 # Error routing: all user errors go to stderr
342 # ---------------------------------------------------------------------------
343
344
345 def test_since_invalid_format_stderr(tmp_path: pathlib.Path) -> None:
346 _init_repo(tmp_path)
347 _make_commit(tmp_path)
348 result = _invoke(["shortlog", "--since", "01-01-2025"], _env(tmp_path))
349 assert result.exit_code != 0
350
351
352 def test_until_invalid_format_stderr(tmp_path: pathlib.Path) -> None:
353 _init_repo(tmp_path)
354 _make_commit(tmp_path)
355 result = _invoke(["shortlog", "--until", "not-a-date"], _env(tmp_path))
356 assert result.exit_code != 0
357
358
359 # ---------------------------------------------------------------------------
360 # JSON schema: _ShortlogJson
361 # ---------------------------------------------------------------------------
362
363
364 def test_json_schema_empty_repo(tmp_path: pathlib.Path) -> None:
365 _init_repo(tmp_path)
366 result = _invoke(["shortlog", "--json"], _env(tmp_path))
367 assert result.exit_code == 0
368 data = _parse_json(result)
369 assert data["repo_id"] == _REPO_ID
370 assert data["branch"] == "main"
371 assert data["groups"] == []
372
373
374 def test_json_schema_all_fields_present(tmp_path: pathlib.Path) -> None:
375 _init_repo(tmp_path)
376 _make_commit(tmp_path, author="Alice", agent_id="bot-1", model_id="gpt-4o")
377 result = _invoke(["shortlog", "--json"], _env(tmp_path))
378 assert result.exit_code == 0
379 data = _parse_json(result)
380 assert data["repo_id"] == _REPO_ID
381 assert data["branch"] == "main"
382 grp = data["groups"][0]
383 assert grp["key"] == "Alice"
384 assert grp["count"] == 1
385 commit_entry = grp["commits"][0]
386 assert "commit_id" in commit_entry
387 assert "message" in commit_entry
388 assert "committed_at" in commit_entry
389 assert "author" in commit_entry
390 assert "agent_id" in commit_entry
391 assert "model_id" in commit_entry
392
393
394 def test_json_schema_repo_id_and_branch_in_output(tmp_path: pathlib.Path) -> None:
395 _init_repo(tmp_path)
396 _make_commit(tmp_path, branch="main")
397 result = _invoke(["shortlog", "--json"], _env(tmp_path))
398 assert result.exit_code == 0
399 data = _parse_json(result)
400 assert data["repo_id"] == _REPO_ID
401 assert data["branch"] == "main"
402
403
404 def test_json_schema_all_branches_label(tmp_path: pathlib.Path) -> None:
405 _init_repo(tmp_path)
406 _make_commit(tmp_path, branch="main")
407 result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path))
408 assert result.exit_code == 0
409 data = _parse_json(result)
410 assert data["branch"] == "__all__"
411
412
413 def test_json_agent_id_and_model_id_present(tmp_path: pathlib.Path) -> None:
414 _init_repo(tmp_path)
415 _make_commit(tmp_path, agent_id="agent-007", model_id="claude-3")
416 result = _invoke(["shortlog", "--json"], _env(tmp_path))
417 assert result.exit_code == 0
418 data = _parse_json(result)
419 entry = data["groups"][0]["commits"][0]
420 assert entry["agent_id"] == "agent-007"
421 assert entry["model_id"] == "claude-3"
422
423
424 # ---------------------------------------------------------------------------
425 # New flag: --group-by
426 # ---------------------------------------------------------------------------
427
428
429 def test_group_by_agent(tmp_path: pathlib.Path) -> None:
430 _init_repo(tmp_path)
431 _make_commit(tmp_path, author="Alice", agent_id="bot-1")
432 _make_commit(tmp_path, author="Bob", agent_id="bot-2")
433 _make_commit(tmp_path, author="Alice", agent_id="bot-1")
434 result = _invoke(["shortlog", "--group-by", "agent", "--json"], _env(tmp_path))
435 assert result.exit_code == 0
436 data = _parse_json(result)
437 keys = {g["key"] for g in data["groups"]}
438 assert "bot-1" in keys
439 assert "bot-2" in keys
440
441
442 def test_group_by_model(tmp_path: pathlib.Path) -> None:
443 _init_repo(tmp_path)
444 _make_commit(tmp_path, model_id="gpt-4o")
445 _make_commit(tmp_path, model_id="claude-3")
446 _make_commit(tmp_path, model_id="gpt-4o")
447 result = _invoke(["shortlog", "--group-by", "model", "--json"], _env(tmp_path))
448 assert result.exit_code == 0
449 data = _parse_json(result)
450 keys = {g["key"] for g in data["groups"]}
451 assert "gpt-4o" in keys
452 assert "claude-3" in keys
453 gpt_count = next(g["count"] for g in data["groups"] if g["key"] == "gpt-4o")
454 assert gpt_count == 2
455
456
457 def test_group_by_branch(tmp_path: pathlib.Path) -> None:
458 _init_repo(tmp_path)
459 _make_commit(tmp_path, branch="main")
460 _make_commit(tmp_path, branch="dev")
461 _make_commit(tmp_path, branch="main")
462 result = _invoke(
463 ["shortlog", "--all", "--group-by", "branch", "--json"], _env(tmp_path)
464 )
465 assert result.exit_code == 0
466 data = _parse_json(result)
467 keys = {g["key"] for g in data["groups"]}
468 assert "main" in keys
469 assert "dev" in keys
470
471
472 def test_group_by_invalid_choice(tmp_path: pathlib.Path) -> None:
473 _init_repo(tmp_path)
474 result = _invoke(["shortlog", "--group-by", "badfield"], _env(tmp_path))
475 assert result.exit_code != 0
476
477
478 # ---------------------------------------------------------------------------
479 # New flag: --summary
480 # ---------------------------------------------------------------------------
481
482
483 def test_summary_suppresses_messages(tmp_path: pathlib.Path) -> None:
484 _init_repo(tmp_path)
485 _make_commit(tmp_path, author="Alice")
486 _make_commit(tmp_path, author="Alice")
487 result = _invoke(["shortlog", "--summary"], _env(tmp_path))
488 assert result.exit_code == 0
489 # Author line should still appear.
490 assert "Alice" in result.output
491 # Individual commit messages should not appear (they start with spaces).
492 assert "msg" not in result.output
493
494
495 def test_summary_with_json_still_includes_commits(tmp_path: pathlib.Path) -> None:
496 """--summary only suppresses messages in text mode; JSON always includes them."""
497 _init_repo(tmp_path)
498 _make_commit(tmp_path, author="Alice")
499 result = _invoke(["shortlog", "--summary", "--json"], _env(tmp_path))
500 assert result.exit_code == 0
501 data = _parse_json(result)
502 assert len(data["groups"][0]["commits"]) >= 1
503
504
505 # ---------------------------------------------------------------------------
506 # New flag: --no-merges
507 # ---------------------------------------------------------------------------
508
509
510 def test_no_merges_excludes_merge_commits(tmp_path: pathlib.Path) -> None:
511 """get_commits_for_branch follows first-parent only.
512
513 Chain: c1 → c2 → c3(merge, parent2=c1) → c4
514 First-parent walk from c4 returns [c4, c3, c2, c1].
515 With --no-merges, c3 is excluded → 3 commits remain.
516 """
517 _init_repo(tmp_path)
518 c1 = _make_commit(tmp_path, author="Alice")
519 c2 = _make_commit(tmp_path, author="Bob") # chains to c1
520 # Merge commit: auto-chains first-parent to c2; parent2 points to c1.
521 _make_commit(tmp_path, author="Alice", parent2_id=c1) # chains to c2
522 _make_commit(tmp_path, author="Bob") # chains to merge
523 result = _invoke(["shortlog", "--no-merges", "--json"], _env(tmp_path))
524 assert result.exit_code == 0
525 data = _parse_json(result)
526 total = sum(g["count"] for g in data["groups"])
527 assert total == 3 # c1, c2, c4 — c3 (merge) excluded
528
529
530 def test_no_merges_with_all_non_merges(tmp_path: pathlib.Path) -> None:
531 _init_repo(tmp_path)
532 for _ in range(5):
533 _make_commit(tmp_path, author="Alice")
534 result = _invoke(["shortlog", "--no-merges", "--json"], _env(tmp_path))
535 assert result.exit_code == 0
536 data = _parse_json(result)
537 assert sum(g["count"] for g in data["groups"]) == 5
538
539
540 # ---------------------------------------------------------------------------
541 # New flags: --since / --until
542 # ---------------------------------------------------------------------------
543
544
545 def test_since_filters_old_commits(tmp_path: pathlib.Path) -> None:
546 _init_repo(tmp_path)
547 old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
548 new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc)
549 _make_commit(tmp_path, author="Old", committed_at=old)
550 _make_commit(tmp_path, author="New", committed_at=new)
551 result = _invoke(["shortlog", "--since", "2025-01-01", "--json"], _env(tmp_path))
552 assert result.exit_code == 0
553 data = _parse_json(result)
554 keys = {g["key"] for g in data["groups"]}
555 assert "New" in keys
556 assert "Old" not in keys
557
558
559 def test_until_filters_future_commits(tmp_path: pathlib.Path) -> None:
560 _init_repo(tmp_path)
561 old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
562 new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc)
563 _make_commit(tmp_path, author="Old", committed_at=old)
564 _make_commit(tmp_path, author="New", committed_at=new)
565 result = _invoke(["shortlog", "--until", "2022-12-31", "--json"], _env(tmp_path))
566 assert result.exit_code == 0
567 data = _parse_json(result)
568 keys = {g["key"] for g in data["groups"]}
569 assert "Old" in keys
570 assert "New" not in keys
571
572
573 def test_since_and_until_window(tmp_path: pathlib.Path) -> None:
574 _init_repo(tmp_path)
575 dates = [
576 datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc),
577 datetime.datetime(2025, 3, 15, tzinfo=datetime.timezone.utc),
578 datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
579 ]
580 authors = ["Before", "Inside", "After"]
581 for a, d in zip(authors, dates):
582 _make_commit(tmp_path, author=a, committed_at=d)
583 result = _invoke(
584 ["shortlog", "--since", "2025-01-01", "--until", "2025-12-31", "--json"],
585 _env(tmp_path),
586 )
587 assert result.exit_code == 0
588 data = _parse_json(result)
589 keys = {g["key"] for g in data["groups"]}
590 assert "Inside" in keys
591 assert "Before" not in keys
592 assert "After" not in keys
593
594
595 def test_since_no_results_returns_empty_json(tmp_path: pathlib.Path) -> None:
596 _init_repo(tmp_path)
597 _make_commit(
598 tmp_path,
599 author="Old",
600 committed_at=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc),
601 )
602 result = _invoke(["shortlog", "--since", "2030-01-01", "--json"], _env(tmp_path))
603 assert result.exit_code == 0
604 data = _parse_json(result)
605 assert data["groups"] == []
606
607
608 # ---------------------------------------------------------------------------
609 # Integration
610 # ---------------------------------------------------------------------------
611
612
613 def test_integration_all_branches_dedup(tmp_path: pathlib.Path) -> None:
614 """A commit reachable from two branches should count once."""
615 _init_repo(tmp_path)
616 shared = _make_commit(tmp_path, author="Alice", branch="main")
617 # Create dev branch pointing at same commit (by writing the ref file).
618 dev_ref = heads_dir(tmp_path) / "dev"
619 dev_ref.write_text(shared, encoding="utf-8")
620 result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path))
621 assert result.exit_code == 0
622 data = _parse_json(result)
623 total = sum(g["count"] for g in data["groups"])
624 assert total == 1 # deduplicated
625
626
627 def test_integration_limit_early_exit(tmp_path: pathlib.Path) -> None:
628 _init_repo(tmp_path)
629 for i in range(50):
630 _make_commit(tmp_path, author=f"Author{i % 5}")
631 result = _invoke(["shortlog", "--limit", "10", "--json"], _env(tmp_path))
632 assert result.exit_code == 0
633 data = _parse_json(result)
634 total = sum(g["count"] for g in data["groups"])
635 assert total <= 10
636
637
638 def test_integration_numbered_combined_with_since(tmp_path: pathlib.Path) -> None:
639 _init_repo(tmp_path)
640 old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
641 new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc)
642 for _ in range(3):
643 _make_commit(tmp_path, author="Prolific", committed_at=new)
644 _make_commit(tmp_path, author="Old", committed_at=old)
645 result = _invoke(
646 ["shortlog", "--since", "2025-01-01", "--numbered", "--json"],
647 _env(tmp_path),
648 )
649 assert result.exit_code == 0
650 data = _parse_json(result)
651 assert data["groups"][0]["key"] == "Prolific"
652 assert "Old" not in {g["key"] for g in data["groups"]}
653
654
655 # ---------------------------------------------------------------------------
656 # E2E: help output
657 # ---------------------------------------------------------------------------
658
659
660 def test_help_shows_new_flags() -> None:
661 result = _invoke(["shortlog", "--help"], {})
662 assert result.exit_code == 0
663 for flag in ("--group-by", "--summary", "--no-merges", "--since", "--until", "--json"):
664 assert flag in result.output, f"Missing flag: {flag}"
665
666
667 def test_help_mentions_group_by_choices() -> None:
668 result = _invoke(["shortlog", "--help"], {})
669 for choice in ("author", "agent", "model", "branch"):
670 assert choice in result.output
671
672
673 # ---------------------------------------------------------------------------
674 # Stress: 500 commits × 5 authors
675 # ---------------------------------------------------------------------------
676
677
678 def test_stress_500_commits(tmp_path: pathlib.Path) -> None:
679 _init_repo(tmp_path)
680 authors = ["Amy", "Ben", "Cleo", "Dan", "Eva"]
681 for i in range(500):
682 _make_commit(tmp_path, author=authors[i % 5])
683 result = _invoke(["shortlog", "--json"], _env(tmp_path))
684 assert result.exit_code == 0
685 data = _parse_json(result)
686 total = sum(g["count"] for g in data["groups"])
687 assert total == 500
688 assert len(data["groups"]) == 5
689
690
691 def test_stress_500_commits_numbered(tmp_path: pathlib.Path) -> None:
692 _init_repo(tmp_path)
693 # Give Alice 300, Bob 200.
694 for _ in range(300):
695 _make_commit(tmp_path, author="Alice")
696 for _ in range(200):
697 _make_commit(tmp_path, author="Bob")
698 result = _invoke(["shortlog", "--numbered", "--json"], _env(tmp_path))
699 assert result.exit_code == 0
700 data = _parse_json(result)
701 assert data["groups"][0]["key"] == "Alice"
702 assert data["groups"][0]["count"] == 300
703
704
705 # ---------------------------------------------------------------------------
706 # JSON schema — duration_ms + exit_code + truncated on every output path
707 # ---------------------------------------------------------------------------
708
709
710 class TestJsonSchema:
711 """Every --json response must carry duration_ms, exit_code, and truncated."""
712
713 def _assert_schema(self, d: Mapping[str, object], *, exit_code: int = 0) -> None:
714 assert "duration_ms" in d, f"duration_ms missing: {d}"
715 assert isinstance(d["duration_ms"], (int, float))
716 assert d["duration_ms"] >= 0
717 assert "exit_code" in d, f"exit_code missing: {d}"
718 assert d["exit_code"] == exit_code
719 assert "truncated" in d, f"truncated missing: {d}"
720
721 def test_normal_output_has_schema(self, tmp_path: pathlib.Path) -> None:
722 _init_repo(tmp_path)
723 _make_commit(tmp_path, author="Alice")
724 result = _invoke(["shortlog", "--json"], _env(tmp_path))
725 assert result.exit_code == 0
726 self._assert_schema(json.loads(result.output))
727
728 def test_empty_repo_json_has_schema(self, tmp_path: pathlib.Path) -> None:
729 _init_repo(tmp_path)
730 result = _invoke(["shortlog", "--json"], _env(tmp_path))
731 assert result.exit_code == 0
732 self._assert_schema(json.loads(result.output))
733
734 def test_all_branches_json_has_schema(self, tmp_path: pathlib.Path) -> None:
735 _init_repo(tmp_path)
736 _make_commit(tmp_path, branch="main")
737 result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path))
738 assert result.exit_code == 0
739 self._assert_schema(json.loads(result.output))
740
741 def test_numbered_json_has_schema(self, tmp_path: pathlib.Path) -> None:
742 _init_repo(tmp_path)
743 _make_commit(tmp_path, author="Alice")
744 _make_commit(tmp_path, author="Bob")
745 result = _invoke(["shortlog", "--numbered", "--json"], _env(tmp_path))
746 assert result.exit_code == 0
747 self._assert_schema(json.loads(result.output))
748
749 def test_since_filtered_empty_has_schema(self, tmp_path: pathlib.Path) -> None:
750 """_emit_empty path (after filtering) must also carry the schema."""
751 _init_repo(tmp_path)
752 _make_commit(
753 tmp_path, author="Old",
754 committed_at=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc),
755 )
756 result = _invoke(["shortlog", "--since", "2030-01-01", "--json"], _env(tmp_path))
757 assert result.exit_code == 0
758 self._assert_schema(json.loads(result.output))
759
760 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
761 _init_repo(tmp_path)
762 _make_commit(tmp_path)
763 result = _invoke(["shortlog", "--json"], _env(tmp_path))
764 d = json.loads(result.output)
765 assert d["exit_code"] == 0
766
767
768 # ---------------------------------------------------------------------------
769 # truncated flag — set when --limit caps the result
770 # ---------------------------------------------------------------------------
771
772
773 class TestTruncated:
774 """truncated:true when --limit hit; false otherwise."""
775
776 def test_truncated_true_when_limit_hit(self, tmp_path: pathlib.Path) -> None:
777 _init_repo(tmp_path)
778 for _ in range(10):
779 _make_commit(tmp_path, author="Alice")
780 result = _invoke(["shortlog", "--limit", "3", "--json"], _env(tmp_path))
781 assert result.exit_code == 0
782 d = json.loads(result.output)
783 assert d["truncated"] is True
784
785 def test_truncated_false_when_under_limit(self, tmp_path: pathlib.Path) -> None:
786 _init_repo(tmp_path)
787 for _ in range(5):
788 _make_commit(tmp_path, author="Alice")
789 result = _invoke(["shortlog", "--limit", "10", "--json"], _env(tmp_path))
790 assert result.exit_code == 0
791 d = json.loads(result.output)
792 assert d["truncated"] is False
793
794 def test_truncated_false_when_no_limit(self, tmp_path: pathlib.Path) -> None:
795 _init_repo(tmp_path)
796 for _ in range(5):
797 _make_commit(tmp_path, author="Alice")
798 result = _invoke(["shortlog", "--json"], _env(tmp_path))
799 assert result.exit_code == 0
800 d = json.loads(result.output)
801 assert d["truncated"] is False
802
803 def test_truncated_false_on_empty_repo(self, tmp_path: pathlib.Path) -> None:
804 _init_repo(tmp_path)
805 result = _invoke(["shortlog", "--json"], _env(tmp_path))
806 assert result.exit_code == 0
807 d = json.loads(result.output)
808 assert d["truncated"] is False
809
810
811 # ---------------------------------------------------------------------------
812 # Error JSON — date parse errors emit structured JSON to stdout with --json
813 # ---------------------------------------------------------------------------
814
815
816 class TestErrorJson:
817 """--since / --until bad dates must emit JSON to stdout when --json is set."""
818
819 def _assert_error(self, result: InvokeResult) -> Mapping[str, object]:
820 assert result.exit_code != 0
821 d = json.loads(result.output)
822 assert "error" in d
823 assert "duration_ms" in d
824 assert "exit_code" in d
825 assert d["exit_code"] != 0
826 return d
827
828 def test_since_bad_date_json_error(self, tmp_path: pathlib.Path) -> None:
829 _init_repo(tmp_path)
830 _make_commit(tmp_path)
831 result = _invoke(["shortlog", "--json", "--since", "not-a-date"], _env(tmp_path))
832 self._assert_error(result)
833
834 def test_until_bad_date_json_error(self, tmp_path: pathlib.Path) -> None:
835 _init_repo(tmp_path)
836 _make_commit(tmp_path)
837 result = _invoke(["shortlog", "--json", "--until", "01/01/2025"], _env(tmp_path))
838 self._assert_error(result)
839
840 def test_date_error_has_message(self, tmp_path: pathlib.Path) -> None:
841 _init_repo(tmp_path)
842 result = _invoke(["shortlog", "--json", "--since", "garbage"], _env(tmp_path))
843 d = self._assert_error(result)
844 assert isinstance(d["message"], str) and len(d["message"]) > 0
845
846
847 # ---------------------------------------------------------------------------
848 # _parse_date refactor — now raises ValueError, not SystemExit
849 # ---------------------------------------------------------------------------
850
851
852 class TestParseDateRefactor:
853 """_parse_date is a pure parser; it raises ValueError, not SystemExit."""
854
855 def test_invalid_date_raises_value_error(self) -> None:
856 with pytest.raises(ValueError):
857 _parse_date("not-a-date", "--since")
858
859 def test_wrong_format_raises_value_error(self) -> None:
860 with pytest.raises(ValueError):
861 _parse_date("15/03/2025", "--since")
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago