gabriel / muse public
test_cmd_semantic_cherry_pick.py python
898 lines 37.5 KB
Raw
sha256:b89fa4fd9ca0d692fc66f6b9aef4c3a0c13c8e9b439faf42da8e91e09f048d4f tests/test_cmd_revert_hardening.py, tests/test_cmd_semantic… Human 4 days ago
1 """Tests for ``muse code semantic-cherry-pick``.
2
3 Coverage layers
4 ---------------
5 Unit
6 _verify_symbol — hit (symbol parseable), miss (symbol not in tree),
7 file read error, corrupt bytes.
8 _apply_symbol — no-separator address, path-traversal, file_missing
9 (obj not in manifest, blob missing), parse_error source,
10 parse_error current, already_current, applied (replace),
11 applied (append), new-file creation, dry-run (no write),
12 diff_lines populated, verified field populated.
13 src_cache — same blob fetched only once across multiple calls.
14
15 Integration (live repo, CliRunner)
16 Exits zero for valid cherry-pick.
17 JSON schema: all required top-level keys present, correct types.
18 JSON: schema_version, branch, from_commit (8-char hex), dry_run,
19 results[], applied, already_current, failed, unverified.
20 Per-result JSON: address, status, detail, old_lines, new_lines,
21 diff_lines, verified.
22 --dry-run: file not written, diff_lines populated in JSON.
23 --dry-run verified is True for valid output.
24 already_current result when symbol body unchanged.
25 ADDRESS without '::' → status not_found.
26 Path-traversal ADDRESS → status not_found.
27 Unknown --from ref → exits non-zero.
28 Symbol not in source commit → status not_found.
29 File not in source snapshot → status file_missing.
30 Multiple addresses in one invocation: all results present.
31 Multiple addresses to same file: blob fetched once (src_cache).
32 Text output contains commit short-hash, applied/failed counts.
33 Missing repo → exits non-zero.
34 unverified list populated when verification fails (monkeypatched).
35
36 E2E (real symbol changes across commits)
37 Applied symbol replaces only target lines; surrounding code unchanged.
38 Applying from earlier commit restores old implementation.
39 Dry-run leaves file unchanged while returning accurate diff_lines.
40 already_current: re-applying the same symbol is idempotent.
41 Symbol absent from working tree is appended at EOF and verifiable.
42 verified=True after clean write.
43 Multi-symbol single invocation applies all independently.
44 Cross-file cherry-pick applies to the correct file.
45
46 Stress
47 50-symbol repo: all cherry-picked in one invocation, all applied.
48 Large file (1 000 lines): only target symbol lines change.
49 Repeated idempotent cherry-pick: outcome stable, no file growth.
50 """
51
52 from __future__ import annotations
53 from collections.abc import Mapping
54
55 type _FileStore = dict[str, bytes]
56
57 import json
58 import pathlib
59 import textwrap
60 import time
61 from typing import TypedDict
62 from unittest import mock
63
64 import pytest
65 from tests.cli_test_helper import CliRunner
66
67 from muse.cli.commands.semantic_cherry_pick import (
68 ApplyStatus,
69 _PickResult,
70 _apply_symbol,
71 _verify_symbol,
72 )
73 from muse.plugins.code.ast_parser import parse_symbols
74 from muse.core.types import Manifest, long_id
75 from muse.core.object_store import object_path
76 from muse.core.paths import muse_dir
77
78 # ---------------------------------------------------------------------------
79 # Shared CLI runner (accepts any first arg for legacy compatibility)
80 # ---------------------------------------------------------------------------
81
82 runner = CliRunner()
83 cli = None # CliRunner always targets muse.cli.app.main
84
85
86 # ---------------------------------------------------------------------------
87 # TypedDicts — strict JSON schema validation
88 # ---------------------------------------------------------------------------
89
90
91 class _ResultEntry(TypedDict):
92 address: str
93 status: str
94 detail: str
95 old_lines: int
96 new_lines: int
97 diff_lines: list[str]
98 verified: bool
99
100
101 class _CherryPickPayload(TypedDict):
102 schema_version: str
103 branch: str
104 from_commit: str
105 dry_run: bool
106 results: list[_ResultEntry]
107 applied: int
108 already_current: int
109 failed: int
110 unverified: list[str]
111
112
113 # ---------------------------------------------------------------------------
114 # Helpers
115 # ---------------------------------------------------------------------------
116
117
118 def _invoke_json(args: list[str]) -> _CherryPickPayload:
119 result = runner.invoke(cli, ["code", "semantic-cherry-pick"] + args + ["--json"])
120 assert result.exit_code == 0, result.output
121 raw: _CherryPickPayload = json.loads(result.output)
122 return raw
123
124
125 # ---------------------------------------------------------------------------
126 # Fixtures
127 # ---------------------------------------------------------------------------
128
129
130 @pytest.fixture
131 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
132 monkeypatch.chdir(tmp_path)
133 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
134 result = runner.invoke(cli, ["init", "--domain", "code"])
135 assert result.exit_code == 0, result.output
136 return tmp_path
137
138
139 @pytest.fixture
140 def two_commit_repo(repo: pathlib.Path) -> tuple[pathlib.Path, str, str, str]:
141 """Repo with two commits with different implementations of compute().
142
143 commit 1 (HEAD~1): compute returns sum(items)
144 commit 2 (HEAD): compute returns sum(items) * 2
145 Returns (root, address, file_rel, HEAD~1_short).
146 """
147 (repo / "billing.py").write_text(textwrap.dedent("""\
148 def header():
149 return "billing"
150
151
152 def compute(items):
153 return sum(items)
154
155
156 def footer():
157 return "end"
158 """))
159 runner.invoke(cli, ["code", "add", "billing.py"])
160 r1 = runner.invoke(cli, ["commit", "-m", "v1"])
161 assert r1.exit_code == 0, r1.output
162
163 (repo / "billing.py").write_text(textwrap.dedent("""\
164 def header():
165 return "billing"
166
167
168 def compute(items):
169 return sum(items) * 2
170
171
172 def footer():
173 return "end"
174 """))
175 runner.invoke(cli, ["code", "add", "billing.py"])
176 r2 = runner.invoke(cli, ["commit", "-m", "v2"])
177 assert r2.exit_code == 0, r2.output
178
179 log_out = runner.invoke(cli, ["log", "--json"])
180 commits: list[Mapping[str, str]] = json.loads(log_out.output)["commits"]
181 head_minus_1 = commits[1]["commit_id"][:8]
182
183 return repo, "billing.py::compute", "billing.py", head_minus_1
184
185
186 @pytest.fixture
187 def multi_file_repo(repo: pathlib.Path) -> pathlib.Path:
188 """Repo with two files, each containing two functions.
189
190 Useful for cross-file and same-file multi-symbol tests.
191 """
192 (repo / "auth.py").write_text(textwrap.dedent("""\
193 def validate_token(tok):
194 return tok == "secret"
195
196
197 def refresh_token(tok):
198 return tok + "_refreshed"
199 """))
200 (repo / "billing.py").write_text(textwrap.dedent("""\
201 def compute(items):
202 return sum(items)
203
204
205 def discount(items):
206 return sum(items) * 0.9
207 """))
208 runner.invoke(cli, ["code", "add", "."])
209 r1 = runner.invoke(cli, ["commit", "-m", "v1"])
210 assert r1.exit_code == 0, r1.output
211
212 # v2 — both files change
213 (repo / "auth.py").write_text(textwrap.dedent("""\
214 def validate_nonce(nonce):
215 return len(nonce) == 64
216
217
218 def refresh_nonce(nonce):
219 return nonce + "_v2"
220 """))
221 (repo / "billing.py").write_text(textwrap.dedent("""\
222 def compute(items):
223 return sum(items) * 2
224
225
226 def discount(items):
227 return sum(items) * 0.8
228 """))
229 runner.invoke(cli, ["code", "add", "."])
230 r2 = runner.invoke(cli, ["commit", "-m", "v2"])
231 assert r2.exit_code == 0, r2.output
232 return repo
233
234
235 # ---------------------------------------------------------------------------
236 # Unit — _verify_symbol
237 # ---------------------------------------------------------------------------
238
239
240 class TestVerifySymbol:
241 def test_valid_symbol_returns_true(self, tmp_path: pathlib.Path) -> None:
242 f = tmp_path / "m.py"
243 f.write_text("def foo():\n return 1\n")
244 assert _verify_symbol(f, "m.py", "m.py::foo") is True
245
246 def test_missing_symbol_returns_false(self, tmp_path: pathlib.Path) -> None:
247 f = tmp_path / "m.py"
248 f.write_text("def bar():\n return 2\n")
249 assert _verify_symbol(f, "m.py", "m.py::foo") is False
250
251 def test_syntax_error_returns_false(self, tmp_path: pathlib.Path) -> None:
252 f = tmp_path / "m.py"
253 f.write_bytes(b"def broken(:\n pass\n")
254 assert _verify_symbol(f, "m.py", "m.py::broken") is False
255
256 def test_missing_file_returns_false(self, tmp_path: pathlib.Path) -> None:
257 f = tmp_path / "nonexistent.py"
258 assert _verify_symbol(f, "nonexistent.py", "nonexistent.py::x") is False
259
260 def test_empty_file_returns_false(self, tmp_path: pathlib.Path) -> None:
261 f = tmp_path / "empty.py"
262 f.write_text("")
263 assert _verify_symbol(f, "empty.py", "empty.py::anything") is False
264
265
266 # ---------------------------------------------------------------------------
267 # Unit — _apply_symbol
268 # ---------------------------------------------------------------------------
269
270
271 class TestApplySymbol:
272 def _manifest_with_blob(
273 self, root: pathlib.Path, file_rel: str, content: bytes
274 ) -> tuple[Mapping[str, str], Mapping[str, bytes]]:
275 """Create a synthetic manifest entry by writing a blob to object store."""
276 from muse.core.types import blob_id
277 from muse.core.object_store import write_object as _wo
278 oid = blob_id(content)
279 _wo(root, oid, content)
280 manifest: Manifest = {file_rel: oid}
281 src_cache: _FileStore = {}
282 return manifest, src_cache
283
284 def test_no_separator_returns_not_found(self, tmp_path: pathlib.Path) -> None:
285 result = _apply_symbol(tmp_path, "nocolon", {}, False, {})
286 assert result.status == "not_found"
287 assert "separator" in result.detail
288
289 def test_path_traversal_returns_not_found(self, tmp_path: pathlib.Path) -> None:
290 muse_dir(tmp_path).mkdir()
291 result = _apply_symbol(tmp_path, "../../etc/shadow::root", {}, False, {})
292 assert result.status == "not_found"
293
294 def test_file_not_in_manifest(self, tmp_path: pathlib.Path) -> None:
295 muse_dir(tmp_path).mkdir()
296 result = _apply_symbol(tmp_path, "missing.py::func", {}, False, {})
297 assert result.status == "file_missing"
298 assert "not in source snapshot" in result.detail
299
300 def test_blob_missing_from_object_store(self, tmp_path: pathlib.Path) -> None:
301 muse_dir(tmp_path).mkdir()
302 manifest: Manifest = {"src.py": long_id("a" * 64)}
303 result = _apply_symbol(tmp_path, "src.py::func", manifest, False, {})
304 assert result.status == "file_missing"
305 assert "missing from object store" in result.detail
306
307 def test_source_parse_error(self, tmp_path: pathlib.Path) -> None:
308 """parse_error is returned when parse_symbols raises (e.g. for non-Python files).
309
310 parse_symbols silently returns {} for bad Python (catches SyntaxError
311 internally), so parse_error is only reachable when parse_symbols itself
312 raises — e.g. due to an unsupported file type or an internal adapter bug.
313 We test this via mocking to verify the error handling path in _apply_symbol.
314 """
315 muse_dir(tmp_path).mkdir()
316 content = b"def foo():\n pass\n"
317 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", content)
318 with mock.patch(
319 "muse.cli.commands.semantic_cherry_pick.parse_symbols",
320 side_effect=RuntimeError("adapter exploded"),
321 ):
322 result = _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
323 assert result.status == "parse_error"
324
325 def test_symbol_not_in_source(self, tmp_path: pathlib.Path) -> None:
326 muse_dir(tmp_path).mkdir()
327 content = b"def other():\n pass\n"
328 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", content)
329 result = _apply_symbol(tmp_path, "src.py::missing_sym", manifest, False, src_cache)
330 assert result.status == "not_found"
331 assert "not found in source commit" in result.detail
332
333 def test_already_current_returns_correct_status(self, tmp_path: pathlib.Path) -> None:
334 muse_dir(tmp_path).mkdir()
335 content = b"def foo():\n return 1\n"
336 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", content)
337 (tmp_path / "src.py").write_bytes(content)
338 result = _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
339 assert result.status == "already_current"
340 assert result.old_lines == 0
341 assert result.new_lines == 0
342
343 def test_applied_replace_writes_new_body(self, tmp_path: pathlib.Path) -> None:
344 muse_dir(tmp_path).mkdir()
345 src_content = b"def foo():\n return 42\n"
346 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", src_content)
347 (tmp_path / "src.py").write_text("def foo():\n return 1\n\ndef bar():\n pass\n")
348 result = _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
349 assert result.status == "applied"
350 text = (tmp_path / "src.py").read_text()
351 assert "return 42" in text
352 assert "bar" in text # surrounding code preserved
353
354 def test_applied_append_when_symbol_missing_in_current(self, tmp_path: pathlib.Path) -> None:
355 muse_dir(tmp_path).mkdir()
356 src_content = b"def new_func():\n return 99\n"
357 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", src_content)
358 (tmp_path / "src.py").write_text("def existing():\n pass\n")
359 result = _apply_symbol(tmp_path, "src.py::new_func", manifest, False, src_cache)
360 assert result.status == "applied"
361 assert "appended" in result.detail
362 text = (tmp_path / "src.py").read_text()
363 assert "new_func" in text
364 assert "existing" in text
365
366 def test_creates_new_file_when_target_absent(self, tmp_path: pathlib.Path) -> None:
367 muse_dir(tmp_path).mkdir()
368 src_content = b"def fresh():\n return 0\n"
369 manifest, src_cache = self._manifest_with_blob(tmp_path, "new_module.py", src_content)
370 result = _apply_symbol(tmp_path, "new_module.py::fresh", manifest, False, src_cache)
371 assert result.status == "applied"
372 assert "created file" in result.detail
373 assert (tmp_path / "new_module.py").exists()
374
375 def test_dry_run_does_not_write(self, tmp_path: pathlib.Path) -> None:
376 muse_dir(tmp_path).mkdir()
377 src_content = b"def foo():\n return 42\n"
378 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", src_content)
379 (tmp_path / "src.py").write_text("def foo():\n return 1\n")
380 _apply_symbol(tmp_path, "src.py::foo", manifest, True, src_cache)
381 assert "return 1" in (tmp_path / "src.py").read_text()
382
383 def test_dry_run_new_file_not_created(self, tmp_path: pathlib.Path) -> None:
384 muse_dir(tmp_path).mkdir()
385 src_content = b"def ghost():\n pass\n"
386 manifest, src_cache = self._manifest_with_blob(tmp_path, "ghost.py", src_content)
387 _apply_symbol(tmp_path, "ghost.py::ghost", manifest, True, src_cache)
388 assert not (tmp_path / "ghost.py").exists()
389
390 def test_diff_lines_populated_on_replace(self, tmp_path: pathlib.Path) -> None:
391 muse_dir(tmp_path).mkdir()
392 src_content = b"def foo():\n return 42\n"
393 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", src_content)
394 (tmp_path / "src.py").write_text("def foo():\n return 1\n")
395 result = _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
396 assert result.status == "applied"
397 assert len(result.diff_lines) > 0
398 diff_text = "\n".join(result.diff_lines)
399 assert "-" in diff_text
400 assert "+" in diff_text
401
402 def test_diff_lines_empty_when_already_current(self, tmp_path: pathlib.Path) -> None:
403 muse_dir(tmp_path).mkdir()
404 content = b"def foo():\n return 1\n"
405 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", content)
406 (tmp_path / "src.py").write_bytes(content)
407 result = _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
408 assert result.diff_lines == []
409
410 def test_verified_true_on_clean_write(self, tmp_path: pathlib.Path) -> None:
411 muse_dir(tmp_path).mkdir()
412 src_content = b"def foo():\n return 42\n"
413 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", src_content)
414 (tmp_path / "src.py").write_text("def foo():\n return 1\n")
415 result = _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
416 assert result.verified is True
417
418 def test_src_cache_prevents_double_fetch(self, tmp_path: pathlib.Path) -> None:
419 """Same blob ID requested twice must be fetched only once."""
420 muse_dir(tmp_path).mkdir()
421 content = b"def foo():\n return 1\ndef bar():\n return 2\n"
422 manifest, src_cache = self._manifest_with_blob(tmp_path, "src.py", content)
423 (tmp_path / "src.py").write_bytes(content)
424
425 call_count = 0
426 original_read = __import__(
427 "muse.core.object_store", fromlist=["read_object"]
428 ).read_object
429
430 def counting_read(root: pathlib.Path, obj_id: str) -> bytes | None:
431 nonlocal call_count
432 call_count += 1
433 result: bytes | None = original_read(root, obj_id)
434 return result
435
436 with mock.patch(
437 "muse.cli.commands.semantic_cherry_pick.read_object", side_effect=counting_read
438 ):
439 _apply_symbol(tmp_path, "src.py::foo", manifest, False, src_cache)
440 _apply_symbol(tmp_path, "src.py::bar", manifest, False, src_cache)
441
442 assert call_count == 1, "Same blob should be fetched only once across calls"
443
444
445 # ---------------------------------------------------------------------------
446 # Integration — CLI runner tests
447 # ---------------------------------------------------------------------------
448
449
450 class TestCherryPickCLI:
451 def test_exit_zero_on_valid_pick(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
452 _, address, _, head_m1 = two_commit_repo
453 result = runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
454 assert result.exit_code == 0
455
456 def test_json_top_level_keys(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
457 _, address, _, head_m1 = two_commit_repo
458 data = _invoke_json([address, "--from", "HEAD~1"])
459 for key in ("schema", "branch", "from_commit", "dry_run", "results",
460 "applied", "already_current", "failed", "unverified"):
461 assert key in data, f"Missing top-level key: {key}"
462
463 def test_json_from_commit_is_full_id(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
464 _, address, _, _ = two_commit_repo
465 data = _invoke_json([address, "--from", "HEAD~1"])
466 assert data["from_commit"].startswith("sha256:")
467 assert len(data["from_commit"]) == 71
468
469 def test_json_result_keys(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
470 _, address, _, _ = two_commit_repo
471 data = _invoke_json([address, "--from", "HEAD~1"])
472 assert len(data["results"]) == 1
473 r = data["results"][0]
474 for key in ("address", "status", "detail", "old_lines", "new_lines", "diff_lines", "verified"):
475 assert key in r, f"Missing result key: {key}"
476
477 def test_json_applied_count(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
478 _, address, _, _ = two_commit_repo
479 data = _invoke_json([address, "--from", "HEAD~1"])
480 assert data["applied"] == 1
481 assert data["failed"] == 0
482
483 def test_json_dry_run_flag(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
484 _, address, _, _ = two_commit_repo
485 result = runner.invoke(
486 cli,
487 ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--dry-run", "--json"],
488 )
489 assert result.exit_code == 0
490 data: _CherryPickPayload = json.loads(result.output)
491 assert data["dry_run"] is True
492
493 def test_dry_run_does_not_write(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
494 root, address, file_rel, _ = two_commit_repo
495 before = (root / file_rel).read_text()
496 runner.invoke(
497 cli,
498 ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--dry-run"],
499 )
500 assert (root / file_rel).read_text() == before
501
502 def test_dry_run_diff_lines_in_json(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
503 _, address, _, _ = two_commit_repo
504 result = runner.invoke(
505 cli,
506 ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--dry-run", "--json"],
507 )
508 data: _CherryPickPayload = json.loads(result.output)
509 r = data["results"][0]
510 assert isinstance(r["diff_lines"], list)
511 assert len(r["diff_lines"]) > 0
512
513 def test_dry_run_verified_true_for_valid(
514 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
515 ) -> None:
516 _, address, _, _ = two_commit_repo
517 result = runner.invoke(
518 cli,
519 ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--dry-run", "--json"],
520 )
521 data: _CherryPickPayload = json.loads(result.output)
522 assert data["results"][0]["verified"] is True
523
524 def test_already_current_on_same_commit(
525 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
526 ) -> None:
527 _, address, _, _ = two_commit_repo
528 data = _invoke_json([address, "--from", "HEAD"])
529 assert data["results"][0]["status"] == "already_current"
530 assert data["already_current"] == 1
531
532 def test_no_separator_address_is_not_found(
533 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
534 ) -> None:
535 _, _, _, _ = two_commit_repo
536 data = _invoke_json(["noseparator", "--from", "HEAD~1"])
537 assert data["results"][0]["status"] == "not_found"
538 assert data["failed"] == 1
539
540 def test_path_traversal_is_not_found(
541 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
542 ) -> None:
543 data = _invoke_json(["../../etc/shadow::passwd", "--from", "HEAD~1"])
544 assert data["results"][0]["status"] == "not_found"
545 assert data["failed"] == 1
546
547 def test_unknown_from_ref_exits_nonzero(
548 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
549 ) -> None:
550 _, address, _, _ = two_commit_repo
551 result = runner.invoke(
552 cli, ["code", "semantic-cherry-pick", address, "--from", "nonexistent-ref"]
553 )
554 assert result.exit_code != 0
555
556 def test_symbol_not_in_source_is_not_found(
557 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
558 ) -> None:
559 _, _, _, _ = two_commit_repo
560 data = _invoke_json(["billing.py::ghost_func", "--from", "HEAD~1"])
561 assert data["results"][0]["status"] == "not_found"
562
563 def test_file_not_in_snapshot_is_file_missing(
564 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
565 ) -> None:
566 data = _invoke_json(["nonexistent_file.py::func", "--from", "HEAD~1"])
567 assert data["results"][0]["status"] == "file_missing"
568
569 def test_multiple_addresses_all_in_results(self, multi_file_repo: pathlib.Path) -> None:
570 data = _invoke_json(
571 ["auth.py::validate_token", "auth.py::refresh_token", "--from", "HEAD~1"]
572 )
573 assert len(data["results"]) == 2
574 addrs = {r["address"] for r in data["results"]}
575 assert "auth.py::validate_token" in addrs
576 assert "auth.py::refresh_token" in addrs
577
578 def test_multiple_addresses_same_file_applied_count(self, multi_file_repo: pathlib.Path) -> None:
579 data = _invoke_json(
580 ["auth.py::validate_token", "auth.py::refresh_token", "--from", "HEAD~1"]
581 )
582 assert data["applied"] == 2
583 assert data["failed"] == 0
584
585 def test_cross_file_addresses_applied(self, multi_file_repo: pathlib.Path) -> None:
586 data = _invoke_json(
587 ["auth.py::validate_token", "billing.py::compute", "--from", "HEAD~1"]
588 )
589 assert data["applied"] == 2
590 assert data["failed"] == 0
591
592 def test_text_output_contains_commit_hash(
593 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
594 ) -> None:
595 _, address, _, head_m1 = two_commit_repo
596 result = runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
597 assert head_m1 in result.output
598
599 def test_text_output_counts(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
600 _, address, _, _ = two_commit_repo
601 result = runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
602 assert "1 applied" in result.output
603 assert "0 failed" in result.output
604
605 def test_missing_repo_exits_nonzero(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
606 monkeypatch.chdir(tmp_path)
607 result = runner.invoke(
608 cli, ["code", "semantic-cherry-pick", "a.py::foo", "--from", "HEAD~1"]
609 )
610 assert result.exit_code != 0
611
612 def test_unverified_populated_when_verify_fails(
613 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
614 ) -> None:
615 _, address, _, _ = two_commit_repo
616 with mock.patch(
617 "muse.cli.commands.semantic_cherry_pick._verify_symbol", return_value=False
618 ):
619 result = runner.invoke(
620 cli,
621 ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--json"],
622 )
623 data: _CherryPickPayload = json.loads(result.output)
624 assert address in data["unverified"]
625 assert data["results"][0]["verified"] is False
626
627
628 # ---------------------------------------------------------------------------
629 # E2E — real symbol diffs across commits
630 # ---------------------------------------------------------------------------
631
632
633 class TestCherryPickE2E:
634 def test_applied_replaces_only_target_lines(
635 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
636 ) -> None:
637 root, address, file_rel, _ = two_commit_repo
638 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
639 text = (root / file_rel).read_text()
640 # Old implementation restored
641 assert "return sum(items)" in text
642 assert "* 2" not in text
643 # Surrounding functions untouched
644 assert "def header" in text
645 assert "def footer" in text
646
647 def test_applied_from_earlier_commit_restores_old_impl(
648 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
649 ) -> None:
650 root, address, file_rel, _ = two_commit_repo
651 before_pick = (root / file_rel).read_text()
652 assert "* 2" in before_pick # HEAD has the * 2 version
653 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
654 after_pick = (root / file_rel).read_text()
655 assert "* 2" not in after_pick
656
657 def test_dry_run_leaves_file_unchanged(
658 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
659 ) -> None:
660 root, address, file_rel, _ = two_commit_repo
661 original = (root / file_rel).read_bytes()
662 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--dry-run"])
663 assert (root / file_rel).read_bytes() == original
664
665 def test_dry_run_diff_lines_accurate(
666 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
667 ) -> None:
668 _, address, _, _ = two_commit_repo
669 result = runner.invoke(
670 cli,
671 ["code", "semantic-cherry-pick", address, "--from", "HEAD~1", "--dry-run", "--json"],
672 )
673 data: _CherryPickPayload = json.loads(result.output)
674 diff_text = "\n".join(data["results"][0]["diff_lines"])
675 # Should remove the * 2 line and restore the plain sum
676 assert "sum(items)" in diff_text
677
678 def test_already_current_is_idempotent(
679 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
680 ) -> None:
681 root, address, file_rel, _ = two_commit_repo
682 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
683 text_after_first = (root / file_rel).read_text()
684 # Apply again — must be a no-op
685 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
686 assert (root / file_rel).read_text() == text_after_first
687
688 def test_second_apply_is_already_current(
689 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
690 ) -> None:
691 _, address, _, _ = two_commit_repo
692 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
693 data = _invoke_json([address, "--from", "HEAD~1"])
694 assert data["results"][0]["status"] == "already_current"
695
696 def test_appended_symbol_parseable(self, two_commit_repo: tuple[pathlib.Path, str, str, str]) -> None:
697 root, _, _, _ = two_commit_repo
698 # billing.py doesn't have 'header2' in source; cherry-pick should append
699 # Use a symbol that exists in source (HEAD~1) but was removed in HEAD.
700 # Add a new symbol in commit 3 to simulate absence in working tree.
701 (root / "utils.py").write_text("def helper():\n return True\n")
702 runner.invoke(cli, ["code", "add", "utils.py"])
703 runner.invoke(cli, ["commit", "-m", "v3"])
704 # Working tree no longer has utils.py (overwrite with something else)
705 (root / "utils.py").write_text("def other():\n return False\n")
706 runner.invoke(cli, ["code", "add", "utils.py"])
707 runner.invoke(cli, ["commit", "-m", "v4"])
708 # Cherry-pick helper from v3 (HEAD~1 now)
709 runner.invoke(cli, ["code", "semantic-cherry-pick", "utils.py::helper", "--from", "HEAD~1"])
710 raw = (root / "utils.py").read_bytes()
711 tree = parse_symbols(raw, "utils.py")
712 assert "utils.py::helper" in tree
713
714 def test_verified_true_after_clean_write(
715 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
716 ) -> None:
717 _, address, _, _ = two_commit_repo
718 data = _invoke_json([address, "--from", "HEAD~1"])
719 assert data["results"][0]["verified"] is True
720
721 def test_multi_symbol_same_file_applies_all(self, multi_file_repo: pathlib.Path) -> None:
722 root = multi_file_repo
723 runner.invoke(
724 cli,
725 ["code", "semantic-cherry-pick", "auth.py::validate_token", "auth.py::refresh_token",
726 "--from", "HEAD~1"],
727 )
728 text = (root / "auth.py").read_text()
729 # Old implementations restored
730 assert 'tok == "secret"' in text
731 assert '"_refreshed"' in text
732
733 def test_cross_file_cherry_pick_correct_files(self, multi_file_repo: pathlib.Path) -> None:
734 root = multi_file_repo
735 runner.invoke(
736 cli,
737 ["code", "semantic-cherry-pick",
738 "auth.py::validate_token", "billing.py::compute",
739 "--from", "HEAD~1"],
740 )
741 auth_text = (root / "auth.py").read_text()
742 bill_text = (root / "billing.py").read_text()
743 assert 'tok == "secret"' in auth_text
744 assert "return sum(items)" in bill_text
745 assert "* 2" not in bill_text
746
747
748 # ---------------------------------------------------------------------------
749 # E2E — regression: all results returned even on mixed success/failure
750 # ---------------------------------------------------------------------------
751
752
753 class TestCherryPickMixedResults:
754 def test_failure_does_not_stop_remaining_addresses(
755 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
756 ) -> None:
757 """All symbols processed; failure in one doesn't skip subsequent ones."""
758 _, _, _, _ = two_commit_repo
759 data = _invoke_json([
760 "billing.py::ghost_func", # not_found
761 "billing.py::compute", # applied
762 "--from", "HEAD~1",
763 ])
764 statuses = {r["address"]: r["status"] for r in data["results"]}
765 assert statuses["billing.py::ghost_func"] == "not_found"
766 assert statuses["billing.py::compute"] == "applied"
767 assert data["applied"] == 1
768 assert data["failed"] == 1
769
770 def test_mixed_results_counts_accurate(
771 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
772 ) -> None:
773 data = _invoke_json([
774 "billing.py::compute", # applied
775 "billing.py::ghost", # not_found
776 "missing_file.py::func", # file_missing
777 "--from", "HEAD~1",
778 ])
779 assert data["applied"] == 1
780 assert data["failed"] == 2
781
782
783 # ---------------------------------------------------------------------------
784 # Stress
785 # ---------------------------------------------------------------------------
786
787
788 class TestCherryPickStress:
789 def test_many_symbols_all_applied(self, repo: pathlib.Path) -> None:
790 """50 distinct functions, all cherry-picked in one invocation."""
791 n = 50
792 funcs = "\n\n".join(f"def func_{i}():\n return {i}" for i in range(n))
793 (repo / "big.py").write_text(f"{funcs}\n")
794 runner.invoke(cli, ["code", "add", "big.py"])
795 r1 = runner.invoke(cli, ["commit", "-m", "v1"])
796 assert r1.exit_code == 0
797
798 new_funcs = "\n\n".join(f"def func_{i}():\n return {i * 10}" for i in range(n))
799 (repo / "big.py").write_text(f"{new_funcs}\n")
800 runner.invoke(cli, ["code", "add", "big.py"])
801 r2 = runner.invoke(cli, ["commit", "-m", "v2"])
802 assert r2.exit_code == 0
803
804 addresses = [f"big.py::func_{i}" for i in range(n)]
805 result = runner.invoke(
806 cli,
807 ["code", "semantic-cherry-pick"] + addresses + ["--from", "HEAD~1", "--json"],
808 )
809 assert result.exit_code == 0
810 data: _CherryPickPayload = json.loads(result.output)
811 # Every symbol should be applied or already_current (no failures)
812 assert data["failed"] == 0
813 assert data["applied"] + data["already_current"] == n
814
815 def test_large_file_only_target_lines_change(self, repo: pathlib.Path) -> None:
816 """1 000-line file: cherry-pick changes exactly target symbol, nothing else."""
817 header = "def noop():\n pass\n\n"
818 target_v1 = "def target():\n return 'v1'\n\n"
819 footer_lines = "".join(f"def pad_{i}():\n pass\n\n" for i in range(100))
820
821 (repo / "large.py").write_text(header + target_v1 + footer_lines)
822 runner.invoke(cli, ["code", "add", "large.py"])
823 runner.invoke(cli, ["commit", "-m", "v1"])
824
825 target_v2 = "def target():\n return 'v2'\n\n"
826 (repo / "large.py").write_text(header + target_v2 + footer_lines)
827 runner.invoke(cli, ["code", "add", "large.py"])
828 runner.invoke(cli, ["commit", "-m", "v2"])
829
830 runner.invoke(cli, ["code", "semantic-cherry-pick", "large.py::target", "--from", "HEAD~1"])
831 text = (repo / "large.py").read_text()
832 assert "'v1'" in text
833 assert "'v2'" not in text
834 for i in range(100):
835 assert f"def pad_{i}" in text
836
837 def test_repeated_cherry_pick_is_idempotent(
838 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
839 ) -> None:
840 root, address, file_rel, _ = two_commit_repo
841 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
842 text_1 = (root / file_rel).read_text()
843 for _ in range(5):
844 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
845 assert (root / file_rel).read_text() == text_1
846
847 def test_repeated_cherry_pick_is_fast(
848 self, two_commit_repo: tuple[pathlib.Path, str, str, str]
849 ) -> None:
850 """Idempotent cherry-picks should complete well within 2 seconds each."""
851 _, address, _, _ = two_commit_repo
852 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
853 start = time.monotonic()
854 for _ in range(10):
855 runner.invoke(cli, ["code", "semantic-cherry-pick", address, "--from", "HEAD~1"])
856 elapsed = time.monotonic() - start
857 assert elapsed < 20.0, f"10 idempotent cherry-picks took {elapsed:.1f}s — too slow"
858
859 def test_src_cache_scales_with_many_same_file_addresses(
860 self, repo: pathlib.Path
861 ) -> None:
862 """Blob fetch count stays 1 regardless of how many symbols target same file."""
863 n = 20
864 content = "\n\n".join(f"def sym_{i}():\n return {i}" for i in range(n))
865 (repo / "cache_test.py").write_text(f"{content}\n")
866 runner.invoke(cli, ["code", "add", "cache_test.py"])
867 runner.invoke(cli, ["commit", "-m", "v1"])
868
869 # Mutate so all symbols differ
870 content_v2 = "\n\n".join(f"def sym_{i}():\n return {i * 100}" for i in range(n))
871 (repo / "cache_test.py").write_text(f"{content_v2}\n")
872 runner.invoke(cli, ["code", "add", "cache_test.py"])
873 runner.invoke(cli, ["commit", "-m", "v2"])
874
875 call_count = 0
876 original_read = __import__(
877 "muse.core.object_store", fromlist=["read_object"]
878 ).read_object
879
880 def counting_read(r: pathlib.Path, obj_id: str) -> bytes | None:
881 nonlocal call_count
882 call_count += 1
883 fetched: bytes | None = original_read(r, obj_id)
884 return fetched
885
886 addresses = [f"cache_test.py::sym_{i}" for i in range(n)]
887 with mock.patch(
888 "muse.cli.commands.semantic_cherry_pick.read_object", side_effect=counting_read
889 ):
890 result = runner.invoke(
891 cli,
892 ["code", "semantic-cherry-pick"] + addresses + ["--from", "HEAD~1", "--json"],
893 )
894 assert result.exit_code == 0
895 # The source blob for cache_test.py must be fetched exactly once
896 assert call_count == 1, (
897 f"Expected 1 blob fetch for {n} symbols in the same file; got {call_count}"
898 )
File History 1 commit
sha256:b89fa4fd9ca0d692fc66f6b9aef4c3a0c13c8e9b439faf42da8e91e09f048d4f tests/test_cmd_revert_hardening.py, tests/test_cmd_semantic… Human 4 days ago