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