gabriel / muse public
test_cmd_snapshot_diff.py python
446 lines 18.5 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
1 """Tests for ``muse snapshot-diff``.
2
3 Verifies categorisation of added/modified/deleted paths, resolution of
4 snapshot IDs, commit IDs, and branch names, text-format output, and error
5 handling for unresolvable refs.
6 """
7
8 from __future__ import annotations
9
10 import datetime
11 import json
12 import pathlib
13
14 from tests.cli_test_helper import CliRunner
15
16 cli = None # argparse migration — CliRunner ignores this arg
17 from muse.core.errors import ExitCode
18 from muse.core.object_store import write_object
19 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
20 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
21 from muse.core._types import Manifest, blob_id
22
23 runner = CliRunner()
24
25
26 # ---------------------------------------------------------------------------
27 # Helpers
28 # ---------------------------------------------------------------------------
29
30
31 def _init_repo(path: pathlib.Path) -> pathlib.Path:
32 muse = path / ".muse"
33 (muse / "commits").mkdir(parents=True)
34 (muse / "snapshots").mkdir(parents=True)
35 (muse / "objects").mkdir(parents=True)
36 (muse / "refs" / "heads").mkdir(parents=True)
37 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
38 (muse / "repo.json").write_text(
39 json.dumps({"repo_id": "test-repo", "domain": "midi"}), encoding="utf-8"
40 )
41 return path
42
43
44 def _env(repo: pathlib.Path) -> Manifest:
45 return {"MUSE_REPO_ROOT": str(repo)}
46
47
48 def _obj(repo: pathlib.Path, content: bytes) -> str:
49 oid = blob_id(content)
50 write_object(repo, oid, content)
51 return oid
52
53
54 def _snap(repo: pathlib.Path, manifest: Manifest) -> str:
55 sid = compute_snapshot_id(manifest)
56 write_snapshot(
57 repo,
58 SnapshotRecord(
59 snapshot_id=sid,
60 manifest=manifest,
61 created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
62 ),
63 )
64 return sid
65
66
67 def _commit(repo: pathlib.Path, tag: str, sid: str, branch: str = "main") -> str:
68 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
69 cid = compute_commit_id([], sid, tag, committed_at.isoformat())
70 write_commit(
71 repo,
72 CommitRecord(
73 commit_id=cid,
74 repo_id="test-repo",
75 branch=branch,
76 snapshot_id=sid,
77 message=tag,
78 committed_at=committed_at,
79 author="tester",
80 parent_commit_id=None,
81 ),
82 )
83 ref = repo / ".muse" / "refs" / "heads" / branch
84 ref.write_text(cid, encoding="utf-8")
85 return cid
86
87
88 # ---------------------------------------------------------------------------
89 # Tests
90 # ---------------------------------------------------------------------------
91
92
93 class TestSnapshotDiff:
94 def test_added_deleted_categorised_correctly(self, tmp_path: pathlib.Path) -> None:
95 repo = _init_repo(tmp_path)
96 shared = _obj(repo, b"shared")
97 new_obj = _obj(repo, b"new")
98 sid_a = _snap(repo, {"shared.mid": shared, "old.mid": shared})
99 sid_b = _snap(repo, {"shared.mid": shared, "new.mid": new_obj})
100 result = runner.invoke(cli, ["snapshot-diff", sid_a, sid_b], env=_env(repo))
101 assert result.exit_code == 0, result.output
102 data = json.loads(result.stdout)
103 assert [e["path"] for e in data["added"]] == ["new.mid"]
104 assert [e["path"] for e in data["deleted"]] == ["old.mid"]
105 assert data["modified"] == []
106 assert data["total_changes"] == 2
107
108 def test_modified_entry_contains_both_object_ids(self, tmp_path: pathlib.Path) -> None:
109 repo = _init_repo(tmp_path)
110 v1 = _obj(repo, b"v1")
111 v2 = _obj(repo, b"v2")
112 sid_a = _snap(repo, {"track.mid": v1})
113 sid_b = _snap(repo, {"track.mid": v2})
114 result = runner.invoke(cli, ["snapshot-diff", sid_a, sid_b], env=_env(repo))
115 assert result.exit_code == 0, result.output
116 data = json.loads(result.stdout)
117 assert len(data["modified"]) == 1
118 mod = data["modified"][0]
119 assert mod["path"] == "track.mid"
120 assert mod["object_id_a"] == v1
121 assert mod["object_id_b"] == v2
122
123 def test_zero_changes_when_snapshots_identical(self, tmp_path: pathlib.Path) -> None:
124 repo = _init_repo(tmp_path)
125 obj = _obj(repo, b"same")
126 sid = _snap(repo, {"f.mid": obj})
127 result = runner.invoke(cli, ["snapshot-diff", sid, sid], env=_env(repo))
128 assert result.exit_code == 0, result.output
129 data = json.loads(result.stdout)
130 assert data["total_changes"] == 0
131
132 def test_resolves_by_branch_name(self, tmp_path: pathlib.Path) -> None:
133 repo = _init_repo(tmp_path)
134 obj_a = _obj(repo, b"a")
135 obj_b = _obj(repo, b"b")
136 _commit(repo, "cmt-main", _snap(repo, {"a.mid": obj_a}), branch="main")
137 _commit(repo, "cmt-dev", _snap(repo, {"b.mid": obj_b}), branch="dev")
138 (repo / ".muse" / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
139 result = runner.invoke(cli, ["snapshot-diff", "main", "dev"], env=_env(repo))
140 assert result.exit_code == 0, result.output
141 data = json.loads(result.stdout)
142 assert data["total_changes"] == 2
143
144 def test_text_format_shows_status_letters(self, tmp_path: pathlib.Path) -> None:
145 repo = _init_repo(tmp_path)
146 shared = _obj(repo, b"s")
147 new_obj = _obj(repo, b"n")
148 sid_a = _snap(repo, {"gone.mid": shared})
149 sid_b = _snap(repo, {"new.mid": new_obj})
150 result = runner.invoke(
151 cli, ["snapshot-diff", "--format", "text", sid_a, sid_b], env=_env(repo)
152 )
153 assert result.exit_code == 0, result.output
154 assert "A new.mid" in result.stdout
155 assert "D gone.mid" in result.stdout
156
157 def test_stat_flag_appends_summary(self, tmp_path: pathlib.Path) -> None:
158 repo = _init_repo(tmp_path)
159 sid_a = _snap(repo, {"gone.mid": _obj(repo, b"g")})
160 sid_b = _snap(repo, {"new.mid": _obj(repo, b"n")})
161 result = runner.invoke(
162 cli,
163 ["snapshot-diff", "--format", "text", "--stat", sid_a, sid_b],
164 env=_env(repo),
165 )
166 assert result.exit_code == 0, result.output
167 assert "added" in result.stdout
168 assert "deleted" in result.stdout
169
170 def test_unresolvable_ref_exits_user_error(self, tmp_path: pathlib.Path) -> None:
171 repo = _init_repo(tmp_path)
172 result = runner.invoke(
173 cli, ["snapshot-diff", "no-such-thing", "also-missing"], env=_env(repo)
174 )
175 assert result.exit_code == ExitCode.USER_ERROR
176 assert "error" in json.loads(result.stdout)
177
178 def test_results_sorted_lexicographically(self, tmp_path: pathlib.Path) -> None:
179 repo = _init_repo(tmp_path)
180 sid_a = _snap(repo, {})
181 sid_b = _snap(
182 repo, {"z.mid": _obj(repo, b"z"), "a.mid": _obj(repo, b"a"), "m.mid": _obj(repo, b"m")}
183 )
184 result = runner.invoke(cli, ["snapshot-diff", sid_a, sid_b], env=_env(repo))
185 assert result.exit_code == 0, result.output
186 data = json.loads(result.stdout)
187 added_paths = [e["path"] for e in data["added"]]
188 assert added_paths == sorted(added_paths)
189
190
191 class TestSnapshotDiffStdin:
192 """Tests for ``--stdin`` batch mode."""
193
194 def test_single_pair_via_stdin_json(self, tmp_path: pathlib.Path) -> None:
195 repo = _init_repo(tmp_path)
196 oid_a = _obj(repo, b"a")
197 oid_b = _obj(repo, b"b")
198 sid_a = _snap(repo, {"a.mid": oid_a})
199 sid_b = _snap(repo, {"b.mid": oid_b})
200 stdin = f"{sid_a} {sid_b}\n"
201 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
202 assert result.exit_code == 0, result.output
203 lines = [ln for ln in result.stdout.strip().splitlines() if ln]
204 assert len(lines) == 1
205 data = json.loads(lines[0])
206 assert data["snapshot_a"] == sid_a
207 assert data["snapshot_b"] == sid_b
208 assert len(data["added"]) == 1
209 assert len(data["deleted"]) == 1
210 assert data["total_changes"] == 2
211
212 def test_multiple_pairs_emit_ndjson(self, tmp_path: pathlib.Path) -> None:
213 repo = _init_repo(tmp_path)
214 oid = _obj(repo, b"x")
215 sid1 = _snap(repo, {"x.mid": oid})
216 sid2 = _snap(repo, {})
217 sid3 = _snap(repo, {"x.mid": oid, "y.mid": _obj(repo, b"y")})
218 stdin = f"{sid1} {sid2}\n{sid2} {sid3}\n"
219 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
220 assert result.exit_code == 0, result.output
221 lines = [ln for ln in result.stdout.strip().splitlines() if ln]
222 assert len(lines) == 2
223 first = json.loads(lines[0])
224 second = json.loads(lines[1])
225 assert first["snapshot_a"] == sid1
226 assert first["snapshot_b"] == sid2
227 assert second["snapshot_a"] == sid2
228 assert second["snapshot_b"] == sid3
229
230 def test_invalid_ref_reported_inline_not_exit_error(self, tmp_path: pathlib.Path) -> None:
231 repo = _init_repo(tmp_path)
232 oid = _obj(repo, b"ok")
233 sid_a = _snap(repo, {"f.mid": oid})
234 sid_b = _snap(repo, {})
235 # First line is bad ref, second is valid
236 bad_ref = "a" * 64 # valid OID format but not in store
237 stdin = f"{bad_ref} {bad_ref}\n{sid_a} {sid_b}\n"
238 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
239 assert result.exit_code == 0 # batch mode always exits 0
240 lines = [ln for ln in result.stdout.strip().splitlines() if ln]
241 assert len(lines) == 2
242 first = json.loads(lines[0])
243 assert "error" in first
244 second = json.loads(lines[1])
245 assert "error" not in second
246 assert second["total_changes"] == 1
247
248 def test_empty_lines_and_comments_skipped(self, tmp_path: pathlib.Path) -> None:
249 repo = _init_repo(tmp_path)
250 sid = _snap(repo, {})
251 stdin = f"\n# this is a comment\n\n{sid} {sid}\n\n"
252 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
253 assert result.exit_code == 0, result.output
254 lines = [ln for ln in result.stdout.strip().splitlines() if ln]
255 assert len(lines) == 1
256 data = json.loads(lines[0])
257 assert data["total_changes"] == 0
258
259 def test_malformed_line_single_token_reported_inline(self, tmp_path: pathlib.Path) -> None:
260 repo = _init_repo(tmp_path)
261 sid = _snap(repo, {})
262 stdin = f"only-one-token\n{sid} {sid}\n"
263 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
264 assert result.exit_code == 0
265 lines = [ln for ln in result.stdout.strip().splitlines() if ln]
266 assert len(lines) == 2
267 first = json.loads(lines[0])
268 assert "error" in first
269 second = json.loads(lines[1])
270 assert "error" not in second
271
272 def test_empty_stdin_produces_no_output(self, tmp_path: pathlib.Path) -> None:
273 repo = _init_repo(tmp_path)
274 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input="")
275 assert result.exit_code == 0
276 assert result.stdout.strip() == ""
277
278 def test_stdin_text_format_blank_line_separated(self, tmp_path: pathlib.Path) -> None:
279 repo = _init_repo(tmp_path)
280 oid_a = _obj(repo, b"a")
281 oid_b = _obj(repo, b"b")
282 sid1 = _snap(repo, {"a.mid": oid_a})
283 sid2 = _snap(repo, {"b.mid": oid_b})
284 sid3 = _snap(repo, {})
285 stdin = f"{sid1} {sid2}\n{sid2} {sid3}\n"
286 result = runner.invoke(
287 cli, ["snapshot-diff", "--stdin", "--format", "text"], env=_env(repo), input=stdin
288 )
289 assert result.exit_code == 0, result.output
290 output = result.stdout
291 # Two diffs separated by a blank line
292 assert "A b.mid" in output or "D a.mid" in output
293 # There should be a blank-line separator between the two pairs
294 blocks = [b.strip() for b in output.split("\n\n") if b.strip()]
295 assert len(blocks) == 2
296
297 def test_stdin_all_errors_still_exits_0(self, tmp_path: pathlib.Path) -> None:
298 repo = _init_repo(tmp_path)
299 bad = "b" * 64 # valid format, not in store
300 stdin = f"{bad} {bad}\n{bad} {bad}\n"
301 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
302 assert result.exit_code == 0
303 lines = [ln for ln in result.stdout.strip().splitlines() if ln]
304 assert all("error" in json.loads(ln) for ln in lines)
305
306 def test_stdin_zero_change_pair_included(self, tmp_path: pathlib.Path) -> None:
307 repo = _init_repo(tmp_path)
308 sid = _snap(repo, {"f.mid": _obj(repo, b"f")})
309 stdin = f"{sid} {sid}\n"
310 result = runner.invoke(cli, ["snapshot-diff", "--stdin"], env=_env(repo), input=stdin)
311 assert result.exit_code == 0, result.output
312 data = json.loads(result.stdout.strip())
313 assert data["total_changes"] == 0
314
315
316 class TestSnapshotDiffEdgeCases:
317 """Edge cases not covered by the primary test classes."""
318
319 def test_bad_format_value_exits_user_error(self, tmp_path: pathlib.Path) -> None:
320 repo = _init_repo(tmp_path)
321 sid = _snap(repo, {})
322 result = runner.invoke(
323 cli, ["snapshot-diff", "--format", "xml", sid, sid], env=_env(repo)
324 )
325 assert result.exit_code == ExitCode.USER_ERROR
326
327 def test_ref_a_provided_ref_b_missing_exits_user_error(self, tmp_path: pathlib.Path) -> None:
328 repo = _init_repo(tmp_path)
329 sid = _snap(repo, {})
330 result = runner.invoke(cli, ["snapshot-diff", sid], env=_env(repo))
331 assert result.exit_code == ExitCode.USER_ERROR
332
333 def test_raw_with_zero_changes_produces_no_diff_lines(self, tmp_path: pathlib.Path) -> None:
334 repo = _init_repo(tmp_path)
335 sid = _snap(repo, {"f.mid": _obj(repo, b"same")})
336 result = runner.invoke(
337 cli, ["snapshot-diff", "--format", "text", "--raw", sid, sid], env=_env(repo)
338 )
339 assert result.exit_code == 0, result.output
340 # No A/M/D lines when there are no changes.
341 for line in result.stdout.splitlines():
342 assert not line.startswith(("A ", "M ", "D "))
343
344 def test_json_shorthand_flag_accepted(self, tmp_path: pathlib.Path) -> None:
345 repo = _init_repo(tmp_path)
346 sid = _snap(repo, {"f.mid": _obj(repo, b"x")})
347 result = runner.invoke(cli, ["snapshot-diff", "--json", sid, sid], env=_env(repo))
348 assert result.exit_code == 0, result.output
349 data = json.loads(result.stdout)
350 assert data["total_changes"] == 0
351
352 def test_no_args_no_stdin_exits_user_error(self, tmp_path: pathlib.Path) -> None:
353 repo = _init_repo(tmp_path)
354 result = runner.invoke(cli, ["snapshot-diff"], env=_env(repo))
355 assert result.exit_code == ExitCode.USER_ERROR
356
357
358 class TestSnapshotDiffRaw:
359 """Tests for ``--raw`` flag (OIDs included in text output)."""
360
361 def test_raw_added_includes_object_id(self, tmp_path: pathlib.Path) -> None:
362 repo = _init_repo(tmp_path)
363 oid = _obj(repo, b"new-content")
364 sid_a = _snap(repo, {})
365 sid_b = _snap(repo, {"new.mid": oid})
366 result = runner.invoke(
367 cli, ["snapshot-diff", "--format", "text", "--raw", sid_a, sid_b], env=_env(repo)
368 )
369 assert result.exit_code == 0, result.output
370 assert oid in result.stdout
371 assert "A" in result.stdout
372 assert "new.mid" in result.stdout
373
374 def test_raw_deleted_includes_object_id(self, tmp_path: pathlib.Path) -> None:
375 repo = _init_repo(tmp_path)
376 oid = _obj(repo, b"old-content")
377 sid_a = _snap(repo, {"gone.mid": oid})
378 sid_b = _snap(repo, {})
379 result = runner.invoke(
380 cli, ["snapshot-diff", "--format", "text", "--raw", sid_a, sid_b], env=_env(repo)
381 )
382 assert result.exit_code == 0, result.output
383 assert oid in result.stdout
384 assert "D" in result.stdout
385 assert "gone.mid" in result.stdout
386
387 def test_raw_modified_includes_both_object_ids(self, tmp_path: pathlib.Path) -> None:
388 repo = _init_repo(tmp_path)
389 oid_a = _obj(repo, b"version-1")
390 oid_b = _obj(repo, b"version-2")
391 sid_a = _snap(repo, {"track.mid": oid_a})
392 sid_b = _snap(repo, {"track.mid": oid_b})
393 result = runner.invoke(
394 cli, ["snapshot-diff", "--format", "text", "--raw", sid_a, sid_b], env=_env(repo)
395 )
396 assert result.exit_code == 0, result.output
397 assert oid_a in result.stdout
398 assert oid_b in result.stdout
399 assert "M" in result.stdout
400 assert "track.mid" in result.stdout
401
402 def test_text_without_raw_omits_object_ids(self, tmp_path: pathlib.Path) -> None:
403 repo = _init_repo(tmp_path)
404 oid = _obj(repo, b"some-content")
405 sid_a = _snap(repo, {})
406 sid_b = _snap(repo, {"file.mid": oid})
407 result = runner.invoke(
408 cli, ["snapshot-diff", "--format", "text", sid_a, sid_b], env=_env(repo)
409 )
410 assert result.exit_code == 0, result.output
411 # OID should NOT appear in non-raw text output
412 assert oid not in result.stdout
413 assert "A file.mid" in result.stdout
414
415 def test_raw_has_no_effect_on_json_output(self, tmp_path: pathlib.Path) -> None:
416 repo = _init_repo(tmp_path)
417 oid_a = _obj(repo, b"va")
418 oid_b = _obj(repo, b"vb")
419 sid_a = _snap(repo, {"t.mid": oid_a})
420 sid_b = _snap(repo, {"t.mid": oid_b})
421 # JSON always includes OIDs; --raw flag is documented as no-op for JSON
422 result_plain = runner.invoke(cli, ["snapshot-diff", sid_a, sid_b], env=_env(repo))
423 result_raw = runner.invoke(cli, ["snapshot-diff", "--raw", sid_a, sid_b], env=_env(repo))
424 assert result_plain.exit_code == 0
425 assert result_raw.exit_code == 0
426 data_plain = json.loads(result_plain.stdout)
427 data_raw = json.loads(result_raw.stdout)
428 # duration_ms will differ between two separate invocations — compare everything else.
429 for key in ("snapshot_a", "snapshot_b", "added", "modified", "deleted", "total_changes"):
430 assert data_plain[key] == data_raw[key]
431
432 def test_raw_stdin_batch_text_includes_oids(self, tmp_path: pathlib.Path) -> None:
433 repo = _init_repo(tmp_path)
434 oid = _obj(repo, b"batch-raw")
435 sid_a = _snap(repo, {})
436 sid_b = _snap(repo, {"r.mid": oid})
437 stdin = f"{sid_a} {sid_b}\n"
438 result = runner.invoke(
439 cli,
440 ["snapshot-diff", "--stdin", "--format", "text", "--raw"],
441 env=_env(repo),
442 input=stdin,
443 )
444 assert result.exit_code == 0, result.output
445 assert oid in result.stdout
446 assert "A" in result.stdout
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor 141 days ago