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