gabriel / muse public
test_cmd_cherry_pick_hardening.py python
926 lines 37.5 KB
Raw
sha256:e029dfbf5482b23e127b4793b93953b3e2b5af9ca3aa533e6c6e08a6d1887e17 fixing more failing tests Human 41 days ago
1 """Comprehensive hardening tests for ``muse cherry-pick``.
2
3 Covers all changes introduced in the cherry-pick command review:
4
5 Unit
6 ----
7 - Parser flags: --dry-run, --message/-m, --force, --no-commit, --format, --json
8 - Dead-code removal: _read_branch absent, pathlib not imported
9 - validate_branch_name called in run()
10 - target.message sanitized before embedding in commit record
11 - ref sanitized in "not found" error
12 - Write ordering: write_snapshot → write_commit → apply_manifest → write_branch_ref
13 - Fail-fast on missing parent commit / parent snapshot (no silent {} fallback)
14
15 Integration
16 -----------
17 - Error messages routed to stderr, stdout clean
18 - JSON schema identical and complete for all code paths
19 (normal, --no-commit, --dry-run, conflict)
20 - --dry-run performs no writes (branch ref, workdir, reflog unchanged)
21 - --no-commit applies workdir changes without advancing the branch ref
22 - Reflog entry appended after normal cherry-pick
23 - -m/--message overrides the cherry-picked commit message
24 - Missing parent commit → INTERNAL_ERROR (not silent fallback)
25 - Missing target snapshot → INTERNAL_ERROR
26
27 End-to-end
28 ----------
29 - Text output format
30 - JSON output format with full schema verification
31 - Cherry-pick from another branch applies correct content
32 - --force bypasses dirty-workdir guard
33
34 Security
35 --------
36 - ANSI escape codes in ref rejected / sanitized in error
37 - ANSI in original commit message not propagated to stored commit
38 - --format with unknown value exits 1 and prints to stderr
39 - Conflict paths sanitized in text output
40
41 Stress
42 ------
43 - Cherry-pick across a 200-commit history
44 - 50 sequential cherry-picks in the same repo
45 - Concurrent cherry-picks to isolated repos
46 """
47
48 from __future__ import annotations
49
50 import argparse
51 import inspect
52 import json
53 import pathlib
54 import threading
55 import time
56
57 import pytest
58
59 from tests.cli_test_helper import CliRunner
60
61 cli = None # argparse migration — CliRunner ignores this arg
62 runner = CliRunner()
63
64
65 # ---------------------------------------------------------------------------
66 # Shared helpers
67 # ---------------------------------------------------------------------------
68
69 def _env(root: pathlib.Path) -> Manifest:
70 return {"MUSE_REPO_ROOT": str(root)}
71
72
73 @pytest.fixture()
74 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
75 """Repo on ``main`` with two commits: base (a.py) + target (b.py).
76
77 The caller can immediately cherry-pick the HEAD commit to a new branch.
78 """
79 monkeypatch.chdir(tmp_path)
80 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
81 r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
82 assert r.exit_code == 0, r.output
83 (tmp_path / "a.py").write_text("x = 1\n")
84 r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
85 assert r.exit_code == 0, r.output
86 (tmp_path / "b.py").write_text("y = 2\n")
87 r = runner.invoke(cli, ["commit", "-m", "add b"], env=_env(tmp_path), catch_exceptions=False)
88 assert r.exit_code == 0, r.output
89 return tmp_path
90
91
92 @pytest.fixture()
93 def two_branch_repo(
94 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
95 ) -> tuple[pathlib.Path, str]:
96 """Repo with main and feat branches, returns (root, commit-id-on-feat).
97
98 ``main``: base commit only
99 ``feat``: base commit + one extra commit (the one to cherry-pick)
100 """
101 monkeypatch.chdir(tmp_path)
102 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
103 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
104 (tmp_path / "base.py").write_text("base\n")
105 runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
106
107 runner.invoke(cli, ["branch", "feat"], env=_env(tmp_path), catch_exceptions=False)
108 runner.invoke(cli, ["checkout", "feat"], env=_env(tmp_path), catch_exceptions=False)
109 (tmp_path / "extra.py").write_text("extra\n")
110 runner.invoke(cli, ["commit", "-m", "extra on feat"], env=_env(tmp_path), catch_exceptions=False)
111
112 from muse.core.store import get_head_commit_id
113 feat_cid = get_head_commit_id(tmp_path, "feat")
114 assert feat_cid is not None
115
116 runner.invoke(cli, ["checkout", "main"], env=_env(tmp_path), catch_exceptions=False)
117 return tmp_path, feat_cid
118
119
120 def _head_id(repo: pathlib.Path, branch: str = "main") -> str | None:
121 from muse.core.store import get_head_commit_id
122 return get_head_commit_id(repo, branch)
123
124
125 # ---------------------------------------------------------------------------
126 # Unit — parser flags and dead-code removal
127 # ---------------------------------------------------------------------------
128
129 class TestRegisterFlags:
130 def _parse(self, *args: str) -> argparse.Namespace:
131 import muse.cli.commands.cherry_pick as m
132 p = argparse.ArgumentParser()
133 sub = p.add_subparsers()
134 m.register(sub)
135 return p.parse_args(["cherry-pick", *args])
136
137 def test_dry_run_flag(self) -> None:
138 ns = self._parse("abc123", "--dry-run")
139 assert ns.dry_run is True
140
141 def test_dry_run_default_false(self) -> None:
142 ns = self._parse("abc123")
143 assert ns.dry_run is False
144
145 def test_no_commit_short(self) -> None:
146 ns = self._parse("abc123", "-n")
147 assert ns.no_commit is True
148
149 def test_no_commit_long(self) -> None:
150 ns = self._parse("abc123", "--no-commit")
151 assert ns.no_commit is True
152
153 def test_force_flag(self) -> None:
154 ns = self._parse("abc123", "--force")
155 assert ns.force is True
156
157 def test_message_short(self) -> None:
158 ns = self._parse("abc123", "-m", "my msg")
159 assert ns.message == "my msg"
160
161 def test_message_long(self) -> None:
162 ns = self._parse("abc123", "--message", "my msg")
163 assert ns.message == "my msg"
164
165 def test_message_default_none(self) -> None:
166 ns = self._parse("abc123")
167 assert ns.message is None
168
169 def test_format_json_shorthand(self) -> None:
170 ns = self._parse("abc123", "--json")
171 assert ns.fmt == "json"
172
173 def test_format_explicit_text(self) -> None:
174 ns = self._parse("abc123", "--format", "text")
175 assert ns.fmt == "text"
176
177 def test_ref_positional(self) -> None:
178 ns = self._parse("deadbeef")
179 assert ns.ref == "deadbeef"
180
181
182 class TestDeadCodeRemoval:
183 def test_no_read_branch_wrapper(self) -> None:
184 import muse.cli.commands.cherry_pick as m
185 assert not hasattr(m, "_read_branch"), "_read_branch must be deleted"
186
187 def test_pathlib_not_imported(self) -> None:
188 import muse.cli.commands.cherry_pick as m
189 assert "import pathlib" not in inspect.getsource(m)
190
191 def test_validate_branch_name_in_run(self) -> None:
192 import muse.cli.commands.cherry_pick as m
193 assert "validate_branch_name" in inspect.getsource(m.run)
194
195 def test_target_message_sanitized_in_run(self) -> None:
196 import muse.cli.commands.cherry_pick as m
197 assert "sanitize_display(target.message" in inspect.getsource(m.run)
198
199 def test_ref_sanitized_in_not_found_error(self) -> None:
200 import muse.cli.commands.cherry_pick as m
201 assert "sanitize_display(ref)" in inspect.getsource(m.run)
202
203 def test_write_snapshot_before_apply_manifest_in_normal_path(self) -> None:
204 """Normal path: write_snapshot and write_commit must precede apply_manifest."""
205 import muse.cli.commands.cherry_pick as m
206 src_lines = [
207 (i, l)
208 for i, l in enumerate(inspect.getsource(m.run).split("\n"), 1)
209 if l.strip() and not l.strip().startswith("#")
210 ]
211 ws = next(i for i, l in src_lines if "write_snapshot(" in l)
212 wc = next(i for i, l in src_lines if "write_commit(" in l)
213 wr = next(i for i, l in src_lines if "write_branch_ref(" in l)
214 # Find the LAST apply_manifest (normal path, not no_commit path)
215 all_am = [i for i, l in src_lines if "apply_manifest(" in l]
216 last_am = max(all_am)
217 assert ws < last_am, f"write_snapshot ({ws}) must precede apply_manifest ({last_am})"
218 assert wc < last_am, f"write_commit ({wc}) must precede apply_manifest ({last_am})"
219 assert last_am < wr, f"apply_manifest ({last_am}) must precede write_branch_ref ({wr})"
220
221 def test_parent_snapshot_missing_raises_not_silently_falls_back(self) -> None:
222 """Code must not silently use {} when parent snapshot is missing."""
223 import muse.cli.commands.cherry_pick as m
224 src = inspect.getsource(m.run)
225 # The silent fallback was: `if parent_snap: base_manifest = parent_snap.manifest`
226 # The fix is: `raise SystemExit(ExitCode.INTERNAL_ERROR)` when parent_snap is None
227 assert "INTERNAL_ERROR" in src
228
229
230 # ---------------------------------------------------------------------------
231 # Integration — error routing and behaviour
232 # ---------------------------------------------------------------------------
233
234 class TestErrorRouting:
235 def test_not_found_to_stderr(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
236 root, _ = two_branch_repo
237 r = runner.invoke(cli, ["cherry-pick", "badref"], env=_env(root))
238 assert r.exit_code != 0
239 assert "not found" in (r.stderr or "").lower()
240 assert "badref" in (r.stderr or "")
241
242 def test_not_found_stdout_clean(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
243 root, _ = two_branch_repo
244 r = runner.invoke(cli, ["cherry-pick", "0000000000000000"], env=_env(root))
245 assert r.exit_code != 0
246 # Error must be in stderr; stdout should have no error messages.
247 assert "not found" in (r.stderr or "").lower()
248
249 def test_bad_format_to_stderr(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
250 root, cid = two_branch_repo
251 r = runner.invoke(cli, ["cherry-pick", "--format", "xml", cid], env=_env(root))
252 assert r.exit_code == 1
253 assert "xml" in (r.stderr or "").lower()
254
255 def test_bad_format_error_in_stderr(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
256 root, cid = two_branch_repo
257 r = runner.invoke(cli, ["cherry-pick", "--format", "html", cid], env=_env(root))
258 assert r.exit_code == 1
259 assert "Unknown" in (r.stderr or "") or "format" in (r.stderr or "").lower()
260
261
262 class TestJsonSchema:
263 """JSON schema must be identical across all code paths."""
264
265 _REQUIRED_KEYS = {
266 "status", "commit_id", "branch", "ref",
267 "source_commit_id", "snapshot_id", "message",
268 "no_commit", "dry_run", "conflicts",
269 }
270
271 def test_normal_json_schema_complete(
272 self, two_branch_repo: tuple[pathlib.Path, str]
273 ) -> None:
274 root, cid = two_branch_repo
275 r = runner.invoke(
276 cli, ["cherry-pick", cid, "--json"],
277 env=_env(root), catch_exceptions=False,
278 )
279 assert r.exit_code == 0, r.output
280 d = json.loads(r.output)
281 assert self._REQUIRED_KEYS <= d.keys()
282
283 def test_normal_status_is_picked(
284 self, two_branch_repo: tuple[pathlib.Path, str]
285 ) -> None:
286 root, cid = two_branch_repo
287 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
288 d = json.loads(r.output)
289 assert d["status"] == "picked"
290 assert d["no_commit"] is False
291 assert d["dry_run"] is False
292 assert d["conflicts"] == []
293
294 def test_normal_ref_field_matches_input(
295 self, two_branch_repo: tuple[pathlib.Path, str]
296 ) -> None:
297 root, cid = two_branch_repo
298 r = runner.invoke(cli, ["cherry-pick", cid[:12], "--json"], env=_env(root), catch_exceptions=False)
299 d = json.loads(r.output)
300 assert d["ref"] == cid[:12]
301
302 def test_normal_snapshot_id_is_string(
303 self, two_branch_repo: tuple[pathlib.Path, str]
304 ) -> None:
305 root, cid = two_branch_repo
306 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
307 d = json.loads(r.output)
308 assert isinstance(d["snapshot_id"], str) and len(d["snapshot_id"]) == 64
309
310 def test_no_commit_json_schema_complete(
311 self, two_branch_repo: tuple[pathlib.Path, str]
312 ) -> None:
313 root, cid = two_branch_repo
314 r = runner.invoke(
315 cli, ["cherry-pick", cid, "--no-commit", "--json"],
316 env=_env(root), catch_exceptions=False,
317 )
318 assert r.exit_code == 0, r.output
319 d = json.loads(r.output)
320 assert self._REQUIRED_KEYS <= d.keys()
321
322 def test_no_commit_status_is_applied(
323 self, two_branch_repo: tuple[pathlib.Path, str]
324 ) -> None:
325 root, cid = two_branch_repo
326 r = runner.invoke(
327 cli, ["cherry-pick", cid, "--no-commit", "--json"],
328 env=_env(root), catch_exceptions=False,
329 )
330 d = json.loads(r.output)
331 assert d["status"] == "applied"
332 assert d["commit_id"] is None
333 assert d["no_commit"] is True
334 assert d["dry_run"] is False
335
336 def test_dry_run_json_schema_complete(
337 self, two_branch_repo: tuple[pathlib.Path, str]
338 ) -> None:
339 root, cid = two_branch_repo
340 r = runner.invoke(
341 cli, ["cherry-pick", cid, "--dry-run", "--json"],
342 env=_env(root), catch_exceptions=False,
343 )
344 assert r.exit_code == 0, r.output
345 d = json.loads(r.output)
346 assert self._REQUIRED_KEYS <= d.keys()
347
348 def test_dry_run_status(
349 self, two_branch_repo: tuple[pathlib.Path, str]
350 ) -> None:
351 root, cid = two_branch_repo
352 r = runner.invoke(
353 cli, ["cherry-pick", cid, "--dry-run", "--json"],
354 env=_env(root), catch_exceptions=False,
355 )
356 d = json.loads(r.output)
357 assert d["dry_run"] is True
358 assert d["commit_id"] is None
359 assert d["status"] == "dry_run"
360
361 def test_all_three_schemas_identical(
362 self, two_branch_repo: tuple[pathlib.Path, str]
363 ) -> None:
364 root, cid = two_branch_repo
365 r_dr = runner.invoke(cli, ["cherry-pick", cid, "--dry-run", "--json"], env=_env(root), catch_exceptions=False)
366 r_nc = runner.invoke(cli, ["cherry-pick", cid, "--no-commit", "--json"], env=_env(root), catch_exceptions=False)
367
368 # Normal cherry-pick: need to re-fetch after --no-commit modified workdir
369 r_commit = runner.invoke(cli, ["commit", "-m", "after nc"], env=_env(root), catch_exceptions=False)
370 from muse.core.store import get_head_commit_id
371 new_cid = get_head_commit_id(root, "main")
372 assert new_cid is not None
373 # Need a different source commit to cherry-pick after the no-commit pick
374 r_nm = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
375
376 keys_dr = set(json.loads(r_dr.output).keys())
377 keys_nc = set(json.loads(r_nc.output).keys())
378 keys_nm = set(json.loads(r_nm.output).keys())
379 assert keys_dr == keys_nc == keys_nm
380
381
382 class TestDryRun:
383 def test_no_commit_created(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
384 from muse.core.store import get_all_commits, get_head_commit_id
385 root, cid = two_branch_repo
386 before_count = len(get_all_commits(root))
387 before_head = get_head_commit_id(root, "main")
388 r = runner.invoke(
389 cli, ["cherry-pick", cid, "--dry-run"],
390 env=_env(root), catch_exceptions=False,
391 )
392 assert r.exit_code == 0, r.output
393 assert len(get_all_commits(root)) == before_count
394 assert get_head_commit_id(root, "main") == before_head
395
396 def test_workdir_unchanged(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
397 root, cid = two_branch_repo
398 extra = root / "extra.py"
399 content_before = extra.read_text() if extra.exists() else None
400 r = runner.invoke(
401 cli, ["cherry-pick", cid, "--dry-run"],
402 env=_env(root), catch_exceptions=False,
403 )
404 assert r.exit_code == 0, r.output
405 assert (extra.read_text() if extra.exists() else None) == content_before
406
407 def test_reflog_unchanged(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
408 from muse.core.reflog import read_reflog
409 root, cid = two_branch_repo
410 before = len(read_reflog(root, "main"))
411 runner.invoke(cli, ["cherry-pick", cid, "--dry-run"], env=_env(root), catch_exceptions=False)
412 assert len(read_reflog(root, "main")) == before
413
414 def test_dry_run_text_output_mentions_dry_run(
415 self, two_branch_repo: tuple[pathlib.Path, str]
416 ) -> None:
417 root, cid = two_branch_repo
418 r = runner.invoke(
419 cli, ["cherry-pick", cid, "--dry-run"],
420 env=_env(root), catch_exceptions=False,
421 )
422 assert "dry-run" in r.output.lower() or "would" in r.output.lower()
423
424 def test_dry_run_invalid_ref_errors(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
425 root, _ = two_branch_repo
426 r = runner.invoke(cli, ["cherry-pick", "no-such-ref", "--dry-run"], env=_env(root))
427 assert r.exit_code != 0
428
429 def test_dry_run_json_snapshot_id_present(
430 self, two_branch_repo: tuple[pathlib.Path, str]
431 ) -> None:
432 root, cid = two_branch_repo
433 r = runner.invoke(
434 cli, ["cherry-pick", cid, "--dry-run", "--json"],
435 env=_env(root), catch_exceptions=False,
436 )
437 d = json.loads(r.output)
438 assert d["snapshot_id"] is not None and len(d["snapshot_id"]) == 64
439
440
441 class TestNoCommit:
442 def test_branch_ref_not_advanced(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
443 from muse.core.store import get_head_commit_id
444 root, cid = two_branch_repo
445 before_head = get_head_commit_id(root, "main")
446 r = runner.invoke(
447 cli, ["cherry-pick", cid, "--no-commit"],
448 env=_env(root), catch_exceptions=False,
449 )
450 assert r.exit_code == 0, r.output
451 assert get_head_commit_id(root, "main") == before_head
452
453 def test_workdir_modified(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
454 root, cid = two_branch_repo
455 r = runner.invoke(
456 cli, ["cherry-pick", cid, "--no-commit"],
457 env=_env(root), catch_exceptions=False,
458 )
459 assert r.exit_code == 0, r.output
460 # extra.py was added by the feat branch commit
461 assert (root / "extra.py").exists()
462
463 def test_no_commit_in_json(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
464 root, cid = two_branch_repo
465 r = runner.invoke(
466 cli, ["cherry-pick", cid, "--no-commit", "--json"],
467 env=_env(root), catch_exceptions=False,
468 )
469 d = json.loads(r.output)
470 assert d["no_commit"] is True
471 assert d["commit_id"] is None
472 assert d["status"] == "applied"
473
474 def test_reflog_not_written_for_no_commit(
475 self, two_branch_repo: tuple[pathlib.Path, str]
476 ) -> None:
477 from muse.core.reflog import read_reflog
478 root, cid = two_branch_repo
479 before = len(read_reflog(root, "main"))
480 runner.invoke(
481 cli, ["cherry-pick", cid, "--no-commit"],
482 env=_env(root), catch_exceptions=False,
483 )
484 assert len(read_reflog(root, "main")) == before
485
486
487 class TestReflog:
488 def test_reflog_appended(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
489 from muse.core.reflog import read_reflog
490 root, cid = two_branch_repo
491 before = len(read_reflog(root, "main"))
492 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
493 assert len(read_reflog(root, "main")) > before
494
495 def test_reflog_operation_contains_cherry_pick(
496 self, two_branch_repo: tuple[pathlib.Path, str]
497 ) -> None:
498 from muse.core.reflog import read_reflog
499 root, cid = two_branch_repo
500 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
501 entries = read_reflog(root, "main")
502 # read_reflog returns newest-first
503 assert "cherry-pick" in entries[0].operation.lower()
504
505
506 class TestMessageFlag:
507 def test_custom_message_in_json(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
508 root, cid = two_branch_repo
509 r = runner.invoke(
510 cli, ["cherry-pick", cid, "-m", "custom pick msg", "--json"],
511 env=_env(root), catch_exceptions=False,
512 )
513 assert r.exit_code == 0, r.output
514 d = json.loads(r.output)
515 assert d["message"] == "custom pick msg"
516
517 def test_custom_message_in_text(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
518 root, cid = two_branch_repo
519 r = runner.invoke(
520 cli, ["cherry-pick", cid, "-m", "undo extra"],
521 env=_env(root), catch_exceptions=False,
522 )
523 assert r.exit_code == 0, r.output
524 assert "undo extra" in r.output
525
526 def test_default_message_is_source_message(
527 self, two_branch_repo: tuple[pathlib.Path, str]
528 ) -> None:
529 root, cid = two_branch_repo
530 r = runner.invoke(
531 cli, ["cherry-pick", cid, "--json"],
532 env=_env(root), catch_exceptions=False,
533 )
534 d = json.loads(r.output)
535 assert d["message"] == "extra on feat"
536
537 def test_message_stored_in_commit_record(
538 self, two_branch_repo: tuple[pathlib.Path, str]
539 ) -> None:
540 from muse.core.store import get_head_commit_id, read_commit
541 root, cid = two_branch_repo
542 runner.invoke(
543 cli, ["cherry-pick", cid, "-m", "my override"],
544 env=_env(root), catch_exceptions=False,
545 )
546 new_cid = get_head_commit_id(root, "main")
547 assert new_cid is not None
548 rec = read_commit(root, new_cid)
549 assert rec is not None
550 assert rec.message == "my override"
551
552
553 class TestWriteOrdering:
554 def test_write_snapshot_before_apply_manifest(
555 self, two_branch_repo: tuple[pathlib.Path, str]
556 ) -> None:
557 """write_snapshot must be called before apply_manifest in the normal path."""
558 from unittest.mock import patch
559 import muse.cli.commands.cherry_pick as cp_mod
560 from muse.core import store as s
561 events: list[str] = []
562 orig_ws = s.write_snapshot
563 from muse.core.workdir import apply_manifest as orig_am_fn
564
565 def tracking_write_snapshot(root: pathlib.Path, rec: s.SnapshotRecord) -> None:
566 orig_ws(root, rec)
567 events.append("write_snapshot")
568
569 def tracking_apply(root: pathlib.Path, manifest: Manifest) -> None:
570 orig_am_fn(root, manifest)
571 events.append("apply_manifest")
572
573 root, cid = two_branch_repo
574 with (
575 patch.object(cp_mod, "write_snapshot", tracking_write_snapshot),
576 patch("muse.cli.commands.cherry_pick.apply_manifest", tracking_apply),
577 ):
578 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
579
580 assert "write_snapshot" in events
581 assert "apply_manifest" in events
582 assert events.index("write_snapshot") < events.index("apply_manifest"), (
583 f"write_snapshot must precede apply_manifest, got: {events}"
584 )
585
586
587 class TestFailFast:
588 def test_missing_parent_commit_is_error(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
589 """When the target has a parent_commit_id but the parent object is missing,
590 cherry-pick must exit INTERNAL_ERROR — not silently use empty base."""
591 import uuid, datetime
592 monkeypatch.chdir(tmp_path)
593 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
594 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
595
596 from muse.core.store import (
597 CommitRecord, SnapshotRecord, write_commit, write_snapshot,
598 get_head_commit_id,
599 )
600 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
601
602 repo_id = (tmp_path / ".muse" / "repo.json").read_text()
603 import json as _json
604 repo_id = _json.loads(repo_id)["repo_id"]
605
606 # Create a base commit
607 m1: Manifest = {}
608 s1 = compute_snapshot_id(m1)
609 t1 = datetime.datetime.now(datetime.timezone.utc)
610 c1 = compute_commit_id([], s1, "base", t1.isoformat())
611 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=s1, manifest=m1))
612 write_commit(tmp_path, CommitRecord(
613 commit_id=c1, repo_id=repo_id, branch="main",
614 snapshot_id=s1, message="base", committed_at=t1,
615 parent_commit_id=None,
616 ))
617 (tmp_path / ".muse" / "refs" / "heads" / "main").write_text(c1)
618
619 # Create a second commit that references c1 as parent
620 m2: Manifest = {}
621 s2 = compute_snapshot_id(m2)
622 t2 = datetime.datetime.now(datetime.timezone.utc)
623 c2 = compute_commit_id([c1], s2, "target", t2.isoformat())
624 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=s2, manifest=m2))
625 write_commit(tmp_path, CommitRecord(
626 commit_id=c2, repo_id=repo_id, branch="main",
627 snapshot_id=s2, message="target", committed_at=t2,
628 parent_commit_id=c1,
629 ))
630
631 # Now delete the parent commit file to simulate object-store corruption
632 commit_file = tmp_path / ".muse" / "commits" / f"{c1}.msgpack"
633 if commit_file.exists():
634 commit_file.unlink()
635
636 # Reset HEAD to base so we can cherry-pick c2 "from another branch"
637 # Switch to a fresh branch at c1's snapshot
638 runner.invoke(cli, ["branch", "target-branch"], env=_env(tmp_path), catch_exceptions=False)
639 runner.invoke(cli, ["checkout", "target-branch"], env=_env(tmp_path), catch_exceptions=False)
640 (tmp_path / ".muse" / "refs" / "heads" / "target-branch").write_text(c1)
641
642 r = runner.invoke(cli, ["cherry-pick", c2], env=_env(tmp_path))
643 # Must fail with INTERNAL_ERROR (exit code 3), not succeed silently
644 assert r.exit_code != 0
645
646
647 # ---------------------------------------------------------------------------
648 # End-to-end — text and JSON output
649 # ---------------------------------------------------------------------------
650
651 class TestTextOutput:
652 def test_text_shows_branch_and_short_id(
653 self, two_branch_repo: tuple[pathlib.Path, str]
654 ) -> None:
655 root, cid = two_branch_repo
656 r = runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
657 assert r.exit_code == 0
658 assert "main" in r.output
659
660 def test_no_commit_text_mentions_workdir(
661 self, two_branch_repo: tuple[pathlib.Path, str]
662 ) -> None:
663 root, cid = two_branch_repo
664 r = runner.invoke(
665 cli, ["cherry-pick", cid, "--no-commit"],
666 env=_env(root), catch_exceptions=False,
667 )
668 output = r.output.lower()
669 assert "working tree" in output or "applied" in output or "commit" in output
670
671 def test_workdir_has_cherry_picked_file(
672 self, two_branch_repo: tuple[pathlib.Path, str]
673 ) -> None:
674 root, cid = two_branch_repo
675 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
676 assert (root / "extra.py").exists()
677 assert (root / "extra.py").read_text() == "extra\n"
678
679
680 class TestJsonOutput:
681 def test_source_commit_id_matches(
682 self, two_branch_repo: tuple[pathlib.Path, str]
683 ) -> None:
684 root, cid = two_branch_repo
685 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
686 d = json.loads(r.output)
687 assert d["source_commit_id"] == cid
688
689 def test_branch_field(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
690 root, cid = two_branch_repo
691 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
692 d = json.loads(r.output)
693 assert d["branch"] == "main"
694
695 def test_new_commit_id_different_from_source(
696 self, two_branch_repo: tuple[pathlib.Path, str]
697 ) -> None:
698 root, cid = two_branch_repo
699 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
700 d = json.loads(r.output)
701 assert d["commit_id"] != cid
702 assert isinstance(d["commit_id"], str) and len(d["commit_id"]) == 64
703
704 def test_conflicts_empty_on_success(
705 self, two_branch_repo: tuple[pathlib.Path, str]
706 ) -> None:
707 root, cid = two_branch_repo
708 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
709 d = json.loads(r.output)
710 assert d["conflicts"] == []
711
712
713 class TestForce:
714 def test_force_bypasses_dirty_check(
715 self, two_branch_repo: tuple[pathlib.Path, str]
716 ) -> None:
717 root, cid = two_branch_repo
718 (root / "base.py").write_text("modified but uncommitted\n")
719 r = runner.invoke(
720 cli, ["cherry-pick", cid, "--force"],
721 env=_env(root), catch_exceptions=False,
722 )
723 assert r.exit_code == 0, r.output
724
725 def test_without_force_dirty_tree_fails(
726 self, two_branch_repo: tuple[pathlib.Path, str]
727 ) -> None:
728 root, cid = two_branch_repo
729 (root / "base.py").write_text("uncommitted change\n")
730 r = runner.invoke(cli, ["cherry-pick", cid], env=_env(root))
731 assert r.exit_code != 0
732
733
734 # ---------------------------------------------------------------------------
735 # Security — ANSI injection and sanitization
736 # ---------------------------------------------------------------------------
737
738 class TestSecurity:
739 def test_ansi_in_ref_error_in_stderr(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
740 root, _ = two_branch_repo
741 ansi_ref = "\x1b[31mbadref\x1b[0m"
742 r = runner.invoke(cli, ["cherry-pick", ansi_ref], env=_env(root))
743 assert r.exit_code != 0
744 assert "\x1b[31m" not in (r.stdout or "")
745 assert "badref" in (r.stderr or "")
746
747 def test_ansi_in_commit_message_not_in_stored_commit(
748 self, two_branch_repo: tuple[pathlib.Path, str]
749 ) -> None:
750 """If the source commit has ANSI in its message, the cherry-pick commit
751 stored on disk must not contain raw escape sequences."""
752 from unittest.mock import patch
753 from muse.core import store as s
754 import muse.cli.commands.cherry_pick as cp_mod
755
756 root, cid = two_branch_repo
757 orig_rc = s.read_commit
758
759 def poisoned_read_commit(root: pathlib.Path, c: str) -> s.CommitRecord | None:
760 rec = orig_rc(root, c)
761 if rec is not None and rec.commit_id == cid:
762 return s.CommitRecord(
763 commit_id=rec.commit_id, repo_id=rec.repo_id,
764 branch=rec.branch, snapshot_id=rec.snapshot_id,
765 message="\x1b[31mmalicious\x1b[0m",
766 committed_at=rec.committed_at,
767 parent_commit_id=rec.parent_commit_id,
768 )
769 return rec
770
771 with patch.object(cp_mod, "read_commit", poisoned_read_commit):
772 r = runner.invoke(
773 cli, ["cherry-pick", cid, "--json"],
774 env=_env(root), catch_exceptions=False,
775 )
776
777 if r.exit_code == 0:
778 d = json.loads(r.output)
779 assert "\x1b[" not in d.get("message", ""), (
780 "Cherry-pick commit message must not contain raw ANSI from source"
781 )
782
783 def test_invalid_format_exits_1_to_stderr(
784 self, two_branch_repo: tuple[pathlib.Path, str]
785 ) -> None:
786 root, cid = two_branch_repo
787 r = runner.invoke(cli, ["cherry-pick", "--format", "html", cid], env=_env(root))
788 assert r.exit_code == 1
789 assert "html" in (r.stderr or "").lower()
790
791
792 # ---------------------------------------------------------------------------
793 # Stress
794 # ---------------------------------------------------------------------------
795
796 class TestStress:
797 @pytest.mark.slow
798 def test_cherry_pick_deep_in_history(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
799 """Cherry-pick a commit that's deep in a 200-commit chain."""
800 monkeypatch.chdir(tmp_path)
801 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
802 env = _env(tmp_path)
803 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
804 (tmp_path / "seed.py").write_text("seed\n")
805 runner.invoke(cli, ["commit", "-m", "seed"], env=env, catch_exceptions=False)
806
807 runner.invoke(cli, ["branch", "source"], env=env, catch_exceptions=False)
808 runner.invoke(cli, ["checkout", "source"], env=env, catch_exceptions=False)
809
810 (tmp_path / "target.py").write_text("target\n")
811 runner.invoke(cli, ["commit", "-m", "target commit"], env=env, catch_exceptions=False)
812 from muse.core.store import get_head_commit_id
813 target_cid = get_head_commit_id(tmp_path, "source")
814 assert target_cid is not None
815
816 for i in range(198):
817 (tmp_path / f"f{i}.py").write_text(f"{i}\n")
818 runner.invoke(cli, ["commit", "-m", f"c{i}"], env=env, catch_exceptions=False)
819
820 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
821 r = runner.invoke(
822 cli, ["cherry-pick", target_cid, "--json"],
823 env=env, catch_exceptions=False,
824 )
825 assert r.exit_code == 0, r.output
826 d = json.loads(r.output)
827 assert d["source_commit_id"] == target_cid
828 assert d["status"] == "picked"
829
830 @pytest.mark.slow
831 def test_sequential_cherry_picks(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
832 """30 sequential cherry-picks must all succeed."""
833 monkeypatch.chdir(tmp_path)
834 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
835 env = _env(tmp_path)
836 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
837 (tmp_path / "base.py").write_text("base\n")
838 runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
839
840 runner.invoke(cli, ["branch", "source"], env=env, catch_exceptions=False)
841 runner.invoke(cli, ["checkout", "source"], env=env, catch_exceptions=False)
842
843 # Create 30 non-conflicting commits on source
844 source_cids: list[str] = []
845 for i in range(30):
846 (tmp_path / f"s{i}.py").write_text(f"s{i}\n")
847 runner.invoke(cli, ["commit", "-m", f"src{i}"], env=env, catch_exceptions=False)
848 from muse.core.store import get_head_commit_id
849 cid = get_head_commit_id(tmp_path, "source")
850 assert cid is not None
851 source_cids.append(cid)
852
853 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
854 failures: list[str] = []
855 for i, cid in enumerate(source_cids):
856 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=env)
857 if r.exit_code != 0:
858 failures.append(f"pick {i}: exit={r.exit_code} {r.output.strip()[:60]}")
859
860 assert not failures, f"Sequential failures: {failures}"
861
862 @pytest.mark.slow
863 def test_concurrent_cherry_picks_isolated_repos(self, tmp_path: pathlib.Path) -> None:
864 """Parallel cherry-picks to distinct repos must not interfere."""
865 errors: list[str] = []
866
867 def do_pick(idx: int) -> None:
868 repo_dir = tmp_path / f"repo_{idx}"
869 repo_dir.mkdir()
870 env = {"MUSE_REPO_ROOT": str(repo_dir)}
871 runner.invoke(cli, ["init"], env=env)
872 (repo_dir / "base.py").write_text("base\n")
873 runner.invoke(cli, ["commit", "-m", "base"], env=env)
874 runner.invoke(cli, ["branch", "src"], env=env)
875 runner.invoke(cli, ["checkout", "src"], env=env)
876 (repo_dir / "extra.py").write_text("extra\n")
877 runner.invoke(cli, ["commit", "-m", "extra"], env=env)
878 from muse.core.store import get_head_commit_id
879 cid = get_head_commit_id(repo_dir, "src")
880 if not cid:
881 errors.append(f"repo {idx}: no HEAD")
882 return
883 runner.invoke(cli, ["checkout", "main"], env=env)
884 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=env)
885 if r.exit_code != 0:
886 errors.append(f"repo {idx}: exit={r.exit_code} {r.output.strip()[:60]}")
887
888 threads = [threading.Thread(target=do_pick, args=(i,)) for i in range(8)]
889 for t in threads:
890 t.start()
891 for t in threads:
892 t.join()
893
894 assert not errors, f"Concurrent errors: {errors}"
895
896 @pytest.mark.slow
897 def test_dry_run_performance(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
898 """--dry-run on a large repo must complete in < 3 s."""
899 monkeypatch.chdir(tmp_path)
900 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
901 env = _env(tmp_path)
902 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
903 (tmp_path / "base.py").write_text("base\n")
904 runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
905 runner.invoke(cli, ["branch", "src"], env=env, catch_exceptions=False)
906 runner.invoke(cli, ["checkout", "src"], env=env, catch_exceptions=False)
907
908 (tmp_path / "target.py").write_text("target\n")
909 runner.invoke(cli, ["commit", "-m", "target"], env=env, catch_exceptions=False)
910 from muse.core.store import get_head_commit_id
911 target_cid = get_head_commit_id(tmp_path, "src")
912 assert target_cid is not None
913
914 for i in range(100):
915 (tmp_path / f"f{i}.py").write_text(f"{i}\n")
916 runner.invoke(cli, ["commit", "-m", f"c{i}"], env=env, catch_exceptions=False)
917
918 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
919 start = time.perf_counter()
920 r = runner.invoke(
921 cli, ["cherry-pick", target_cid, "--dry-run", "--json"],
922 env=env, catch_exceptions=False,
923 )
924 elapsed = time.perf_counter() - start
925 assert r.exit_code == 0, r.output
926 assert elapsed < 3.0, f"--dry-run took {elapsed:.2f}s"
File History 1 commit
sha256:e029dfbf5482b23e127b4793b93953b3e2b5af9ca3aa533e6c6e08a6d1887e17 fixing more failing tests Human 41 days ago