"""Hardening test suite for ``muse shortlog``. Coverage: - Unit: _branch_names (symlink guard), _group_key (all four modes), _build_groups (email flag, dedup), _parse_date (valid + invalid) - Security: ANSI in author/message sanitized in text, raw in JSON; symlink inside refs/heads is skipped - Error routing: all user errors go to stderr - JSON schema: _ShortlogJson shape (repo_id, branch, groups), all fields - New flags: --group-by (agent, model, branch), --summary, --no-merges, --since, --until, combined filters - --json: empty, single group, multi-group, provenance fields - Integration: --all branches with dedup, --limit early-exit, date range - E2E: help output, combined flags - Stress: 500 commits × 5 authors, 50-branch repo, concurrent reads """ from __future__ import annotations from collections.abc import Mapping import datetime import json import os import pathlib from typing import TypedDict from unittest.mock import patch import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.cli.commands.shortlog import _branch_names, _build_groups, _group_key, _parse_date from muse.core.object_store import write_object from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, blob_id from muse.core.paths import heads_dir, muse_dir runner = CliRunner() _REPO_ID = "shortlog-hard-test" # Tracks the latest commit_id per (str(root), branch) so _make_commit # can auto-chain without callers needing to pass parent_id explicitly. _branch_heads_map: Manifest = {} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path, *, domain: str = "code") -> pathlib.Path: muse = muse_dir(path) for sub in ("commits", "snapshots", "objects", "refs/heads"): (muse / sub).mkdir(parents=True, exist_ok=True) (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (muse / "repo.json").write_text( json.dumps({"repo_id": _REPO_ID, "domain": domain}), encoding="utf-8", ) return path _commit_counter = 0 def _make_commit( root: pathlib.Path, *, author: str = "Alice", agent_id: str | None = None, model_id: str | None = None, branch: str = "main", parent_id: str | None = None, parent2_id: str | None = None, committed_at: datetime.datetime | None = None, ) -> str: """Create and store a commit, auto-chaining to the previous on the same branch.""" global _commit_counter _commit_counter += 1 content = f"c{_commit_counter}".encode() obj_id = blob_id(content) write_object(root, obj_id, content) manifest = {f"f{_commit_counter}.txt": obj_id} snap_id = hash_snapshot(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) ts = committed_at or datetime.datetime.now(datetime.timezone.utc) # Auto-chain: if caller didn't provide parent_id, use the last known head. effective_parent = parent_id if effective_parent is None: effective_parent = _branch_heads_map.get(f"{root}:{branch}") pids = [pid for pid in (effective_parent, parent2_id) if pid is not None] commit_id = hash_commit( parent_ids=pids, snapshot_id=snap_id, message=f"msg {_commit_counter}", committed_at_iso=ts.isoformat(), author=author, ) rec = CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=f"msg {_commit_counter}", committed_at=ts, parent_commit_id=effective_parent, parent2_commit_id=parent2_id, author=author, agent_id=agent_id or "", model_id=model_id or "", ) write_commit(root, rec) ref_dir = heads_dir(root) ref_file = ref_dir / branch ref_file.parent.mkdir(parents=True, exist_ok=True) ref_file.write_text(commit_id, encoding="utf-8") _branch_heads_map[f"{root}:{branch}"] = commit_id return commit_id def _env(repo: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(repo)} def _invoke(args: list[str], env: Manifest) -> InvokeResult: return runner.invoke(None, args, env=env) class _GroupOut(TypedDict): key: str count: int commits: list[Mapping[str, str | None]] class _ShortlogOut(TypedDict): repo_id: str branch: str groups: list[_GroupOut] def _parse_json(result: InvokeResult) -> _ShortlogOut: raw = json.loads(result.output.strip()) groups: list[_GroupOut] = [ _GroupOut( key=g["key"], count=g["count"], commits=g["commits"], ) for g in raw["groups"] ] return _ShortlogOut( repo_id=raw["repo_id"], branch=raw["branch"], groups=groups, ) # --------------------------------------------------------------------------- # Unit: _branch_names — symlink guard # --------------------------------------------------------------------------- def test_branch_names_returns_normal_branches(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, branch="main") _make_commit(tmp_path, branch="dev") names = _branch_names(tmp_path) assert "main" in names assert "dev" in names def test_branch_names_skips_symlinks(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, branch="main") h_dir = heads_dir(tmp_path) malicious = h_dir / "malicious-branch" try: malicious.symlink_to(tmp_path / "some_other_file") except OSError: pytest.skip("filesystem does not support symlinks") names = _branch_names(tmp_path) assert "malicious-branch" not in names assert "main" in names def test_branch_names_missing_heads_dir(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) import shutil shutil.rmtree(heads_dir(tmp_path)) assert _branch_names(tmp_path) == [] # --------------------------------------------------------------------------- # Unit: _group_key # --------------------------------------------------------------------------- def _make_rec( *, author: str = "", agent_id: str = "", model_id: str = "", branch: str = "main", ) -> CommitRecord: return CommitRecord( commit_id="aaa", branch=branch, snapshot_id="snap", message="x", committed_at=datetime.datetime.now(datetime.timezone.utc), author=author, agent_id=agent_id, model_id=model_id, ) def test_group_key_author_with_author() -> None: rec = _make_rec(author="Alice") assert _group_key(rec, "author") == "Alice" def test_group_key_author_fallback_to_agent() -> None: rec = _make_rec(agent_id="bot-1") assert _group_key(rec, "author") == "bot-1 (agent)" def test_group_key_author_unknown() -> None: rec = _make_rec() assert _group_key(rec, "author") == "(unknown)" def test_group_key_agent() -> None: rec = _make_rec(agent_id="gpt-agent") assert _group_key(rec, "agent") == "gpt-agent" def test_group_key_agent_no_agent() -> None: rec = _make_rec() assert _group_key(rec, "agent") == "(no agent)" def test_group_key_model() -> None: rec = _make_rec(model_id="gpt-4o") assert _group_key(rec, "model") == "gpt-4o" def test_group_key_model_no_model() -> None: rec = _make_rec() assert _group_key(rec, "model") == "(no model)" def test_group_key_branch() -> None: rec = _make_rec(branch="feat/my-thing") assert _group_key(rec, "branch") == "feat/my-thing" # --------------------------------------------------------------------------- # Unit: _parse_date # --------------------------------------------------------------------------- def test_parse_date_valid() -> None: dt = _parse_date("2025-03-15", "--since") assert dt.year == 2025 assert dt.month == 3 assert dt.day == 15 assert dt.tzinfo == datetime.timezone.utc def test_parse_date_invalid_exits() -> None: with pytest.raises(ValueError): _parse_date("not-a-date", "--since") def test_parse_date_wrong_format_exits() -> None: with pytest.raises(ValueError): _parse_date("15/03/2025", "--since") # --------------------------------------------------------------------------- # Security: ANSI injection # --------------------------------------------------------------------------- def test_ansi_in_author_name_stripped_text(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Evil\x1b[31mRED\x1b[0m") result = _invoke(["shortlog"], _env(tmp_path)) assert result.exit_code == 0 assert "\x1b[31m" not in result.output def test_ansi_in_author_name_raw_in_json(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Evil\x1b[31mRED\x1b[0m") result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["groups"][0]["key"] == "Evil\x1b[31mRED\x1b[0m" def test_ansi_in_message_stripped_text(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) commit_id = _make_commit(tmp_path) # Directly overwrite message in stored commit to contain ANSI. from muse.core.commits import read_commit original = read_commit(tmp_path, commit_id) assert original is not None from muse.core.ids import hash_commit from muse.core.commits import write_commit malicious_msg = "fix: \x1b[1mBOLD\x1b[0m thing" parent_ids = [original.parent_commit_id] if original.parent_commit_id else [] new_cid = hash_commit( parent_ids=parent_ids, snapshot_id=original.snapshot_id, message=malicious_msg, committed_at_iso=original.committed_at.isoformat(), author=original.author, ) patched = CommitRecord( commit_id=new_cid, branch=original.branch, snapshot_id=original.snapshot_id, message=malicious_msg, committed_at=original.committed_at, author=original.author, ) write_commit(tmp_path, patched) (heads_dir(tmp_path) / "main").write_text(new_cid) result = _invoke(["shortlog"], _env(tmp_path)) assert "\x1b[1m" not in result.output # --------------------------------------------------------------------------- # Error routing: all user errors go to stderr # --------------------------------------------------------------------------- def test_since_invalid_format_stderr(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["shortlog", "--since", "01-01-2025"], _env(tmp_path)) assert result.exit_code != 0 def test_until_invalid_format_stderr(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["shortlog", "--until", "not-a-date"], _env(tmp_path)) assert result.exit_code != 0 # --------------------------------------------------------------------------- # JSON schema: _ShortlogJson # --------------------------------------------------------------------------- def test_json_schema_empty_repo(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["repo_id"] == _REPO_ID assert data["branch"] == "main" assert data["groups"] == [] def test_json_schema_all_fields_present(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Alice", agent_id="bot-1", model_id="gpt-4o") result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["repo_id"] == _REPO_ID assert data["branch"] == "main" grp = data["groups"][0] assert grp["key"] == "Alice" assert grp["count"] == 1 commit_entry = grp["commits"][0] assert "commit_id" in commit_entry assert "message" in commit_entry assert "committed_at" in commit_entry assert "author" in commit_entry assert "agent_id" in commit_entry assert "model_id" in commit_entry def test_json_schema_repo_id_and_branch_in_output(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, branch="main") result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["repo_id"] == _REPO_ID assert data["branch"] == "main" def test_json_schema_all_branches_label(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, branch="main") result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["branch"] == "__all__" def test_json_agent_id_and_model_id_present(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, agent_id="agent-007", model_id="claude-3") result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) entry = data["groups"][0]["commits"][0] assert entry["agent_id"] == "agent-007" assert entry["model_id"] == "claude-3" # --------------------------------------------------------------------------- # New flag: --group-by # --------------------------------------------------------------------------- def test_group_by_agent(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Alice", agent_id="bot-1") _make_commit(tmp_path, author="Bob", agent_id="bot-2") _make_commit(tmp_path, author="Alice", agent_id="bot-1") result = _invoke(["shortlog", "--group-by", "agent", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) keys = {g["key"] for g in data["groups"]} assert "bot-1" in keys assert "bot-2" in keys def test_group_by_model(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, model_id="gpt-4o") _make_commit(tmp_path, model_id="claude-3") _make_commit(tmp_path, model_id="gpt-4o") result = _invoke(["shortlog", "--group-by", "model", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) keys = {g["key"] for g in data["groups"]} assert "gpt-4o" in keys assert "claude-3" in keys gpt_count = next(g["count"] for g in data["groups"] if g["key"] == "gpt-4o") assert gpt_count == 2 def test_group_by_branch(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, branch="main") _make_commit(tmp_path, branch="dev") _make_commit(tmp_path, branch="main") result = _invoke( ["shortlog", "--all", "--group-by", "branch", "--json"], _env(tmp_path) ) assert result.exit_code == 0 data = _parse_json(result) keys = {g["key"] for g in data["groups"]} assert "main" in keys assert "dev" in keys def test_group_by_invalid_choice(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["shortlog", "--group-by", "badfield"], _env(tmp_path)) assert result.exit_code != 0 # --------------------------------------------------------------------------- # New flag: --summary # --------------------------------------------------------------------------- def test_summary_suppresses_messages(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Alice") _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--summary"], _env(tmp_path)) assert result.exit_code == 0 # Author line should still appear. assert "Alice" in result.output # Individual commit messages should not appear (they start with spaces). assert "msg" not in result.output def test_summary_with_json_still_includes_commits(tmp_path: pathlib.Path) -> None: """--summary only suppresses messages in text mode; JSON always includes them.""" _init_repo(tmp_path) _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--summary", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert len(data["groups"][0]["commits"]) >= 1 # --------------------------------------------------------------------------- # New flag: --no-merges # --------------------------------------------------------------------------- def test_no_merges_excludes_merge_commits(tmp_path: pathlib.Path) -> None: """get_commits_for_branch follows first-parent only. Chain: c1 → c2 → c3(merge, parent2=c1) → c4 First-parent walk from c4 returns [c4, c3, c2, c1]. With --no-merges, c3 is excluded → 3 commits remain. """ _init_repo(tmp_path) c1 = _make_commit(tmp_path, author="Alice") c2 = _make_commit(tmp_path, author="Bob") # chains to c1 # Merge commit: auto-chains first-parent to c2; parent2 points to c1. _make_commit(tmp_path, author="Alice", parent2_id=c1) # chains to c2 _make_commit(tmp_path, author="Bob") # chains to merge result = _invoke(["shortlog", "--no-merges", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) total = sum(g["count"] for g in data["groups"]) assert total == 3 # c1, c2, c4 — c3 (merge) excluded def test_no_merges_with_all_non_merges(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for _ in range(5): _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--no-merges", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert sum(g["count"] for g in data["groups"]) == 5 # --------------------------------------------------------------------------- # New flags: --since / --until # --------------------------------------------------------------------------- def test_since_filters_old_commits(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc) new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc) _make_commit(tmp_path, author="Old", committed_at=old) _make_commit(tmp_path, author="New", committed_at=new) result = _invoke(["shortlog", "--since", "2025-01-01", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) keys = {g["key"] for g in data["groups"]} assert "New" in keys assert "Old" not in keys def test_until_filters_future_commits(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc) new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc) _make_commit(tmp_path, author="Old", committed_at=old) _make_commit(tmp_path, author="New", committed_at=new) result = _invoke(["shortlog", "--until", "2022-12-31", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) keys = {g["key"] for g in data["groups"]} assert "Old" in keys assert "New" not in keys def test_since_and_until_window(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) dates = [ datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 3, 15, tzinfo=datetime.timezone.utc), datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), ] authors = ["Before", "Inside", "After"] for a, d in zip(authors, dates): _make_commit(tmp_path, author=a, committed_at=d) result = _invoke( ["shortlog", "--since", "2025-01-01", "--until", "2025-12-31", "--json"], _env(tmp_path), ) assert result.exit_code == 0 data = _parse_json(result) keys = {g["key"] for g in data["groups"]} assert "Inside" in keys assert "Before" not in keys assert "After" not in keys def test_since_no_results_returns_empty_json(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit( tmp_path, author="Old", committed_at=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc), ) result = _invoke(["shortlog", "--since", "2030-01-01", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["groups"] == [] # --------------------------------------------------------------------------- # Integration # --------------------------------------------------------------------------- def test_integration_all_branches_dedup(tmp_path: pathlib.Path) -> None: """A commit reachable from two branches should count once.""" _init_repo(tmp_path) shared = _make_commit(tmp_path, author="Alice", branch="main") # Create dev branch pointing at same commit (by writing the ref file). dev_ref = heads_dir(tmp_path) / "dev" dev_ref.write_text(shared, encoding="utf-8") result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) total = sum(g["count"] for g in data["groups"]) assert total == 1 # deduplicated def test_integration_limit_early_exit(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for i in range(50): _make_commit(tmp_path, author=f"Author{i % 5}") result = _invoke(["shortlog", "--limit", "10", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) total = sum(g["count"] for g in data["groups"]) assert total <= 10 def test_integration_numbered_combined_with_since(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) old = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc) new = datetime.datetime(2025, 6, 1, tzinfo=datetime.timezone.utc) for _ in range(3): _make_commit(tmp_path, author="Prolific", committed_at=new) _make_commit(tmp_path, author="Old", committed_at=old) result = _invoke( ["shortlog", "--since", "2025-01-01", "--numbered", "--json"], _env(tmp_path), ) assert result.exit_code == 0 data = _parse_json(result) assert data["groups"][0]["key"] == "Prolific" assert "Old" not in {g["key"] for g in data["groups"]} # --------------------------------------------------------------------------- # E2E: help output # --------------------------------------------------------------------------- def test_help_shows_new_flags() -> None: result = _invoke(["shortlog", "--help"], {}) assert result.exit_code == 0 for flag in ("--group-by", "--summary", "--no-merges", "--since", "--until", "--json"): assert flag in result.output, f"Missing flag: {flag}" def test_help_mentions_group_by_choices() -> None: result = _invoke(["shortlog", "--help"], {}) for choice in ("author", "agent", "model", "branch"): assert choice in result.output # --------------------------------------------------------------------------- # Stress: 500 commits × 5 authors # --------------------------------------------------------------------------- def test_stress_500_commits(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) authors = ["Amy", "Ben", "Cleo", "Dan", "Eva"] for i in range(500): _make_commit(tmp_path, author=authors[i % 5]) result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) total = sum(g["count"] for g in data["groups"]) assert total == 500 assert len(data["groups"]) == 5 def test_stress_500_commits_numbered(tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) # Give Alice 300, Bob 200. for _ in range(300): _make_commit(tmp_path, author="Alice") for _ in range(200): _make_commit(tmp_path, author="Bob") result = _invoke(["shortlog", "--numbered", "--json"], _env(tmp_path)) assert result.exit_code == 0 data = _parse_json(result) assert data["groups"][0]["key"] == "Alice" assert data["groups"][0]["count"] == 300 # --------------------------------------------------------------------------- # JSON schema — duration_ms + exit_code + truncated on every output path # --------------------------------------------------------------------------- class TestJsonSchema: """Every --json response must carry duration_ms, exit_code, and truncated.""" def _assert_schema(self, d: Mapping[str, object], *, exit_code: int = 0) -> None: assert "duration_ms" in d, f"duration_ms missing: {d}" assert isinstance(d["duration_ms"], (int, float)) assert d["duration_ms"] >= 0 assert "exit_code" in d, f"exit_code missing: {d}" assert d["exit_code"] == exit_code assert "truncated" in d, f"truncated missing: {d}" def test_normal_output_has_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 self._assert_schema(json.loads(result.output)) def test_empty_repo_json_has_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 self._assert_schema(json.loads(result.output)) def test_all_branches_json_has_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, branch="main") result = _invoke(["shortlog", "--all", "--json"], _env(tmp_path)) assert result.exit_code == 0 self._assert_schema(json.loads(result.output)) def test_numbered_json_has_schema(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path, author="Alice") _make_commit(tmp_path, author="Bob") result = _invoke(["shortlog", "--numbered", "--json"], _env(tmp_path)) assert result.exit_code == 0 self._assert_schema(json.loads(result.output)) def test_since_filtered_empty_has_schema(self, tmp_path: pathlib.Path) -> None: """_emit_empty path (after filtering) must also carry the schema.""" _init_repo(tmp_path) _make_commit( tmp_path, author="Old", committed_at=datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc), ) result = _invoke(["shortlog", "--since", "2030-01-01", "--json"], _env(tmp_path)) assert result.exit_code == 0 self._assert_schema(json.loads(result.output)) def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["shortlog", "--json"], _env(tmp_path)) d = json.loads(result.output) assert d["exit_code"] == 0 # --------------------------------------------------------------------------- # truncated flag — set when --limit caps the result # --------------------------------------------------------------------------- class TestTruncated: """truncated:true when --limit hit; false otherwise.""" def test_truncated_true_when_limit_hit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for _ in range(10): _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--limit", "3", "--json"], _env(tmp_path)) assert result.exit_code == 0 d = json.loads(result.output) assert d["truncated"] is True def test_truncated_false_when_under_limit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for _ in range(5): _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--limit", "10", "--json"], _env(tmp_path)) assert result.exit_code == 0 d = json.loads(result.output) assert d["truncated"] is False def test_truncated_false_when_no_limit(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) for _ in range(5): _make_commit(tmp_path, author="Alice") result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 d = json.loads(result.output) assert d["truncated"] is False def test_truncated_false_on_empty_repo(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["shortlog", "--json"], _env(tmp_path)) assert result.exit_code == 0 d = json.loads(result.output) assert d["truncated"] is False # --------------------------------------------------------------------------- # Error JSON — date parse errors emit structured JSON to stdout with --json # --------------------------------------------------------------------------- class TestErrorJson: """--since / --until bad dates must emit JSON to stdout when --json is set.""" def _assert_error(self, result: InvokeResult) -> Mapping[str, object]: assert result.exit_code != 0 d = json.loads(result.output) assert "error" in d assert "duration_ms" in d assert "exit_code" in d assert d["exit_code"] != 0 return d def test_since_bad_date_json_error(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["shortlog", "--json", "--since", "not-a-date"], _env(tmp_path)) self._assert_error(result) def test_until_bad_date_json_error(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) _make_commit(tmp_path) result = _invoke(["shortlog", "--json", "--until", "01/01/2025"], _env(tmp_path)) self._assert_error(result) def test_date_error_has_message(self, tmp_path: pathlib.Path) -> None: _init_repo(tmp_path) result = _invoke(["shortlog", "--json", "--since", "garbage"], _env(tmp_path)) d = self._assert_error(result) assert isinstance(d["message"], str) and len(d["message"]) > 0 # --------------------------------------------------------------------------- # _parse_date refactor — now raises ValueError, not SystemExit # --------------------------------------------------------------------------- class TestParseDateRefactor: """_parse_date is a pure parser; it raises ValueError, not SystemExit.""" def test_invalid_date_raises_value_error(self) -> None: with pytest.raises(ValueError): _parse_date("not-a-date", "--since") def test_wrong_format_raises_value_error(self) -> None: with pytest.raises(ValueError): _parse_date("15/03/2025", "--since")