"""Supercharge tests for ``muse format-patch``. TDD — sections labelled [RED] contain tests that fail until the feature lands. Sections labelled [GREEN] fill gaps against existing behavior. New features under test ----------------------- - ``--agent-id `` [RED] embed agent provenance in the patch record - ``--model-id `` [RED] embed model provenance in the patch record - ``--intent `` [RED] embed an intent description in the patch record - ``--no-blobs`` [RED] omit base64 blob content from the patch file - Rename detection [RED] same-oid delete+insert → rename op in files_renamed Gap-fill coverage ----------------- - Register-flag parser shape - Unit tests for _sem_ver_bump, _breaking_changes, _make_patch_filename, _action_label - Blob content verification (decoded bytes match source) - from/to manifest delta correctness - Initial-commit sentinel (from_snapshot_id = sha256:000…, from_commit_id = "") - Required-objects sorted + sha256: prefix - ops count === files_added + files_modified + files_deleted - Default stdout output is valid JSON - Stress: 50-file commit - Security: path-traversal and ANSI in treeish - Performance: duration_ms plausible """ from __future__ import annotations from collections.abc import Mapping import argparse import datetime import json import pathlib import time import pytest from muse.cli.commands.format_patch import ( _action_label, _breaking_changes, _build_file_level_ops, _make_patch_filename, _sem_ver_bump, register, ) from muse.core.object_store import write_object from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import NULL_LONG_ID, blob_id, long_id from muse.core.paths import ref_path, muse_dir runner = CliRunner() # --------------------------------------------------------------------------- # Repo / commit helpers (shared) # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path) -> pathlib.Path: dot_muse = muse_dir(path) for sub in ("commits", "snapshots", "objects", "refs/heads"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"})) return path def _write_obj(repo: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(repo, oid, content) return oid _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) def _commit( repo: pathlib.Path, msg: str, manifest: dict[str, str], *, branch: str = "main", parent: str | None = None, ) -> str: sid = compute_snapshot_id(manifest) write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest, created_at=_TS)) parent_ids = [parent] if parent else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=sid, message=msg, committed_at_iso=_TS.isoformat(), author="gabriel",) write_commit(repo, CommitRecord( commit_id=cid, branch=branch, snapshot_id=sid, message=msg, committed_at=_TS, author="gabriel", parent_commit_id=parent, parent2_commit_id=None, )) ref = ref_path(repo, branch) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(cid) return cid def _fp(repo: pathlib.Path, *args: str) -> InvokeResult: return runner.invoke(None, ["format-patch", *args], env={"MUSE_REPO_ROOT": str(repo)}) def _json_out(r: InvokeResult) -> Mapping[str, object]: for line in r.output.splitlines(): line = line.strip() if line.startswith("{"): return json.loads(line) raise ValueError(f"No JSON line in output:\n{r.output!r}") # --------------------------------------------------------------------------- # Register flags [GREEN] # --------------------------------------------------------------------------- class TestRegisterFlags: """Parser shape — verify flags are wired correctly.""" def _parse(self, *args: str) -> argparse.Namespace: p = argparse.ArgumentParser() subs = p.add_subparsers() register(subs) return p.parse_args(["format-patch", *args]) def test_treeish_defaults_to_head(self) -> None: ns = self._parse() assert ns.treeish == "HEAD" def test_treeish_positional(self) -> None: ns = self._parse("main") assert ns.treeish == "main" def test_output_dir_flag(self) -> None: ns = self._parse("--output-dir", "/tmp") assert ns.output_dir == "/tmp" def test_output_dir_short_flag(self) -> None: ns = self._parse("-o", "/tmp") assert ns.output_dir == "/tmp" def test_json_flag(self) -> None: ns = self._parse("--json") assert ns.json_out is True def test_json_default_false(self) -> None: ns = self._parse() assert ns.json_out is False def test_output_dir_default_none(self) -> None: ns = self._parse() assert ns.output_dir is None # [RED] — these flags don't exist yet def test_agent_id_flag(self) -> None: ns = self._parse("--agent-id", "claude-code") assert ns.agent_id == "claude-code" def test_agent_id_default_empty(self) -> None: ns = self._parse() assert ns.agent_id == "" def test_model_id_flag(self) -> None: ns = self._parse("--model-id", "claude-sonnet-4-6") assert ns.model_id == "claude-sonnet-4-6" def test_model_id_default_empty(self) -> None: ns = self._parse() assert ns.model_id == "" def test_intent_flag(self) -> None: ns = self._parse("--intent", "add login flow") assert ns.intent == "add login flow" def test_intent_default_empty(self) -> None: ns = self._parse() assert ns.intent == "" def test_no_blobs_flag(self) -> None: ns = self._parse("--no-blobs") assert ns.no_blobs is True def test_no_blobs_default_false(self) -> None: ns = self._parse() assert ns.no_blobs is False # --------------------------------------------------------------------------- # _sem_ver_bump unit tests [GREEN] # --------------------------------------------------------------------------- class TestSemVerBump: def test_break_prefix_is_major(self) -> None: assert _sem_ver_bump("break: remove old API", [], []) == "major" def test_feat_bang_is_major(self) -> None: assert _sem_ver_bump("feat!: overhaul auth", [], []) == "major" def test_breaking_change_body_is_major(self) -> None: assert _sem_ver_bump("refactor: cleanup\n\nBREAKING CHANGE: old param removed", [], []) == "major" def test_breaking_change_case_insensitive(self) -> None: assert _sem_ver_bump("breaking change in behavior", [], []) == "major" def test_feat_prefix_is_minor(self) -> None: assert _sem_ver_bump("feat: add endpoint", [], []) == "minor" def test_files_added_is_minor(self) -> None: assert _sem_ver_bump("chore: misc", ["new_file.py"], []) == "minor" def test_fix_prefix_is_patch(self) -> None: assert _sem_ver_bump("fix: off-by-one", [], []) == "patch" def test_chore_no_additions_is_patch(self) -> None: assert _sem_ver_bump("chore: update deps", [], []) == "patch" def test_empty_message_is_patch(self) -> None: assert _sem_ver_bump("", [], []) == "patch" def test_files_deleted_alone_is_patch(self) -> None: assert _sem_ver_bump("chore: cleanup", [], ["old.py"]) == "patch" def test_feat_prefix_beats_files_added(self) -> None: # Both trigger minor — result is still minor assert _sem_ver_bump("feat: add stuff", ["new.py"], []) == "minor" def test_break_prefix_beats_files_added(self) -> None: assert _sem_ver_bump("break: remove", ["new.py"], []) == "major" def test_result_is_one_of_three_values(self) -> None: for msg in ["anything", "feat: x", "break: y"]: result = _sem_ver_bump(msg, [], []) assert result in ("major", "minor", "patch") # --------------------------------------------------------------------------- # _breaking_changes unit tests [GREEN] # --------------------------------------------------------------------------- class TestBreakingChanges: def test_empty_message_returns_empty(self) -> None: assert _breaking_changes("") == [] def test_no_breaking_change_returns_empty(self) -> None: assert _breaking_changes("feat: add endpoint") == [] def test_single_breaking_change(self) -> None: msg = "refactor: cleanup\n\nBREAKING CHANGE: removed --legacy flag" result = _breaking_changes(msg) assert result == ["removed --legacy flag"] def test_multiple_breaking_changes(self) -> None: msg = "refactor:\n\nBREAKING CHANGE: first\nBREAKING CHANGE: second" result = _breaking_changes(msg) assert result == ["first", "second"] def test_leading_trailing_whitespace_stripped(self) -> None: msg = "BREAKING CHANGE: trimmed " result = _breaking_changes(msg) assert result == ["trimmed"] def test_not_at_start_of_line_ignored(self) -> None: # Mid-sentence "BREAKING CHANGE" not at start of line msg = "This has BREAKING CHANGE: in the middle" # The implementation checks stripped.upper().startswith("BREAKING CHANGE:") # so it WOULD match if the stripped line starts with it — it does here # because "This has..." stripped starts with "This" not "BREAKING CHANGE" result = _breaking_changes(msg) assert result == [] def test_returns_list(self) -> None: assert isinstance(_breaking_changes("anything"), list) # --------------------------------------------------------------------------- # _make_patch_filename unit tests [GREEN] # --------------------------------------------------------------------------- class TestMakePatchFilename: def test_basic_subject(self) -> None: assert _make_patch_filename("feat: add hello") == "feat-add-hello.mpatch" def test_ends_with_mpatch(self) -> None: name = _make_patch_filename("anything") assert name.endswith(".mpatch") def test_empty_subject_returns_patch(self) -> None: assert _make_patch_filename("") == "patch.mpatch" def test_slash_replaced(self) -> None: name = _make_patch_filename("fix/my-bug") assert "/" not in name def test_backslash_replaced(self) -> None: name = _make_patch_filename("fix\\my-bug") assert "\\" not in name def test_dot_replaced(self) -> None: # dots → dashes in the slug portion (before the .mpatch extension) name = _make_patch_filename("v1.2.3 release") slug = name.removesuffix(".mpatch") assert "." not in slug def test_long_subject_truncated(self) -> None: long_msg = "x" * 100 name = _make_patch_filename(long_msg) slug = name.removesuffix(".mpatch") assert len(slug) <= 52 def test_unicode_stripped(self) -> None: name = _make_patch_filename("feat: émoji 🚀 add") # Non-ASCII removed, but ASCII words remain assert "feat" in name def test_whitespace_replaced_with_dash(self) -> None: name = _make_patch_filename("add multiple spaces") assert " " not in name def test_no_leading_trailing_dashes_in_slug(self) -> None: slug = _make_patch_filename(" spaces around ").removesuffix(".mpatch") assert not slug.startswith("-") assert not slug.endswith("-") def test_no_consecutive_dashes_in_slug(self) -> None: slug = _make_patch_filename("a!!b").removesuffix(".mpatch") assert "--" not in slug # --------------------------------------------------------------------------- # _action_label unit tests [GREEN] # --------------------------------------------------------------------------- class TestActionLabel: def test_insert_is_inserted(self) -> None: assert _action_label("insert") == "inserted" def test_delete_is_deleted(self) -> None: assert _action_label("delete") == "deleted" def test_replace_is_modified(self) -> None: assert _action_label("replace") == "modified" def test_mutate_is_modified(self) -> None: assert _action_label("mutate") == "modified" def test_patch_is_modified(self) -> None: assert _action_label("patch") == "modified" def test_move_is_moved(self) -> None: assert _action_label("move") == "moved" def test_rename_is_renamed(self) -> None: assert _action_label("rename") == "renamed" def test_unknown_defaults_to_modified(self) -> None: assert _action_label("frob") == "modified" assert _action_label("") == "modified" assert _action_label("UPDATE") == "modified" # --------------------------------------------------------------------------- # _build_file_level_ops — internal unit tests [GREEN + RED for rename] # --------------------------------------------------------------------------- class TestBuildFileOps: def test_added_file_in_ops(self) -> None: base: dict[str, str] = {} target = {"new.py": long_id("a" * 64)} ops, added, modified, deleted, *_ = _build_file_level_ops(base, target) assert any(op["address"] == "new.py" and op["op"] == "insert" for op in ops) def test_deleted_file_in_ops(self) -> None: base = {"old.py": long_id("a" * 64)} target: dict[str, str] = {} ops, added, modified, deleted, *_ = _build_file_level_ops(base, target) assert any(op["address"] == "old.py" and op["op"] == "delete" for op in ops) def test_modified_file_in_ops(self) -> None: oid_a = long_id("a" * 64) oid_b = long_id("b" * 64) ops, added, modified, deleted, *_ = _build_file_level_ops( {"f.py": oid_a}, {"f.py": oid_b} ) assert any(op["address"] == "f.py" and op["op"] == "replace" for op in ops) def test_added_list_sorted(self) -> None: base: dict[str, str] = {} target = {"z.py": long_id("z" * 64), "a.py": long_id("a" * 64)} _, added, _, _, *_ = _build_file_level_ops(base, target) assert added == sorted(added) def test_deleted_list_sorted(self) -> None: base = {"z.py": long_id("z" * 64), "a.py": long_id("a" * 64)} _, _, _, deleted, *_ = _build_file_level_ops(base, {}) assert deleted == sorted(deleted) def test_modified_list_sorted(self) -> None: oid_a = long_id("a" * 64) oid_b = long_id("b" * 64) base = {"z.py": oid_a, "a.py": oid_a} target = {"z.py": oid_b, "a.py": oid_b} _, _, modified, _, *_ = _build_file_level_ops(base, target) assert modified == sorted(modified) def test_unchanged_file_not_in_ops(self) -> None: oid = long_id("a" * 64) ops, _, _, _, *_ = _build_file_level_ops({"f.py": oid}, {"f.py": oid}) addresses = [op["address"] for op in ops] assert "f.py" not in addresses # [RED] rename detection — same oid deleted + added at different path = rename def test_rename_detected(self) -> None: oid = long_id("a" * 64) base = {"old.py": oid} target = {"new.py": oid} ops, added, modified, deleted, renamed = _build_file_level_ops(base, target) assert "old.py" in renamed assert renamed["old.py"] == "new.py" def test_rename_not_in_files_added(self) -> None: oid = long_id("a" * 64) _, added, _, _, renamed = _build_file_level_ops({"old.py": oid}, {"new.py": oid}) assert "new.py" not in added def test_rename_not_in_files_deleted(self) -> None: oid = long_id("a" * 64) _, _, _, deleted, renamed = _build_file_level_ops({"old.py": oid}, {"new.py": oid}) assert "old.py" not in deleted def test_rename_op_present_in_ops(self) -> None: oid = long_id("a" * 64) ops, _, _, _, _ = _build_file_level_ops({"old.py": oid}, {"new.py": oid}) rename_ops = [op for op in ops if op.get("op") == "move"] assert len(rename_ops) == 1 def test_rename_op_action_label_is_moved(self) -> None: oid = long_id("a" * 64) ops, _, _, _, _ = _build_file_level_ops({"old.py": oid}, {"new.py": oid}) rename_ops = [op for op in ops if op.get("op") == "move"] assert rename_ops[0]["action_label"] == "moved" def test_different_oid_not_a_rename(self) -> None: oid_a = long_id("a" * 64) oid_b = long_id("b" * 64) _, added, _, deleted, renamed = _build_file_level_ops( {"old.py": oid_a}, {"new.py": oid_b} ) assert not renamed assert "old.py" in deleted assert "new.py" in added def test_empty_renamed_dict_when_no_renames(self) -> None: oid_a = long_id("a" * 64) oid_b = long_id("b" * 64) _, _, _, _, renamed = _build_file_level_ops({"f.py": oid_a}, {"f.py": oid_b}) assert renamed == {} # --------------------------------------------------------------------------- # Blob embedding [GREEN] # --------------------------------------------------------------------------- class TestBlobEmbedding: def test_blobs_field_present(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"hello blob") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert "blobs" in data def test_blobs_is_dict(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"hello blob") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert isinstance(data["blobs"], dict) def test_blob_key_matches_required_object(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) content = b"blob content" oid = _write_obj(repo, content) _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert oid in data["blobs"] def test_blob_decodes_to_original_content(self, tmp_path: pathlib.Path) -> None: import base64 repo = _init_repo(tmp_path) content = b"exact bytes\x00\x01\x02" oid = _write_obj(repo, content) _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) decoded = base64.b64decode(data["blobs"][oid]) assert decoded == content def test_blobs_is_base64_valid_string(self, tmp_path: pathlib.Path) -> None: import base64 repo = _init_repo(tmp_path) oid = _write_obj(repo, b"any content") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) for val in data["blobs"].values(): assert isinstance(val, str) base64.b64decode(val) # must not raise def test_unmodified_objects_not_in_blobs(self, tmp_path: pathlib.Path) -> None: """Blobs only contains objects in to_manifest (new/modified), not deleted.""" repo = _init_repo(tmp_path) oid_a = _write_obj(repo, b"a") oid_b = _write_obj(repo, b"b") c1 = _commit(repo, "c1", {"a.py": oid_a, "b.py": oid_b}) oid_c = _write_obj(repo, b"c") _commit(repo, "c2", {"a.py": oid_a, "c.py": oid_c}, parent=c1) # b.py deleted → oid_b not in to_manifest → not in blobs data = _json_out(_fp(repo, "--json")) assert oid_b not in data["blobs"] # --------------------------------------------------------------------------- # Required objects [GREEN] # --------------------------------------------------------------------------- class TestRequiredObjects: def test_all_sha256_prefixed(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) for rid in data["required_objects"]: assert rid.startswith("sha256:") def test_required_objects_is_sorted(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid_a = _write_obj(repo, b"aaa") oid_b = _write_obj(repo, b"bbb") _commit(repo, "init", {"a.py": oid_a, "b.py": oid_b}) data = _json_out(_fp(repo, "--json")) ro = data["required_objects"] assert ro == sorted(ro) def test_required_objects_subset_of_to_manifest(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"y") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) to_vals = set(data["to_manifest"].values()) for rid in data["required_objects"]: assert rid in to_vals def test_required_objects_empty_for_no_change(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"z") c1 = _commit(repo, "c1", {"f.py": oid}) _commit(repo, "c2 no-op", {"f.py": oid}, parent=c1) data = _json_out(_fp(repo, "--json")) assert data["required_objects"] == [] # --------------------------------------------------------------------------- # Manifest delta correctness [GREEN] # --------------------------------------------------------------------------- class TestManifestDelta: def test_added_path_in_to_manifest(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"new") _commit(repo, "init", {"new.py": oid}) data = _json_out(_fp(repo, "--json")) assert "new.py" in data["to_manifest"] def test_added_path_not_in_from_manifest(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"new") _commit(repo, "init", {"new.py": oid}) data = _json_out(_fp(repo, "--json")) assert "new.py" not in data["from_manifest"] def test_deleted_path_in_from_manifest(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"old") c1 = _commit(repo, "c1", {"old.py": oid}) _commit(repo, "c2", {}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "old.py" in data["from_manifest"] def test_deleted_path_not_in_to_manifest(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"old") c1 = _commit(repo, "c1", {"old.py": oid}) _commit(repo, "c2", {}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "old.py" not in data["to_manifest"] def test_modified_path_in_both_manifests(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid_a = _write_obj(repo, b"v1") c1 = _commit(repo, "c1", {"f.py": oid_a}) oid_b = _write_obj(repo, b"v2") _commit(repo, "c2", {"f.py": oid_b}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "f.py" in data["from_manifest"] assert "f.py" in data["to_manifest"] assert data["from_manifest"]["f.py"] != data["to_manifest"]["f.py"] def test_unchanged_path_not_in_either_manifest(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid_keep = _write_obj(repo, b"keep") oid_chg = _write_obj(repo, b"v1") c1 = _commit(repo, "c1", {"keep.py": oid_keep, "chg.py": oid_chg}) oid_chg2 = _write_obj(repo, b"v2") _commit(repo, "c2", {"keep.py": oid_keep, "chg.py": oid_chg2}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "keep.py" not in data["from_manifest"] assert "keep.py" not in data["to_manifest"] # --------------------------------------------------------------------------- # Initial commit sentinel [GREEN] # --------------------------------------------------------------------------- class TestInitialCommit: def test_from_snapshot_id_is_sentinel_for_initial(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) # Sentinel for initial commit is sha256:000...000 (64 zeros) assert data["from_snapshot_id"] == NULL_LONG_ID def test_from_commit_id_empty_for_initial(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert data["from_commit_id"] == "" def test_all_files_in_files_added_for_initial(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid_a = _write_obj(repo, b"a") oid_b = _write_obj(repo, b"b") _commit(repo, "init", {"a.py": oid_a, "b.py": oid_b}) data = _json_out(_fp(repo, "--json")) assert "a.py" in data["files_added"] assert "b.py" in data["files_added"] assert data["files_modified"] == [] assert data["files_deleted"] == [] def test_from_snapshot_id_set_for_second_commit(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"v1") c1 = _commit(repo, "c1", {"f.py": oid}) oid2 = _write_obj(repo, b"v2") _commit(repo, "c2", {"f.py": oid2}, parent=c1) data = _json_out(_fp(repo, "--json")) # Non-initial: from_snapshot_id should NOT be the sentinel assert data["from_snapshot_id"] != NULL_LONG_ID # --------------------------------------------------------------------------- # Agent provenance flags [RED] — --agent-id, --model-id, --intent # --------------------------------------------------------------------------- class TestAgentProvenance: def test_agent_id_set_in_output(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json", "--agent-id", "claude-code")) assert data["agent_id"] == "claude-code" def test_model_id_set_in_output(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json", "--model-id", "claude-sonnet-4-6")) assert data["model_id"] == "claude-sonnet-4-6" def test_intent_set_in_output(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json", "--intent", "bootstrap project")) assert data["intent"] == "bootstrap project" def test_agent_id_in_mpatch_file(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) out_dir = tmp_path / "patches" out_dir.mkdir() r = _fp(repo, "--output-dir", str(out_dir), "--agent-id", "claude-code") assert r.exit_code == 0 patch_file = list(out_dir.glob("*.mpatch"))[0] data = json.loads(patch_file.read_bytes()) assert data["agent_id"] == "claude-code" def test_agent_id_affects_patch_id(self, tmp_path: pathlib.Path) -> None: """Different agent_id → different patch_id (agent_id is part of canonical JSON).""" repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) pid_no_agent = _json_out(_fp(repo, "--json"))["patch_id"] pid_with_agent = _json_out(_fp(repo, "--json", "--agent-id", "claude-code"))["patch_id"] assert pid_no_agent != pid_with_agent def test_no_agent_flags_leaves_fields_empty(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert data["agent_id"] == "" assert data["model_id"] == "" assert data["intent"] == "" def test_all_provenance_flags_together(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json", "--agent-id", "claude-code", "--model-id", "claude-sonnet-4-6", "--intent", "add login endpoint")) assert data["agent_id"] == "claude-code" assert data["model_id"] == "claude-sonnet-4-6" assert data["intent"] == "add login endpoint" # --------------------------------------------------------------------------- # --no-blobs flag [RED] # --------------------------------------------------------------------------- class TestNoBlobs: def test_no_blobs_empties_blobs_dict(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"blob content here") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json", "--no-blobs")) assert data["blobs"] == {} def test_no_blobs_preserves_required_objects(self, tmp_path: pathlib.Path) -> None: """required_objects still lists what the target needs even without inline blobs.""" repo = _init_repo(tmp_path) oid = _write_obj(repo, b"blob content here") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json", "--no-blobs")) assert oid in data["required_objects"] def test_no_blobs_in_mpatch_file(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"some bytes") _commit(repo, "init", {"f.py": oid}) out_dir = tmp_path / "patches" out_dir.mkdir() r = _fp(repo, "--output-dir", str(out_dir), "--no-blobs") assert r.exit_code == 0 data = json.loads(list(out_dir.glob("*.mpatch"))[0].read_bytes()) assert data["blobs"] == {} def test_default_has_blobs(self, tmp_path: pathlib.Path) -> None: """Without --no-blobs, blobs are embedded (existing behavior).""" repo = _init_repo(tmp_path) oid = _write_obj(repo, b"keep me") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert len(data["blobs"]) > 0 def test_no_blobs_output_smaller_than_with_blobs(self, tmp_path: pathlib.Path) -> None: """--no-blobs patch should be smaller (no base64 content).""" repo = _init_repo(tmp_path) content = b"x" * 1024 # 1KB object oid = _write_obj(repo, content) _commit(repo, "init", {"f.py": oid}) r_with = _fp(repo, "--json") r_no = _fp(repo, "--json", "--no-blobs") assert len(r_no.output) < len(r_with.output) # --------------------------------------------------------------------------- # Rename detection via CLI [RED] # --------------------------------------------------------------------------- class TestRenameDetectionCLI: def test_rename_in_files_renamed(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"shared content") c1 = _commit(repo, "c1", {"old.py": oid}) _commit(repo, "c2 rename", {"new.py": oid}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "old.py" in data["files_renamed"] assert data["files_renamed"]["old.py"] == "new.py" def test_rename_not_in_files_added(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"shared content") c1 = _commit(repo, "c1", {"old.py": oid}) _commit(repo, "c2 rename", {"new.py": oid}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "new.py" not in data["files_added"] def test_rename_not_in_files_deleted(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"shared content") c1 = _commit(repo, "c1", {"old.py": oid}) _commit(repo, "c2 rename", {"new.py": oid}, parent=c1) data = _json_out(_fp(repo, "--json")) assert "old.py" not in data["files_deleted"] def test_genuine_add_and_delete_not_confused_for_rename(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid_a = _write_obj(repo, b"content A") oid_b = _write_obj(repo, b"content B") c1 = _commit(repo, "c1", {"a.py": oid_a}) _commit(repo, "c2", {"b.py": oid_b}, parent=c1) data = _json_out(_fp(repo, "--json")) assert data["files_renamed"] == {} assert "b.py" in data["files_added"] assert "a.py" in data["files_deleted"] # --------------------------------------------------------------------------- # Default stdout output [GREEN] # --------------------------------------------------------------------------- class TestDefaultOutput: def test_default_output_is_valid_json(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo) assert r.exit_code == 0 data = json.loads(r.output.strip()) assert "patch_id" in data def test_default_output_has_patch_id(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo) data = json.loads(r.output.strip()) assert data["patch_id"].startswith("sha256:") # --------------------------------------------------------------------------- # ops completeness [GREEN] # --------------------------------------------------------------------------- class TestOpsCompleteness: def test_ops_count_equals_sum_of_file_lists(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid1 = _write_obj(repo, b"a") oid2 = _write_obj(repo, b"b") oid3 = _write_obj(repo, b"c") c1 = _commit(repo, "c1", {"a.py": oid1, "b.py": oid2, "c.py": oid3}) oid4 = _write_obj(repo, b"a-modified") _commit(repo, "c2", {"a.py": oid4, "b.py": oid2}, parent=c1) data = _json_out(_fp(repo, "--json")) # c.py deleted, a.py modified, b.py unchanged total_file_changes = ( len(data["files_added"]) + len(data["files_modified"]) + len(data["files_deleted"]) + len(data["files_renamed"]) ) # Each changed file has exactly one op (excluding renames which have one move op) assert len(data["ops"]) == total_file_changes def test_each_op_has_required_fields(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid_a = _write_obj(repo, b"a") oid_b = _write_obj(repo, b"b") c1 = _commit(repo, "c1", {"a.py": oid_a}) oid_c = _write_obj(repo, b"a-mod") _commit(repo, "c2", {"a.py": oid_c, "b.py": oid_b}, parent=c1) data = _json_out(_fp(repo, "--json")) for op in data["ops"]: assert "op" in op assert "address" in op assert "action_label" in op # --------------------------------------------------------------------------- # Stress [GREEN] # --------------------------------------------------------------------------- class TestStress: def test_50_files_added(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) manifest = {} for i in range(50): content = f"# file {i}\n".encode() * 10 oid = _write_obj(repo, content) manifest[f"src/file_{i:02d}.py"] = oid _commit(repo, "feat: add 50 files", manifest) r = _fp(repo, "--json") assert r.exit_code == 0 data = _json_out(r) assert len(data["files_added"]) == 50 assert len(data["ops"]) == 50 def test_mixed_50_file_commit(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) manifest_c1 = {} for i in range(40): oid = _write_obj(repo, f"v1-{i}".encode()) manifest_c1[f"f{i:02d}.py"] = oid c1 = _commit(repo, "c1", manifest_c1) manifest_c2 = {} # Keep 20, modify 10, delete 10, add 10 new oids = list(manifest_c1.items()) for path, oid in oids[:20]: manifest_c2[path] = oid for path, _ in oids[20:30]: manifest_c2[path] = _write_obj(repo, f"v2-{path}".encode()) # oids[30:40] deleted for i in range(10): manifest_c2[f"new{i}.py"] = _write_obj(repo, f"new-{i}".encode()) _commit(repo, "c2 mixed", manifest_c2, parent=c1) r = _fp(repo, "--json") assert r.exit_code == 0 data = _json_out(r) assert len(data["files_added"]) == 10 assert len(data["files_modified"]) == 10 assert len(data["files_deleted"]) == 10 # --------------------------------------------------------------------------- # Security [GREEN] # --------------------------------------------------------------------------- class TestSecurity: def test_path_traversal_in_treeish_rejected(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo, "../../etc/passwd", "--json") assert r.exit_code != 0 def test_ansi_escape_in_treeish_rejected(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo, "\x1b[31mbad\x1b[0m", "--json") assert r.exit_code != 0 def test_very_long_treeish_rejected(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo, "a" * 300, "--json") assert r.exit_code != 0 def test_null_byte_in_treeish_rejected(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo, "main\x00malicious", "--json") assert r.exit_code != 0 def test_error_goes_to_stderr_not_stdout(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) r = _fp(repo, "--json") # empty repo → error assert r.exit_code != 0 assert "❌" in r.stderr or "error" in r.stderr.lower() or r.exit_code != 0 def test_no_traceback_on_bad_ref(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) r = _fp(repo, "no-such-ref", "--json") assert "Traceback" not in r.output assert "Traceback" not in r.stderr # --------------------------------------------------------------------------- # Performance [GREEN] # --------------------------------------------------------------------------- class TestPerformance: def test_duration_ms_under_two_seconds(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) manifest = {} for i in range(20): oid = _write_obj(repo, f"content {i}".encode() * 50) manifest[f"file_{i}.py"] = oid _commit(repo, "feat: 20 files", manifest) data = _json_out(_fp(repo, "--json")) assert data["duration_ms"] < 2000.0 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) oid = _write_obj(repo, b"x") _commit(repo, "init", {"f.py": oid}) data = _json_out(_fp(repo, "--json")) assert data["duration_ms"] >= 0.0