gabriel / muse public
test_cmd_shortlog.py python
268 lines 8.7 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago
1 """Tests for ``muse shortlog``.
2
3 Covers: empty repo, single author, multiple authors, --numbered sort,
4 --email flag, --format json, --all branches, --limit, short flags,
5 stress: 200 commits across 3 authors.
6 """
7
8 from __future__ import annotations
9
10 import datetime
11 import json
12 import pathlib
13
14 import pytest
15 from tests.cli_test_helper import CliRunner
16
17 cli = None # argparse migration — CliRunner ignores this arg
18 from muse.core.object_store import write_object
19 from muse.core.ids import hash_commit, hash_snapshot
20 from muse.core.commits import (
21 CommitRecord,
22 write_commit,
23 )
24 from muse.core.snapshots import (
25 SnapshotRecord,
26 write_snapshot,
27 )
28 from muse.core.types import Manifest, blob_id
29 from muse.core.paths import muse_dir, ref_path
30
31 runner = CliRunner()
32
33 _REPO_ID = "shortlog-test"
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41
42
43 def _init_repo(path: pathlib.Path) -> pathlib.Path:
44 dot_muse = muse_dir(path)
45 for d in ("commits", "snapshots", "objects", "refs/heads"):
46 (dot_muse / d).mkdir(parents=True, exist_ok=True)
47 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
48 (dot_muse / "repo.json").write_text(
49 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
50 )
51 return path
52
53
54 def _env(repo: pathlib.Path) -> Manifest:
55 return {"MUSE_REPO_ROOT": str(repo)}
56
57
58 _counter = 0
59
60 # Per-branch tracking of the latest commit so tests can chain automatically.
61 _branch_heads: Manifest = {}
62
63
64 def _make_commit(
65 root: pathlib.Path,
66 author: str = "Alice",
67 parent_id: str | None = None,
68 branch: str = "main",
69 ) -> str:
70 """Create a commit, automatically chaining to the previous commit on the branch."""
71 global _counter
72 _counter += 1
73 # Auto-chain: if no explicit parent, use the last commit on this branch.
74 if parent_id is None:
75 parent_id = _branch_heads.get(f"{str(root)}:{branch}")
76 content = f"content-{_counter}".encode()
77 obj_id = blob_id(content)
78 write_object(root, obj_id, content)
79 manifest = {f"file_{_counter}.txt": obj_id}
80 snap_id = hash_snapshot(manifest)
81 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
82 committed_at = datetime.datetime.now(datetime.timezone.utc)
83 parent_ids = [parent_id] if parent_id else []
84 commit_id = hash_commit(
85 parent_ids=parent_ids,
86 snapshot_id=snap_id,
87 message=f"commit by {author} #{_counter}",
88 committed_at_iso=committed_at.isoformat(),
89 author=author,
90 )
91 write_commit(root, CommitRecord(
92 commit_id=commit_id,
93 branch=branch,
94 snapshot_id=snap_id,
95 message=f"commit by {author} #{_counter}",
96 committed_at=committed_at,
97 parent_commit_id=parent_id,
98 author=author,
99 ))
100 (ref_path(root, branch)).write_text(commit_id, encoding="utf-8")
101 _branch_heads[f"{str(root)}:{branch}"] = commit_id
102 return commit_id
103
104
105 # ---------------------------------------------------------------------------
106 # Unit: empty repo
107 # ---------------------------------------------------------------------------
108
109
110 def test_shortlog_empty_repo(tmp_path: pathlib.Path) -> None:
111 _init_repo(tmp_path)
112 result = runner.invoke(cli, ["shortlog"], env=_env(tmp_path))
113 assert result.exit_code == 0
114 assert "no commits" in result.output.lower()
115
116
117 def test_shortlog_help() -> None:
118 result = runner.invoke(cli, ["shortlog", "--help"])
119 assert result.exit_code == 0
120 assert "--numbered" in result.output or "-n" in result.output
121
122
123 # ---------------------------------------------------------------------------
124 # Unit: single author
125 # ---------------------------------------------------------------------------
126
127
128 def test_shortlog_single_author(tmp_path: pathlib.Path) -> None:
129 _init_repo(tmp_path)
130 _make_commit(tmp_path, author="Alice")
131 _make_commit(tmp_path, author="Alice")
132 result = runner.invoke(cli, ["shortlog"], env=_env(tmp_path))
133 assert result.exit_code == 0
134 assert "Alice" in result.output
135 assert "(2)" in result.output
136
137
138 # ---------------------------------------------------------------------------
139 # Unit: multiple authors
140 # ---------------------------------------------------------------------------
141
142
143 def test_shortlog_multiple_authors(tmp_path: pathlib.Path) -> None:
144 _init_repo(tmp_path)
145 _make_commit(tmp_path, author="Alice")
146 _make_commit(tmp_path, author="Bob")
147 _make_commit(tmp_path, author="Alice")
148 result = runner.invoke(cli, ["shortlog"], env=_env(tmp_path))
149 assert result.exit_code == 0
150 assert "Alice" in result.output
151 assert "Bob" in result.output
152
153
154 # ---------------------------------------------------------------------------
155 # Unit: --numbered sorts by count
156 # ---------------------------------------------------------------------------
157
158
159 def test_shortlog_numbered(tmp_path: pathlib.Path) -> None:
160 _init_repo(tmp_path)
161 _make_commit(tmp_path, author="Bob")
162 _make_commit(tmp_path, author="Alice")
163 _make_commit(tmp_path, author="Alice")
164 _make_commit(tmp_path, author="Alice")
165 result = runner.invoke(cli, ["shortlog", "--numbered"], env=_env(tmp_path))
166 assert result.exit_code == 0
167 alice_pos = result.output.index("Alice")
168 bob_pos = result.output.index("Bob")
169 assert alice_pos < bob_pos # Alice has more commits, should appear first
170
171
172 # ---------------------------------------------------------------------------
173 # Unit: --format json
174 # ---------------------------------------------------------------------------
175
176
177 def test_shortlog_json_output(tmp_path: pathlib.Path) -> None:
178 _init_repo(tmp_path)
179 _make_commit(tmp_path, author="Charlie")
180 result = runner.invoke(cli, ["shortlog", "--json"], env=_env(tmp_path))
181 assert result.exit_code == 0
182 data = json.loads(result.output)
183 assert isinstance(data, dict)
184 groups = data["groups"]
185 assert len(groups) >= 1
186 assert groups[0]["key"] == "Charlie"
187 assert groups[0]["count"] >= 1
188
189
190 # ---------------------------------------------------------------------------
191 # Unit: --limit
192 # ---------------------------------------------------------------------------
193
194
195 def test_shortlog_limit(tmp_path: pathlib.Path) -> None:
196 _init_repo(tmp_path)
197 for _ in range(20):
198 _make_commit(tmp_path, author="Dave")
199 result = runner.invoke(cli, ["shortlog", "--limit", "5", "--json"], env=_env(tmp_path))
200 assert result.exit_code == 0
201 data = json.loads(result.output)
202 total_commits = sum(g["count"] for g in data["groups"])
203 assert total_commits <= 5
204
205
206 # ---------------------------------------------------------------------------
207 # Unit: short flags
208 # ---------------------------------------------------------------------------
209
210
211 def test_shortlog_short_flags(tmp_path: pathlib.Path) -> None:
212 _init_repo(tmp_path)
213 _make_commit(tmp_path, author="Eve")
214 result = runner.invoke(cli, ["shortlog", "--numbered", "--json"], env=_env(tmp_path))
215 assert result.exit_code == 0
216 data = json.loads(result.output)
217 assert len(data["groups"]) >= 1
218
219
220 # ---------------------------------------------------------------------------
221 # Stress: 200 commits across 3 authors
222 # ---------------------------------------------------------------------------
223
224
225 def test_shortlog_stress_200_commits(tmp_path: pathlib.Path) -> None:
226 _init_repo(tmp_path)
227 authors = ["Frank", "Grace", "Heidi"]
228 for i in range(200):
229 _make_commit(tmp_path, author=authors[i % 3])
230
231 result = runner.invoke(cli, ["shortlog", "--json"], env=_env(tmp_path))
232 assert result.exit_code == 0
233 data = json.loads(result.output)
234 total = sum(g["count"] for g in data["groups"])
235 assert total == 200
236 names = {g["key"] for g in data["groups"]}
237 assert "Frank" in names
238 assert "Grace" in names
239 assert "Heidi" in names
240
241
242 class TestRegisterFlags:
243 def test_default_json_out_is_false(self) -> None:
244 import argparse
245 from muse.cli.commands.shortlog import register
246 p = argparse.ArgumentParser()
247 subs = p.add_subparsers()
248 register(subs)
249 args = p.parse_args(["shortlog"])
250 assert args.json_out is False
251
252 def test_json_flag_sets_json_out(self) -> None:
253 import argparse
254 from muse.cli.commands.shortlog import register
255 p = argparse.ArgumentParser()
256 subs = p.add_subparsers()
257 register(subs)
258 args = p.parse_args(["shortlog", "--json"])
259 assert args.json_out is True
260
261 def test_j_shorthand_sets_json_out(self) -> None:
262 import argparse
263 from muse.cli.commands.shortlog import register
264 p = argparse.ArgumentParser()
265 subs = p.add_subparsers()
266 register(subs)
267 args = p.parse_args(["shortlog", "-j"])
268 assert args.json_out is True
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 7 days ago