gabriel / muse public
test_cmd_format_patch.py python
492 lines 20.0 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 129 days ago
1 """Tests for ``muse format-patch`` — content-addressed, domain-aware Muse patch export.
2
3 Test tiers
4 ----------
5 - Unit: output schema, required fields, JSON format, action_label on ops
6 - Integration: initial commit, two-commit delta, multi-file changes
7 - Data integrity: patch_id sha256-prefixed, duration_ms/exit_code present
8 - Security: error to stderr, malicious ref rejected
9 - Performance: duration_ms plausible
10 - Edge: empty diff (no file changes), --output-dir creates .mpatch file
11 """
12 from __future__ import annotations
13 from collections.abc import Mapping
14
15 import datetime
16 import json
17 import pathlib
18
19 import pytest
20
21 from tests.cli_test_helper import CliRunner, InvokeResult
22 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
23 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
24 from muse.core.object_store import write_object
25 from muse.core.types import long_id, split_id, blob_id
26 from muse.core.paths import muse_dir, ref_path
27
28 runner = CliRunner()
29
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35
36 def _init_repo(path: pathlib.Path) -> pathlib.Path:
37 dot_muse = muse_dir(path)
38 for sub in ("commits", "snapshots", "objects", "refs/heads"):
39 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
40 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
41 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
42 return path
43
44
45 def _write_object(repo: pathlib.Path, content: bytes) -> str:
46 """Write bytes to the object store and return the sha256: prefixed ID."""
47 oid = blob_id(content)
48 write_object(repo, oid, content)
49 return oid
50
51
52 def _commit(
53 repo: pathlib.Path,
54 msg: str,
55 manifest: dict[str, str],
56 branch: str = "main",
57 parent: str | None = None,
58 ts: datetime.datetime | None = None,
59 ) -> str:
60 ts = ts or datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
61 sid = compute_snapshot_id(manifest)
62 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest, created_at=ts))
63 parent_ids = [parent] if parent else []
64 cid = compute_commit_id( parent_ids=parent_ids,
65 snapshot_id=sid,
66 message=msg,
67 committed_at_iso=ts.isoformat(),
68 author="gabriel",
69 )
70 write_commit(repo, CommitRecord(
71 commit_id=cid, repo_id="test-repo", branch=branch,
72 snapshot_id=sid, message=msg, committed_at=ts,
73 author="gabriel", parent_commit_id=parent, parent2_commit_id=None,
74 ))
75 branch_ref = ref_path(repo, branch)
76 branch_ref.parent.mkdir(parents=True, exist_ok=True)
77 branch_ref.write_text(cid)
78 return cid
79
80
81 def _fp(repo: pathlib.Path, *args: str) -> InvokeResult:
82 return runner.invoke(None, ["format-patch", *args],
83 env={"MUSE_REPO_ROOT": str(repo)})
84
85
86 def _json(r: InvokeResult) -> Mapping[str, object]:
87 return json.loads(r.output)
88
89
90 # ---------------------------------------------------------------------------
91 # JSON output schema
92 # ---------------------------------------------------------------------------
93
94
95 class TestJsonSchema:
96 def test_exits_zero_on_success(self, tmp_path: pathlib.Path) -> None:
97 repo = _init_repo(tmp_path)
98 oid = _write_object(repo, b"print('hello')\n")
99 _commit(repo, "init: add hello.py", {"hello.py": oid})
100 r = _fp(repo, "--json")
101 assert r.exit_code == 0
102
103 def test_has_patch_id(self, tmp_path: pathlib.Path) -> None:
104 repo = _init_repo(tmp_path)
105 oid = _write_object(repo, b"x = 1\n")
106 _commit(repo, "init", {"hello.py": oid})
107 assert "patch_id" in _json(_fp(repo, "--json"))
108
109 def test_patch_id_sha256_prefix(self, tmp_path: pathlib.Path) -> None:
110 repo = _init_repo(tmp_path)
111 oid = _write_object(repo, b"x = 1\n")
112 _commit(repo, "init", {"hello.py": oid})
113 assert _json(_fp(repo, "--json"))["patch_id"].startswith("sha256:")
114
115 def test_patch_id_full_length(self, tmp_path: pathlib.Path) -> None:
116 repo = _init_repo(tmp_path)
117 oid = _write_object(repo, b"x = 1\n")
118 _commit(repo, "init", {"hello.py": oid})
119 pid = _json(_fp(repo, "--json"))["patch_id"]
120 assert len(pid) == 71 # "sha256:" (7) + 64 hex
121
122 def test_has_from_snapshot_id(self, tmp_path: pathlib.Path) -> None:
123 repo = _init_repo(tmp_path)
124 oid = _write_object(repo, b"x = 1\n")
125 _commit(repo, "init", {"hello.py": oid})
126 data = _json(_fp(repo, "--json"))
127 assert "from_snapshot_id" in data
128
129 def test_has_to_snapshot_id(self, tmp_path: pathlib.Path) -> None:
130 repo = _init_repo(tmp_path)
131 oid = _write_object(repo, b"x = 1\n")
132 _commit(repo, "init", {"hello.py": oid})
133 data = _json(_fp(repo, "--json"))
134 assert "to_snapshot_id" in data
135 assert data["to_snapshot_id"].startswith("sha256:")
136
137 def test_has_from_commit_id(self, tmp_path: pathlib.Path) -> None:
138 repo = _init_repo(tmp_path)
139 oid1 = _write_object(repo, b"x = 1\n")
140 c1 = _commit(repo, "c1", {"a.py": oid1})
141 oid2 = _write_object(repo, b"x = 2\n")
142 _commit(repo, "c2", {"a.py": oid2}, parent=c1)
143 data = _json(_fp(repo, "--json"))
144 assert "from_commit_id" in data
145
146 def test_has_to_commit_id(self, tmp_path: pathlib.Path) -> None:
147 repo = _init_repo(tmp_path)
148 oid = _write_object(repo, b"x = 1\n")
149 _commit(repo, "init", {"hello.py": oid})
150 data = _json(_fp(repo, "--json"))
151 assert "to_commit_id" in data
152 assert data["to_commit_id"].startswith("sha256:")
153
154 def test_domain_matches_repo(self, tmp_path: pathlib.Path) -> None:
155 repo = _init_repo(tmp_path)
156 oid = _write_object(repo, b"x = 1\n")
157 _commit(repo, "init", {"hello.py": oid})
158 data = _json(_fp(repo, "--json"))
159 assert data["domain"] == "code"
160
161 def test_format_version_is_1_0(self, tmp_path: pathlib.Path) -> None:
162 repo = _init_repo(tmp_path)
163 oid = _write_object(repo, b"x = 1\n")
164 _commit(repo, "init", {"hello.py": oid})
165 assert _json(_fp(repo, "--json"))["format_version"] == "1.0"
166
167 def test_duration_ms_present(self, tmp_path: pathlib.Path) -> None:
168 repo = _init_repo(tmp_path)
169 oid = _write_object(repo, b"x = 1\n")
170 _commit(repo, "init", {"hello.py": oid})
171 assert "duration_ms" in _json(_fp(repo, "--json"))
172
173 def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
174 repo = _init_repo(tmp_path)
175 oid = _write_object(repo, b"x = 1\n")
176 _commit(repo, "init", {"hello.py": oid})
177 v = _json(_fp(repo, "--json"))["duration_ms"]
178 assert isinstance(v, float)
179 assert v >= 0.0
180
181 def test_duration_ms_six_decimal_places(self, tmp_path: pathlib.Path) -> None:
182 repo = _init_repo(tmp_path)
183 oid = _write_object(repo, b"x = 1\n")
184 _commit(repo, "init", {"hello.py": oid})
185 v = _json(_fp(repo, "--json"))["duration_ms"]
186 assert v == round(v, 6)
187
188 def test_exit_code_zero_in_json(self, tmp_path: pathlib.Path) -> None:
189 repo = _init_repo(tmp_path)
190 oid = _write_object(repo, b"x = 1\n")
191 _commit(repo, "init", {"hello.py": oid})
192 assert _json(_fp(repo, "--json"))["exit_code"] == 0
193
194 def test_exit_code_is_int_not_bool(self, tmp_path: pathlib.Path) -> None:
195 repo = _init_repo(tmp_path)
196 oid = _write_object(repo, b"x = 1\n")
197 _commit(repo, "init", {"hello.py": oid})
198 assert type(_json(_fp(repo, "--json"))["exit_code"]) is int
199
200 def test_sem_ver_bump_present(self, tmp_path: pathlib.Path) -> None:
201 repo = _init_repo(tmp_path)
202 oid = _write_object(repo, b"x = 1\n")
203 _commit(repo, "init", {"hello.py": oid})
204 data = _json(_fp(repo, "--json"))
205 assert "sem_ver_bump" in data
206 assert data["sem_ver_bump"] in ("major", "minor", "patch")
207
208 def test_summary_is_string(self, tmp_path: pathlib.Path) -> None:
209 repo = _init_repo(tmp_path)
210 oid = _write_object(repo, b"x = 1\n")
211 _commit(repo, "init", {"hello.py": oid})
212 assert isinstance(_json(_fp(repo, "--json"))["summary"], str)
213
214 def test_ops_is_list(self, tmp_path: pathlib.Path) -> None:
215 repo = _init_repo(tmp_path)
216 oid = _write_object(repo, b"x = 1\n")
217 _commit(repo, "init", {"hello.py": oid})
218 assert isinstance(_json(_fp(repo, "--json"))["ops"], list)
219
220 def test_applicability_has_requires_snapshot(self, tmp_path: pathlib.Path) -> None:
221 repo = _init_repo(tmp_path)
222 oid = _write_object(repo, b"x = 1\n")
223 _commit(repo, "init", {"hello.py": oid})
224 data = _json(_fp(repo, "--json"))
225 assert "requires_snapshot" in data["applicability"]
226
227 def test_applicability_has_conflict_free(self, tmp_path: pathlib.Path) -> None:
228 repo = _init_repo(tmp_path)
229 oid = _write_object(repo, b"x = 1\n")
230 _commit(repo, "init", {"hello.py": oid})
231 data = _json(_fp(repo, "--json"))
232 assert isinstance(data["applicability"]["conflict_free"], bool)
233
234 def test_applicability_has_independent_dimensions(self, tmp_path: pathlib.Path) -> None:
235 repo = _init_repo(tmp_path)
236 oid = _write_object(repo, b"x = 1\n")
237 _commit(repo, "init", {"hello.py": oid})
238 data = _json(_fp(repo, "--json"))
239 assert isinstance(data["applicability"]["independent_dimensions"], list)
240
241 def test_output_is_compact_single_line(self, tmp_path: pathlib.Path) -> None:
242 repo = _init_repo(tmp_path)
243 oid = _write_object(repo, b"x = 1\n")
244 _commit(repo, "init", {"hello.py": oid})
245 r = _fp(repo, "--json")
246 assert len(r.output.strip().splitlines()) == 1
247
248
249 # ---------------------------------------------------------------------------
250 # File change tracking
251 # ---------------------------------------------------------------------------
252
253
254 class TestFileChanges:
255 def test_initial_commit_files_in_files_added(self, tmp_path: pathlib.Path) -> None:
256 repo = _init_repo(tmp_path)
257 oid = _write_object(repo, b"x = 1\n")
258 _commit(repo, "init", {"hello.py": oid})
259 assert "hello.py" in _json(_fp(repo, "--json"))["files_added"]
260
261 def test_modified_file_in_files_modified(self, tmp_path: pathlib.Path) -> None:
262 repo = _init_repo(tmp_path)
263 oid1 = _write_object(repo, b"x = 1\n")
264 c1 = _commit(repo, "c1", {"a.py": oid1})
265 oid2 = _write_object(repo, b"x = 2\n")
266 _commit(repo, "c2", {"a.py": oid2}, parent=c1)
267 assert "a.py" in _json(_fp(repo, "--json"))["files_modified"]
268
269 def test_removed_file_in_files_deleted(self, tmp_path: pathlib.Path) -> None:
270 repo = _init_repo(tmp_path)
271 oid1 = _write_object(repo, b"x = 1\n")
272 oid2 = _write_object(repo, b"y = 2\n")
273 c1 = _commit(repo, "c1", {"old.py": oid1, "keep.py": oid2})
274 _commit(repo, "c2", {"keep.py": oid2}, parent=c1)
275 assert "old.py" in _json(_fp(repo, "--json"))["files_deleted"]
276
277 def test_empty_diff_all_lists_empty(self, tmp_path: pathlib.Path) -> None:
278 repo = _init_repo(tmp_path)
279 oid = _write_object(repo, b"x = 1\n")
280 c1 = _commit(repo, "c1", {"a.py": oid})
281 _commit(repo, "c2", {"a.py": oid}, parent=c1)
282 data = _json(_fp(repo, "--json"))
283 assert data["files_added"] == []
284 assert data["files_modified"] == []
285 assert data["files_deleted"] == []
286
287 def test_required_objects_is_list(self, tmp_path: pathlib.Path) -> None:
288 repo = _init_repo(tmp_path)
289 oid = _write_object(repo, b"x = 1\n")
290 _commit(repo, "init", {"hello.py": oid})
291 assert isinstance(_json(_fp(repo, "--json"))["required_objects"], list)
292
293 def test_from_manifest_is_dict(self, tmp_path: pathlib.Path) -> None:
294 repo = _init_repo(tmp_path)
295 oid = _write_object(repo, b"x = 1\n")
296 _commit(repo, "init", {"hello.py": oid})
297 assert isinstance(_json(_fp(repo, "--json"))["from_manifest"], dict)
298
299 def test_to_manifest_is_dict(self, tmp_path: pathlib.Path) -> None:
300 repo = _init_repo(tmp_path)
301 oid = _write_object(repo, b"x = 1\n")
302 _commit(repo, "init", {"hello.py": oid})
303 assert isinstance(_json(_fp(repo, "--json"))["to_manifest"], dict)
304
305 def test_applicability_requires_snapshot_matches_from_snapshot(self, tmp_path: pathlib.Path) -> None:
306 repo = _init_repo(tmp_path)
307 oid1 = _write_object(repo, b"x = 1\n")
308 c1 = _commit(repo, "c1", {"a.py": oid1})
309 oid2 = _write_object(repo, b"x = 2\n")
310 _commit(repo, "c2", {"a.py": oid2}, parent=c1)
311 data = _json(_fp(repo, "--json"))
312 assert data["applicability"]["requires_snapshot"] == data["from_snapshot_id"]
313
314 def test_count_equals_len_refs(self, tmp_path: pathlib.Path) -> None:
315 """files_added + files_modified + files_deleted must account for all changed paths."""
316 repo = _init_repo(tmp_path)
317 oid1 = _write_object(repo, b"x = 1\n")
318 oid2 = _write_object(repo, b"y = 2\n")
319 c1 = _commit(repo, "c1", {"a.py": oid1, "b.py": oid2})
320 oid3 = _write_object(repo, b"x = 99\n")
321 _commit(repo, "c2", {"a.py": oid3}, parent=c1) # modify a, delete b
322 data = _json(_fp(repo, "--json"))
323 total = len(data["files_added"]) + len(data["files_modified"]) + len(data["files_deleted"])
324 assert total == 2 # a.py modified, b.py deleted
325
326
327 # ---------------------------------------------------------------------------
328 # Cohen action labels on ops
329 # ---------------------------------------------------------------------------
330
331
332 class TestCohenActionLabels:
333 def test_inserted_label_on_added_file(self, tmp_path: pathlib.Path) -> None:
334 repo = _init_repo(tmp_path)
335 oid = _write_object(repo, b"x = 1\n")
336 _commit(repo, "init", {"hello.py": oid})
337 labels = [op.get("action_label") for op in _json(_fp(repo, "--json"))["ops"]]
338 assert "inserted" in labels
339
340 def test_deleted_label_on_removed_file(self, tmp_path: pathlib.Path) -> None:
341 repo = _init_repo(tmp_path)
342 oid = _write_object(repo, b"x = 1\n")
343 c1 = _commit(repo, "c1", {"old.py": oid})
344 _commit(repo, "c2", {}, parent=c1)
345 labels = [op.get("action_label") for op in _json(_fp(repo, "--json"))["ops"]]
346 assert "deleted" in labels
347
348 def test_modified_label_on_changed_file(self, tmp_path: pathlib.Path) -> None:
349 repo = _init_repo(tmp_path)
350 oid1 = _write_object(repo, b"x = 1\n")
351 c1 = _commit(repo, "c1", {"a.py": oid1})
352 oid2 = _write_object(repo, b"x = 2\n")
353 _commit(repo, "c2", {"a.py": oid2}, parent=c1)
354 labels = [op.get("action_label") for op in _json(_fp(repo, "--json"))["ops"]]
355 assert "modified" in labels
356
357 def test_all_ops_have_action_label(self, tmp_path: pathlib.Path) -> None:
358 repo = _init_repo(tmp_path)
359 oid1 = _write_object(repo, b"x = 1\n")
360 oid2 = _write_object(repo, b"y = 2\n")
361 _commit(repo, "init", {"a.py": oid1, "b.py": oid2})
362 data = _json(_fp(repo, "--json"))
363 for op in data["ops"]:
364 assert "action_label" in op, f"missing action_label in op: {op}"
365
366 def test_action_label_values_are_valid(self, tmp_path: pathlib.Path) -> None:
367 repo = _init_repo(tmp_path)
368 oid1 = _write_object(repo, b"x = 1\n")
369 _commit(repo, "init", {"a.py": oid1})
370 valid = {"inserted", "deleted", "modified", "moved", "renamed"}
371 for op in _json(_fp(repo, "--json"))["ops"]:
372 assert op["action_label"] in valid
373
374
375 # ---------------------------------------------------------------------------
376 # --output-dir writes .mpatch file
377 # ---------------------------------------------------------------------------
378
379
380 class TestOutputDir:
381 def test_writes_mpatch_file(self, tmp_path: pathlib.Path) -> None:
382 repo = _init_repo(tmp_path)
383 oid = _write_object(repo, b"x = 1\n")
384 _commit(repo, "init", {"hello.py": oid})
385 out_dir = tmp_path / "patches"
386 out_dir.mkdir()
387 r = _fp(repo, "--output-dir", str(out_dir))
388 assert r.exit_code == 0
389 assert len(list(out_dir.glob("*.mpatch"))) == 1
390
391 def test_mpatch_file_is_valid_json(self, tmp_path: pathlib.Path) -> None:
392 repo = _init_repo(tmp_path)
393 oid = _write_object(repo, b"x = 1\n")
394 _commit(repo, "init", {"hello.py": oid})
395 out_dir = tmp_path / "patches"
396 out_dir.mkdir()
397 _fp(repo, "--output-dir", str(out_dir))
398 patch_file = list(out_dir.glob("*.mpatch"))[0]
399 data = json.loads(patch_file.read_bytes())
400 assert "patch_id" in data
401 assert "domain" in data
402
403 def test_mpatch_filename_includes_subject(self, tmp_path: pathlib.Path) -> None:
404 repo = _init_repo(tmp_path)
405 oid = _write_object(repo, b"x = 1\n")
406 _commit(repo, "feat: add hello", {"hello.py": oid})
407 out_dir = tmp_path / "patches"
408 out_dir.mkdir()
409 _fp(repo, "--output-dir", str(out_dir))
410 names = [f.name for f in out_dir.glob("*.mpatch")]
411 assert any("feat" in n for n in names)
412
413 def test_nonexistent_output_dir_fails(self, tmp_path: pathlib.Path) -> None:
414 repo = _init_repo(tmp_path)
415 oid = _write_object(repo, b"x = 1\n")
416 _commit(repo, "init", {"hello.py": oid})
417 r = _fp(repo, "--output-dir", str(tmp_path / "does_not_exist"))
418 assert r.exit_code != 0
419
420
421 # ---------------------------------------------------------------------------
422 # Patch ID stability (data integrity)
423 # ---------------------------------------------------------------------------
424
425
426 class TestPatchIdStability:
427 def test_same_commit_same_patch_id(self, tmp_path: pathlib.Path) -> None:
428 repo = _init_repo(tmp_path)
429 oid = _write_object(repo, b"x = 1\n")
430 _commit(repo, "init", {"hello.py": oid})
431 pid1 = _json(_fp(repo, "--json"))["patch_id"]
432 pid2 = _json(_fp(repo, "--json"))["patch_id"]
433 assert pid1 == pid2
434
435 def test_different_commits_different_patch_ids(self, tmp_path: pathlib.Path) -> None:
436 repo = _init_repo(tmp_path)
437 oid1 = _write_object(repo, b"x = 1\n")
438 c1 = _commit(repo, "c1", {"a.py": oid1})
439 oid2 = _write_object(repo, b"x = 2\n")
440 _commit(repo, "c2", {"a.py": oid2}, parent=c1)
441 # HEAD is c2; compare its patch_id to c1's patch_id via explicit ref
442 pid_c2 = _json(_fp(repo, "--json"))["patch_id"]
443 pid_c1 = _json(_fp(repo, split_id(c1)[1], "--json"))["patch_id"]
444 assert pid_c2 != pid_c1
445
446
447 # ---------------------------------------------------------------------------
448 # Error paths
449 # ---------------------------------------------------------------------------
450
451
452 class TestErrorPaths:
453 def test_empty_repo_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
454 repo = _init_repo(tmp_path)
455 r = _fp(repo, "--json")
456 assert r.exit_code != 0
457
458 def test_bad_ref_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
459 repo = _init_repo(tmp_path)
460 oid = _write_object(repo, b"x = 1\n")
461 _commit(repo, "init", {"hello.py": oid})
462 r = _fp(repo, "nonexistent-branch", "--json")
463 assert r.exit_code != 0
464
465
466 class TestRegisterFlags:
467 def test_default_json_out_is_false(self) -> None:
468 import argparse
469 from muse.cli.commands.format_patch import register
470 p = argparse.ArgumentParser()
471 subs = p.add_subparsers()
472 register(subs)
473 args = p.parse_args(["format-patch"])
474 assert args.json_out is False
475
476 def test_json_flag_sets_json_out(self) -> None:
477 import argparse
478 from muse.cli.commands.format_patch import register
479 p = argparse.ArgumentParser()
480 subs = p.add_subparsers()
481 register(subs)
482 args = p.parse_args(["format-patch", "--json"])
483 assert args.json_out is True
484
485 def test_j_shorthand_sets_json_out(self) -> None:
486 import argparse
487 from muse.cli.commands.format_patch import register
488 p = argparse.ArgumentParser()
489 subs = p.add_subparsers()
490 register(subs)
491 args = p.parse_args(["format-patch", "-j"])
492 assert args.json_out is True
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 129 days ago