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