gabriel / muse public
test_cmd_range_diff.py python
422 lines 18.3 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 149 days ago
1 """Tests for ``muse range-diff`` — compare two versions of a commit series.
2
3 Coverage tiers:
4 - Unit: range parsing, patch-id computation per commit, pairing logic
5 (identical series, single changed, dropped commit, added commit,
6 reordered, empty series)
7 - Integration: JSON schema, text output, nonexistent ref exits nonzero,
8 trivially equivalent flag, creation-factor=0 no fuzzy pairing,
9 both ranges empty
10 - Security: ANSI in range arg rejected
11 - Stress: 50-commit series fully equivalent (sub-second); 50-commit with
12 25 changed, 10 dropped, 10 added
13 """
14
15 from __future__ import annotations
16
17 import datetime
18 import hashlib
19 import json
20 import pathlib
21
22 import pytest
23
24 from tests.cli_test_helper import CliRunner
25 from muse.core.object_store import write_object
26 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
27 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
28 from muse.core._types import Manifest
29
30 runner = CliRunner()
31
32 _REPO_ID = "range-diff-test"
33 _counter = 0
34
35
36 # ---------------------------------------------------------------------------
37 # Helpers
38 # ---------------------------------------------------------------------------
39
40
41 def _sha(data: bytes) -> str:
42 return hashlib.sha256(data).hexdigest()
43
44
45 def _init_repo(path: pathlib.Path) -> pathlib.Path:
46 muse = path / ".muse"
47 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
48 (muse / d).mkdir(parents=True, exist_ok=True)
49 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
50 (muse / "repo.json").write_text(
51 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
52 )
53 return path
54
55
56 def _env(repo: pathlib.Path) -> dict[str, str]:
57 return {"MUSE_REPO_ROOT": str(repo)}
58
59
60 def _write_files(root: pathlib.Path, files: dict[str, bytes]) -> Manifest:
61 manifest: Manifest = {}
62 for rel, content in files.items():
63 oid = _sha(content)
64 write_object(root, oid, content)
65 manifest[rel] = oid
66 p = root / rel
67 p.parent.mkdir(parents=True, exist_ok=True)
68 p.write_bytes(content)
69 return manifest
70
71
72 def _commit(
73 root: pathlib.Path,
74 files: dict[str, bytes],
75 branch: str = "main",
76 parent_id: str | None = None,
77 message: str | None = None,
78 ) -> str:
79 global _counter
80 _counter += 1
81 manifest = _write_files(root, files)
82 snap_id = compute_snapshot_id(manifest)
83 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
84 committed_at = datetime.datetime.now(datetime.timezone.utc)
85 msg = message or f"commit {_counter}"
86 commit_id = compute_commit_id(
87 [parent_id] if parent_id else [], snap_id, msg, committed_at.isoformat()
88 )
89 write_commit(root, CommitRecord(
90 commit_id=commit_id, repo_id=_REPO_ID, branch=branch,
91 snapshot_id=snap_id, message=msg, committed_at=committed_at,
92 parent_commit_id=parent_id,
93 ))
94 ref_path = root / ".muse" / "refs" / "heads" / branch
95 ref_path.parent.mkdir(parents=True, exist_ok=True)
96 ref_path.write_text(commit_id, encoding="utf-8")
97 return commit_id
98
99
100 def _invoke(repo: pathlib.Path, *args: str):
101 from muse.cli.app import main as cli
102 return runner.invoke(cli, ["range-diff", *args], env=_env(repo))
103
104
105 # ---------------------------------------------------------------------------
106 # Unit — range parsing (imported directly)
107 # ---------------------------------------------------------------------------
108
109
110 def test_parse_range_with_dotdot() -> None:
111 from muse.cli.commands.range_diff import _parse_range
112 base, tip = _parse_range("abc..def")
113 assert base == "abc"
114 assert tip == "def"
115
116
117 def test_parse_range_no_dotdot() -> None:
118 from muse.cli.commands.range_diff import _parse_range
119 base, tip = _parse_range("main")
120 assert base is None
121 assert tip == "main"
122
123
124 def test_parse_range_preserves_whitespace_stripped() -> None:
125 from muse.cli.commands.range_diff import _parse_range
126 base, tip = _parse_range("base .. tip")
127 assert base == "base"
128 assert tip == "tip"
129
130
131 # ---------------------------------------------------------------------------
132 # Unit — pairing logic
133 # ---------------------------------------------------------------------------
134
135
136 def test_identical_series_all_equivalent(tmp_path: pathlib.Path) -> None:
137 root = _init_repo(tmp_path)
138 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
139 # old series: 3 commits
140 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
141 c2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="old", parent_id=c1)
142 c3 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n", "c.py": b"c=3\n"}, branch="old", parent_id=c2)
143 # new series: identical content → same patch-ids
144 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
145 n2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="new", parent_id=n1)
146 n3 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n", "c.py": b"c=3\n"}, branch="new", parent_id=n2)
147 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
148 assert result.exit_code == 0
149 data = json.loads(result.stdout)
150 assert data["trivially_equivalent"] is True
151 assert all(p["status"] == "equivalent" for p in data["pairs"])
152
153
154 def test_single_commit_changed(tmp_path: pathlib.Path) -> None:
155 root = _init_repo(tmp_path)
156 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
157 # old: add a.py with content v1
158 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
159 # new: add a.py with content v2 (different patch-id)
160 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
161 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
162 data = json.loads(result.stdout)
163 assert data["trivially_equivalent"] is False
164 # One pair, status should be "changed" (different patch-ids, positionally paired)
165 assert len(data["pairs"]) == 1
166 assert data["pairs"][0]["status"] == "changed"
167
168
169 def test_commit_dropped(tmp_path: pathlib.Path) -> None:
170 """Old series has 2 commits; new series only has 1 (one was dropped/squashed)."""
171 root = _init_repo(tmp_path)
172 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
173 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base, message="add a")
174 c2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="old", parent_id=c1, message="add b")
175 # new: squashed into one commit with same final content as old c2
176 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "b.py": b"b=2\n"}, branch="new", parent_id=base, message="add a and b")
177 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
178 data = json.loads(result.stdout)
179 statuses = {p["status"] for p in data["pairs"]}
180 # At least one commit should be dropped or the squash results in a "changed" pair
181 assert "dropped" in statuses or "changed" in statuses
182
183
184 def test_commit_added(tmp_path: pathlib.Path) -> None:
185 """New series has an extra commit not in old series."""
186 root = _init_repo(tmp_path)
187 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
188 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
189 # new: same first commit + an extra one
190 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
191 n2 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n", "extra.py": b"e=9\n"}, branch="new", parent_id=n1)
192 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
193 data = json.loads(result.stdout)
194 statuses = [p["status"] for p in data["pairs"]]
195 assert "added" in statuses
196 assert "equivalent" in statuses # c1 ↔ n1
197
198
199 def test_empty_old_series_all_added(tmp_path: pathlib.Path) -> None:
200 root = _init_repo(tmp_path)
201 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
202 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
203 # old range is base..base → empty
204 result = _invoke(root, f"{base}..{base}", f"{base}..new", "--json")
205 data = json.loads(result.stdout)
206 assert all(p["status"] == "added" for p in data["pairs"])
207
208
209 def test_empty_new_series_all_dropped(tmp_path: pathlib.Path) -> None:
210 root = _init_repo(tmp_path)
211 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
212 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
213 # new range is base..base → empty
214 result = _invoke(root, f"{base}..old", f"{base}..{base}", "--json")
215 data = json.loads(result.stdout)
216 assert all(p["status"] == "dropped" for p in data["pairs"])
217
218
219 def test_both_empty_trivially_equivalent(tmp_path: pathlib.Path) -> None:
220 root = _init_repo(tmp_path)
221 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
222 result = _invoke(root, f"{base}..{base}", f"{base}..{base}", "--json")
223 assert result.exit_code == 0
224 data = json.loads(result.stdout)
225 assert data["trivially_equivalent"] is True
226 assert data["pairs"] == []
227
228
229 # ---------------------------------------------------------------------------
230 # Integration — JSON schema
231 # ---------------------------------------------------------------------------
232
233
234 def test_json_schema_all_fields(tmp_path: pathlib.Path) -> None:
235 root = _init_repo(tmp_path)
236 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
237 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
238 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
239 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
240 assert result.exit_code == 0
241 data = json.loads(result.stdout)
242 for key in ("pairs", "trivially_equivalent", "old_range", "new_range"):
243 assert key in data
244 pair = data["pairs"][0]
245 for key in ("old", "new", "status"):
246 assert key in pair
247 if pair["old"] is not None:
248 for key in ("commit_id", "patch_id", "subject"):
249 assert key in pair["old"]
250
251
252 def test_json_pair_commit_info(tmp_path: pathlib.Path) -> None:
253 root = _init_repo(tmp_path)
254 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
255 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base, message="add a")
256 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base, message="add a")
257 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
258 data = json.loads(result.stdout)
259 pair = data["pairs"][0]
260 assert pair["old"]["subject"] == "add a"
261 assert pair["new"]["subject"] == "add a"
262 assert pair["old"]["commit_id"] == c1
263 assert pair["new"]["commit_id"] == n1
264 assert pair["status"] == "equivalent"
265
266
267 # ---------------------------------------------------------------------------
268 # Integration — text output
269 # ---------------------------------------------------------------------------
270
271
272 def test_text_output_shows_equivalent(tmp_path: pathlib.Path) -> None:
273 root = _init_repo(tmp_path)
274 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
275 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base, message="add a")
276 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base, message="add a")
277 result = _invoke(root, f"{base}..old", f"{base}..new")
278 assert result.exit_code == 0
279 assert "equivalent" in result.stdout.lower() or "=" in result.stdout
280
281
282 def test_text_output_shows_changed(tmp_path: pathlib.Path) -> None:
283 root = _init_repo(tmp_path)
284 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
285 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base, message="add a v1")
286 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base, message="add a v2")
287 result = _invoke(root, f"{base}..old", f"{base}..new")
288 assert result.exit_code != 0 # has changes
289 assert "changed" in result.stdout.lower() or "!" in result.stdout
290
291
292 # ---------------------------------------------------------------------------
293 # Integration — creation-factor
294 # ---------------------------------------------------------------------------
295
296
297 def test_creation_factor_zero_no_fuzzy_pairing(tmp_path: pathlib.Path) -> None:
298 """With --creation-factor=0.0, only exact patch-id matches are paired."""
299 root = _init_repo(tmp_path)
300 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
301 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
302 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
303 result = _invoke(root, f"{base}..old", f"{base}..new", "--creation-factor", "0.0", "--json")
304 data = json.loads(result.stdout)
305 # No exact patch-id match, creation-factor=0 → dropped + added
306 statuses = {p["status"] for p in data["pairs"]}
307 assert "changed" not in statuses
308 assert "dropped" in statuses or "added" in statuses
309
310
311 def test_creation_factor_one_all_positionally_paired(tmp_path: pathlib.Path) -> None:
312 root = _init_repo(tmp_path)
313 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
314 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
315 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
316 result = _invoke(root, f"{base}..old", f"{base}..new", "--creation-factor", "1.0", "--json")
317 data = json.loads(result.stdout)
318 statuses = {p["status"] for p in data["pairs"]}
319 assert "changed" in statuses # positionally paired → changed
320
321
322 # ---------------------------------------------------------------------------
323 # Integration — error cases
324 # ---------------------------------------------------------------------------
325
326
327 def test_nonexistent_old_ref_exits_nonzero(tmp_path: pathlib.Path) -> None:
328 root = _init_repo(tmp_path)
329 base = _commit(root, {"a.py": b"x\n"}, branch="main")
330 result = _invoke(root, "ghost..no-such", f"{base}..main")
331 assert result.exit_code != 0
332
333
334 def test_nonexistent_new_ref_exits_nonzero(tmp_path: pathlib.Path) -> None:
335 root = _init_repo(tmp_path)
336 base = _commit(root, {"a.py": b"x\n"}, branch="main")
337 result = _invoke(root, f"{base}..main", "ghost..no-such")
338 assert result.exit_code != 0
339
340
341 def test_exit_zero_on_trivially_equivalent(tmp_path: pathlib.Path) -> None:
342 root = _init_repo(tmp_path)
343 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
344 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="old", parent_id=base)
345 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"a=1\n"}, branch="new", parent_id=base)
346 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
347 assert result.exit_code == 0
348
349
350 def test_exit_nonzero_on_differences(tmp_path: pathlib.Path) -> None:
351 root = _init_repo(tmp_path)
352 base = _commit(root, {"readme.txt": b"base\n"}, branch="main")
353 c1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v1\n"}, branch="old", parent_id=base)
354 n1 = _commit(root, {"readme.txt": b"base\n", "a.py": b"v2\n"}, branch="new", parent_id=base)
355 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
356 assert result.exit_code != 0
357
358
359 # ---------------------------------------------------------------------------
360 # Security
361 # ---------------------------------------------------------------------------
362
363
364 def test_ansi_in_old_range_rejected(tmp_path: pathlib.Path) -> None:
365 root = _init_repo(tmp_path)
366 result = _invoke(root, "\x1b[31mbad\x1b[0m..main", "main..main")
367 assert result.exit_code != 0
368
369
370 def test_ansi_in_new_range_rejected(tmp_path: pathlib.Path) -> None:
371 root = _init_repo(tmp_path)
372 result = _invoke(root, "main..main", "\x1b[31mbad\x1b[0m..main")
373 assert result.exit_code != 0
374
375
376 # ---------------------------------------------------------------------------
377 # Stress — 50 commits
378 # ---------------------------------------------------------------------------
379
380
381 def test_stress_50_commits_trivially_equivalent(tmp_path: pathlib.Path) -> None:
382 """50-commit series on each side, all equivalent — must complete quickly."""
383 root = _init_repo(tmp_path)
384 base = _commit(root, {"base.py": b"base\n"}, branch="main")
385 old_id = base
386 new_id = base
387 for i in range(50):
388 content = f"v = {i}\n".encode()
389 old_id = _commit(root, {f"f{i}.py": content}, branch="old", parent_id=old_id, message=f"add f{i}")
390 new_id = _commit(root, {f"f{i}.py": content}, branch="new", parent_id=new_id, message=f"add f{i}")
391 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
392 assert result.exit_code == 0
393 data = json.loads(result.stdout)
394 assert data["trivially_equivalent"] is True
395 assert len(data["pairs"]) == 50
396
397
398 def test_stress_50_commits_mixed(tmp_path: pathlib.Path) -> None:
399 """50-commit old series, new has 25 equivalent + 15 changed + 10 added."""
400 root = _init_repo(tmp_path)
401 base = _commit(root, {"base.py": b"base\n"}, branch="main")
402 old_id = base
403 new_id = base
404 # First 25: identical
405 for i in range(25):
406 content = f"v = {i}\n".encode()
407 old_id = _commit(root, {f"f{i}.py": content}, branch="old", parent_id=old_id)
408 new_id = _commit(root, {f"f{i}.py": content}, branch="new", parent_id=new_id)
409 # Next 25: different content
410 for i in range(25, 50):
411 old_id = _commit(root, {f"f{i}.py": f"old_{i}\n".encode()}, branch="old", parent_id=old_id)
412 new_id = _commit(root, {f"f{i}.py": f"new_{i}\n".encode()}, branch="new", parent_id=new_id)
413 # New has 10 extra commits
414 for i in range(50, 60):
415 new_id = _commit(root, {f"extra{i}.py": b"extra\n"}, branch="new", parent_id=new_id)
416 result = _invoke(root, f"{base}..old", f"{base}..new", "--json")
417 assert result.exit_code != 0 # has changes
418 data = json.loads(result.stdout)
419 equivalent = [p for p in data["pairs"] if p["status"] == "equivalent"]
420 assert len(equivalent) == 25
421 added = [p for p in data["pairs"] if p["status"] == "added"]
422 assert len(added) == 10
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor 149 days ago