gabriel / muse public
test_cmd_cherry_pick.py python
195 lines 7.6 KB
Raw
sha256:f6cd81bc71702f5c1c6890bd39aaba994fe58c75f019d7c03934724fa2739bb4 fix: carry dev changes harmony dropped in merge — detached … Sonnet 4.6 minor ⚠ breaking 17 days ago
1 """Comprehensive tests for ``muse cherry-pick``.
2
3 Covers:
4 - E2E: cherry-pick a specific commit onto current branch
5 - Integration: commit is replayed, creates new commit
6 - Security: sanitized output for conflict paths
7 - Stress: cherry-pick many commits
8 """
9
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 import pytest
17 from tests.cli_test_helper import CliRunner
18 from muse.core.types import long_id, fake_id, blob_id
19 from muse.core.paths import heads_dir, ref_path, muse_dir
20
21 cli = None # argparse migration — CliRunner ignores this arg
22
23 runner = CliRunner()
24
25
26 # ---------------------------------------------------------------------------
27 # Shared helpers
28 # ---------------------------------------------------------------------------
29
30 def _env(root: pathlib.Path) -> Manifest:
31 return {"MUSE_REPO_ROOT": str(root)}
32
33
34 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
35 dot_muse = muse_dir(tmp_path)
36 dot_muse.mkdir()
37 repo_id = fake_id("repo")
38 (dot_muse / "repo.json").write_text(json.dumps({
39 "repo_id": repo_id,
40 "domain": "code",
41 "default_branch": "main",
42 "created_at": "2025-01-01T00:00:00+00:00",
43 }), encoding="utf-8")
44 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
45 (dot_muse / "refs" / "heads").mkdir(parents=True)
46 (dot_muse / "snapshots").mkdir()
47 (dot_muse / "commits").mkdir()
48 (dot_muse / "objects").mkdir()
49 return tmp_path, repo_id
50
51
52 def _make_commit(root: pathlib.Path, repo_id: str, branch: str = "main",
53 message: str = "test",
54 manifest: Manifest | None = None) -> str:
55 from muse.core.commits import (
56 CommitRecord,
57 write_commit,
58 )
59 from muse.core.snapshots import (
60 SnapshotRecord,
61 write_snapshot,
62 )
63 from muse.core.ids import hash_commit, hash_snapshot
64
65 ref_file = ref_path(root, branch)
66 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
67 m = manifest or {}
68 snap_id = hash_snapshot(m)
69 committed_at = datetime.datetime.now(datetime.timezone.utc)
70 commit_id = hash_commit( parent_ids=[parent_id] if parent_id else [],
71 snapshot_id=snap_id, message=message,
72 committed_at_iso=committed_at.isoformat(),
73 )
74 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m))
75 write_commit(root, CommitRecord(
76 commit_id=commit_id, branch=branch,
77 snapshot_id=snap_id, message=message, committed_at=committed_at,
78 parent_commit_id=parent_id,
79 ))
80 ref_file.parent.mkdir(parents=True, exist_ok=True)
81 ref_file.write_text(commit_id, encoding="utf-8")
82 return commit_id
83
84
85 def _write_object(root: pathlib.Path, content: bytes) -> str:
86 from muse.core.object_store import write_object
87 oid = blob_id(content)
88 write_object(root, oid, content)
89 return oid
90
91
92 # ---------------------------------------------------------------------------
93 # Parser flag tests
94 # ---------------------------------------------------------------------------
95
96 class TestRegisterFlags:
97 def _parse(self, *args: str) -> "argparse.Namespace":
98 import argparse
99 from muse.cli.commands.cherry_pick import register
100 p = argparse.ArgumentParser()
101 sub = p.add_subparsers()
102 register(sub)
103 return p.parse_args(["cherry-pick", *args])
104
105 def test_default_json_out_is_false(self) -> None:
106 ns = self._parse("abc123")
107 assert ns.json_out is False
108
109 def test_json_flag_sets_json_out(self) -> None:
110 ns = self._parse("--json", "abc123")
111 assert ns.json_out is True
112
113 def test_j_shorthand_sets_json_out(self) -> None:
114 ns = self._parse("-j", "abc123")
115 assert ns.json_out is True
116
117
118 # ---------------------------------------------------------------------------
119 # Tests
120 # ---------------------------------------------------------------------------
121
122 class TestCherryPickCLI:
123 def test_cherry_pick_commit_from_another_branch(self, tmp_path: pathlib.Path) -> None:
124 root, repo_id = _init_repo(tmp_path)
125 base = _make_commit(root, repo_id, branch="main", message="base")
126 (heads_dir(root) / "feature").write_text(base)
127 obj = _write_object(root, b"feature content")
128 feature_commit = _make_commit(root, repo_id, branch="feature",
129 message="feature work",
130 manifest={"new.mid": obj})
131 result = runner.invoke(
132 cli, ["cherry-pick", feature_commit], env=_env(root), catch_exceptions=False
133 )
134 assert result.exit_code == 0
135
136 def test_cherry_pick_invalid_commit_fails(self, tmp_path: pathlib.Path) -> None:
137 root, repo_id = _init_repo(tmp_path)
138 _make_commit(root, repo_id)
139 result = runner.invoke(cli, ["cherry-pick", "deadbeef" * 8], env=_env(root))
140 assert result.exit_code != 0
141
142 def test_cherry_pick_creates_new_commit(self, tmp_path: pathlib.Path) -> None:
143 root, repo_id = _init_repo(tmp_path)
144 base = _make_commit(root, repo_id, branch="main", message="base")
145 (heads_dir(root) / "feature").write_text(base)
146 obj = _write_object(root, b"cherry content")
147 feature_commit = _make_commit(root, repo_id, branch="feature",
148 message="cherry", manifest={"c.mid": obj})
149 original_head = (heads_dir(root) / "main").read_text().strip()
150 runner.invoke(cli, ["cherry-pick", feature_commit], env=_env(root), catch_exceptions=False)
151 new_head = (heads_dir(root) / "main").read_text().strip()
152 assert new_head != original_head
153
154 def test_cherry_pick_format_json(self, tmp_path: pathlib.Path) -> None:
155 root, repo_id = _init_repo(tmp_path)
156 base = _make_commit(root, repo_id, branch="main", message="base")
157 (heads_dir(root) / "feature").write_text(base)
158 obj = _write_object(root, b"json pick")
159 feature_commit = _make_commit(root, repo_id, branch="feature",
160 message="json", manifest={"j.mid": obj})
161 result = runner.invoke(
162 cli, ["cherry-pick", "--json", feature_commit],
163 env=_env(root), catch_exceptions=False
164 )
165 assert result.exit_code == 0
166 data = json.loads(result.output)
167 assert isinstance(data, dict)
168
169 def test_cherry_pick_output_sanitized(self, tmp_path: pathlib.Path) -> None:
170 root, repo_id = _init_repo(tmp_path)
171 base = _make_commit(root, repo_id, branch="main", message="base")
172 (heads_dir(root) / "feature").write_text(base)
173 obj = _write_object(root, b"safe content")
174 feature_commit = _make_commit(root, repo_id, branch="feature",
175 message="safe", manifest={"s.mid": obj})
176 result = runner.invoke(cli, ["cherry-pick", feature_commit], env=_env(root), catch_exceptions=False)
177 assert "\x1b" not in result.output
178
179
180 class TestCherryPickStress:
181 def test_cherry_pick_sequence(self, tmp_path: pathlib.Path) -> None:
182 root, repo_id = _init_repo(tmp_path)
183 base = _make_commit(root, repo_id, branch="main", message="base")
184 (heads_dir(root) / "feature").write_text(base)
185 commits = []
186 for i in range(5):
187 obj = _write_object(root, f"content {i}".encode())
188 c = _make_commit(root, repo_id, branch="feature",
189 message=f"commit {i}", manifest={f"f{i}.mid": obj})
190 commits.append(c)
191 for commit_id in commits:
192 result = runner.invoke(
193 cli, ["cherry-pick", commit_id], env=_env(root), catch_exceptions=False
194 )
195 assert result.exit_code == 0
File History 3 commits
sha256:43c82f6d4fa2e85dd9ed9dd1a31199ec6b481191517aba66dfa9da275dbfa1af Merge branch 'dev' into main Human 3 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 22 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 74 days ago