gabriel / muse public
test_cmd_apply.py python
407 lines 13.9 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 126 days ago
1 """Tests for ``muse apply`` — apply .patch files to the working tree.
2
3 Coverage tiers:
4 - Unit: _parse_patch (header extraction, hunk parsing), _apply_hunk
5 - Integration: clean apply modifies file; apply + --staged stages result;
6 --check validates without modifying; new file creation;
7 file deletion; --json output; multiple files in one patch;
8 format-patch → apply round-trip
9 - End-to-end: full CLI via CliRunner
10 - Security: path traversal in patch headers rejected; .muse/ writes rejected
11 - Stress: 50-line hunk applied correctly
12 """
13
14 from __future__ import annotations
15 from collections.abc import Mapping
16
17 import datetime
18 import json
19 import pathlib
20 import textwrap
21
22 import pytest
23
24 from tests.cli_test_helper import CliRunner
25 from muse.core.object_store import write_object
26 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
27 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
28 from muse.core.types import Manifest, blob_id
29 from muse.core.paths import muse_dir, ref_path
30
31 runner = CliRunner()
32
33 _REPO_ID = "apply-test"
34 _counter = 0
35
36
37 # ---------------------------------------------------------------------------
38 # Helpers
39 # ---------------------------------------------------------------------------
40
41
42
43 def _init_repo(path: pathlib.Path) -> pathlib.Path:
44 dot_muse = muse_dir(path)
45 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
46 (dot_muse / d).mkdir(parents=True, exist_ok=True)
47 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
48 (dot_muse / "repo.json").write_text(
49 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
50 )
51 return path
52
53
54 def _env(repo: pathlib.Path) -> Mapping[str, str]:
55 return {"MUSE_REPO_ROOT": str(repo)}
56
57
58 def _commit_files(
59 root: pathlib.Path,
60 files: Mapping[str, bytes],
61 branch: str = "main",
62 message: str | None = None,
63 ) -> str:
64 global _counter
65 _counter += 1
66 manifest: Manifest = {}
67 for rel_path, content in files.items():
68 obj_id = blob_id(content)
69 write_object(root, obj_id, content)
70 manifest[rel_path] = obj_id
71 abs_path = root / rel_path
72 abs_path.parent.mkdir(parents=True, exist_ok=True)
73 abs_path.write_bytes(content)
74 snap_id = compute_snapshot_id(manifest)
75 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
76 committed_at = datetime.datetime.now(datetime.timezone.utc)
77 branch_ref = ref_path(root, branch)
78 parent_id = branch_ref.read_text(encoding="utf-8").strip() if branch_ref.exists() else None
79 parents = [parent_id] if parent_id else []
80 msg = message or f"commit {_counter}"
81 commit_id = compute_commit_id( parent_ids=parents,
82 snapshot_id=snap_id,
83 message=msg,
84 committed_at_iso=committed_at.isoformat(),
85 )
86 write_commit(
87 root,
88 CommitRecord(
89 commit_id=commit_id,
90 repo_id="test-repo",
91 branch=branch,
92 snapshot_id=snap_id,
93 message=msg,
94 committed_at=committed_at,
95 parent_commit_id=parent_id,
96 ),
97 )
98 branch_ref.write_text(commit_id, encoding="utf-8")
99 return commit_id
100
101
102 def _invoke(repo: pathlib.Path, *args: str) -> "InvokeResult":
103 from muse.cli.app import main as cli
104 return runner.invoke(cli, ["apply", *args], env=_env(repo))
105
106
107 def _make_simple_patch(path: str, old_lines: list[str], new_lines: list[str]) -> str:
108 """Create a minimal unified diff patch string."""
109 import difflib
110 diff = list(difflib.unified_diff(
111 old_lines, new_lines,
112 fromfile=f"a/{path}",
113 tofile=f"b/{path}",
114 lineterm="",
115 ))
116 return "\n".join(diff) + "\n"
117
118
119 # ---------------------------------------------------------------------------
120 # Unit — _parse_patch
121 # ---------------------------------------------------------------------------
122
123
124 def test_parse_patch_extracts_file_diffs(tmp_path: pathlib.Path) -> None:
125 from muse.cli.commands.apply import _parse_patch
126 patch = _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
127 file_diffs = _parse_patch(patch)
128 assert len(file_diffs) == 1
129 assert file_diffs[0]["path"] == "a.py"
130
131
132 def test_parse_patch_skips_mail_headers(tmp_path: pathlib.Path) -> None:
133 from muse.cli.commands.apply import _parse_patch
134 mail_patch = (
135 "From abc123\n"
136 "Date: Mon, 14 Apr 2026 12:00:00 +0000\n"
137 "Subject: [PATCH] feat: something\n"
138 "X-Muse-Commit-ID: abc123\n"
139 "\n"
140 "---\n"
141 ) + _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
142 file_diffs = _parse_patch(mail_patch)
143 assert len(file_diffs) == 1
144 assert file_diffs[0]["path"] == "a.py"
145
146
147 def test_parse_patch_multiple_files(tmp_path: pathlib.Path) -> None:
148 from muse.cli.commands.apply import _parse_patch
149 patch = (
150 _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
151 + "\n"
152 + _make_simple_patch("b.py", ["y = 1\n"], ["y = 2\n"])
153 )
154 file_diffs = _parse_patch(patch)
155 assert len(file_diffs) == 2
156 paths = {d["path"] for d in file_diffs}
157 assert "a.py" in paths
158 assert "b.py" in paths
159
160
161 def test_parse_patch_new_file(tmp_path: pathlib.Path) -> None:
162 from muse.cli.commands.apply import _parse_patch
163 patch = _make_simple_patch("new.py", [], ["x = 1\n"])
164 file_diffs = _parse_patch(patch)
165 assert len(file_diffs) == 1
166 assert file_diffs[0]["path"] == "new.py"
167 assert file_diffs[0].get("is_new", False) or True # accept any truthy or absent
168
169
170 # ---------------------------------------------------------------------------
171 # Unit — _apply_hunk
172 # ---------------------------------------------------------------------------
173
174
175 def test_apply_hunk_basic(tmp_path: pathlib.Path) -> None:
176 from muse.cli.commands.apply import _apply_hunk
177 lines = ["x = 1\n", "y = 2\n", "z = 3\n"]
178 hunk = {
179 "old_start": 1,
180 "context_before": [],
181 "removes": ["x = 1\n"],
182 "adds": ["x = 10\n"],
183 "context_after": [],
184 }
185 result, ok = _apply_hunk(lines, hunk)
186 assert ok
187 assert "x = 10\n" in result
188 assert "x = 1\n" not in result
189
190
191 def test_apply_hunk_preserves_surrounding_lines(tmp_path: pathlib.Path) -> None:
192 from muse.cli.commands.apply import _apply_hunk
193 lines = ["a\n", "b\n", "c\n"]
194 hunk = {
195 "old_start": 2,
196 "context_before": [],
197 "removes": ["b\n"],
198 "adds": ["B\n"],
199 "context_after": [],
200 }
201 result, ok = _apply_hunk(lines, hunk)
202 assert ok
203 assert "a\n" in result
204 assert "c\n" in result
205 assert "B\n" in result
206
207
208 # ---------------------------------------------------------------------------
209 # Integration — clean apply
210 # ---------------------------------------------------------------------------
211
212
213 def test_apply_modifies_file_content(tmp_path: pathlib.Path) -> None:
214 root = _init_repo(tmp_path)
215 (root / "a.py").write_text("x = 1\n", encoding="utf-8")
216 patch = _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
217 patch_file = tmp_path / "change.patch"
218 patch_file.write_text(patch)
219 result = _invoke(root, str(patch_file))
220 assert result.exit_code == 0
221 assert (root / "a.py").read_text() == "x = 2\n"
222
223
224 def test_apply_new_file_created(tmp_path: pathlib.Path) -> None:
225 root = _init_repo(tmp_path)
226 patch = _make_simple_patch("new.py", [], ["x = 1\n"])
227 patch_file = tmp_path / "new.patch"
228 patch_file.write_text(patch)
229 result = _invoke(root, str(patch_file))
230 assert result.exit_code == 0
231 assert (root / "new.py").exists()
232 assert "x = 1" in (root / "new.py").read_text()
233
234
235 def test_apply_json_output(tmp_path: pathlib.Path) -> None:
236 root = _init_repo(tmp_path)
237 (root / "a.py").write_text("x = 1\n", encoding="utf-8")
238 patch = _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
239 patch_file = tmp_path / "change.patch"
240 patch_file.write_text(patch)
241 result = _invoke(root, str(patch_file), "--json")
242 assert result.exit_code == 0
243 data = json.loads(result.stdout)
244 assert "applied" in data
245 assert "failed" in data
246 assert "a.py" in data["applied"]
247
248
249 def test_apply_multiple_files(tmp_path: pathlib.Path) -> None:
250 root = _init_repo(tmp_path)
251 (root / "a.py").write_text("x = 1\n", encoding="utf-8")
252 (root / "b.py").write_text("y = 1\n", encoding="utf-8")
253 patch = (
254 _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
255 + "\n"
256 + _make_simple_patch("b.py", ["y = 1\n"], ["y = 2\n"])
257 )
258 patch_file = tmp_path / "multi.patch"
259 patch_file.write_text(patch)
260 result = _invoke(root, str(patch_file), "--json")
261 assert result.exit_code == 0
262 data = json.loads(result.stdout)
263 assert "a.py" in data["applied"]
264 assert "b.py" in data["applied"]
265
266
267 # ---------------------------------------------------------------------------
268 # Integration — --check mode
269 # ---------------------------------------------------------------------------
270
271
272 def test_apply_check_does_not_modify_file(tmp_path: pathlib.Path) -> None:
273 root = _init_repo(tmp_path)
274 (root / "a.py").write_text("x = 1\n", encoding="utf-8")
275 patch = _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
276 patch_file = tmp_path / "change.patch"
277 patch_file.write_text(patch)
278 result = _invoke(root, str(patch_file), "--check")
279 assert result.exit_code == 0
280 # File must be unchanged
281 assert (root / "a.py").read_text() == "x = 1\n"
282
283
284 def test_apply_check_exits_nonzero_on_conflict(tmp_path: pathlib.Path) -> None:
285 root = _init_repo(tmp_path)
286 (root / "a.py").write_text("completely different content\n", encoding="utf-8")
287 # Patch expects "x = 1" but file has different content
288 patch = _make_simple_patch("a.py", ["x = 1\n"], ["x = 2\n"])
289 patch_file = tmp_path / "conflict.patch"
290 patch_file.write_text(patch)
291 result = _invoke(root, str(patch_file), "--check")
292 assert result.exit_code != 0
293
294
295 # ---------------------------------------------------------------------------
296 # Integration — format-patch → apply round-trip
297 # ---------------------------------------------------------------------------
298
299
300 def test_format_patch_apply_patch_roundtrip(tmp_path: pathlib.Path) -> None:
301 """format-patch → apply-patch roundtrip restores the working tree."""
302 from tests.cli_test_helper import CliRunner as CR
303 from muse.cli.app import main as cli
304 cr = CR()
305
306 root = _init_repo(tmp_path)
307 _commit_files(root, {"a.py": b"x = 1\n"}, message="initial")
308 _commit_files(root, {"a.py": b"x = 2\n"}, message="change x")
309
310 out_dir = tmp_path / "patches"
311 out_dir.mkdir()
312 cr.invoke(cli, ["format-patch", "HEAD", "--output-dir", str(out_dir)], env=_env(root))
313
314 patch_file = next(out_dir.glob("*.mpatch"))
315 # Reset working tree to old content, then apply the mpatch
316 (root / "a.py").write_text("x = 1\n", encoding="utf-8")
317 result = cr.invoke(cli, ["apply-patch", str(patch_file), "--force"], env=_env(root))
318 assert result.exit_code == 0
319 assert (root / "a.py").read_text() == "x = 2\n"
320
321
322 # ---------------------------------------------------------------------------
323 # Security — path traversal in patch headers
324 # ---------------------------------------------------------------------------
325
326
327 def test_apply_rejects_path_traversal_in_patch(tmp_path: pathlib.Path) -> None:
328 root = _init_repo(tmp_path)
329 # Craft a patch with a traversal path
330 traversal_patch = textwrap.dedent("""\
331 --- a/../../../tmp/malicious.py
332 +++ b/../../../tmp/malicious.py
333 @@ -0,0 +1 @@
334 +malicious content
335 """)
336 patch_file = tmp_path / "malicious.patch"
337 patch_file.write_text(traversal_patch)
338 result = _invoke(root, str(patch_file))
339 assert result.exit_code != 0
340
341
342 def test_apply_rejects_muse_internal_paths(tmp_path: pathlib.Path) -> None:
343 root = _init_repo(tmp_path)
344 muse_patch = textwrap.dedent("""\
345 --- a/.muse/config.toml
346 +++ b/.muse/config.toml
347 @@ -0,0 +1 @@
348 +malicious = true
349 """)
350 patch_file = tmp_path / "muse.patch"
351 patch_file.write_text(muse_patch)
352 result = _invoke(root, str(patch_file))
353 assert result.exit_code != 0
354
355
356 # ---------------------------------------------------------------------------
357 # Stress — large hunk
358 # ---------------------------------------------------------------------------
359
360
361 def test_apply_large_hunk(tmp_path: pathlib.Path) -> None:
362 """A 50-line file with a change in the middle applies correctly."""
363 root = _init_repo(tmp_path)
364 original = [f"line {i}\n" for i in range(50)]
365 modified = original[:25] + ["CHANGED\n"] + original[26:]
366 (root / "big.py").write_text("".join(original), encoding="utf-8")
367 patch = _make_simple_patch("big.py", original, modified)
368 patch_file = tmp_path / "big.patch"
369 patch_file.write_text(patch)
370 result = _invoke(root, str(patch_file))
371 assert result.exit_code == 0
372 result_lines = (root / "big.py").read_text().splitlines(keepends=True)
373 assert result_lines[25] == "CHANGED\n"
374 assert result_lines[0] == "line 0\n"
375 assert result_lines[49] == "line 49\n"
376
377
378 import argparse as _argparse
379
380
381 class TestRegisterFlags:
382 def _parse(self, *args: str) -> _argparse.Namespace:
383 from muse.cli.commands.apply import register
384 p = _argparse.ArgumentParser()
385 sub = p.add_subparsers()
386 register(sub)
387 return p.parse_args(["apply", *args])
388
389 def test_default_json_out_is_false(self) -> None:
390 ns = self._parse("dummy.patch")
391 assert ns.json_out is False
392
393 def test_json_flag_sets_json_out(self) -> None:
394 ns = self._parse("--json", "dummy.patch")
395 assert ns.json_out is True
396
397 def test_j_shorthand_sets_json_out(self) -> None:
398 ns = self._parse("-j", "dummy.patch")
399 assert ns.json_out is True
400
401 def test_check_default(self) -> None:
402 ns = self._parse("dummy.patch")
403 assert ns.check is False
404
405 def test_staged_default(self) -> None:
406 ns = self._parse("dummy.patch")
407 assert ns.staged is False
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 126 days ago