gabriel / muse public
test_cmd_cherry_pick_hardening.py python
1,176 lines 46.8 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 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 muse.core.types import fake_id, long_id, short_id, split_id
60 from tests.cli_test_helper import CliRunner
61
62 cli = None # argparse migration — CliRunner ignores this arg
63 runner = CliRunner()
64
65
66 # ---------------------------------------------------------------------------
67 # Shared helpers
68 # ---------------------------------------------------------------------------
69
70 def _env(root: pathlib.Path) -> Manifest:
71 return {"MUSE_REPO_ROOT": str(root)}
72
73
74 @pytest.fixture()
75 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
76 """Repo on ``main`` with two commits: base (a.py) + target (b.py).
77
78 The caller can immediately cherry-pick the HEAD commit to a new branch.
79 """
80 monkeypatch.chdir(tmp_path)
81 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
82 r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
83 assert r.exit_code == 0, r.output
84 (tmp_path / "a.py").write_text("x = 1\n")
85 r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
86 assert r.exit_code == 0, r.output
87 (tmp_path / "b.py").write_text("y = 2\n")
88 r = runner.invoke(cli, ["commit", "-m", "add b"], env=_env(tmp_path), catch_exceptions=False)
89 assert r.exit_code == 0, r.output
90 return tmp_path
91
92
93 @pytest.fixture()
94 def two_branch_repo(
95 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
96 ) -> tuple[pathlib.Path, str]:
97 """Repo with main and feat branches, returns (root, commit-id-on-feat).
98
99 ``main``: base commit only
100 ``feat``: base commit + one extra commit (the one to cherry-pick)
101 """
102 monkeypatch.chdir(tmp_path)
103 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
104 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
105 (tmp_path / "base.py").write_text("base\n")
106 runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
107
108 runner.invoke(cli, ["branch", "feat"], env=_env(tmp_path), catch_exceptions=False)
109 runner.invoke(cli, ["checkout", "feat"], env=_env(tmp_path), catch_exceptions=False)
110 (tmp_path / "extra.py").write_text("extra\n")
111 runner.invoke(cli, ["commit", "-m", "extra on feat"], env=_env(tmp_path), catch_exceptions=False)
112
113 from muse.core.store import get_head_commit_id
114 feat_cid = get_head_commit_id(tmp_path, "feat")
115 assert feat_cid is not None
116
117 runner.invoke(cli, ["checkout", "main"], env=_env(tmp_path), catch_exceptions=False)
118 return tmp_path, feat_cid
119
120
121 def _head_id(repo: pathlib.Path, branch: str = "main") -> str | None:
122 from muse.core.store import get_head_commit_id
123 return get_head_commit_id(repo, branch)
124
125
126 # ---------------------------------------------------------------------------
127 # Unit — parser flags and dead-code removal
128 # ---------------------------------------------------------------------------
129
130 class TestRegisterFlags:
131 def _parse(self, *args: str) -> argparse.Namespace:
132 import muse.cli.commands.cherry_pick as m
133 p = argparse.ArgumentParser()
134 sub = p.add_subparsers()
135 m.register(sub)
136 return p.parse_args(["cherry-pick", *args])
137
138 def test_dry_run_flag(self) -> None:
139 ns = self._parse("abc123", "--dry-run")
140 assert ns.dry_run is True
141
142 def test_dry_run_default_false(self) -> None:
143 ns = self._parse("abc123")
144 assert ns.dry_run is False
145
146 def test_no_commit_short(self) -> None:
147 ns = self._parse("abc123", "-n")
148 assert ns.no_commit is True
149
150 def test_no_commit_long(self) -> None:
151 ns = self._parse("abc123", "--no-commit")
152 assert ns.no_commit is True
153
154 def test_force_flag(self) -> None:
155 ns = self._parse("abc123", "--force")
156 assert ns.force is True
157
158 def test_message_short(self) -> None:
159 ns = self._parse("abc123", "-m", "my msg")
160 assert ns.message == "my msg"
161
162 def test_message_long(self) -> None:
163 ns = self._parse("abc123", "--message", "my msg")
164 assert ns.message == "my msg"
165
166 def test_message_default_none(self) -> None:
167 ns = self._parse("abc123")
168 assert ns.message is None
169
170 def test_format_json_shorthand(self) -> None:
171 ns = self._parse("abc123", "--json")
172 assert ns.fmt == "json"
173
174 def test_format_explicit_text(self) -> None:
175 ns = self._parse("abc123", "--format", "text")
176 assert ns.fmt == "text"
177
178 def test_ref_positional(self) -> None:
179 ns = self._parse("deadbeef")
180 assert ns.ref == "deadbeef"
181
182
183 class TestDeadCodeRemoval:
184 def test_no_read_branch_wrapper(self) -> None:
185 import muse.cli.commands.cherry_pick as m
186 assert not hasattr(m, "_read_branch"), "_read_branch must be deleted"
187
188 def test_pathlib_not_imported(self) -> None:
189 import muse.cli.commands.cherry_pick as m
190 assert "import pathlib" not in inspect.getsource(m)
191
192 def test_validate_branch_name_in_run(self) -> None:
193 import muse.cli.commands.cherry_pick as m
194 assert "validate_branch_name" in inspect.getsource(m.run)
195
196 def test_target_message_sanitized_in_run(self) -> None:
197 import muse.cli.commands.cherry_pick as m
198 assert "sanitize_display(target.message" in inspect.getsource(m.run)
199
200 def test_ref_sanitized_in_not_found_error(self) -> None:
201 import muse.cli.commands.cherry_pick as m
202 assert "sanitize_display(ref)" in inspect.getsource(m.run)
203
204 def test_write_snapshot_before_apply_manifest_in_normal_path(self) -> None:
205 """Normal path: write_snapshot and write_commit must precede apply_manifest."""
206 import muse.cli.commands.cherry_pick as m
207 src_lines = [
208 (i, l)
209 for i, l in enumerate(inspect.getsource(m.run).split("\n"), 1)
210 if l.strip() and not l.strip().startswith("#")
211 ]
212 ws = next(i for i, l in src_lines if "write_snapshot(" in l)
213 wc = next(i for i, l in src_lines if "write_commit(" in l)
214 wr = next(i for i, l in src_lines if "write_branch_ref(" in l)
215 # Find the LAST apply_manifest (normal path, not no_commit path)
216 all_am = [i for i, l in src_lines if "apply_manifest(" in l]
217 last_am = max(all_am)
218 assert ws < last_am, f"write_snapshot ({ws}) must precede apply_manifest ({last_am})"
219 assert wc < last_am, f"write_commit ({wc}) must precede apply_manifest ({last_am})"
220 assert last_am < wr, f"apply_manifest ({last_am}) must precede write_branch_ref ({wr})"
221
222 def test_parent_snapshot_missing_raises_not_silently_falls_back(self) -> None:
223 """Code must not silently use {} when parent snapshot is missing."""
224 import muse.cli.commands.cherry_pick as m
225 src = inspect.getsource(m.run)
226 # The silent fallback was: `if parent_snap: base_manifest = parent_snap.manifest`
227 # The fix is: `raise SystemExit(ExitCode.INTERNAL_ERROR)` when parent_snap is None
228 assert "INTERNAL_ERROR" in src
229
230
231 # ---------------------------------------------------------------------------
232 # Integration — error routing and behaviour
233 # ---------------------------------------------------------------------------
234
235 class TestErrorRouting:
236 def test_not_found_to_stderr(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
237 root, _ = two_branch_repo
238 r = runner.invoke(cli, ["cherry-pick", "badref"], env=_env(root))
239 assert r.exit_code != 0
240 assert "not found" in (r.stderr or "").lower()
241 assert "badref" in (r.stderr or "")
242
243 def test_not_found_stdout_clean(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
244 root, _ = two_branch_repo
245 r = runner.invoke(cli, ["cherry-pick", "0000000000000000"], env=_env(root))
246 assert r.exit_code != 0
247 # Error must be in stderr; stdout should have no error messages.
248 assert "not found" in (r.stderr or "").lower()
249
250
251
252 class TestJsonSchema:
253 """JSON schema must be identical across all code paths."""
254
255 _REQUIRED_KEYS = {
256 "status", "commit_id", "branch", "ref",
257 "source_commit_id", "snapshot_id", "message",
258 "no_commit", "dry_run", "conflicts",
259 }
260
261 def test_normal_json_schema_complete(
262 self, two_branch_repo: tuple[pathlib.Path, str]
263 ) -> None:
264 root, cid = two_branch_repo
265 r = runner.invoke(
266 cli, ["cherry-pick", cid, "--json"],
267 env=_env(root), catch_exceptions=False,
268 )
269 assert r.exit_code == 0, r.output
270 d = json.loads(r.output)
271 assert self._REQUIRED_KEYS <= d.keys()
272
273 def test_normal_status_is_picked(
274 self, two_branch_repo: tuple[pathlib.Path, str]
275 ) -> None:
276 root, cid = two_branch_repo
277 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
278 d = json.loads(r.output)
279 assert d["status"] == "picked"
280 assert d["no_commit"] is False
281 assert d["dry_run"] is False
282 assert d["conflicts"] == []
283
284 def test_normal_ref_field_matches_input(
285 self, two_branch_repo: tuple[pathlib.Path, str]
286 ) -> None:
287 root, cid = two_branch_repo
288 r = runner.invoke(cli, ["cherry-pick", short_id(cid), "--json"], env=_env(root), catch_exceptions=False)
289 d = json.loads(r.output)
290 assert d["ref"] == short_id(cid)
291
292 def test_normal_snapshot_id_is_string(
293 self, two_branch_repo: tuple[pathlib.Path, str]
294 ) -> None:
295 root, cid = two_branch_repo
296 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
297 d = json.loads(r.output)
298 # Canonical Muse IDs are "sha256:<64 hex chars>" = 71 chars total
299 assert isinstance(d["snapshot_id"], str)
300 assert d["snapshot_id"].startswith("sha256:")
301 assert len(d["snapshot_id"]) == 71
302
303 def test_no_commit_json_schema_complete(
304 self, two_branch_repo: tuple[pathlib.Path, str]
305 ) -> None:
306 root, cid = two_branch_repo
307 r = runner.invoke(
308 cli, ["cherry-pick", cid, "--no-commit", "--json"],
309 env=_env(root), catch_exceptions=False,
310 )
311 assert r.exit_code == 0, r.output
312 d = json.loads(r.output)
313 assert self._REQUIRED_KEYS <= d.keys()
314
315 def test_no_commit_status_is_applied(
316 self, two_branch_repo: tuple[pathlib.Path, str]
317 ) -> None:
318 root, cid = two_branch_repo
319 r = runner.invoke(
320 cli, ["cherry-pick", cid, "--no-commit", "--json"],
321 env=_env(root), catch_exceptions=False,
322 )
323 d = json.loads(r.output)
324 assert d["status"] == "applied"
325 assert d["commit_id"] is None
326 assert d["no_commit"] is True
327 assert d["dry_run"] is False
328
329 def test_dry_run_json_schema_complete(
330 self, two_branch_repo: tuple[pathlib.Path, str]
331 ) -> None:
332 root, cid = two_branch_repo
333 r = runner.invoke(
334 cli, ["cherry-pick", cid, "--dry-run", "--json"],
335 env=_env(root), catch_exceptions=False,
336 )
337 assert r.exit_code == 0, r.output
338 d = json.loads(r.output)
339 assert self._REQUIRED_KEYS <= d.keys()
340
341 def test_dry_run_status(
342 self, two_branch_repo: tuple[pathlib.Path, str]
343 ) -> None:
344 root, cid = two_branch_repo
345 r = runner.invoke(
346 cli, ["cherry-pick", cid, "--dry-run", "--json"],
347 env=_env(root), catch_exceptions=False,
348 )
349 d = json.loads(r.output)
350 assert d["dry_run"] is True
351 assert d["commit_id"] is None
352 assert d["status"] == "dry_run"
353
354 def test_all_three_schemas_identical(
355 self, two_branch_repo: tuple[pathlib.Path, str]
356 ) -> None:
357 root, cid = two_branch_repo
358 r_dr = runner.invoke(cli, ["cherry-pick", cid, "--dry-run", "--json"], env=_env(root), catch_exceptions=False)
359 r_nc = runner.invoke(cli, ["cherry-pick", cid, "--no-commit", "--json"], env=_env(root), catch_exceptions=False)
360
361 # Normal cherry-pick: need to re-fetch after --no-commit modified workdir
362 r_commit = runner.invoke(cli, ["commit", "-m", "after nc"], env=_env(root), catch_exceptions=False)
363 from muse.core.store import get_head_commit_id
364 new_cid = get_head_commit_id(root, "main")
365 assert new_cid is not None
366 # Need a different source commit to cherry-pick after the no-commit pick
367 r_nm = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
368
369 keys_dr = set(json.loads(r_dr.output).keys())
370 keys_nc = set(json.loads(r_nc.output).keys())
371 keys_nm = set(json.loads(r_nm.output).keys())
372 assert keys_dr == keys_nc == keys_nm
373
374
375 class TestDryRun:
376 def test_no_commit_created(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
377 from muse.core.store import get_all_commits, get_head_commit_id
378 root, cid = two_branch_repo
379 before_count = len(get_all_commits(root))
380 before_head = get_head_commit_id(root, "main")
381 r = runner.invoke(
382 cli, ["cherry-pick", cid, "--dry-run"],
383 env=_env(root), catch_exceptions=False,
384 )
385 assert r.exit_code == 0, r.output
386 assert len(get_all_commits(root)) == before_count
387 assert get_head_commit_id(root, "main") == before_head
388
389 def test_workdir_unchanged(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
390 root, cid = two_branch_repo
391 extra = root / "extra.py"
392 content_before = extra.read_text() if extra.exists() else None
393 r = runner.invoke(
394 cli, ["cherry-pick", cid, "--dry-run"],
395 env=_env(root), catch_exceptions=False,
396 )
397 assert r.exit_code == 0, r.output
398 assert (extra.read_text() if extra.exists() else None) == content_before
399
400 def test_reflog_unchanged(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
401 from muse.core.reflog import read_reflog
402 root, cid = two_branch_repo
403 before = len(read_reflog(root, "main"))
404 runner.invoke(cli, ["cherry-pick", cid, "--dry-run"], env=_env(root), catch_exceptions=False)
405 assert len(read_reflog(root, "main")) == before
406
407 def test_dry_run_text_output_mentions_dry_run(
408 self, two_branch_repo: tuple[pathlib.Path, str]
409 ) -> None:
410 root, cid = two_branch_repo
411 r = runner.invoke(
412 cli, ["cherry-pick", cid, "--dry-run"],
413 env=_env(root), catch_exceptions=False,
414 )
415 assert "dry-run" in r.output.lower() or "would" in r.output.lower()
416
417 def test_dry_run_invalid_ref_errors(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
418 root, _ = two_branch_repo
419 r = runner.invoke(cli, ["cherry-pick", "no-such-ref", "--dry-run"], env=_env(root))
420 assert r.exit_code != 0
421
422 def test_dry_run_json_snapshot_id_present(
423 self, two_branch_repo: tuple[pathlib.Path, str]
424 ) -> None:
425 root, cid = two_branch_repo
426 r = runner.invoke(
427 cli, ["cherry-pick", cid, "--dry-run", "--json"],
428 env=_env(root), catch_exceptions=False,
429 )
430 d = json.loads(r.output)
431 # Canonical Muse IDs are "sha256:<64 hex chars>" = 71 chars total
432 assert d["snapshot_id"] is not None
433 assert d["snapshot_id"].startswith("sha256:")
434 assert len(d["snapshot_id"]) == 71
435
436
437 class TestNoCommit:
438 def test_branch_ref_not_advanced(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
439 from muse.core.store import get_head_commit_id
440 root, cid = two_branch_repo
441 before_head = get_head_commit_id(root, "main")
442 r = runner.invoke(
443 cli, ["cherry-pick", cid, "--no-commit"],
444 env=_env(root), catch_exceptions=False,
445 )
446 assert r.exit_code == 0, r.output
447 assert get_head_commit_id(root, "main") == before_head
448
449 def test_workdir_modified(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
450 root, cid = two_branch_repo
451 r = runner.invoke(
452 cli, ["cherry-pick", cid, "--no-commit"],
453 env=_env(root), catch_exceptions=False,
454 )
455 assert r.exit_code == 0, r.output
456 # extra.py was added by the feat branch commit
457 assert (root / "extra.py").exists()
458
459 def test_no_commit_in_json(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
460 root, cid = two_branch_repo
461 r = runner.invoke(
462 cli, ["cherry-pick", cid, "--no-commit", "--json"],
463 env=_env(root), catch_exceptions=False,
464 )
465 d = json.loads(r.output)
466 assert d["no_commit"] is True
467 assert d["commit_id"] is None
468 assert d["status"] == "applied"
469
470 def test_reflog_not_written_for_no_commit(
471 self, two_branch_repo: tuple[pathlib.Path, str]
472 ) -> None:
473 from muse.core.reflog import read_reflog
474 root, cid = two_branch_repo
475 before = len(read_reflog(root, "main"))
476 runner.invoke(
477 cli, ["cherry-pick", cid, "--no-commit"],
478 env=_env(root), catch_exceptions=False,
479 )
480 assert len(read_reflog(root, "main")) == before
481
482
483 class TestReflog:
484 def test_reflog_appended(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
485 from muse.core.reflog import read_reflog
486 root, cid = two_branch_repo
487 before = len(read_reflog(root, "main"))
488 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
489 assert len(read_reflog(root, "main")) > before
490
491 def test_reflog_operation_contains_cherry_pick(
492 self, two_branch_repo: tuple[pathlib.Path, str]
493 ) -> None:
494 from muse.core.reflog import read_reflog
495 root, cid = two_branch_repo
496 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
497 entries = read_reflog(root, "main")
498 # read_reflog returns newest-first
499 assert "cherry-pick" in entries[0].operation.lower()
500
501
502 class TestMessageFlag:
503 def test_custom_message_in_json(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
504 root, cid = two_branch_repo
505 r = runner.invoke(
506 cli, ["cherry-pick", cid, "-m", "custom pick msg", "--json"],
507 env=_env(root), catch_exceptions=False,
508 )
509 assert r.exit_code == 0, r.output
510 d = json.loads(r.output)
511 assert d["message"] == "custom pick msg"
512
513 def test_custom_message_in_text(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
514 root, cid = two_branch_repo
515 r = runner.invoke(
516 cli, ["cherry-pick", cid, "-m", "undo extra"],
517 env=_env(root), catch_exceptions=False,
518 )
519 assert r.exit_code == 0, r.output
520 assert "undo extra" in r.output
521
522 def test_default_message_is_source_message(
523 self, two_branch_repo: tuple[pathlib.Path, str]
524 ) -> None:
525 root, cid = two_branch_repo
526 r = runner.invoke(
527 cli, ["cherry-pick", cid, "--json"],
528 env=_env(root), catch_exceptions=False,
529 )
530 d = json.loads(r.output)
531 assert d["message"] == "extra on feat"
532
533 def test_message_stored_in_commit_record(
534 self, two_branch_repo: tuple[pathlib.Path, str]
535 ) -> None:
536 from muse.core.store import get_head_commit_id, read_commit
537 root, cid = two_branch_repo
538 runner.invoke(
539 cli, ["cherry-pick", cid, "-m", "my override"],
540 env=_env(root), catch_exceptions=False,
541 )
542 new_cid = get_head_commit_id(root, "main")
543 assert new_cid is not None
544 rec = read_commit(root, new_cid)
545 assert rec is not None
546 assert rec.message == "my override"
547
548
549 class TestWriteOrdering:
550 def test_write_snapshot_before_apply_manifest(
551 self, two_branch_repo: tuple[pathlib.Path, str]
552 ) -> None:
553 """write_snapshot must be called before apply_manifest in the normal path."""
554 from unittest.mock import patch
555 import muse.cli.commands.cherry_pick as cp_mod
556 from muse.core import store as s
557 events: list[str] = []
558 orig_ws = s.write_snapshot
559 from muse.core.workdir import apply_manifest as orig_am_fn
560
561 def tracking_write_snapshot(root: pathlib.Path, rec: s.SnapshotRecord) -> None:
562 orig_ws(root, rec)
563 events.append("write_snapshot")
564
565 def tracking_apply(root: pathlib.Path, prev: Manifest, manifest: Manifest) -> None:
566 orig_am_fn(root, prev, manifest)
567 events.append("apply_manifest")
568
569 root, cid = two_branch_repo
570 with (
571 patch.object(cp_mod, "write_snapshot", tracking_write_snapshot),
572 patch("muse.cli.commands.cherry_pick.apply_manifest", tracking_apply),
573 ):
574 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
575
576 assert "write_snapshot" in events
577 assert "apply_manifest" in events
578 assert events.index("write_snapshot") < events.index("apply_manifest"), (
579 f"write_snapshot must precede apply_manifest, got: {events}"
580 )
581
582
583 class TestFailFast:
584 def test_missing_parent_commit_is_error(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
585 """When the target has a parent_commit_id but the parent object is missing,
586 cherry-pick must exit INTERNAL_ERROR — not silently use empty base."""
587 import datetime
588 monkeypatch.chdir(tmp_path)
589 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
590 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
591
592 from muse.core.store import (
593 CommitRecord, SnapshotRecord, write_commit, write_snapshot,
594 get_head_commit_id,
595 )
596 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
597
598 repo_id = (repo_json_path(tmp_path)).read_text()
599 import json as _json
600 repo_id = _json.loads(repo_id)["repo_id"]
601
602 # Create a base commit
603 m1: Manifest = {}
604 s1 = compute_snapshot_id(m1)
605 t1 = datetime.datetime.now(datetime.timezone.utc)
606 c1 = compute_commit_id( parent_ids=[],
607 snapshot_id=s1,
608 message="base",
609 committed_at_iso=t1.isoformat(),
610 )
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 (heads_dir(tmp_path) / "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( parent_ids=[c1],
624 snapshot_id=s2,
625 message="target",
626 committed_at_iso=t2.isoformat(),
627 )
628 write_snapshot(tmp_path, SnapshotRecord(snapshot_id=s2, manifest=m2))
629 write_commit(tmp_path, CommitRecord(
630 commit_id=c2, repo_id=repo_id, branch="main",
631 snapshot_id=s2, message="target", committed_at=t2,
632 parent_commit_id=c1,
633 ))
634
635 # Now delete the parent commit file to simulate object-store corruption
636 # Path shape: .muse/commits/<algo>/<hex>.msgpack
637 algo, hex_str = split_id(c1)
638 commit_file = commits_dir(tmp_path) / algo / f"{hex_str}.msgpack"
639 if commit_file.exists():
640 commit_file.unlink()
641
642 # Reset HEAD to base so we can cherry-pick c2 "from another branch"
643 # Switch to a fresh branch at c1's snapshot
644 runner.invoke(cli, ["branch", "target-branch"], env=_env(tmp_path), catch_exceptions=False)
645 runner.invoke(cli, ["checkout", "target-branch"], env=_env(tmp_path), catch_exceptions=False)
646 (heads_dir(tmp_path) / "target-branch").write_text(c1)
647
648 r = runner.invoke(cli, ["cherry-pick", c2], env=_env(tmp_path))
649 # Must fail with INTERNAL_ERROR (exit code 3), not succeed silently
650 assert r.exit_code != 0
651
652
653 # ---------------------------------------------------------------------------
654 # End-to-end — text and JSON output
655 # ---------------------------------------------------------------------------
656
657 class TestTextOutput:
658 def test_text_shows_branch_and_short_id(
659 self, two_branch_repo: tuple[pathlib.Path, str]
660 ) -> None:
661 root, cid = two_branch_repo
662 r = runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
663 assert r.exit_code == 0
664 assert "main" in r.output
665
666 def test_no_commit_text_mentions_workdir(
667 self, two_branch_repo: tuple[pathlib.Path, str]
668 ) -> None:
669 root, cid = two_branch_repo
670 r = runner.invoke(
671 cli, ["cherry-pick", cid, "--no-commit"],
672 env=_env(root), catch_exceptions=False,
673 )
674 output = r.output.lower()
675 assert "working tree" in output or "applied" in output or "commit" in output
676
677 def test_workdir_has_cherry_picked_file(
678 self, two_branch_repo: tuple[pathlib.Path, str]
679 ) -> None:
680 root, cid = two_branch_repo
681 runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
682 assert (root / "extra.py").exists()
683 assert (root / "extra.py").read_text() == "extra\n"
684
685
686 class TestJsonOutput:
687 def test_source_commit_id_matches(
688 self, two_branch_repo: tuple[pathlib.Path, str]
689 ) -> 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["source_commit_id"] == cid
694
695 def test_branch_field(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
696 root, cid = two_branch_repo
697 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
698 d = json.loads(r.output)
699 assert d["branch"] == "main"
700
701 def test_new_commit_id_different_from_source(
702 self, two_branch_repo: tuple[pathlib.Path, str]
703 ) -> None:
704 root, cid = two_branch_repo
705 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
706 d = json.loads(r.output)
707 assert d["commit_id"] != cid
708 # Canonical Muse IDs are "sha256:<64 hex chars>" = 71 chars total
709 assert isinstance(d["commit_id"], str)
710 assert d["commit_id"].startswith("sha256:")
711 assert len(d["commit_id"]) == 71
712
713 def test_conflicts_empty_on_success(
714 self, two_branch_repo: tuple[pathlib.Path, str]
715 ) -> None:
716 root, cid = two_branch_repo
717 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
718 d = json.loads(r.output)
719 assert d["conflicts"] == []
720
721
722 class TestForce:
723 def test_force_bypasses_dirty_check(
724 self, two_branch_repo: tuple[pathlib.Path, str]
725 ) -> None:
726 root, cid = two_branch_repo
727 (root / "base.py").write_text("modified but uncommitted\n")
728 r = runner.invoke(
729 cli, ["cherry-pick", cid, "--force"],
730 env=_env(root), catch_exceptions=False,
731 )
732 assert r.exit_code == 0, r.output
733
734 def test_without_force_dirty_tree_fails(
735 self, two_branch_repo: tuple[pathlib.Path, str]
736 ) -> None:
737 root, cid = two_branch_repo
738 (root / "base.py").write_text("uncommitted change\n")
739 r = runner.invoke(cli, ["cherry-pick", cid], env=_env(root))
740 assert r.exit_code != 0
741
742
743 # ---------------------------------------------------------------------------
744 # Security — ANSI injection and sanitization
745 # ---------------------------------------------------------------------------
746
747 class TestSecurity:
748 def test_ansi_in_ref_error_in_stderr(self, two_branch_repo: tuple[pathlib.Path, str]) -> None:
749 root, _ = two_branch_repo
750 ansi_ref = "\x1b[31mbadref\x1b[0m"
751 r = runner.invoke(cli, ["cherry-pick", ansi_ref], env=_env(root))
752 assert r.exit_code != 0
753 assert "\x1b[31m" not in (r.stdout or "")
754 assert "badref" in (r.stderr or "")
755
756 def test_ansi_in_commit_message_not_in_stored_commit(
757 self, two_branch_repo: tuple[pathlib.Path, str]
758 ) -> None:
759 """If the source commit has ANSI in its message, the cherry-pick commit
760 stored on disk must not contain raw escape sequences."""
761 from unittest.mock import patch
762 from muse.core import store as s
763 import muse.cli.commands.cherry_pick as cp_mod
764
765 root, cid = two_branch_repo
766 orig_rc = s.read_commit
767
768 def poisoned_read_commit(root: pathlib.Path, c: str) -> s.CommitRecord | None:
769 rec = orig_rc(root, c)
770 if rec is not None and rec.commit_id == cid:
771 return s.CommitRecord(
772 commit_id=rec.commit_id, repo_id=rec.repo_id,
773 branch=rec.branch, snapshot_id=rec.snapshot_id,
774 message="\x1b[31mmalicious\x1b[0m",
775 committed_at=rec.committed_at,
776 parent_commit_id=rec.parent_commit_id,
777 )
778 return rec
779
780 with patch.object(cp_mod, "read_commit", poisoned_read_commit):
781 r = runner.invoke(
782 cli, ["cherry-pick", cid, "--json"],
783 env=_env(root), catch_exceptions=False,
784 )
785
786 if r.exit_code == 0:
787 d = json.loads(r.output)
788 assert "\x1b[" not in d.get("message", ""), (
789 "Cherry-pick commit message must not contain raw ANSI from source"
790 )
791
792
793
794 # ---------------------------------------------------------------------------
795 # Stress
796 # ---------------------------------------------------------------------------
797
798 class TestStress:
799 @pytest.mark.slow
800 def test_cherry_pick_deep_in_history(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
801 """Cherry-pick a commit that's deep in a 200-commit chain."""
802 monkeypatch.chdir(tmp_path)
803 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
804 env = _env(tmp_path)
805 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
806 (tmp_path / "seed.py").write_text("seed\n")
807 runner.invoke(cli, ["commit", "-m", "seed"], env=env, catch_exceptions=False)
808
809 runner.invoke(cli, ["branch", "source"], env=env, catch_exceptions=False)
810 runner.invoke(cli, ["checkout", "source"], env=env, catch_exceptions=False)
811
812 (tmp_path / "target.py").write_text("target\n")
813 runner.invoke(cli, ["commit", "-m", "target commit"], env=env, catch_exceptions=False)
814 from muse.core.store import get_head_commit_id
815 target_cid = get_head_commit_id(tmp_path, "source")
816 assert target_cid is not None
817
818 for i in range(198):
819 (tmp_path / f"f{i}.py").write_text(f"{i}\n")
820 runner.invoke(cli, ["commit", "-m", f"c{i}"], env=env, catch_exceptions=False)
821
822 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
823 r = runner.invoke(
824 cli, ["cherry-pick", target_cid, "--json"],
825 env=env, catch_exceptions=False,
826 )
827 assert r.exit_code == 0, r.output
828 d = json.loads(r.output)
829 assert d["source_commit_id"] == target_cid
830 assert d["status"] == "picked"
831
832 @pytest.mark.slow
833 def test_sequential_cherry_picks(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
834 """30 sequential cherry-picks must all succeed."""
835 monkeypatch.chdir(tmp_path)
836 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
837 env = _env(tmp_path)
838 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
839 (tmp_path / "base.py").write_text("base\n")
840 runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
841
842 runner.invoke(cli, ["branch", "source"], env=env, catch_exceptions=False)
843 runner.invoke(cli, ["checkout", "source"], env=env, catch_exceptions=False)
844
845 # Create 30 non-conflicting commits on source
846 source_cids: list[str] = []
847 for i in range(30):
848 (tmp_path / f"s{i}.py").write_text(f"s{i}\n")
849 runner.invoke(cli, ["commit", "-m", f"src{i}"], env=env, catch_exceptions=False)
850 from muse.core.store import get_head_commit_id
851 cid = get_head_commit_id(tmp_path, "source")
852 assert cid is not None
853 source_cids.append(cid)
854
855 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
856 failures: list[str] = []
857 for i, cid in enumerate(source_cids):
858 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=env)
859 if r.exit_code != 0:
860 failures.append(f"pick {i}: exit={r.exit_code} {r.output.strip()[:60]}")
861
862 assert not failures, f"Sequential failures: {failures}"
863
864
865 @pytest.mark.slow
866 def test_dry_run_performance(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
867 """--dry-run on a large repo must complete in < 3 s."""
868 monkeypatch.chdir(tmp_path)
869 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
870 env = _env(tmp_path)
871 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
872 (tmp_path / "base.py").write_text("base\n")
873 runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
874 runner.invoke(cli, ["branch", "src"], env=env, catch_exceptions=False)
875 runner.invoke(cli, ["checkout", "src"], env=env, catch_exceptions=False)
876
877 (tmp_path / "target.py").write_text("target\n")
878 runner.invoke(cli, ["commit", "-m", "target"], env=env, catch_exceptions=False)
879 from muse.core.store import get_head_commit_id
880 target_cid = get_head_commit_id(tmp_path, "src")
881 assert target_cid is not None
882
883 for i in range(100):
884 (tmp_path / f"f{i}.py").write_text(f"{i}\n")
885 runner.invoke(cli, ["commit", "-m", f"c{i}"], env=env, catch_exceptions=False)
886
887 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
888 start = time.perf_counter()
889 r = runner.invoke(
890 cli, ["cherry-pick", target_cid, "--dry-run", "--json"],
891 env=env, catch_exceptions=False,
892 )
893 elapsed = time.perf_counter() - start
894 assert r.exit_code == 0, r.output
895 assert elapsed < 3.0, f"--dry-run took {elapsed:.2f}s"
896
897
898 # ---------------------------------------------------------------------------
899 # Agent supercharge — duration_ms and exit_code in every JSON output
900 # ---------------------------------------------------------------------------
901
902
903 class TestElapsed:
904 """Every JSON output path must include ``duration_ms`` as a float."""
905
906 def test_picked_json_has_elapsed(
907 self, two_branch_repo: tuple[pathlib.Path, str]
908 ) -> None:
909 root, cid = two_branch_repo
910 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
911 d = json.loads(r.output)
912 assert "duration_ms" in d
913 assert isinstance(d["duration_ms"], float)
914
915 def test_applied_json_has_elapsed(
916 self, two_branch_repo: tuple[pathlib.Path, str]
917 ) -> None:
918 root, cid = two_branch_repo
919 r = runner.invoke(
920 cli, ["cherry-pick", cid, "--no-commit", "--json"],
921 env=_env(root), catch_exceptions=False,
922 )
923 d = json.loads(r.output)
924 assert "duration_ms" in d
925 assert isinstance(d["duration_ms"], float)
926
927 def test_dry_run_json_has_elapsed(
928 self, two_branch_repo: tuple[pathlib.Path, str]
929 ) -> None:
930 root, cid = two_branch_repo
931 r = runner.invoke(
932 cli, ["cherry-pick", cid, "--dry-run", "--json"],
933 env=_env(root), catch_exceptions=False,
934 )
935 d = json.loads(r.output)
936 assert "duration_ms" in d
937 assert isinstance(d["duration_ms"], float)
938
939 def test_conflict_json_has_elapsed(
940 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
941 ) -> None:
942 """The conflict JSON path must also include duration_ms."""
943 monkeypatch.chdir(tmp_path)
944 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
945 env = _env(tmp_path)
946 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
947
948 (tmp_path / "shared.py").write_text("line1\nline2\nline3\n")
949 runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
950
951 runner.invoke(cli, ["branch", "src"], env=env, catch_exceptions=False)
952 runner.invoke(cli, ["checkout", "src"], env=env, catch_exceptions=False)
953 (tmp_path / "shared.py").write_text("line1\nSRC_LINE2\nline3\n")
954 runner.invoke(cli, ["commit", "-m", "src change"], env=env, catch_exceptions=False)
955 from muse.core.store import get_head_commit_id as _gci
956 src_cid = _gci(tmp_path, "src")
957 assert src_cid is not None
958
959 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
960 (tmp_path / "shared.py").write_text("line1\nMAIN_LINE2\nline3\n")
961 runner.invoke(cli, ["commit", "-m", "main change"], env=env, catch_exceptions=False)
962
963 r = runner.invoke(cli, ["cherry-pick", src_cid, "--json"], env=env)
964 # Conflict should produce JSON with duration_ms even on exit 1
965 assert r.exit_code == 1
966 d = json.loads(r.output)
967 assert "duration_ms" in d
968 assert isinstance(d["duration_ms"], float)
969
970
971 class TestExitCode:
972 """Every successful JSON path includes ``exit_code: 0``; conflict path has ``exit_code: 1``."""
973
974 def test_picked_json_exit_code_0(
975 self, two_branch_repo: tuple[pathlib.Path, str]
976 ) -> None:
977 root, cid = two_branch_repo
978 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
979 d = json.loads(r.output)
980 assert d["exit_code"] == 0
981
982 def test_applied_json_exit_code_0(
983 self, two_branch_repo: tuple[pathlib.Path, str]
984 ) -> None:
985 root, cid = two_branch_repo
986 r = runner.invoke(
987 cli, ["cherry-pick", cid, "--no-commit", "--json"],
988 env=_env(root), catch_exceptions=False,
989 )
990 d = json.loads(r.output)
991 assert d["exit_code"] == 0
992
993 def test_dry_run_json_exit_code_0(
994 self, two_branch_repo: tuple[pathlib.Path, str]
995 ) -> None:
996 root, cid = two_branch_repo
997 r = runner.invoke(
998 cli, ["cherry-pick", cid, "--dry-run", "--json"],
999 env=_env(root), catch_exceptions=False,
1000 )
1001 d = json.loads(r.output)
1002 assert d["exit_code"] == 0
1003
1004 def test_conflict_json_exit_code_1(
1005 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1006 ) -> None:
1007 """Conflict JSON must report exit_code: 1, mirroring the process exit."""
1008 monkeypatch.chdir(tmp_path)
1009 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
1010 env = _env(tmp_path)
1011 runner.invoke(cli, ["init"], env=env, catch_exceptions=False)
1012
1013 (tmp_path / "shared.py").write_text("line1\nline2\nline3\n")
1014 runner.invoke(cli, ["commit", "-m", "base"], env=env, catch_exceptions=False)
1015
1016 runner.invoke(cli, ["branch", "src"], env=env, catch_exceptions=False)
1017 runner.invoke(cli, ["checkout", "src"], env=env, catch_exceptions=False)
1018 (tmp_path / "shared.py").write_text("line1\nSRC_LINE2\nline3\n")
1019 runner.invoke(cli, ["commit", "-m", "src change"], env=env, catch_exceptions=False)
1020 from muse.core.store import get_head_commit_id as _gci
1021 src_cid = _gci(tmp_path, "src")
1022 assert src_cid is not None
1023
1024 runner.invoke(cli, ["checkout", "main"], env=env, catch_exceptions=False)
1025 (tmp_path / "shared.py").write_text("line1\nMAIN_LINE2\nline3\n")
1026 runner.invoke(cli, ["commit", "-m", "main change"], env=env, catch_exceptions=False)
1027
1028 r = runner.invoke(cli, ["cherry-pick", src_cid, "--json"], env=env)
1029 assert r.exit_code == 1
1030 d = json.loads(r.output)
1031 assert d["exit_code"] == 1
1032
1033
1034 class TestJsonSchemaComplete:
1035 """``duration_ms`` and ``exit_code`` must be in every JSON output."""
1036
1037 _FULL_KEYS = {
1038 "status", "commit_id", "branch", "ref",
1039 "source_commit_id", "snapshot_id", "message",
1040 "no_commit", "dry_run", "conflicts",
1041 "duration_ms", "exit_code",
1042 "muse_version", "schema", "timestamp", "warnings",
1043 }
1044
1045 def test_picked_has_complete_schema(
1046 self, two_branch_repo: tuple[pathlib.Path, str]
1047 ) -> None:
1048 root, cid = two_branch_repo
1049 r = runner.invoke(cli, ["cherry-pick", cid, "--json"], env=_env(root), catch_exceptions=False)
1050 d = json.loads(r.output)
1051 missing = self._FULL_KEYS - d.keys()
1052 assert not missing, f"Missing keys in 'picked' JSON: {missing}"
1053
1054 def test_applied_has_complete_schema(
1055 self, two_branch_repo: tuple[pathlib.Path, str]
1056 ) -> None:
1057 root, cid = two_branch_repo
1058 r = runner.invoke(
1059 cli, ["cherry-pick", cid, "--no-commit", "--json"],
1060 env=_env(root), catch_exceptions=False,
1061 )
1062 d = json.loads(r.output)
1063 missing = self._FULL_KEYS - d.keys()
1064 assert not missing, f"Missing keys in 'applied' JSON: {missing}"
1065
1066 def test_dry_run_has_complete_schema(
1067 self, two_branch_repo: tuple[pathlib.Path, str]
1068 ) -> None:
1069 root, cid = two_branch_repo
1070 r = runner.invoke(
1071 cli, ["cherry-pick", cid, "--dry-run", "--json"],
1072 env=_env(root), catch_exceptions=False,
1073 )
1074 d = json.loads(r.output)
1075 missing = self._FULL_KEYS - d.keys()
1076 assert not missing, f"Missing keys in 'dry_run' JSON: {missing}"
1077
1078 def test_all_schemas_identical(
1079 self, two_branch_repo: tuple[pathlib.Path, str]
1080 ) -> None:
1081 """All three success paths must have identical key sets."""
1082 root, cid = two_branch_repo
1083 r_dr = runner.invoke(
1084 cli, ["cherry-pick", cid, "--dry-run", "--json"],
1085 env=_env(root), catch_exceptions=False,
1086 )
1087 r_nc = runner.invoke(
1088 cli, ["cherry-pick", cid, "--no-commit", "--json"],
1089 env=_env(root), catch_exceptions=False,
1090 )
1091 runner.invoke(cli, ["commit", "-m", "after nc"], env=_env(root), catch_exceptions=False)
1092 r_nm = runner.invoke(
1093 cli, ["cherry-pick", cid, "--json"],
1094 env=_env(root), catch_exceptions=False,
1095 )
1096 keys_dr = set(json.loads(r_dr.output).keys())
1097 keys_nc = set(json.loads(r_nc.output).keys())
1098 keys_nm = set(json.loads(r_nm.output).keys())
1099 assert keys_dr == keys_nc == keys_nm == self._FULL_KEYS
1100
1101
1102 class TestTextOutputHex:
1103 """Text output must show sha256: prefix + 8 hex chars — canonical and algorithm-identifying."""
1104
1105 def test_picked_text_shows_prefixed_short_id(
1106 self, two_branch_repo: tuple[pathlib.Path, str]
1107 ) -> None:
1108 root, cid = two_branch_repo
1109 r = runner.invoke(cli, ["cherry-pick", cid], env=_env(root), catch_exceptions=False)
1110 assert r.exit_code == 0
1111 from muse.core.store import get_head_commit_id
1112 new_cid = get_head_commit_id(root, "main")
1113 assert new_cid is not None
1114 short = new_cid[:len("sha256:") + 8]
1115 assert short in r.output, (
1116 f"Expected '{short}' in cherry-pick output, got: {r.output!r}"
1117 )
1118
1119 def test_dry_run_text_shows_prefixed_full_id(
1120 self, two_branch_repo: tuple[pathlib.Path, str]
1121 ) -> None:
1122 import re
1123 root, cid = two_branch_repo
1124 r = runner.invoke(
1125 cli, ["cherry-pick", cid, "--dry-run"],
1126 env=_env(root), catch_exceptions=False,
1127 )
1128 assert r.exit_code == 0
1129 # The dry-run text shows the source full ID in parens: (sha256:<64hex>)
1130 match = re.search(r'\(sha256:([0-9a-f]{64})\)', r.output)
1131 assert match is not None, (
1132 f"Expected '(sha256:<64hex>)' in dry-run output, got: {r.output!r}"
1133 )
1134 assert long_id(match.group(1)) == cid, (
1135 f"Expected {cid} in parens, got {long_id(match.group(1))}"
1136 )
1137
1138
1139 # ---------------------------------------------------------------------------
1140 # Flag registration tests
1141 # ---------------------------------------------------------------------------
1142
1143 import argparse as _argparse
1144 from muse.cli.commands.cherry_pick import register as _register_cherry_pick
1145 from muse.core.paths import commits_dir, heads_dir, repo_json_path
1146
1147
1148 def _parse_cp(*args: str) -> _argparse.Namespace:
1149 """Build an argument parser via register() and parse args."""
1150 root_p = _argparse.ArgumentParser()
1151 subs = root_p.add_subparsers(dest="cmd")
1152 _register_cherry_pick(subs)
1153 return root_p.parse_args(["cherry-pick", *args])
1154
1155
1156 class TestRegisterFlags:
1157 def test_default_json_out_is_false(self) -> None:
1158 ns = _parse_cp(fake_id("a"))
1159 assert ns.json_out is False
1160
1161 def test_json_flag_sets_json_out(self) -> None:
1162 ns = _parse_cp(fake_id("a"), "--json")
1163 assert ns.json_out is True
1164
1165 def test_j_shorthand_sets_json_out(self) -> None:
1166 ns = _parse_cp(fake_id("a"), "-j")
1167 assert ns.json_out is True
1168
1169 def test_no_commit_flag(self) -> None:
1170 ns = _parse_cp(fake_id("a"), "--no-commit")
1171 assert ns.no_commit is True
1172
1173 def test_no_commit_has_no_n_shorthand(self) -> None:
1174 import pytest
1175 with pytest.raises(SystemExit):
1176 _parse_cp(fake_id("a"), "-n")
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago