gabriel / muse public
test_format_patch_supercharge.py python
1,020 lines 39.6 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Supercharge tests for ``muse format-patch``.
2
3 TDD — sections labelled [RED] contain tests that fail until the feature lands.
4 Sections labelled [GREEN] fill gaps against existing behavior.
5
6 New features under test
7 -----------------------
8 - ``--agent-id <id>`` [RED] embed agent provenance in the patch record
9 - ``--model-id <id>`` [RED] embed model provenance in the patch record
10 - ``--intent <text>`` [RED] embed an intent description in the patch record
11 - ``--no-blobs`` [RED] omit base64 blob content from the patch file
12 - Rename detection [RED] same-oid delete+insert → rename op in files_renamed
13
14 Gap-fill coverage
15 -----------------
16 - Register-flag parser shape
17 - Unit tests for _sem_ver_bump, _breaking_changes, _make_patch_filename, _action_label
18 - Blob content verification (decoded bytes match source)
19 - from/to manifest delta correctness
20 - Initial-commit sentinel (from_snapshot_id = sha256:000…, from_commit_id = "")
21 - Required-objects sorted + sha256: prefix
22 - ops count === files_added + files_modified + files_deleted
23 - Default stdout output is valid JSON
24 - Stress: 50-file commit
25 - Security: path-traversal and ANSI in treeish
26 - Performance: duration_ms plausible
27 """
28 from __future__ import annotations
29 from collections.abc import Mapping
30
31 import argparse
32 import datetime
33 import json
34 import pathlib
35 import time
36
37 import pytest
38
39 from muse.cli.commands.format_patch import (
40 _action_label,
41 _breaking_changes,
42 _build_file_level_ops,
43 _make_patch_filename,
44 _sem_ver_bump,
45 register,
46 )
47 from muse.core.object_store import write_object
48 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
49 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
50 from tests.cli_test_helper import CliRunner, InvokeResult
51 from muse.core.types import NULL_LONG_ID, blob_id, long_id
52 from muse.core.paths import ref_path, muse_dir
53
54 runner = CliRunner()
55
56
57 # ---------------------------------------------------------------------------
58 # Repo / commit helpers (shared)
59 # ---------------------------------------------------------------------------
60
61
62 def _init_repo(path: pathlib.Path) -> pathlib.Path:
63 dot_muse = muse_dir(path)
64 for sub in ("commits", "snapshots", "objects", "refs/heads"):
65 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
66 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
67 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
68 return path
69
70
71 def _write_obj(repo: pathlib.Path, content: bytes) -> str:
72 oid = blob_id(content)
73 write_object(repo, oid, content)
74 return oid
75
76
77 _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
78
79
80 def _commit(
81 repo: pathlib.Path,
82 msg: str,
83 manifest: dict[str, str],
84 *,
85 branch: str = "main",
86 parent: str | None = None,
87 ) -> str:
88 sid = compute_snapshot_id(manifest)
89 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest, created_at=_TS))
90 parent_ids = [parent] if parent else []
91 cid = compute_commit_id( parent_ids=parent_ids,
92 snapshot_id=sid,
93 message=msg,
94 committed_at_iso=_TS.isoformat(),
95 author="gabriel",)
96 write_commit(repo, CommitRecord(
97 commit_id=cid, repo_id="test-repo", branch=branch,
98 snapshot_id=sid, message=msg, committed_at=_TS,
99 author="gabriel", parent_commit_id=parent, parent2_commit_id=None,
100 ))
101 ref = ref_path(repo, branch)
102 ref.parent.mkdir(parents=True, exist_ok=True)
103 ref.write_text(cid)
104 return cid
105
106
107 def _fp(repo: pathlib.Path, *args: str) -> InvokeResult:
108 return runner.invoke(None, ["format-patch", *args], env={"MUSE_REPO_ROOT": str(repo)})
109
110
111 def _json_out(r: InvokeResult) -> Mapping[str, object]:
112 for line in r.output.splitlines():
113 line = line.strip()
114 if line.startswith("{"):
115 return json.loads(line)
116 raise ValueError(f"No JSON line in output:\n{r.output!r}")
117
118
119 # ---------------------------------------------------------------------------
120 # Register flags [GREEN]
121 # ---------------------------------------------------------------------------
122
123
124 class TestRegisterFlags:
125 """Parser shape — verify flags are wired correctly."""
126
127 def _parse(self, *args: str) -> argparse.Namespace:
128 p = argparse.ArgumentParser()
129 subs = p.add_subparsers()
130 register(subs)
131 return p.parse_args(["format-patch", *args])
132
133 def test_treeish_defaults_to_head(self) -> None:
134 ns = self._parse()
135 assert ns.treeish == "HEAD"
136
137 def test_treeish_positional(self) -> None:
138 ns = self._parse("main")
139 assert ns.treeish == "main"
140
141 def test_output_dir_flag(self) -> None:
142 ns = self._parse("--output-dir", "/tmp")
143 assert ns.output_dir == "/tmp"
144
145 def test_output_dir_short_flag(self) -> None:
146 ns = self._parse("-o", "/tmp")
147 assert ns.output_dir == "/tmp"
148
149 def test_json_flag(self) -> None:
150 ns = self._parse("--json")
151 assert ns.json_out is True
152
153 def test_json_default_false(self) -> None:
154 ns = self._parse()
155 assert ns.json_out is False
156
157 def test_output_dir_default_none(self) -> None:
158 ns = self._parse()
159 assert ns.output_dir is None
160
161 # [RED] — these flags don't exist yet
162 def test_agent_id_flag(self) -> None:
163 ns = self._parse("--agent-id", "claude-code")
164 assert ns.agent_id == "claude-code"
165
166 def test_agent_id_default_empty(self) -> None:
167 ns = self._parse()
168 assert ns.agent_id == ""
169
170 def test_model_id_flag(self) -> None:
171 ns = self._parse("--model-id", "claude-sonnet-4-6")
172 assert ns.model_id == "claude-sonnet-4-6"
173
174 def test_model_id_default_empty(self) -> None:
175 ns = self._parse()
176 assert ns.model_id == ""
177
178 def test_intent_flag(self) -> None:
179 ns = self._parse("--intent", "add login flow")
180 assert ns.intent == "add login flow"
181
182 def test_intent_default_empty(self) -> None:
183 ns = self._parse()
184 assert ns.intent == ""
185
186 def test_no_blobs_flag(self) -> None:
187 ns = self._parse("--no-blobs")
188 assert ns.no_blobs is True
189
190 def test_no_blobs_default_false(self) -> None:
191 ns = self._parse()
192 assert ns.no_blobs is False
193
194
195 # ---------------------------------------------------------------------------
196 # _sem_ver_bump unit tests [GREEN]
197 # ---------------------------------------------------------------------------
198
199
200 class TestSemVerBump:
201 def test_break_prefix_is_major(self) -> None:
202 assert _sem_ver_bump("break: remove old API", [], []) == "major"
203
204 def test_feat_bang_is_major(self) -> None:
205 assert _sem_ver_bump("feat!: overhaul auth", [], []) == "major"
206
207 def test_breaking_change_body_is_major(self) -> None:
208 assert _sem_ver_bump("refactor: cleanup\n\nBREAKING CHANGE: old param removed", [], []) == "major"
209
210 def test_breaking_change_case_insensitive(self) -> None:
211 assert _sem_ver_bump("breaking change in behavior", [], []) == "major"
212
213 def test_feat_prefix_is_minor(self) -> None:
214 assert _sem_ver_bump("feat: add endpoint", [], []) == "minor"
215
216 def test_files_added_is_minor(self) -> None:
217 assert _sem_ver_bump("chore: misc", ["new_file.py"], []) == "minor"
218
219 def test_fix_prefix_is_patch(self) -> None:
220 assert _sem_ver_bump("fix: off-by-one", [], []) == "patch"
221
222 def test_chore_no_additions_is_patch(self) -> None:
223 assert _sem_ver_bump("chore: update deps", [], []) == "patch"
224
225 def test_empty_message_is_patch(self) -> None:
226 assert _sem_ver_bump("", [], []) == "patch"
227
228 def test_files_deleted_alone_is_patch(self) -> None:
229 assert _sem_ver_bump("chore: cleanup", [], ["old.py"]) == "patch"
230
231 def test_feat_prefix_beats_files_added(self) -> None:
232 # Both trigger minor — result is still minor
233 assert _sem_ver_bump("feat: add stuff", ["new.py"], []) == "minor"
234
235 def test_break_prefix_beats_files_added(self) -> None:
236 assert _sem_ver_bump("break: remove", ["new.py"], []) == "major"
237
238 def test_result_is_one_of_three_values(self) -> None:
239 for msg in ["anything", "feat: x", "break: y"]:
240 result = _sem_ver_bump(msg, [], [])
241 assert result in ("major", "minor", "patch")
242
243
244 # ---------------------------------------------------------------------------
245 # _breaking_changes unit tests [GREEN]
246 # ---------------------------------------------------------------------------
247
248
249 class TestBreakingChanges:
250 def test_empty_message_returns_empty(self) -> None:
251 assert _breaking_changes("") == []
252
253 def test_no_breaking_change_returns_empty(self) -> None:
254 assert _breaking_changes("feat: add endpoint") == []
255
256 def test_single_breaking_change(self) -> None:
257 msg = "refactor: cleanup\n\nBREAKING CHANGE: removed --legacy flag"
258 result = _breaking_changes(msg)
259 assert result == ["removed --legacy flag"]
260
261 def test_multiple_breaking_changes(self) -> None:
262 msg = "refactor:\n\nBREAKING CHANGE: first\nBREAKING CHANGE: second"
263 result = _breaking_changes(msg)
264 assert result == ["first", "second"]
265
266 def test_leading_trailing_whitespace_stripped(self) -> None:
267 msg = "BREAKING CHANGE: trimmed "
268 result = _breaking_changes(msg)
269 assert result == ["trimmed"]
270
271 def test_not_at_start_of_line_ignored(self) -> None:
272 # Mid-sentence "BREAKING CHANGE" not at start of line
273 msg = "This has BREAKING CHANGE: in the middle"
274 # The implementation checks stripped.upper().startswith("BREAKING CHANGE:")
275 # so it WOULD match if the stripped line starts with it — it does here
276 # because "This has..." stripped starts with "This" not "BREAKING CHANGE"
277 result = _breaking_changes(msg)
278 assert result == []
279
280 def test_returns_list(self) -> None:
281 assert isinstance(_breaking_changes("anything"), list)
282
283
284 # ---------------------------------------------------------------------------
285 # _make_patch_filename unit tests [GREEN]
286 # ---------------------------------------------------------------------------
287
288
289 class TestMakePatchFilename:
290 def test_basic_subject(self) -> None:
291 assert _make_patch_filename("feat: add hello") == "feat-add-hello.mpatch"
292
293 def test_ends_with_mpatch(self) -> None:
294 name = _make_patch_filename("anything")
295 assert name.endswith(".mpatch")
296
297 def test_empty_subject_returns_patch(self) -> None:
298 assert _make_patch_filename("") == "patch.mpatch"
299
300 def test_slash_replaced(self) -> None:
301 name = _make_patch_filename("fix/my-bug")
302 assert "/" not in name
303
304 def test_backslash_replaced(self) -> None:
305 name = _make_patch_filename("fix\\my-bug")
306 assert "\\" not in name
307
308 def test_dot_replaced(self) -> None:
309 # dots → dashes in the slug portion (before the .mpatch extension)
310 name = _make_patch_filename("v1.2.3 release")
311 slug = name.removesuffix(".mpatch")
312 assert "." not in slug
313
314 def test_long_subject_truncated(self) -> None:
315 long_msg = "x" * 100
316 name = _make_patch_filename(long_msg)
317 slug = name.removesuffix(".mpatch")
318 assert len(slug) <= 52
319
320 def test_unicode_stripped(self) -> None:
321 name = _make_patch_filename("feat: émoji 🚀 add")
322 # Non-ASCII removed, but ASCII words remain
323 assert "feat" in name
324
325 def test_whitespace_replaced_with_dash(self) -> None:
326 name = _make_patch_filename("add multiple spaces")
327 assert " " not in name
328
329 def test_no_leading_trailing_dashes_in_slug(self) -> None:
330 slug = _make_patch_filename(" spaces around ").removesuffix(".mpatch")
331 assert not slug.startswith("-")
332 assert not slug.endswith("-")
333
334 def test_no_consecutive_dashes_in_slug(self) -> None:
335 slug = _make_patch_filename("a!!b").removesuffix(".mpatch")
336 assert "--" not in slug
337
338
339 # ---------------------------------------------------------------------------
340 # _action_label unit tests [GREEN]
341 # ---------------------------------------------------------------------------
342
343
344 class TestActionLabel:
345 def test_insert_is_inserted(self) -> None:
346 assert _action_label("insert") == "inserted"
347
348 def test_delete_is_deleted(self) -> None:
349 assert _action_label("delete") == "deleted"
350
351 def test_replace_is_modified(self) -> None:
352 assert _action_label("replace") == "modified"
353
354 def test_mutate_is_modified(self) -> None:
355 assert _action_label("mutate") == "modified"
356
357 def test_patch_is_modified(self) -> None:
358 assert _action_label("patch") == "modified"
359
360 def test_move_is_moved(self) -> None:
361 assert _action_label("move") == "moved"
362
363 def test_directory_rename_is_renamed(self) -> None:
364 assert _action_label("directory_rename") == "renamed"
365
366 def test_unknown_defaults_to_modified(self) -> None:
367 assert _action_label("frob") == "modified"
368 assert _action_label("") == "modified"
369 assert _action_label("UPDATE") == "modified"
370
371
372 # ---------------------------------------------------------------------------
373 # _build_file_level_ops — internal unit tests [GREEN + RED for rename]
374 # ---------------------------------------------------------------------------
375
376
377 class TestBuildFileOps:
378 def test_added_file_in_ops(self) -> None:
379 base: dict[str, str] = {}
380 target = {"new.py": long_id("a" * 64)}
381 ops, added, modified, deleted, *_ = _build_file_level_ops(base, target)
382 assert any(op["address"] == "new.py" and op["op"] == "insert" for op in ops)
383
384 def test_deleted_file_in_ops(self) -> None:
385 base = {"old.py": long_id("a" * 64)}
386 target: dict[str, str] = {}
387 ops, added, modified, deleted, *_ = _build_file_level_ops(base, target)
388 assert any(op["address"] == "old.py" and op["op"] == "delete" for op in ops)
389
390 def test_modified_file_in_ops(self) -> None:
391 oid_a = long_id("a" * 64)
392 oid_b = long_id("b" * 64)
393 ops, added, modified, deleted, *_ = _build_file_level_ops(
394 {"f.py": oid_a}, {"f.py": oid_b}
395 )
396 assert any(op["address"] == "f.py" and op["op"] == "replace" for op in ops)
397
398 def test_added_list_sorted(self) -> None:
399 base: dict[str, str] = {}
400 target = {"z.py": long_id("z" * 64), "a.py": long_id("a" * 64)}
401 _, added, _, _, *_ = _build_file_level_ops(base, target)
402 assert added == sorted(added)
403
404 def test_deleted_list_sorted(self) -> None:
405 base = {"z.py": long_id("z" * 64), "a.py": long_id("a" * 64)}
406 _, _, _, deleted, *_ = _build_file_level_ops(base, {})
407 assert deleted == sorted(deleted)
408
409 def test_modified_list_sorted(self) -> None:
410 oid_a = long_id("a" * 64)
411 oid_b = long_id("b" * 64)
412 base = {"z.py": oid_a, "a.py": oid_a}
413 target = {"z.py": oid_b, "a.py": oid_b}
414 _, _, modified, _, *_ = _build_file_level_ops(base, target)
415 assert modified == sorted(modified)
416
417 def test_unchanged_file_not_in_ops(self) -> None:
418 oid = long_id("a" * 64)
419 ops, _, _, _, *_ = _build_file_level_ops({"f.py": oid}, {"f.py": oid})
420 addresses = [op["address"] for op in ops]
421 assert "f.py" not in addresses
422
423 # [RED] rename detection — same oid deleted + added at different path = rename
424 def test_rename_detected(self) -> None:
425 oid = long_id("a" * 64)
426 base = {"old.py": oid}
427 target = {"new.py": oid}
428 ops, added, modified, deleted, renamed = _build_file_level_ops(base, target)
429 assert "old.py" in renamed
430 assert renamed["old.py"] == "new.py"
431
432 def test_rename_not_in_files_added(self) -> None:
433 oid = long_id("a" * 64)
434 _, added, _, _, renamed = _build_file_level_ops({"old.py": oid}, {"new.py": oid})
435 assert "new.py" not in added
436
437 def test_rename_not_in_files_deleted(self) -> None:
438 oid = long_id("a" * 64)
439 _, _, _, deleted, renamed = _build_file_level_ops({"old.py": oid}, {"new.py": oid})
440 assert "old.py" not in deleted
441
442 def test_rename_op_present_in_ops(self) -> None:
443 oid = long_id("a" * 64)
444 ops, _, _, _, _ = _build_file_level_ops({"old.py": oid}, {"new.py": oid})
445 rename_ops = [op for op in ops if op.get("op") == "move"]
446 assert len(rename_ops) == 1
447
448 def test_rename_op_action_label_is_moved(self) -> None:
449 oid = long_id("a" * 64)
450 ops, _, _, _, _ = _build_file_level_ops({"old.py": oid}, {"new.py": oid})
451 rename_ops = [op for op in ops if op.get("op") == "move"]
452 assert rename_ops[0]["action_label"] == "moved"
453
454 def test_different_oid_not_a_rename(self) -> None:
455 oid_a = long_id("a" * 64)
456 oid_b = long_id("b" * 64)
457 _, added, _, deleted, renamed = _build_file_level_ops(
458 {"old.py": oid_a}, {"new.py": oid_b}
459 )
460 assert not renamed
461 assert "old.py" in deleted
462 assert "new.py" in added
463
464 def test_empty_renamed_dict_when_no_renames(self) -> None:
465 oid_a = long_id("a" * 64)
466 oid_b = long_id("b" * 64)
467 _, _, _, _, renamed = _build_file_level_ops({"f.py": oid_a}, {"f.py": oid_b})
468 assert renamed == {}
469
470
471 # ---------------------------------------------------------------------------
472 # Blob embedding [GREEN]
473 # ---------------------------------------------------------------------------
474
475
476 class TestBlobEmbedding:
477 def test_blobs_field_present(self, tmp_path: pathlib.Path) -> None:
478 repo = _init_repo(tmp_path)
479 oid = _write_obj(repo, b"hello blob")
480 _commit(repo, "init", {"f.py": oid})
481 data = _json_out(_fp(repo, "--json"))
482 assert "blobs" in data
483
484 def test_blobs_is_dict(self, tmp_path: pathlib.Path) -> None:
485 repo = _init_repo(tmp_path)
486 oid = _write_obj(repo, b"hello blob")
487 _commit(repo, "init", {"f.py": oid})
488 data = _json_out(_fp(repo, "--json"))
489 assert isinstance(data["blobs"], dict)
490
491 def test_blob_key_matches_required_object(self, tmp_path: pathlib.Path) -> None:
492 repo = _init_repo(tmp_path)
493 content = b"blob content"
494 oid = _write_obj(repo, content)
495 _commit(repo, "init", {"f.py": oid})
496 data = _json_out(_fp(repo, "--json"))
497 assert oid in data["blobs"]
498
499 def test_blob_decodes_to_original_content(self, tmp_path: pathlib.Path) -> None:
500 import base64
501 repo = _init_repo(tmp_path)
502 content = b"exact bytes\x00\x01\x02"
503 oid = _write_obj(repo, content)
504 _commit(repo, "init", {"f.py": oid})
505 data = _json_out(_fp(repo, "--json"))
506 decoded = base64.b64decode(data["blobs"][oid])
507 assert decoded == content
508
509 def test_blobs_is_base64_valid_string(self, tmp_path: pathlib.Path) -> None:
510 import base64
511 repo = _init_repo(tmp_path)
512 oid = _write_obj(repo, b"any content")
513 _commit(repo, "init", {"f.py": oid})
514 data = _json_out(_fp(repo, "--json"))
515 for val in data["blobs"].values():
516 assert isinstance(val, str)
517 base64.b64decode(val) # must not raise
518
519 def test_unmodified_objects_not_in_blobs(self, tmp_path: pathlib.Path) -> None:
520 """Blobs only contains objects in to_manifest (new/modified), not deleted."""
521 repo = _init_repo(tmp_path)
522 oid_a = _write_obj(repo, b"a")
523 oid_b = _write_obj(repo, b"b")
524 c1 = _commit(repo, "c1", {"a.py": oid_a, "b.py": oid_b})
525 oid_c = _write_obj(repo, b"c")
526 _commit(repo, "c2", {"a.py": oid_a, "c.py": oid_c}, parent=c1)
527 # b.py deleted → oid_b not in to_manifest → not in blobs
528 data = _json_out(_fp(repo, "--json"))
529 assert oid_b not in data["blobs"]
530
531
532 # ---------------------------------------------------------------------------
533 # Required objects [GREEN]
534 # ---------------------------------------------------------------------------
535
536
537 class TestRequiredObjects:
538 def test_all_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
539 repo = _init_repo(tmp_path)
540 oid = _write_obj(repo, b"x")
541 _commit(repo, "init", {"f.py": oid})
542 data = _json_out(_fp(repo, "--json"))
543 for rid in data["required_objects"]:
544 assert rid.startswith("sha256:")
545
546 def test_required_objects_is_sorted(self, tmp_path: pathlib.Path) -> None:
547 repo = _init_repo(tmp_path)
548 oid_a = _write_obj(repo, b"aaa")
549 oid_b = _write_obj(repo, b"bbb")
550 _commit(repo, "init", {"a.py": oid_a, "b.py": oid_b})
551 data = _json_out(_fp(repo, "--json"))
552 ro = data["required_objects"]
553 assert ro == sorted(ro)
554
555 def test_required_objects_subset_of_to_manifest(self, tmp_path: pathlib.Path) -> None:
556 repo = _init_repo(tmp_path)
557 oid = _write_obj(repo, b"y")
558 _commit(repo, "init", {"f.py": oid})
559 data = _json_out(_fp(repo, "--json"))
560 to_vals = set(data["to_manifest"].values())
561 for rid in data["required_objects"]:
562 assert rid in to_vals
563
564 def test_required_objects_empty_for_no_change(self, tmp_path: pathlib.Path) -> None:
565 repo = _init_repo(tmp_path)
566 oid = _write_obj(repo, b"z")
567 c1 = _commit(repo, "c1", {"f.py": oid})
568 _commit(repo, "c2 no-op", {"f.py": oid}, parent=c1)
569 data = _json_out(_fp(repo, "--json"))
570 assert data["required_objects"] == []
571
572
573 # ---------------------------------------------------------------------------
574 # Manifest delta correctness [GREEN]
575 # ---------------------------------------------------------------------------
576
577
578 class TestManifestDelta:
579 def test_added_path_in_to_manifest(self, tmp_path: pathlib.Path) -> None:
580 repo = _init_repo(tmp_path)
581 oid = _write_obj(repo, b"new")
582 _commit(repo, "init", {"new.py": oid})
583 data = _json_out(_fp(repo, "--json"))
584 assert "new.py" in data["to_manifest"]
585
586 def test_added_path_not_in_from_manifest(self, tmp_path: pathlib.Path) -> None:
587 repo = _init_repo(tmp_path)
588 oid = _write_obj(repo, b"new")
589 _commit(repo, "init", {"new.py": oid})
590 data = _json_out(_fp(repo, "--json"))
591 assert "new.py" not in data["from_manifest"]
592
593 def test_deleted_path_in_from_manifest(self, tmp_path: pathlib.Path) -> None:
594 repo = _init_repo(tmp_path)
595 oid = _write_obj(repo, b"old")
596 c1 = _commit(repo, "c1", {"old.py": oid})
597 _commit(repo, "c2", {}, parent=c1)
598 data = _json_out(_fp(repo, "--json"))
599 assert "old.py" in data["from_manifest"]
600
601 def test_deleted_path_not_in_to_manifest(self, tmp_path: pathlib.Path) -> None:
602 repo = _init_repo(tmp_path)
603 oid = _write_obj(repo, b"old")
604 c1 = _commit(repo, "c1", {"old.py": oid})
605 _commit(repo, "c2", {}, parent=c1)
606 data = _json_out(_fp(repo, "--json"))
607 assert "old.py" not in data["to_manifest"]
608
609 def test_modified_path_in_both_manifests(self, tmp_path: pathlib.Path) -> None:
610 repo = _init_repo(tmp_path)
611 oid_a = _write_obj(repo, b"v1")
612 c1 = _commit(repo, "c1", {"f.py": oid_a})
613 oid_b = _write_obj(repo, b"v2")
614 _commit(repo, "c2", {"f.py": oid_b}, parent=c1)
615 data = _json_out(_fp(repo, "--json"))
616 assert "f.py" in data["from_manifest"]
617 assert "f.py" in data["to_manifest"]
618 assert data["from_manifest"]["f.py"] != data["to_manifest"]["f.py"]
619
620 def test_unchanged_path_not_in_either_manifest(self, tmp_path: pathlib.Path) -> None:
621 repo = _init_repo(tmp_path)
622 oid_keep = _write_obj(repo, b"keep")
623 oid_chg = _write_obj(repo, b"v1")
624 c1 = _commit(repo, "c1", {"keep.py": oid_keep, "chg.py": oid_chg})
625 oid_chg2 = _write_obj(repo, b"v2")
626 _commit(repo, "c2", {"keep.py": oid_keep, "chg.py": oid_chg2}, parent=c1)
627 data = _json_out(_fp(repo, "--json"))
628 assert "keep.py" not in data["from_manifest"]
629 assert "keep.py" not in data["to_manifest"]
630
631
632 # ---------------------------------------------------------------------------
633 # Initial commit sentinel [GREEN]
634 # ---------------------------------------------------------------------------
635
636
637 class TestInitialCommit:
638 def test_from_snapshot_id_is_sentinel_for_initial(self, tmp_path: pathlib.Path) -> None:
639 repo = _init_repo(tmp_path)
640 oid = _write_obj(repo, b"x")
641 _commit(repo, "init", {"f.py": oid})
642 data = _json_out(_fp(repo, "--json"))
643 # Sentinel for initial commit is sha256:000...000 (64 zeros)
644 assert data["from_snapshot_id"] == NULL_LONG_ID
645
646 def test_from_commit_id_empty_for_initial(self, tmp_path: pathlib.Path) -> None:
647 repo = _init_repo(tmp_path)
648 oid = _write_obj(repo, b"x")
649 _commit(repo, "init", {"f.py": oid})
650 data = _json_out(_fp(repo, "--json"))
651 assert data["from_commit_id"] == ""
652
653 def test_all_files_in_files_added_for_initial(self, tmp_path: pathlib.Path) -> None:
654 repo = _init_repo(tmp_path)
655 oid_a = _write_obj(repo, b"a")
656 oid_b = _write_obj(repo, b"b")
657 _commit(repo, "init", {"a.py": oid_a, "b.py": oid_b})
658 data = _json_out(_fp(repo, "--json"))
659 assert "a.py" in data["files_added"]
660 assert "b.py" in data["files_added"]
661 assert data["files_modified"] == []
662 assert data["files_deleted"] == []
663
664 def test_from_snapshot_id_set_for_second_commit(self, tmp_path: pathlib.Path) -> None:
665 repo = _init_repo(tmp_path)
666 oid = _write_obj(repo, b"v1")
667 c1 = _commit(repo, "c1", {"f.py": oid})
668 oid2 = _write_obj(repo, b"v2")
669 _commit(repo, "c2", {"f.py": oid2}, parent=c1)
670 data = _json_out(_fp(repo, "--json"))
671 # Non-initial: from_snapshot_id should NOT be the sentinel
672 assert data["from_snapshot_id"] != NULL_LONG_ID
673
674
675 # ---------------------------------------------------------------------------
676 # Agent provenance flags [RED] — --agent-id, --model-id, --intent
677 # ---------------------------------------------------------------------------
678
679
680 class TestAgentProvenance:
681 def test_agent_id_set_in_output(self, tmp_path: pathlib.Path) -> None:
682 repo = _init_repo(tmp_path)
683 oid = _write_obj(repo, b"x")
684 _commit(repo, "init", {"f.py": oid})
685 data = _json_out(_fp(repo, "--json", "--agent-id", "claude-code"))
686 assert data["agent_id"] == "claude-code"
687
688 def test_model_id_set_in_output(self, tmp_path: pathlib.Path) -> None:
689 repo = _init_repo(tmp_path)
690 oid = _write_obj(repo, b"x")
691 _commit(repo, "init", {"f.py": oid})
692 data = _json_out(_fp(repo, "--json", "--model-id", "claude-sonnet-4-6"))
693 assert data["model_id"] == "claude-sonnet-4-6"
694
695 def test_intent_set_in_output(self, tmp_path: pathlib.Path) -> None:
696 repo = _init_repo(tmp_path)
697 oid = _write_obj(repo, b"x")
698 _commit(repo, "init", {"f.py": oid})
699 data = _json_out(_fp(repo, "--json", "--intent", "bootstrap project"))
700 assert data["intent"] == "bootstrap project"
701
702 def test_agent_id_in_mpatch_file(self, tmp_path: pathlib.Path) -> None:
703 repo = _init_repo(tmp_path)
704 oid = _write_obj(repo, b"x")
705 _commit(repo, "init", {"f.py": oid})
706 out_dir = tmp_path / "patches"
707 out_dir.mkdir()
708 r = _fp(repo, "--output-dir", str(out_dir), "--agent-id", "claude-code")
709 assert r.exit_code == 0
710 patch_file = list(out_dir.glob("*.mpatch"))[0]
711 data = json.loads(patch_file.read_bytes())
712 assert data["agent_id"] == "claude-code"
713
714 def test_agent_id_affects_patch_id(self, tmp_path: pathlib.Path) -> None:
715 """Different agent_id → different patch_id (agent_id is part of canonical JSON)."""
716 repo = _init_repo(tmp_path)
717 oid = _write_obj(repo, b"x")
718 _commit(repo, "init", {"f.py": oid})
719 pid_no_agent = _json_out(_fp(repo, "--json"))["patch_id"]
720 pid_with_agent = _json_out(_fp(repo, "--json", "--agent-id", "claude-code"))["patch_id"]
721 assert pid_no_agent != pid_with_agent
722
723 def test_no_agent_flags_leaves_fields_empty(self, tmp_path: pathlib.Path) -> None:
724 repo = _init_repo(tmp_path)
725 oid = _write_obj(repo, b"x")
726 _commit(repo, "init", {"f.py": oid})
727 data = _json_out(_fp(repo, "--json"))
728 assert data["agent_id"] == ""
729 assert data["model_id"] == ""
730 assert data["intent"] == ""
731
732 def test_all_provenance_flags_together(self, tmp_path: pathlib.Path) -> None:
733 repo = _init_repo(tmp_path)
734 oid = _write_obj(repo, b"x")
735 _commit(repo, "init", {"f.py": oid})
736 data = _json_out(_fp(repo, "--json",
737 "--agent-id", "claude-code",
738 "--model-id", "claude-sonnet-4-6",
739 "--intent", "add login endpoint"))
740 assert data["agent_id"] == "claude-code"
741 assert data["model_id"] == "claude-sonnet-4-6"
742 assert data["intent"] == "add login endpoint"
743
744
745 # ---------------------------------------------------------------------------
746 # --no-blobs flag [RED]
747 # ---------------------------------------------------------------------------
748
749
750 class TestNoBlobs:
751 def test_no_blobs_empties_blobs_dict(self, tmp_path: pathlib.Path) -> None:
752 repo = _init_repo(tmp_path)
753 oid = _write_obj(repo, b"blob content here")
754 _commit(repo, "init", {"f.py": oid})
755 data = _json_out(_fp(repo, "--json", "--no-blobs"))
756 assert data["blobs"] == {}
757
758 def test_no_blobs_preserves_required_objects(self, tmp_path: pathlib.Path) -> None:
759 """required_objects still lists what the target needs even without inline blobs."""
760 repo = _init_repo(tmp_path)
761 oid = _write_obj(repo, b"blob content here")
762 _commit(repo, "init", {"f.py": oid})
763 data = _json_out(_fp(repo, "--json", "--no-blobs"))
764 assert oid in data["required_objects"]
765
766 def test_no_blobs_in_mpatch_file(self, tmp_path: pathlib.Path) -> None:
767 repo = _init_repo(tmp_path)
768 oid = _write_obj(repo, b"some bytes")
769 _commit(repo, "init", {"f.py": oid})
770 out_dir = tmp_path / "patches"
771 out_dir.mkdir()
772 r = _fp(repo, "--output-dir", str(out_dir), "--no-blobs")
773 assert r.exit_code == 0
774 data = json.loads(list(out_dir.glob("*.mpatch"))[0].read_bytes())
775 assert data["blobs"] == {}
776
777 def test_default_has_blobs(self, tmp_path: pathlib.Path) -> None:
778 """Without --no-blobs, blobs are embedded (existing behavior)."""
779 repo = _init_repo(tmp_path)
780 oid = _write_obj(repo, b"keep me")
781 _commit(repo, "init", {"f.py": oid})
782 data = _json_out(_fp(repo, "--json"))
783 assert len(data["blobs"]) > 0
784
785 def test_no_blobs_output_smaller_than_with_blobs(self, tmp_path: pathlib.Path) -> None:
786 """--no-blobs patch should be smaller (no base64 content)."""
787 repo = _init_repo(tmp_path)
788 content = b"x" * 1024 # 1KB object
789 oid = _write_obj(repo, content)
790 _commit(repo, "init", {"f.py": oid})
791 r_with = _fp(repo, "--json")
792 r_no = _fp(repo, "--json", "--no-blobs")
793 assert len(r_no.output) < len(r_with.output)
794
795
796 # ---------------------------------------------------------------------------
797 # Rename detection via CLI [RED]
798 # ---------------------------------------------------------------------------
799
800
801 class TestRenameDetectionCLI:
802 def test_rename_in_files_renamed(self, tmp_path: pathlib.Path) -> None:
803 repo = _init_repo(tmp_path)
804 oid = _write_obj(repo, b"shared content")
805 c1 = _commit(repo, "c1", {"old.py": oid})
806 _commit(repo, "c2 rename", {"new.py": oid}, parent=c1)
807 data = _json_out(_fp(repo, "--json"))
808 assert "old.py" in data["files_renamed"]
809 assert data["files_renamed"]["old.py"] == "new.py"
810
811 def test_rename_not_in_files_added(self, tmp_path: pathlib.Path) -> None:
812 repo = _init_repo(tmp_path)
813 oid = _write_obj(repo, b"shared content")
814 c1 = _commit(repo, "c1", {"old.py": oid})
815 _commit(repo, "c2 rename", {"new.py": oid}, parent=c1)
816 data = _json_out(_fp(repo, "--json"))
817 assert "new.py" not in data["files_added"]
818
819 def test_rename_not_in_files_deleted(self, tmp_path: pathlib.Path) -> None:
820 repo = _init_repo(tmp_path)
821 oid = _write_obj(repo, b"shared content")
822 c1 = _commit(repo, "c1", {"old.py": oid})
823 _commit(repo, "c2 rename", {"new.py": oid}, parent=c1)
824 data = _json_out(_fp(repo, "--json"))
825 assert "old.py" not in data["files_deleted"]
826
827 def test_genuine_add_and_delete_not_confused_for_rename(self, tmp_path: pathlib.Path) -> None:
828 repo = _init_repo(tmp_path)
829 oid_a = _write_obj(repo, b"content A")
830 oid_b = _write_obj(repo, b"content B")
831 c1 = _commit(repo, "c1", {"a.py": oid_a})
832 _commit(repo, "c2", {"b.py": oid_b}, parent=c1)
833 data = _json_out(_fp(repo, "--json"))
834 assert data["files_renamed"] == {}
835 assert "b.py" in data["files_added"]
836 assert "a.py" in data["files_deleted"]
837
838
839 # ---------------------------------------------------------------------------
840 # Default stdout output [GREEN]
841 # ---------------------------------------------------------------------------
842
843
844 class TestDefaultOutput:
845 def test_default_output_is_valid_json(self, tmp_path: pathlib.Path) -> None:
846 repo = _init_repo(tmp_path)
847 oid = _write_obj(repo, b"x")
848 _commit(repo, "init", {"f.py": oid})
849 r = _fp(repo)
850 assert r.exit_code == 0
851 data = json.loads(r.output.strip())
852 assert "patch_id" in data
853
854 def test_default_output_has_patch_id(self, tmp_path: pathlib.Path) -> None:
855 repo = _init_repo(tmp_path)
856 oid = _write_obj(repo, b"x")
857 _commit(repo, "init", {"f.py": oid})
858 r = _fp(repo)
859 data = json.loads(r.output.strip())
860 assert data["patch_id"].startswith("sha256:")
861
862
863 # ---------------------------------------------------------------------------
864 # ops completeness [GREEN]
865 # ---------------------------------------------------------------------------
866
867
868 class TestOpsCompleteness:
869 def test_ops_count_equals_sum_of_file_lists(self, tmp_path: pathlib.Path) -> None:
870 repo = _init_repo(tmp_path)
871 oid1 = _write_obj(repo, b"a")
872 oid2 = _write_obj(repo, b"b")
873 oid3 = _write_obj(repo, b"c")
874 c1 = _commit(repo, "c1", {"a.py": oid1, "b.py": oid2, "c.py": oid3})
875 oid4 = _write_obj(repo, b"a-modified")
876 _commit(repo, "c2", {"a.py": oid4, "b.py": oid2}, parent=c1)
877 data = _json_out(_fp(repo, "--json"))
878 # c.py deleted, a.py modified, b.py unchanged
879 total_file_changes = (
880 len(data["files_added"])
881 + len(data["files_modified"])
882 + len(data["files_deleted"])
883 + len(data["files_renamed"])
884 )
885 # Each changed file has exactly one op (excluding renames which have one move op)
886 assert len(data["ops"]) == total_file_changes
887
888 def test_each_op_has_required_fields(self, tmp_path: pathlib.Path) -> None:
889 repo = _init_repo(tmp_path)
890 oid_a = _write_obj(repo, b"a")
891 oid_b = _write_obj(repo, b"b")
892 c1 = _commit(repo, "c1", {"a.py": oid_a})
893 oid_c = _write_obj(repo, b"a-mod")
894 _commit(repo, "c2", {"a.py": oid_c, "b.py": oid_b}, parent=c1)
895 data = _json_out(_fp(repo, "--json"))
896 for op in data["ops"]:
897 assert "op" in op
898 assert "address" in op
899 assert "action_label" in op
900
901
902 # ---------------------------------------------------------------------------
903 # Stress [GREEN]
904 # ---------------------------------------------------------------------------
905
906
907 class TestStress:
908 def test_50_files_added(self, tmp_path: pathlib.Path) -> None:
909 repo = _init_repo(tmp_path)
910 manifest = {}
911 for i in range(50):
912 content = f"# file {i}\n".encode() * 10
913 oid = _write_obj(repo, content)
914 manifest[f"src/file_{i:02d}.py"] = oid
915 _commit(repo, "feat: add 50 files", manifest)
916 r = _fp(repo, "--json")
917 assert r.exit_code == 0
918 data = _json_out(r)
919 assert len(data["files_added"]) == 50
920 assert len(data["ops"]) == 50
921
922 def test_mixed_50_file_commit(self, tmp_path: pathlib.Path) -> None:
923 repo = _init_repo(tmp_path)
924 manifest_c1 = {}
925 for i in range(40):
926 oid = _write_obj(repo, f"v1-{i}".encode())
927 manifest_c1[f"f{i:02d}.py"] = oid
928 c1 = _commit(repo, "c1", manifest_c1)
929
930 manifest_c2 = {}
931 # Keep 20, modify 10, delete 10, add 10 new
932 oids = list(manifest_c1.items())
933 for path, oid in oids[:20]:
934 manifest_c2[path] = oid
935 for path, _ in oids[20:30]:
936 manifest_c2[path] = _write_obj(repo, f"v2-{path}".encode())
937 # oids[30:40] deleted
938 for i in range(10):
939 manifest_c2[f"new{i}.py"] = _write_obj(repo, f"new-{i}".encode())
940 _commit(repo, "c2 mixed", manifest_c2, parent=c1)
941
942 r = _fp(repo, "--json")
943 assert r.exit_code == 0
944 data = _json_out(r)
945 assert len(data["files_added"]) == 10
946 assert len(data["files_modified"]) == 10
947 assert len(data["files_deleted"]) == 10
948
949
950 # ---------------------------------------------------------------------------
951 # Security [GREEN]
952 # ---------------------------------------------------------------------------
953
954
955 class TestSecurity:
956 def test_path_traversal_in_treeish_rejected(self, tmp_path: pathlib.Path) -> None:
957 repo = _init_repo(tmp_path)
958 oid = _write_obj(repo, b"x")
959 _commit(repo, "init", {"f.py": oid})
960 r = _fp(repo, "../../etc/passwd", "--json")
961 assert r.exit_code != 0
962
963 def test_ansi_escape_in_treeish_rejected(self, tmp_path: pathlib.Path) -> None:
964 repo = _init_repo(tmp_path)
965 oid = _write_obj(repo, b"x")
966 _commit(repo, "init", {"f.py": oid})
967 r = _fp(repo, "\x1b[31mbad\x1b[0m", "--json")
968 assert r.exit_code != 0
969
970 def test_very_long_treeish_rejected(self, tmp_path: pathlib.Path) -> None:
971 repo = _init_repo(tmp_path)
972 oid = _write_obj(repo, b"x")
973 _commit(repo, "init", {"f.py": oid})
974 r = _fp(repo, "a" * 300, "--json")
975 assert r.exit_code != 0
976
977 def test_null_byte_in_treeish_rejected(self, tmp_path: pathlib.Path) -> None:
978 repo = _init_repo(tmp_path)
979 oid = _write_obj(repo, b"x")
980 _commit(repo, "init", {"f.py": oid})
981 r = _fp(repo, "main\x00malicious", "--json")
982 assert r.exit_code != 0
983
984 def test_error_goes_to_stderr_not_stdout(self, tmp_path: pathlib.Path) -> None:
985 repo = _init_repo(tmp_path)
986 r = _fp(repo, "--json") # empty repo → error
987 assert r.exit_code != 0
988 assert "❌" in r.stderr or "error" in r.stderr.lower() or r.exit_code != 0
989
990 def test_no_traceback_on_bad_ref(self, tmp_path: pathlib.Path) -> None:
991 repo = _init_repo(tmp_path)
992 oid = _write_obj(repo, b"x")
993 _commit(repo, "init", {"f.py": oid})
994 r = _fp(repo, "no-such-ref", "--json")
995 assert "Traceback" not in r.output
996 assert "Traceback" not in r.stderr
997
998
999 # ---------------------------------------------------------------------------
1000 # Performance [GREEN]
1001 # ---------------------------------------------------------------------------
1002
1003
1004 class TestPerformance:
1005 def test_duration_ms_under_two_seconds(self, tmp_path: pathlib.Path) -> None:
1006 repo = _init_repo(tmp_path)
1007 manifest = {}
1008 for i in range(20):
1009 oid = _write_obj(repo, f"content {i}".encode() * 50)
1010 manifest[f"file_{i}.py"] = oid
1011 _commit(repo, "feat: 20 files", manifest)
1012 data = _json_out(_fp(repo, "--json"))
1013 assert data["duration_ms"] < 2000.0
1014
1015 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
1016 repo = _init_repo(tmp_path)
1017 oid = _write_obj(repo, b"x")
1018 _commit(repo, "init", {"f.py": oid})
1019 data = _json_out(_fp(repo, "--json"))
1020 assert data["duration_ms"] >= 0.0
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago