test_cmd_plan_merge.py
python
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠ breaking
122 days ago
| 1 | """Tests for ``muse coord plan-merge``. |
| 2 | |
| 3 | Coverage matrix |
| 4 | --------------- |
| 5 | Unit |
| 6 | ~~~~ |
| 7 | * :func:`_classify_change` — all six change classifications |
| 8 | * :func:`_classify_conflict` — all three-way matrix cells |
| 9 | (base/ours/theirs combinations: 8 cases + rename, symbol_edit_overlap) |
| 10 | * :func:`_find_renames_and_moves` — rename vs move discrimination, trivial body skipped |
| 11 | * :func:`_find_delete_use_conflicts` — new callers detected, call graph errors handled |
| 12 | * :func:`_find_dependency_conflicts` — transitive dependency detection, errors handled |
| 13 | * :class:`_MergeItem` — to_dict, slots |
| 14 | |
| 15 | Integration (mock-based — no real commits required) |
| 16 | ~~~~~~~~~~~ |
| 17 | * ``plan-merge OURS THEIRS`` — ref not found → exit 1 |
| 18 | * ``plan-merge OURS THEIRS`` — theirs ref not found → exit 1 |
| 19 | * ``plan-merge OURS THEIRS --base MISSING`` — base ref not found → exit 1 |
| 20 | * ``plan-merge OURS THEIRS`` — base auto-computed |
| 21 | * ``plan-merge OURS THEIRS --base BASE_REF`` |
| 22 | * ``plan-merge OURS THEIRS --skip-call-graph`` |
| 23 | * ``plan-merge OURS THEIRS --format json`` — schema complete |
| 24 | * ``plan-merge OURS THEIRS --json`` — shorthand |
| 25 | * JSON: ``base_auto_computed``, ``call_graph_available``, ``warnings``, |
| 26 | ``duration_ms``, full commit IDs, ``conflicts_by_type`` breakdown |
| 27 | * JSON output is compact (no indent — single line) |
| 28 | * Text output: conflict summary, warnings visible, elapsed present |
| 29 | * ``symbol_edit_overlap`` detected (both changed, content differs) |
| 30 | * ``rename_edit`` detected via body_hash matching |
| 31 | * ``move_edit`` detected via body_hash + file prefix mismatch |
| 32 | * ``delete_use`` detected via forward call graph |
| 33 | * ``dependency_conflict`` detected via reverse call graph |
| 34 | * ``no_conflict`` for unilateral changes (three-way correctness) |
| 35 | * call_graph unavailable → warning, not crash |
| 36 | * unexpected call graph exception propagates |
| 37 | |
| 38 | Error shapes |
| 39 | ~~~~~~~~~~~~ |
| 40 | * JSON error has ``{"error": ..., "status": "error"}`` |
| 41 | * Text error uses ``❌`` prefix on stderr |
| 42 | * theirs-ref not-found JSON error shape |
| 43 | * base-ref not-found JSON error shape |
| 44 | |
| 45 | Security |
| 46 | ~~~~~~~~ |
| 47 | * ANSI sequences in ref names, addresses, change descriptions stripped |
| 48 | * Path traversal in ref args → resolve_commit_ref handles it (no FS access) |
| 49 | |
| 50 | Stress |
| 51 | ~~~~~~ |
| 52 | * 1000 symbols across ours + theirs → Pass 1 in < 2 s |
| 53 | * 200 renames → Pass 2 in < 1 s |
| 54 | * delete_use with 100 deleted symbols → < 2 s |
| 55 | * conflicts_by_type counts correct with mixed conflict types at scale |
| 56 | |
| 57 | E2E |
| 58 | ~~~ |
| 59 | * Same commit for ours and theirs → 0 conflicts |
| 60 | * Three distinct conflict types in one plan → all appear in conflicts_by_type |
| 61 | """ |
| 62 | |
| 63 | from __future__ import annotations |
| 64 | |
| 65 | import argparse |
| 66 | import json |
| 67 | import os |
| 68 | import pathlib |
| 69 | import sys |
| 70 | import time |
| 71 | from unittest.mock import MagicMock, patch |
| 72 | |
| 73 | import pytest |
| 74 | |
| 75 | from muse.core.types import Manifest, fake_id |
| 76 | from muse.core.paths import muse_dir |
| 77 | from muse.plugins.code.ast_parser import SymbolRecord, SymbolTree |
| 78 | type SymbolMap = dict[str, SymbolRecord] |
| 79 | type SymbolsByFile = dict[str, SymbolMap] |
| 80 | from muse.plugins.code._callgraph import ForwardGraph |
| 81 | |
| 82 | |
| 83 | # ── Helpers ─────────────────────────────────────────────────────────────────── |
| 84 | |
| 85 | |
| 86 | def _sym( |
| 87 | name: str, |
| 88 | content_id: str | None = None, |
| 89 | body_hash: str | None = None, |
| 90 | signature_id: str | None = None, |
| 91 | metadata_id: str = "", |
| 92 | ) -> SymbolRecord: |
| 93 | """Build a minimal SymbolRecord for testing.""" |
| 94 | h = content_id or f"cid-{name}" |
| 95 | b = body_hash or f"bh-{name}" |
| 96 | s = signature_id or f"sid-{name}" |
| 97 | return { |
| 98 | "kind": "function", |
| 99 | "name": name, |
| 100 | "qualified_name": name, |
| 101 | "content_id": h, |
| 102 | "body_hash": b, |
| 103 | "signature_id": s, |
| 104 | "metadata_id": metadata_id, |
| 105 | "canonical_key": f"f.py##{name}", |
| 106 | } |
| 107 | |
| 108 | |
| 109 | def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 110 | dot_muse = muse_dir(tmp_path) |
| 111 | dot_muse.mkdir() |
| 112 | (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") |
| 113 | (dot_muse / "repo.json").write_text( |
| 114 | json.dumps({"repo_id": fake_id("repo"), "name": "test-repo"}) |
| 115 | ) |
| 116 | return tmp_path |
| 117 | |
| 118 | |
| 119 | @pytest.fixture() |
| 120 | def repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 121 | return _make_repo(tmp_path) |
| 122 | |
| 123 | |
| 124 | def _mock_commit(cid: str | None = None) -> MagicMock: |
| 125 | m = MagicMock() |
| 126 | m.commit_id = cid or fake_id("commit") |
| 127 | return m |
| 128 | |
| 129 | |
| 130 | def _run_plan_merge( |
| 131 | repo: pathlib.Path, |
| 132 | ours_ref: str = "HEAD", |
| 133 | theirs_ref: str = "main", |
| 134 | base_ref: str | None = None, |
| 135 | skip_call_graph: bool = True, |
| 136 | fmt: str = "json", |
| 137 | mock_ours_commit: MagicMock | None = None, |
| 138 | mock_theirs_commit: MagicMock | None = None, |
| 139 | mock_base_cid: str | None = None, |
| 140 | mock_ours_syms: SymbolMap | None = None, |
| 141 | mock_theirs_syms: SymbolMap | None = None, |
| 142 | mock_base_syms: SymbolMap | None = None, |
| 143 | ) -> tuple[int, str]: |
| 144 | """Run plan-merge with mocked commits/symbols. Returns (exit_code, stdout).""" |
| 145 | from muse.cli.commands.plan_merge import run as pm_run |
| 146 | |
| 147 | ours_c = mock_ours_commit or _mock_commit(f"aaa{'0' * 61}") |
| 148 | theirs_c = mock_theirs_commit or _mock_commit(f"bbb{'0' * 61}") |
| 149 | base_cid = mock_base_cid or f"ccc{'0' * 61}" |
| 150 | |
| 151 | ours_syms = mock_ours_syms or {} |
| 152 | theirs_syms = mock_theirs_syms or {} |
| 153 | base_syms = mock_base_syms or {} |
| 154 | |
| 155 | def _sym_for_snapshot(root: pathlib.Path, manifest: Manifest, **kwargs: str) -> SymbolsByFile: |
| 156 | if manifest.get("_branch") == "ours": |
| 157 | return {"f.py": ours_syms} |
| 158 | if manifest.get("_branch") == "theirs": |
| 159 | return {"f.py": theirs_syms} |
| 160 | if manifest.get("_branch") == "base": |
| 161 | return {"f.py": base_syms} |
| 162 | return {} |
| 163 | |
| 164 | ns = argparse.Namespace( |
| 165 | ours_ref=ours_ref, |
| 166 | theirs_ref=theirs_ref, |
| 167 | base_ref=base_ref, |
| 168 | skip_call_graph=skip_call_graph, |
| 169 | fmt=fmt, |
| 170 | json_out=(fmt == "json"), |
| 171 | ) |
| 172 | old = os.getcwd() |
| 173 | os.chdir(repo) |
| 174 | import io, sys |
| 175 | captured = io.StringIO() |
| 176 | old_stdout = sys.stdout |
| 177 | sys.stdout = captured |
| 178 | |
| 179 | exit_code = 0 |
| 180 | try: |
| 181 | with ( |
| 182 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 183 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="test-repo"), |
| 184 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 185 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", |
| 186 | side_effect=lambda root, repo_id, branch, ref: ( |
| 187 | ours_c if ref in (ours_ref, None) else |
| 188 | theirs_c if ref == theirs_ref else |
| 189 | _mock_commit(mock_base_cid) if ref == base_ref else None |
| 190 | )), |
| 191 | patch("muse.cli.commands.plan_merge.find_merge_base", |
| 192 | return_value=base_cid), |
| 193 | patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", |
| 194 | side_effect=lambda root, cid: ( |
| 195 | {"_branch": "ours"} if cid == ours_c.commit_id else |
| 196 | {"_branch": "theirs"} if cid == theirs_c.commit_id else |
| 197 | {"_branch": "base"} |
| 198 | )), |
| 199 | patch("muse.cli.commands.plan_merge.symbols_for_snapshot", |
| 200 | side_effect=_sym_for_snapshot), |
| 201 | ): |
| 202 | pm_run(ns) |
| 203 | except SystemExit as exc: |
| 204 | exit_code = exc.code or 0 |
| 205 | finally: |
| 206 | sys.stdout = old_stdout |
| 207 | os.chdir(old) |
| 208 | |
| 209 | return exit_code, captured.getvalue() |
| 210 | |
| 211 | |
| 212 | # ───────────────────────────────────────────────────────────────────────────── |
| 213 | # Unit tests — _classify_change |
| 214 | # ───────────────────────────────────────────────────────────────────────────── |
| 215 | |
| 216 | |
| 217 | class TestClassifyChange: |
| 218 | def test_unchanged(self) -> None: |
| 219 | from muse.cli.commands.plan_merge import _classify_change |
| 220 | s = _sym("fn", content_id="X") |
| 221 | assert _classify_change(s, s) == "unchanged" |
| 222 | |
| 223 | def test_metadata_only(self) -> None: |
| 224 | from muse.cli.commands.plan_merge import _classify_change |
| 225 | base = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG") |
| 226 | target = _sym("fn", content_id="Y", body_hash="BH", signature_id="SIG") |
| 227 | assert _classify_change(base, target) == "metadata_only" |
| 228 | |
| 229 | def test_signature_only(self) -> None: |
| 230 | from muse.cli.commands.plan_merge import _classify_change |
| 231 | base = _sym("fn", body_hash="BH", signature_id="SIG1") |
| 232 | target = _sym("fn", content_id="X2", body_hash="BH", signature_id="SIG2") |
| 233 | assert _classify_change(base, target) == "signature_only" |
| 234 | |
| 235 | def test_impl_only(self) -> None: |
| 236 | from muse.cli.commands.plan_merge import _classify_change |
| 237 | base = _sym("fn", body_hash="BH1", signature_id="SIG") |
| 238 | target = _sym("fn", content_id="X2", body_hash="BH2", signature_id="SIG") |
| 239 | assert _classify_change(base, target) == "impl_only" |
| 240 | |
| 241 | def test_rename_modify(self) -> None: |
| 242 | from muse.cli.commands.plan_merge import _classify_change |
| 243 | base = {**_sym("fn_old"), "name": "fn_old"} |
| 244 | target = {**_sym("fn_new", content_id="X2", body_hash="BH2"), "name": "fn_new"} |
| 245 | assert _classify_change(base, target) == "rename+modify" |
| 246 | |
| 247 | def test_full_rewrite(self) -> None: |
| 248 | from muse.cli.commands.plan_merge import _classify_change |
| 249 | base = _sym("fn") |
| 250 | target = _sym("fn", content_id="X2", body_hash="BH2", signature_id="SIG2") |
| 251 | assert _classify_change(base, target) == "full_rewrite" |
| 252 | |
| 253 | |
| 254 | # ───────────────────────────────────────────────────────────────────────────── |
| 255 | # Unit tests — _classify_conflict (three-way matrix) |
| 256 | # ───────────────────────────────────────────────────────────────────────────── |
| 257 | |
| 258 | |
| 259 | class TestClassifyConflict: |
| 260 | def test_both_absent_no_conflict(self) -> None: |
| 261 | from muse.cli.commands.plan_merge import _classify_conflict |
| 262 | item = _classify_conflict("x.py::fn", None, None, None) |
| 263 | assert item.conflict_type == "no_conflict" |
| 264 | |
| 265 | def test_ours_new_no_base_no_conflict(self) -> None: |
| 266 | from muse.cli.commands.plan_merge import _classify_conflict |
| 267 | item = _classify_conflict("x.py::fn", None, _sym("fn"), None) |
| 268 | assert item.conflict_type == "no_conflict" |
| 269 | assert item.ours_change == "added" |
| 270 | |
| 271 | def test_theirs_new_no_base_no_conflict(self) -> None: |
| 272 | from muse.cli.commands.plan_merge import _classify_conflict |
| 273 | item = _classify_conflict("x.py::fn", None, None, _sym("fn")) |
| 274 | assert item.conflict_type == "no_conflict" |
| 275 | assert item.theirs_change == "added" |
| 276 | |
| 277 | def test_only_ours_changed_three_way(self) -> None: |
| 278 | """base=X, ours=Y, theirs=X → only ours changed → no_conflict.""" |
| 279 | from muse.cli.commands.plan_merge import _classify_conflict |
| 280 | base = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG") |
| 281 | ours = _sym("fn", content_id="Y", body_hash="BH2", signature_id="SIG") |
| 282 | theirs = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG") |
| 283 | item = _classify_conflict("x.py::fn", base, ours, theirs) |
| 284 | assert item.conflict_type == "no_conflict" |
| 285 | assert "fast-forward" in item.recommendation |
| 286 | |
| 287 | def test_only_theirs_changed_three_way(self) -> None: |
| 288 | """base=X, ours=X, theirs=Y → only theirs changed → no_conflict.""" |
| 289 | from muse.cli.commands.plan_merge import _classify_conflict |
| 290 | base = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG") |
| 291 | ours = _sym("fn", content_id="X", body_hash="BH1", signature_id="SIG") |
| 292 | theirs = _sym("fn", content_id="Y", body_hash="BH2", signature_id="SIG") |
| 293 | item = _classify_conflict("x.py::fn", base, ours, theirs) |
| 294 | assert item.conflict_type == "no_conflict" |
| 295 | |
| 296 | def test_both_changed_three_way_overlap(self) -> None: |
| 297 | """base=X, ours=Y, theirs=Z → both changed → symbol_edit_overlap.""" |
| 298 | from muse.cli.commands.plan_merge import _classify_conflict |
| 299 | base = _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG0") |
| 300 | ours = _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG0") |
| 301 | theirs = _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG0") |
| 302 | item = _classify_conflict("x.py::fn", base, ours, theirs) |
| 303 | assert item.conflict_type == "symbol_edit_overlap" |
| 304 | |
| 305 | def test_identical_on_both_no_conflict(self) -> None: |
| 306 | from muse.cli.commands.plan_merge import _classify_conflict |
| 307 | s = _sym("fn", content_id="SAME") |
| 308 | item = _classify_conflict("x.py::fn", _sym("fn"), s, s) |
| 309 | assert item.conflict_type == "no_conflict" |
| 310 | assert "identical" in item.recommendation |
| 311 | |
| 312 | def test_ours_deleted_theirs_unchanged_no_conflict(self) -> None: |
| 313 | """base=X, ours=None, theirs=X (unchanged) → no_conflict.""" |
| 314 | from muse.cli.commands.plan_merge import _classify_conflict |
| 315 | base = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG") |
| 316 | theirs = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG") |
| 317 | item = _classify_conflict("x.py::fn", base, None, theirs) |
| 318 | assert item.conflict_type == "no_conflict" |
| 319 | |
| 320 | def test_ours_deleted_theirs_modified_review(self) -> None: |
| 321 | """base=X, ours=None, theirs=Y → potentially delete_use → review.""" |
| 322 | from muse.cli.commands.plan_merge import _classify_conflict |
| 323 | base = _sym("fn", content_id="X", body_hash="BH", signature_id="SIG") |
| 324 | theirs = _sym("fn", content_id="Y", body_hash="BH2", signature_id="SIG") |
| 325 | item = _classify_conflict("x.py::fn", base, None, theirs) |
| 326 | # delete_use is detected in Pass 3; this is a "review" no_conflict. |
| 327 | assert item.conflict_type == "no_conflict" |
| 328 | assert "review" in item.recommendation |
| 329 | |
| 330 | def test_rename_edit_detected_ours_renamed(self) -> None: |
| 331 | """Ours renamed (same body, different name) + theirs modified → rename_edit.""" |
| 332 | from muse.cli.commands.plan_merge import _classify_conflict |
| 333 | base_rec = {**_sym("fn_old", body_hash="BH", signature_id="SIG"), "name": "fn_old"} |
| 334 | # Ours: same body hash, but name changed (signature_only = rename) |
| 335 | ours_rec = { |
| 336 | **_sym("fn_new", content_id="C2", body_hash="BH", signature_id="SIG2"), |
| 337 | "name": "fn_new", |
| 338 | } |
| 339 | theirs_rec = { |
| 340 | **_sym("fn_old", content_id="C3", body_hash="BH3", signature_id="SIG"), |
| 341 | "name": "fn_old", |
| 342 | } |
| 343 | item = _classify_conflict("x.py::fn_old", base_rec, ours_rec, theirs_rec) |
| 344 | assert item.conflict_type == "rename_edit" |
| 345 | |
| 346 | def test_both_added_same_content_no_conflict(self) -> None: |
| 347 | """No base, both added same symbol → no_conflict.""" |
| 348 | from muse.cli.commands.plan_merge import _classify_conflict |
| 349 | s = _sym("fn", content_id="SAME") |
| 350 | item = _classify_conflict("x.py::fn", None, s, s) |
| 351 | assert item.conflict_type == "no_conflict" |
| 352 | |
| 353 | def test_both_added_different_content_overlap(self) -> None: |
| 354 | """No base, both added different content → symbol_edit_overlap.""" |
| 355 | from muse.cli.commands.plan_merge import _classify_conflict |
| 356 | item = _classify_conflict( |
| 357 | "x.py::fn", None, |
| 358 | _sym("fn", content_id="X"), |
| 359 | _sym("fn", content_id="Y"), |
| 360 | ) |
| 361 | assert item.conflict_type == "symbol_edit_overlap" |
| 362 | |
| 363 | def test_merge_item_slots_and_to_dict(self) -> None: |
| 364 | from muse.cli.commands.plan_merge import _MergeItem |
| 365 | m = _MergeItem("addr", "no_conflict", "a", "b", "rec") |
| 366 | d = m.to_dict() |
| 367 | assert set(d.keys()) == {"address", "conflict_type", "ours_change", "theirs_change", "recommendation"} |
| 368 | assert d["conflict_type"] == "no_conflict" |
| 369 | |
| 370 | |
| 371 | # ───────────────────────────────────────────────────────────────────────────── |
| 372 | # Unit tests — _find_renames_and_moves |
| 373 | # ───────────────────────────────────────────────────────────────────────────── |
| 374 | |
| 375 | |
| 376 | class TestFindRenamesAndMoves: |
| 377 | def test_rename_detected_same_file(self) -> None: |
| 378 | from muse.cli.commands.plan_merge import _find_renames_and_moves |
| 379 | base = {"file.py::fn_old": _sym("fn_old", body_hash="BODYHASH123456")} |
| 380 | branch = {"file.py::fn_new": {**_sym("fn_new", body_hash="BODYHASH123456"), "name": "fn_new"}} |
| 381 | renames, moves = _find_renames_and_moves(base, branch) |
| 382 | assert "file.py::fn_old" in renames |
| 383 | assert renames["file.py::fn_old"] == "file.py::fn_new" |
| 384 | assert not moves |
| 385 | |
| 386 | def test_move_detected_different_file(self) -> None: |
| 387 | from muse.cli.commands.plan_merge import _find_renames_and_moves |
| 388 | base = {"old/file.py::fn": _sym("fn", body_hash="BODYHASH123456")} |
| 389 | branch = {"new/file.py::fn": _sym("fn", body_hash="BODYHASH123456")} |
| 390 | renames, moves = _find_renames_and_moves(base, branch) |
| 391 | assert not renames |
| 392 | assert "old/file.py::fn" in moves |
| 393 | |
| 394 | def test_same_file_preferred_over_other_file(self) -> None: |
| 395 | from muse.cli.commands.plan_merge import _find_renames_and_moves |
| 396 | base = {"file.py::fn_old": _sym("fn_old", body_hash="BODYHASH123456")} |
| 397 | branch = { |
| 398 | "file.py::fn_new": _sym("fn_new", body_hash="BODYHASH123456"), |
| 399 | "other.py::fn_copy": _sym("fn_copy", body_hash="BODYHASH123456"), |
| 400 | } |
| 401 | renames, moves = _find_renames_and_moves(base, branch) |
| 402 | # Same file candidate should win → rename, not move. |
| 403 | assert "file.py::fn_old" in renames |
| 404 | assert "file.py::fn_old" not in moves |
| 405 | |
| 406 | def test_trivial_body_hash_skipped(self) -> None: |
| 407 | """Very short body hashes are skipped to avoid false positives.""" |
| 408 | from muse.cli.commands.plan_merge import _find_renames_and_moves |
| 409 | base = {"file.py::fn": _sym("fn", body_hash="X")} # Too short. |
| 410 | branch = {"file.py::fn_new": _sym("fn_new", body_hash="X")} |
| 411 | renames, moves = _find_renames_and_moves(base, branch) |
| 412 | assert not renames |
| 413 | assert not moves |
| 414 | |
| 415 | def test_still_present_not_a_rename(self) -> None: |
| 416 | from muse.cli.commands.plan_merge import _find_renames_and_moves |
| 417 | sym = _sym("fn", body_hash="BODY1234567890") |
| 418 | base = {"file.py::fn": sym} |
| 419 | branch = {"file.py::fn": sym, "file.py::fn2": _sym("fn2", body_hash="BODY1234567890")} |
| 420 | renames, moves = _find_renames_and_moves(base, branch) |
| 421 | # Original still present → not a rename. |
| 422 | assert "file.py::fn" not in renames |
| 423 | |
| 424 | |
| 425 | # ───────────────────────────────────────────────────────────────────────────── |
| 426 | # Unit tests — _find_delete_use_conflicts |
| 427 | # ───────────────────────────────────────────────────────────────────────────── |
| 428 | |
| 429 | |
| 430 | class TestFindDeleteUseConflicts: |
| 431 | def _make_manifests(self) -> tuple[Manifest, Manifest, Manifest]: |
| 432 | return {"base": "m"}, {"ours": "m"}, {"theirs": "m"} |
| 433 | |
| 434 | def test_delete_use_detected(self, repo: pathlib.Path) -> None: |
| 435 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 436 | base_syms = {"src/api.py::fn": _sym("fn")} |
| 437 | ours_syms = {} # deleted on ours |
| 438 | theirs_syms = {"src/api.py::fn": _sym("fn")} # still in theirs |
| 439 | |
| 440 | # Theirs has a new caller of fn that wasn't in base. |
| 441 | base_fg = {"caller_base.py::existing": frozenset()} |
| 442 | ours_fg = {} |
| 443 | theirs_fg = { |
| 444 | "caller_new.py::new_caller": frozenset({"fn"}), # new! |
| 445 | } |
| 446 | with ( |
| 447 | patch("muse.plugins.code._callgraph.build_forward_graph", |
| 448 | side_effect=[base_fg, ours_fg, theirs_fg]), |
| 449 | ): |
| 450 | items, ok, warn = _find_delete_use_conflicts( |
| 451 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 452 | ) |
| 453 | |
| 454 | assert ok is True |
| 455 | assert warn is None |
| 456 | assert len(items) == 1 |
| 457 | assert items[0].conflict_type == "delete_use" |
| 458 | assert "caller_new.py::new_caller" in items[0].theirs_change |
| 459 | |
| 460 | def test_no_delete_use_when_no_deletions(self, repo: pathlib.Path) -> None: |
| 461 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 462 | base_syms = {"src/api.py::fn": _sym("fn")} |
| 463 | ours_syms = {"src/api.py::fn": _sym("fn")} |
| 464 | theirs_syms = {"src/api.py::fn": _sym("fn")} |
| 465 | items, ok, warn = _find_delete_use_conflicts( |
| 466 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 467 | ) |
| 468 | assert ok is True |
| 469 | assert items == [] |
| 470 | |
| 471 | def test_call_graph_unavailable_warns(self, repo: pathlib.Path) -> None: |
| 472 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 473 | base_syms = {"src/api.py::fn": _sym("fn")} |
| 474 | ours_syms = {} |
| 475 | theirs_syms = {"src/api.py::fn": _sym("fn")} |
| 476 | |
| 477 | with patch("muse.plugins.code._callgraph.build_forward_graph", |
| 478 | side_effect=OSError("no index")): |
| 479 | items, ok, warn = _find_delete_use_conflicts( |
| 480 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 481 | ) |
| 482 | |
| 483 | assert ok is False |
| 484 | assert warn is not None |
| 485 | assert "call graph unavailable" in warn |
| 486 | assert items == [] |
| 487 | |
| 488 | def test_keyerror_from_call_graph_warns(self, repo: pathlib.Path) -> None: |
| 489 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 490 | base_syms = {"src/api.py::fn": _sym("fn")} |
| 491 | ours_syms = {} |
| 492 | theirs_syms = {"src/api.py::fn": _sym("fn")} |
| 493 | |
| 494 | with patch("muse.plugins.code._callgraph.build_forward_graph", |
| 495 | side_effect=KeyError("missing")): |
| 496 | items, ok, warn = _find_delete_use_conflicts( |
| 497 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 498 | ) |
| 499 | |
| 500 | assert ok is False |
| 501 | |
| 502 | def test_unexpected_exception_propagates(self, repo: pathlib.Path) -> None: |
| 503 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 504 | base_syms = {"src/api.py::fn": _sym("fn")} |
| 505 | ours_syms = {} |
| 506 | theirs_syms = {"src/api.py::fn": _sym("fn")} |
| 507 | |
| 508 | with patch("muse.plugins.code._callgraph.build_forward_graph", |
| 509 | side_effect=MemoryError("OOM")): |
| 510 | with pytest.raises(MemoryError): |
| 511 | _find_delete_use_conflicts( |
| 512 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 513 | ) |
| 514 | |
| 515 | def test_caller_preview_truncated_at_3(self, repo: pathlib.Path) -> None: |
| 516 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 517 | base_syms = {"src/api.py::fn": _sym("fn")} |
| 518 | ours_syms = {} |
| 519 | theirs_syms = {"src/api.py::fn": _sym("fn")} |
| 520 | |
| 521 | theirs_fg = {f"mod{i}.py::caller{i}": frozenset({"fn"}) for i in range(10)} |
| 522 | base_fg = {} |
| 523 | ours_fg = {} |
| 524 | |
| 525 | with patch("muse.plugins.code._callgraph.build_forward_graph", |
| 526 | side_effect=[base_fg, ours_fg, theirs_fg]): |
| 527 | items, ok, _ = _find_delete_use_conflicts( |
| 528 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 529 | ) |
| 530 | |
| 531 | assert len(items) == 1 |
| 532 | assert "+7 more" in items[0].theirs_change |
| 533 | |
| 534 | |
| 535 | # ───────────────────────────────────────────────────────────────────────────── |
| 536 | # Unit tests — _find_dependency_conflicts |
| 537 | # ───────────────────────────────────────────────────────────────────────────── |
| 538 | |
| 539 | |
| 540 | class TestFindDependencyConflicts: |
| 541 | def test_dependency_detected(self, repo: pathlib.Path) -> None: |
| 542 | from muse.cli.commands.plan_merge import _find_dependency_conflicts |
| 543 | # Ours changed fn_a; theirs changed fn_b which calls fn_a. |
| 544 | ours_changed = {"src/x.py::fn_a"} |
| 545 | theirs_changed = {"src/y.py::fn_b"} |
| 546 | reverse = {"fn_a": ["src/y.py::fn_b"]} # fn_b calls fn_a. |
| 547 | |
| 548 | with ( |
| 549 | patch("muse.plugins.code._callgraph.build_reverse_graph", return_value=reverse), |
| 550 | patch("muse.plugins.code._callgraph.transitive_callers", |
| 551 | return_value={1: ["src/y.py::fn_b"]}), |
| 552 | ): |
| 553 | items, ok, warn = _find_dependency_conflicts( |
| 554 | repo, {}, {}, ours_changed, theirs_changed, |
| 555 | ) |
| 556 | |
| 557 | assert ok is True |
| 558 | assert len(items) == 1 |
| 559 | assert items[0].conflict_type == "dependency_conflict" |
| 560 | assert "src/x.py::fn_a" == items[0].address |
| 561 | |
| 562 | def test_no_conflict_when_no_changes(self, repo: pathlib.Path) -> None: |
| 563 | from muse.cli.commands.plan_merge import _find_dependency_conflicts |
| 564 | items, ok, warn = _find_dependency_conflicts(repo, {}, {}, set(), set()) |
| 565 | assert items == [] |
| 566 | assert ok is True |
| 567 | |
| 568 | def test_call_graph_error_warns(self, repo: pathlib.Path) -> None: |
| 569 | from muse.cli.commands.plan_merge import _find_dependency_conflicts |
| 570 | with patch("muse.plugins.code._callgraph.build_reverse_graph", |
| 571 | side_effect=ValueError("bad data")): |
| 572 | items, ok, warn = _find_dependency_conflicts( |
| 573 | repo, {}, {}, {"x.py::fn"}, {"y.py::fn"}, |
| 574 | ) |
| 575 | assert ok is False |
| 576 | assert warn is not None |
| 577 | |
| 578 | def test_deduplication(self, repo: pathlib.Path) -> None: |
| 579 | """Same (ours_addr, theirs_addr) pair not added twice.""" |
| 580 | from muse.cli.commands.plan_merge import _find_dependency_conflicts |
| 581 | ours_changed = {"x.py::fn_a"} |
| 582 | theirs_changed = {"y.py::fn_b"} |
| 583 | reverse = {"fn_a": ["y.py::fn_b", "y.py::fn_b"]} # Duplicate. |
| 584 | |
| 585 | with ( |
| 586 | patch("muse.plugins.code._callgraph.build_reverse_graph", return_value=reverse), |
| 587 | patch("muse.plugins.code._callgraph.transitive_callers", |
| 588 | return_value={1: ["y.py::fn_b", "y.py::fn_b"]}), |
| 589 | ): |
| 590 | items, _, _ = _find_dependency_conflicts(repo, {}, {}, ours_changed, theirs_changed) |
| 591 | |
| 592 | assert len(items) == 1 # Not duplicated. |
| 593 | |
| 594 | |
| 595 | # ───────────────────────────────────────────────────────────────────────────── |
| 596 | # Integration tests — full CLI (mock-based) |
| 597 | # ───────────────────────────────────────────────────────────────────────────── |
| 598 | |
| 599 | |
| 600 | class TestPlanMergeIntegration: |
| 601 | def test_ref_not_found_exits_1(self, repo: pathlib.Path) -> None: |
| 602 | from muse.cli.commands.plan_merge import run as pm_run |
| 603 | ns = argparse.Namespace( |
| 604 | ours_ref="nonexistent", theirs_ref="main", |
| 605 | base_ref=None, skip_call_graph=True, fmt="json", |
| 606 | json_out=True, |
| 607 | ) |
| 608 | old = os.getcwd() |
| 609 | os.chdir(repo) |
| 610 | try: |
| 611 | with ( |
| 612 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 613 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 614 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 615 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None), |
| 616 | ): |
| 617 | with pytest.raises(SystemExit) as exc: |
| 618 | pm_run(ns) |
| 619 | finally: |
| 620 | os.chdir(old) |
| 621 | assert exc.value.code == 1 |
| 622 | |
| 623 | def test_json_schema_complete(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 624 | code, out = _run_plan_merge(repo) |
| 625 | data = json.loads(out) |
| 626 | required = { |
| 627 | "schema", "ours", "theirs", "base", "base_auto_computed", |
| 628 | "call_graph_available", "call_graph_skipped", "warnings", |
| 629 | "total_symbols", "conflicts", "clean", "items", "duration_ms", |
| 630 | } |
| 631 | assert required.issubset(data.keys()) |
| 632 | |
| 633 | def test_duration_ms_is_non_negative_float(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 634 | _, out = _run_plan_merge(repo) |
| 635 | data = json.loads(out) |
| 636 | assert isinstance(data["duration_ms"], float) |
| 637 | assert data["duration_ms"] >= 0 |
| 638 | |
| 639 | def test_full_commit_ids_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 640 | ours_c = _mock_commit("a" * 64) |
| 641 | theirs_c = _mock_commit("b" * 64) |
| 642 | _, out = _run_plan_merge( |
| 643 | repo, |
| 644 | mock_ours_commit=ours_c, |
| 645 | mock_theirs_commit=theirs_c, |
| 646 | ) |
| 647 | data = json.loads(out) |
| 648 | assert data["ours"] == "a" * 64 |
| 649 | assert data["theirs"] == "b" * 64 |
| 650 | |
| 651 | def test_base_auto_computed_true_when_no_base_arg(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 652 | _, out = _run_plan_merge(repo) |
| 653 | data = json.loads(out) |
| 654 | assert data["base_auto_computed"] is True |
| 655 | |
| 656 | def test_base_auto_computed_false_when_base_arg_given(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 657 | from muse.cli.commands.plan_merge import run as pm_run |
| 658 | ours_c = _mock_commit("a" * 64) |
| 659 | theirs_c = _mock_commit("b" * 64) |
| 660 | base_c = _mock_commit("c" * 64) |
| 661 | |
| 662 | ns = argparse.Namespace( |
| 663 | ours_ref="HEAD", theirs_ref="main", |
| 664 | base_ref="base-ref", skip_call_graph=True, fmt="json", |
| 665 | json_out=True, |
| 666 | ) |
| 667 | import io, sys |
| 668 | captured = io.StringIO() |
| 669 | old_stdout = sys.stdout |
| 670 | sys.stdout = captured |
| 671 | old = os.getcwd() |
| 672 | os.chdir(repo) |
| 673 | try: |
| 674 | with ( |
| 675 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 676 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 677 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 678 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", |
| 679 | side_effect=lambda *a, **kw: ( |
| 680 | ours_c if a[3] in ("HEAD", None) else |
| 681 | theirs_c if a[3] == "main" else |
| 682 | base_c if a[3] == "base-ref" else None |
| 683 | )), |
| 684 | patch("muse.cli.commands.plan_merge.find_merge_base", return_value="c" * 64), |
| 685 | patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", return_value={}), |
| 686 | patch("muse.cli.commands.plan_merge.symbols_for_snapshot", return_value={}), |
| 687 | ): |
| 688 | pm_run(ns) |
| 689 | except SystemExit: |
| 690 | pass |
| 691 | finally: |
| 692 | sys.stdout = old_stdout |
| 693 | os.chdir(old) |
| 694 | |
| 695 | data = json.loads(captured.getvalue()) |
| 696 | assert data["base_auto_computed"] is False |
| 697 | |
| 698 | def test_call_graph_skipped_flag(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 699 | _, out = _run_plan_merge(repo, skip_call_graph=True) |
| 700 | data = json.loads(out) |
| 701 | assert data["call_graph_skipped"] is True |
| 702 | |
| 703 | def test_warnings_list_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 704 | _, out = _run_plan_merge(repo) |
| 705 | data = json.loads(out) |
| 706 | assert isinstance(data["warnings"], list) |
| 707 | |
| 708 | def test_no_conflicts_empty_swarm(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 709 | _, out = _run_plan_merge(repo) |
| 710 | data = json.loads(out) |
| 711 | assert data["conflicts"] == 0 |
| 712 | assert data["items"] == [] |
| 713 | |
| 714 | def test_symbol_edit_overlap_detected(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 715 | """Three-way: base=X, ours=Y, theirs=Z → symbol_edit_overlap.""" |
| 716 | base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 717 | ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 718 | theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")} |
| 719 | _, out = _run_plan_merge( |
| 720 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 721 | ) |
| 722 | data = json.loads(out) |
| 723 | assert data["conflicts"] == 1 |
| 724 | assert data["items"][0]["conflict_type"] == "symbol_edit_overlap" |
| 725 | |
| 726 | def test_no_false_positive_unilateral_change(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 727 | """Three-way: base=X, ours=Y, theirs=X → only ours changed → no_conflict.""" |
| 728 | base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 729 | ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 730 | theirs = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 731 | _, out = _run_plan_merge( |
| 732 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 733 | ) |
| 734 | data = json.loads(out) |
| 735 | assert data["conflicts"] == 0, ( |
| 736 | "Three-way should detect no conflict when only ours changed" |
| 737 | ) |
| 738 | |
| 739 | def test_rename_edit_via_pass2_body_hash(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 740 | """Pass 2 upgrades symbol_edit_overlap → rename_edit via body_hash matching.""" |
| 741 | body = "REAL_BODY_HASH_12345" |
| 742 | base = {"f.py::fn_old": {**_sym("fn_old", content_id="X", body_hash=body), "name": "fn_old"}} |
| 743 | # Ours: fn_old deleted, fn_new added with same body (rename) |
| 744 | ours = {"f.py::fn_new": {**_sym("fn_new", content_id="Y", body_hash=body), "name": "fn_new"}} |
| 745 | # Theirs: fn_old modified (impl change) |
| 746 | theirs = {"f.py::fn_old": {**_sym("fn_old", content_id="Z", body_hash="OTHER"), "name": "fn_old"}} |
| 747 | _, out = _run_plan_merge( |
| 748 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 749 | ) |
| 750 | data = json.loads(out) |
| 751 | conflict_types = [i["conflict_type"] for i in data["items"]] |
| 752 | assert "rename_edit" in conflict_types |
| 753 | |
| 754 | def test_move_edit_via_pass2(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 755 | """Pass 2 detects move_edit when ours moved to different file and theirs modified.""" |
| 756 | body = "MOVED_BODY_HASH_12345" |
| 757 | base = {"old/file.py::fn": _sym("fn", content_id="X", body_hash=body)} |
| 758 | # Ours: fn moved to new/file.py |
| 759 | ours = {"new/file.py::fn": _sym("fn", content_id="X", body_hash=body)} |
| 760 | # Theirs: fn modified in original location |
| 761 | theirs = {"old/file.py::fn": _sym("fn", content_id="Z", body_hash="OTHER")} |
| 762 | _, out = _run_plan_merge( |
| 763 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 764 | ) |
| 765 | data = json.loads(out) |
| 766 | conflict_types = [i["conflict_type"] for i in data["items"]] |
| 767 | assert "move_edit" in conflict_types |
| 768 | |
| 769 | def test_text_output_format(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 770 | _, out = _run_plan_merge(repo, fmt="text") |
| 771 | assert "Semantic merge plan" in out |
| 772 | assert "base:" in out |
| 773 | |
| 774 | def test_text_output_shows_conflicts(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 775 | base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 776 | ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 777 | theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")} |
| 778 | _, out = _run_plan_merge( |
| 779 | repo, fmt="text", |
| 780 | mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 781 | ) |
| 782 | assert "symbol_edit_overlap" in out |
| 783 | assert "ours:" in out |
| 784 | assert "theirs:" in out |
| 785 | |
| 786 | def test_text_output_shows_elapsed(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 787 | _, out = _run_plan_merge(repo, fmt="text") |
| 788 | assert "s)" in out |
| 789 | |
| 790 | def test_skip_call_graph_omits_delete_use(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 791 | base = {"f.py::fn": _sym("fn")} |
| 792 | ours = {} # deleted |
| 793 | theirs = {"f.py::fn": _sym("fn")} |
| 794 | _, out = _run_plan_merge( |
| 795 | repo, skip_call_graph=True, |
| 796 | mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 797 | ) |
| 798 | data = json.loads(out) |
| 799 | conflict_types = [i["conflict_type"] for i in data["items"]] |
| 800 | assert "delete_use" not in conflict_types |
| 801 | |
| 802 | def test_base_none_warning_in_json(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 803 | """When find_merge_base returns None, warnings includes a notice.""" |
| 804 | from muse.cli.commands.plan_merge import run as pm_run |
| 805 | ours_c = _mock_commit("a" * 64) |
| 806 | theirs_c = _mock_commit("b" * 64) |
| 807 | ns = argparse.Namespace( |
| 808 | ours_ref="HEAD", theirs_ref="main", |
| 809 | base_ref=None, skip_call_graph=True, fmt="json", |
| 810 | json_out=True, |
| 811 | ) |
| 812 | import io, sys |
| 813 | captured = io.StringIO() |
| 814 | old = os.getcwd() |
| 815 | os.chdir(repo) |
| 816 | sys.stdout = captured |
| 817 | try: |
| 818 | with ( |
| 819 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 820 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 821 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 822 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", |
| 823 | side_effect=lambda *a: ours_c if a[3] in ("HEAD", None) else theirs_c), |
| 824 | patch("muse.cli.commands.plan_merge.find_merge_base", return_value=None), |
| 825 | patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", return_value={}), |
| 826 | patch("muse.cli.commands.plan_merge.symbols_for_snapshot", return_value={}), |
| 827 | ): |
| 828 | pm_run(ns) |
| 829 | except SystemExit: |
| 830 | pass |
| 831 | finally: |
| 832 | sys.stdout = sys.__stdout__ |
| 833 | os.chdir(old) |
| 834 | |
| 835 | data = json.loads(captured.getvalue()) |
| 836 | assert any("no common ancestor" in w for w in data["warnings"]) |
| 837 | assert data["base"] is None |
| 838 | |
| 839 | def test_format_json_shorthand(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 840 | """--json is equivalent to --format json.""" |
| 841 | from muse.cli.commands.plan_merge import run as pm_run |
| 842 | ours_c = _mock_commit("a" * 64) |
| 843 | theirs_c = _mock_commit("b" * 64) |
| 844 | ns = argparse.Namespace( |
| 845 | ours_ref="HEAD", theirs_ref="main", |
| 846 | base_ref=None, skip_call_graph=True, fmt="json", # same as --json |
| 847 | json_out=True, |
| 848 | ) |
| 849 | import io, sys |
| 850 | captured = io.StringIO() |
| 851 | old = os.getcwd() |
| 852 | os.chdir(repo) |
| 853 | sys.stdout = captured |
| 854 | try: |
| 855 | with ( |
| 856 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 857 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 858 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 859 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", |
| 860 | side_effect=lambda *a: ours_c if a[3] in ("HEAD", None) else theirs_c), |
| 861 | patch("muse.cli.commands.plan_merge.find_merge_base", return_value="c" * 64), |
| 862 | patch("muse.cli.commands.plan_merge.get_commit_snapshot_manifest", return_value={}), |
| 863 | patch("muse.cli.commands.plan_merge.symbols_for_snapshot", return_value={}), |
| 864 | ): |
| 865 | pm_run(ns) |
| 866 | except SystemExit: |
| 867 | pass |
| 868 | finally: |
| 869 | sys.stdout = sys.__stdout__ |
| 870 | os.chdir(old) |
| 871 | |
| 872 | data = json.loads(captured.getvalue()) |
| 873 | assert "conflicts" in data |
| 874 | |
| 875 | |
| 876 | # ───────────────────────────────────────────────────────────────────────────── |
| 877 | # Security tests |
| 878 | # ───────────────────────────────────────────────────────────────────────────── |
| 879 | |
| 880 | |
| 881 | class TestPlanMergeSecurity: |
| 882 | def test_ansi_in_address_stripped_text_output(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 883 | """ANSI escape codes in symbol addresses are stripped before display.""" |
| 884 | ansi_addr = "\x1b[31msrc/malicious.py::fn\x1b[0m" |
| 885 | base = {ansi_addr: _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 886 | ours = {ansi_addr: _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 887 | theirs = {ansi_addr: _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")} |
| 888 | _, out = _run_plan_merge( |
| 889 | repo, fmt="text", |
| 890 | mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 891 | ) |
| 892 | assert "\x1b[" not in out |
| 893 | assert "src/malicious.py::fn" in out # sanitized content still shown |
| 894 | |
| 895 | def test_ansi_in_recommendation_stripped(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 896 | base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 897 | ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 898 | theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")} |
| 899 | _, out = _run_plan_merge( |
| 900 | repo, fmt="text", |
| 901 | mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 902 | ) |
| 903 | assert "\x1b[" not in out |
| 904 | |
| 905 | def test_control_chars_in_ref_not_escape_fs(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 906 | """Control characters in ref names are sanitised before display in error output.""" |
| 907 | from muse.cli.commands.plan_merge import run as pm_run |
| 908 | import io, sys |
| 909 | captured = io.StringIO() |
| 910 | malicious_ref = "ref\x00/../../../etc" |
| 911 | ns = argparse.Namespace( |
| 912 | ours_ref=malicious_ref, theirs_ref="main", |
| 913 | base_ref=None, skip_call_graph=True, fmt="json", |
| 914 | json_out=True, |
| 915 | ) |
| 916 | old = os.getcwd() |
| 917 | os.chdir(repo) |
| 918 | sys.stdout = captured |
| 919 | try: |
| 920 | with ( |
| 921 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 922 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 923 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 924 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None), |
| 925 | ): |
| 926 | with pytest.raises(SystemExit) as exc: |
| 927 | pm_run(ns) |
| 928 | finally: |
| 929 | sys.stdout = sys.__stdout__ |
| 930 | os.chdir(old) |
| 931 | assert exc.value.code == 1 |
| 932 | out = captured.getvalue() |
| 933 | # Output must not contain raw null bytes. |
| 934 | assert "\x00" not in out |
| 935 | |
| 936 | |
| 937 | # ───────────────────────────────────────────────────────────────────────────── |
| 938 | # Stress tests |
| 939 | # ───────────────────────────────────────────────────────────────────────────── |
| 940 | |
| 941 | |
| 942 | class TestPlanMergeStress: |
| 943 | def test_1000_symbols_pass1_under_2s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 944 | """Pass 1 with 1000 symbols completes in < 2 s.""" |
| 945 | N = 1000 |
| 946 | base = {f"f{i}.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)} |
| 947 | # Ours and theirs both change half the symbols differently. |
| 948 | ours = { |
| 949 | f"f{i}.py::fn{i}": ( |
| 950 | _sym(f"fn{i}", content_id=f"Y{i}") if i % 2 == 0 |
| 951 | else _sym(f"fn{i}", content_id=f"X{i}") |
| 952 | ) |
| 953 | for i in range(N) |
| 954 | } |
| 955 | theirs = { |
| 956 | f"f{i}.py::fn{i}": ( |
| 957 | _sym(f"fn{i}", content_id=f"X{i}") if i % 2 == 0 |
| 958 | else _sym(f"fn{i}", content_id=f"Z{i}") |
| 959 | ) |
| 960 | for i in range(N) |
| 961 | } |
| 962 | |
| 963 | t0 = time.monotonic() |
| 964 | _, out = _run_plan_merge( |
| 965 | repo, |
| 966 | mock_ours_syms=ours, |
| 967 | mock_theirs_syms=theirs, |
| 968 | mock_base_syms=base, |
| 969 | ) |
| 970 | elapsed = time.monotonic() - t0 |
| 971 | |
| 972 | assert elapsed < 2.0, f"Pass 1 took {elapsed:.2f}s — too slow" |
| 973 | data = json.loads(out) |
| 974 | assert data["total_symbols"] == N |
| 975 | |
| 976 | def test_200_renames_pass2_under_1s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 977 | """Pass 2 rename detection with 200 renames completes in < 1 s.""" |
| 978 | N = 200 |
| 979 | body_hashes = [f"BODYHASH{i:08d}" for i in range(N)] |
| 980 | base = { |
| 981 | f"f.py::fn_old_{i}": {**_sym(f"fn_old_{i}", body_hash=body_hashes[i]), "name": f"fn_old_{i}"} |
| 982 | for i in range(N) |
| 983 | } |
| 984 | # Ours: all renamed |
| 985 | ours = { |
| 986 | f"f.py::fn_new_{i}": {**_sym(f"fn_new_{i}", body_hash=body_hashes[i]), "name": f"fn_new_{i}"} |
| 987 | for i in range(N) |
| 988 | } |
| 989 | # Theirs: all modified (different body) |
| 990 | theirs = { |
| 991 | f"f.py::fn_old_{i}": {**_sym(f"fn_old_{i}", content_id=f"Z{i}", body_hash=f"DIFF{i}"), "name": f"fn_old_{i}"} |
| 992 | for i in range(N) |
| 993 | } |
| 994 | |
| 995 | t0 = time.monotonic() |
| 996 | _, out = _run_plan_merge( |
| 997 | repo, |
| 998 | mock_ours_syms=ours, |
| 999 | mock_theirs_syms=theirs, |
| 1000 | mock_base_syms=base, |
| 1001 | ) |
| 1002 | elapsed = time.monotonic() - t0 |
| 1003 | |
| 1004 | assert elapsed < 1.0, f"Pass 2 took {elapsed:.2f}s — too slow" |
| 1005 | data = json.loads(out) |
| 1006 | conflict_types = [i["conflict_type"] for i in data["items"]] |
| 1007 | assert "rename_edit" in conflict_types |
| 1008 | |
| 1009 | def test_100_deletes_delete_use_detection_under_2s(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 1010 | """delete_use detection with 100 deleted symbols completes in < 2 s.""" |
| 1011 | from muse.cli.commands.plan_merge import _find_delete_use_conflicts |
| 1012 | |
| 1013 | N = 100 |
| 1014 | base_syms = {f"f{i}.py::fn{i}": _sym(f"fn{i}") for i in range(N)} |
| 1015 | ours_syms: SymbolTree = {} # All deleted on ours. |
| 1016 | theirs_syms = dict(base_syms) # All present on theirs. |
| 1017 | |
| 1018 | # Mock: each fn has a new caller on theirs. |
| 1019 | base_fg: ForwardGraph = {} |
| 1020 | ours_fg: ForwardGraph = {} |
| 1021 | theirs_fg = {f"new_caller{i}.py::caller{i}": frozenset({f"fn{i}"}) for i in range(N)} |
| 1022 | |
| 1023 | t0 = time.monotonic() |
| 1024 | with patch("muse.plugins.code._callgraph.build_forward_graph", |
| 1025 | side_effect=[base_fg, ours_fg, theirs_fg]): |
| 1026 | items, ok, warn = _find_delete_use_conflicts( |
| 1027 | repo, {}, {}, {}, base_syms, ours_syms, theirs_syms, |
| 1028 | ) |
| 1029 | elapsed = time.monotonic() - t0 |
| 1030 | |
| 1031 | assert elapsed < 2.0, f"delete_use took {elapsed:.2f}s — too slow" |
| 1032 | assert ok is True |
| 1033 | assert len(items) == N |
| 1034 | |
| 1035 | def test_correctness_three_way_no_false_positives(self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 1036 | """Regression: three-way diff must not produce false positives on unilateral changes.""" |
| 1037 | N = 500 |
| 1038 | # Ours changes even-indexed symbols, theirs changes odd-indexed. |
| 1039 | # Each symbol is changed by only ONE side → all should be no_conflict. |
| 1040 | base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"BASE{i}") for i in range(N)} |
| 1041 | ours = { |
| 1042 | f"f.py::fn{i}": ( |
| 1043 | _sym(f"fn{i}", content_id=f"OUR{i}") if i % 2 == 0 |
| 1044 | else _sym(f"fn{i}", content_id=f"BASE{i}") |
| 1045 | ) |
| 1046 | for i in range(N) |
| 1047 | } |
| 1048 | theirs = { |
| 1049 | f"f.py::fn{i}": ( |
| 1050 | _sym(f"fn{i}", content_id=f"BASE{i}") if i % 2 == 0 |
| 1051 | else _sym(f"fn{i}", content_id=f"THEIR{i}") |
| 1052 | ) |
| 1053 | for i in range(N) |
| 1054 | } |
| 1055 | |
| 1056 | _, out = _run_plan_merge( |
| 1057 | repo, |
| 1058 | mock_ours_syms=ours, |
| 1059 | mock_theirs_syms=theirs, |
| 1060 | mock_base_syms=base, |
| 1061 | ) |
| 1062 | data = json.loads(out) |
| 1063 | assert data["conflicts"] == 0, ( |
| 1064 | f"Three-way correctness: expected 0 conflicts, got {data['conflicts']}" |
| 1065 | ) |
| 1066 | |
| 1067 | |
| 1068 | # ───────────────────────────────────────────────────────────────────────────── |
| 1069 | # Error shape tests |
| 1070 | # ───────────────────────────────────────────────────────────────────────────── |
| 1071 | |
| 1072 | |
| 1073 | class TestPlanMergeErrorShapes: |
| 1074 | """Verify error output shapes are consistent across text and JSON modes.""" |
| 1075 | |
| 1076 | def _run_with_no_ours(self, repo: pathlib.Path, fmt: str = "json") -> tuple[int, str, str]: |
| 1077 | """Run plan-merge where ours-ref resolves to None.""" |
| 1078 | from muse.cli.commands.plan_merge import run as pm_run |
| 1079 | ns = argparse.Namespace( |
| 1080 | ours_ref="missing-ref", theirs_ref="main", |
| 1081 | base_ref=None, skip_call_graph=True, fmt=fmt, |
| 1082 | json_out=(fmt == "json"), |
| 1083 | ) |
| 1084 | import io |
| 1085 | captured = io.StringIO() |
| 1086 | old_stdout = sys.stdout |
| 1087 | old_stderr = sys.stderr |
| 1088 | sys.stdout = captured |
| 1089 | err_captured = io.StringIO() |
| 1090 | sys.stderr = err_captured |
| 1091 | old = os.getcwd() |
| 1092 | os.chdir(repo) |
| 1093 | exit_code = 0 |
| 1094 | try: |
| 1095 | with ( |
| 1096 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 1097 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 1098 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 1099 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None), |
| 1100 | ): |
| 1101 | with pytest.raises(SystemExit) as exc: |
| 1102 | pm_run(ns) |
| 1103 | exit_code = exc.value.code |
| 1104 | finally: |
| 1105 | sys.stdout = old_stdout |
| 1106 | sys.stderr = old_stderr |
| 1107 | os.chdir(old) |
| 1108 | return exit_code, captured.getvalue(), err_captured.getvalue() |
| 1109 | |
| 1110 | def _run_with_no_theirs(self, repo: pathlib.Path, fmt: str = "json") -> tuple[int, str, str]: |
| 1111 | from muse.cli.commands.plan_merge import run as pm_run |
| 1112 | ours_c = _mock_commit("a" * 64) |
| 1113 | ns = argparse.Namespace( |
| 1114 | ours_ref="HEAD", theirs_ref="missing-branch", |
| 1115 | base_ref=None, skip_call_graph=True, fmt=fmt, |
| 1116 | json_out=(fmt == "json"), |
| 1117 | ) |
| 1118 | import io |
| 1119 | captured = io.StringIO() |
| 1120 | err_captured = io.StringIO() |
| 1121 | old_stdout, old_stderr = sys.stdout, sys.stderr |
| 1122 | sys.stdout = captured |
| 1123 | sys.stderr = err_captured |
| 1124 | old = os.getcwd() |
| 1125 | os.chdir(repo) |
| 1126 | try: |
| 1127 | with ( |
| 1128 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 1129 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 1130 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 1131 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", |
| 1132 | side_effect=lambda *a: ours_c if a[3] == "HEAD" else None), |
| 1133 | ): |
| 1134 | with pytest.raises(SystemExit) as exc: |
| 1135 | pm_run(ns) |
| 1136 | exit_code = exc.value.code |
| 1137 | finally: |
| 1138 | sys.stdout = old_stdout |
| 1139 | sys.stderr = old_stderr |
| 1140 | os.chdir(old) |
| 1141 | return exit_code, captured.getvalue(), err_captured.getvalue() |
| 1142 | |
| 1143 | def _run_with_no_base(self, repo: pathlib.Path, fmt: str = "json") -> tuple[int, str, str]: |
| 1144 | from muse.cli.commands.plan_merge import run as pm_run |
| 1145 | ours_c = _mock_commit("a" * 64) |
| 1146 | theirs_c = _mock_commit("b" * 64) |
| 1147 | ns = argparse.Namespace( |
| 1148 | ours_ref="HEAD", theirs_ref="main", |
| 1149 | base_ref="missing-base", skip_call_graph=True, fmt=fmt, |
| 1150 | json_out=(fmt == "json"), |
| 1151 | ) |
| 1152 | import io |
| 1153 | captured = io.StringIO() |
| 1154 | err_captured = io.StringIO() |
| 1155 | old_stdout, old_stderr = sys.stdout, sys.stderr |
| 1156 | sys.stdout = captured |
| 1157 | sys.stderr = err_captured |
| 1158 | old = os.getcwd() |
| 1159 | os.chdir(repo) |
| 1160 | try: |
| 1161 | with ( |
| 1162 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 1163 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 1164 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 1165 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", |
| 1166 | side_effect=lambda *a: ( |
| 1167 | ours_c if a[3] == "HEAD" else |
| 1168 | theirs_c if a[3] == "main" else |
| 1169 | None # base-ref not found |
| 1170 | )), |
| 1171 | ): |
| 1172 | with pytest.raises(SystemExit) as exc: |
| 1173 | pm_run(ns) |
| 1174 | exit_code = exc.value.code |
| 1175 | finally: |
| 1176 | sys.stdout = old_stdout |
| 1177 | sys.stderr = old_stderr |
| 1178 | os.chdir(old) |
| 1179 | return exit_code, captured.getvalue(), err_captured.getvalue() |
| 1180 | |
| 1181 | # ── ours-ref not found ──────────────────────────────────────────────────── |
| 1182 | |
| 1183 | def test_ours_not_found_json_has_error_and_status(self, repo: pathlib.Path) -> None: |
| 1184 | code, out, _ = self._run_with_no_ours(repo, fmt="json") |
| 1185 | assert code == 1 |
| 1186 | data = json.loads(out.strip()) |
| 1187 | assert "error" in data |
| 1188 | assert data["status"] == "error" |
| 1189 | |
| 1190 | def test_ours_not_found_json_error_mentions_ref(self, repo: pathlib.Path) -> None: |
| 1191 | code, out, _ = self._run_with_no_ours(repo, fmt="json") |
| 1192 | data = json.loads(out.strip()) |
| 1193 | assert "missing-ref" in data["error"] |
| 1194 | |
| 1195 | def test_ours_not_found_text_uses_tick_prefix(self, repo: pathlib.Path) -> None: |
| 1196 | code, _, err = self._run_with_no_ours(repo, fmt="text") |
| 1197 | assert code == 1 |
| 1198 | assert "❌" in err |
| 1199 | |
| 1200 | def test_ours_not_found_text_no_output_on_stdout(self, repo: pathlib.Path) -> None: |
| 1201 | code, out, _ = self._run_with_no_ours(repo, fmt="text") |
| 1202 | assert out == "" |
| 1203 | |
| 1204 | # ── theirs-ref not found ────────────────────────────────────────────────── |
| 1205 | |
| 1206 | def test_theirs_not_found_exits_1(self, repo: pathlib.Path) -> None: |
| 1207 | code, _, _ = self._run_with_no_theirs(repo) |
| 1208 | assert code == 1 |
| 1209 | |
| 1210 | def test_theirs_not_found_json_has_status(self, repo: pathlib.Path) -> None: |
| 1211 | code, out, _ = self._run_with_no_theirs(repo, fmt="json") |
| 1212 | data = json.loads(out.strip()) |
| 1213 | assert data["status"] == "error" |
| 1214 | |
| 1215 | def test_theirs_not_found_error_mentions_ref(self, repo: pathlib.Path) -> None: |
| 1216 | code, out, _ = self._run_with_no_theirs(repo, fmt="json") |
| 1217 | data = json.loads(out.strip()) |
| 1218 | assert "missing-branch" in data["error"] |
| 1219 | |
| 1220 | def test_theirs_not_found_text_uses_tick_prefix(self, repo: pathlib.Path) -> None: |
| 1221 | code, _, err = self._run_with_no_theirs(repo, fmt="text") |
| 1222 | assert "❌" in err |
| 1223 | |
| 1224 | # ── base-ref not found ──────────────────────────────────────────────────── |
| 1225 | |
| 1226 | def test_base_not_found_exits_1(self, repo: pathlib.Path) -> None: |
| 1227 | code, _, _ = self._run_with_no_base(repo) |
| 1228 | assert code == 1 |
| 1229 | |
| 1230 | def test_base_not_found_json_has_status(self, repo: pathlib.Path) -> None: |
| 1231 | code, out, _ = self._run_with_no_base(repo, fmt="json") |
| 1232 | data = json.loads(out.strip()) |
| 1233 | assert data["status"] == "error" |
| 1234 | |
| 1235 | def test_base_not_found_error_mentions_ref(self, repo: pathlib.Path) -> None: |
| 1236 | code, out, _ = self._run_with_no_base(repo, fmt="json") |
| 1237 | data = json.loads(out.strip()) |
| 1238 | assert "missing-base" in data["error"] |
| 1239 | |
| 1240 | def test_base_not_found_text_uses_tick_prefix(self, repo: pathlib.Path) -> None: |
| 1241 | code, _, err = self._run_with_no_base(repo, fmt="text") |
| 1242 | assert "❌" in err |
| 1243 | |
| 1244 | |
| 1245 | # ───────────────────────────────────────────────────────────────────────────── |
| 1246 | # Compact JSON and conflicts_by_type |
| 1247 | # ───────────────────────────────────────────────────────────────────────────── |
| 1248 | |
| 1249 | |
| 1250 | class TestPlanMergeJsonOutput: |
| 1251 | """Verify JSON output shape, compactness, and new fields.""" |
| 1252 | |
| 1253 | def test_json_is_single_line(self, repo: pathlib.Path) -> None: |
| 1254 | """Output must be compact — no embedded newlines from indent=2.""" |
| 1255 | _, out = _run_plan_merge(repo) |
| 1256 | lines = [ln for ln in out.splitlines() if ln.strip()] |
| 1257 | assert len(lines) == 1, f"JSON output must be a single line, got {len(lines)} lines" |
| 1258 | |
| 1259 | def test_json_is_valid(self, repo: pathlib.Path) -> None: |
| 1260 | _, out = _run_plan_merge(repo) |
| 1261 | data = json.loads(out) # raises if invalid |
| 1262 | assert isinstance(data, dict) |
| 1263 | |
| 1264 | def test_json_includes_conflicts_by_type_empty(self, repo: pathlib.Path) -> None: |
| 1265 | """When no conflicts, conflicts_by_type is an empty dict.""" |
| 1266 | _, out = _run_plan_merge(repo) |
| 1267 | data = json.loads(out) |
| 1268 | assert "conflicts_by_type" in data |
| 1269 | assert data["conflicts_by_type"] == {} |
| 1270 | |
| 1271 | def test_json_conflicts_by_type_single_type(self, repo: pathlib.Path) -> None: |
| 1272 | """Single conflict type → one entry in conflicts_by_type.""" |
| 1273 | base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 1274 | ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 1275 | theirs = {"f.py::fn": _sym("fn", content_id="Z", body_hash="BH2", signature_id="SIG")} |
| 1276 | _, out = _run_plan_merge( |
| 1277 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1278 | ) |
| 1279 | data = json.loads(out) |
| 1280 | assert data["conflicts_by_type"].get("symbol_edit_overlap", 0) == 1 |
| 1281 | |
| 1282 | def test_json_conflicts_by_type_count_matches_conflicts_field(self, repo: pathlib.Path) -> None: |
| 1283 | """Sum of conflicts_by_type values must equal the 'conflicts' field.""" |
| 1284 | N = 5 |
| 1285 | base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)} |
| 1286 | ours = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Y{i}") for i in range(N)} |
| 1287 | theirs = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Z{i}") for i in range(N)} |
| 1288 | _, out = _run_plan_merge( |
| 1289 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1290 | ) |
| 1291 | data = json.loads(out) |
| 1292 | assert sum(data["conflicts_by_type"].values()) == data["conflicts"] |
| 1293 | |
| 1294 | def test_json_schema_includes_all_required_fields(self, repo: pathlib.Path) -> None: |
| 1295 | """Backward-compat check: all documented schema fields are present.""" |
| 1296 | _, out = _run_plan_merge(repo) |
| 1297 | data = json.loads(out) |
| 1298 | required = { |
| 1299 | "schema", "ours", "theirs", "base", "base_auto_computed", |
| 1300 | "call_graph_available", "call_graph_skipped", "warnings", |
| 1301 | "total_symbols", "conflicts", "clean", "conflicts_by_type", |
| 1302 | "items", "duration_ms", |
| 1303 | } |
| 1304 | missing = required - data.keys() |
| 1305 | assert not missing, f"Missing JSON fields: {missing}" |
| 1306 | |
| 1307 | def test_json_items_contains_only_conflicts(self, repo: pathlib.Path) -> None: |
| 1308 | """items must contain only conflicting symbols, not clean ones.""" |
| 1309 | base = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 1310 | ours = {"f.py::fn": _sym("fn", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 1311 | theirs = {"f.py::fn": _sym("fn", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 1312 | _, out = _run_plan_merge( |
| 1313 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1314 | ) |
| 1315 | data = json.loads(out) |
| 1316 | assert data["conflicts"] == 0 |
| 1317 | assert data["items"] == [] |
| 1318 | |
| 1319 | def test_json_error_shape_has_status(self, repo: pathlib.Path) -> None: |
| 1320 | """Error JSON must include 'status' = 'error' (agent-parseable error shape).""" |
| 1321 | from muse.cli.commands.plan_merge import run as pm_run |
| 1322 | ns = argparse.Namespace( |
| 1323 | ours_ref="bad", theirs_ref="main", |
| 1324 | base_ref=None, skip_call_graph=True, fmt="json", |
| 1325 | json_out=True, |
| 1326 | ) |
| 1327 | import io |
| 1328 | captured = io.StringIO() |
| 1329 | old = os.getcwd() |
| 1330 | os.chdir(repo) |
| 1331 | old_stdout = sys.stdout |
| 1332 | sys.stdout = captured |
| 1333 | try: |
| 1334 | with ( |
| 1335 | patch("muse.cli.commands.plan_merge.require_repo", return_value=repo), |
| 1336 | patch("muse.cli.commands.plan_merge.read_repo_id", return_value="r"), |
| 1337 | patch("muse.cli.commands.plan_merge.read_current_branch", return_value="main"), |
| 1338 | patch("muse.cli.commands.plan_merge.resolve_commit_ref", return_value=None), |
| 1339 | ): |
| 1340 | with pytest.raises(SystemExit): |
| 1341 | pm_run(ns) |
| 1342 | finally: |
| 1343 | sys.stdout = old_stdout |
| 1344 | os.chdir(old) |
| 1345 | data = json.loads(captured.getvalue().strip()) |
| 1346 | assert data["status"] == "error" |
| 1347 | assert "error" in data |
| 1348 | |
| 1349 | |
| 1350 | # ───────────────────────────────────────────────────────────────────────────── |
| 1351 | # Stress tests — conflicts_by_type correctness at scale |
| 1352 | # ───────────────────────────────────────────────────────────────────────────── |
| 1353 | |
| 1354 | |
| 1355 | class TestPlanMergeConflictsByTypeStress: |
| 1356 | def test_conflicts_by_type_counts_correct_at_scale(self, repo: pathlib.Path) -> None: |
| 1357 | """100 symbol_edit_overlap conflicts → conflicts_by_type["symbol_edit_overlap"] == 100.""" |
| 1358 | N = 100 |
| 1359 | base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)} |
| 1360 | ours = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Y{i}") for i in range(N)} |
| 1361 | theirs = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Z{i}") for i in range(N)} |
| 1362 | _, out = _run_plan_merge( |
| 1363 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1364 | ) |
| 1365 | data = json.loads(out) |
| 1366 | assert data["conflicts_by_type"]["symbol_edit_overlap"] == N |
| 1367 | assert sum(data["conflicts_by_type"].values()) == N |
| 1368 | |
| 1369 | def test_json_is_still_compact_with_many_items(self, repo: pathlib.Path) -> None: |
| 1370 | """Even with 500 conflict items, JSON output is a single line.""" |
| 1371 | N = 500 |
| 1372 | base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"X{i}") for i in range(N)} |
| 1373 | ours = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Y{i}") for i in range(N)} |
| 1374 | theirs = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"Z{i}") for i in range(N)} |
| 1375 | _, out = _run_plan_merge( |
| 1376 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1377 | ) |
| 1378 | lines = [ln for ln in out.splitlines() if ln.strip()] |
| 1379 | assert len(lines) == 1 |
| 1380 | |
| 1381 | def test_total_symbols_counts_union_of_all_three(self, repo: pathlib.Path) -> None: |
| 1382 | """total_symbols is |ours ∪ theirs ∪ base| — not just conflicts.""" |
| 1383 | # 5 shared (conflicts) + 5 ours-only (clean) + 5 theirs-only (clean) |
| 1384 | base = {f"f.py::shared{i}": _sym(f"shared{i}") for i in range(5)} |
| 1385 | ours_extra = {f"f.py::ours{i}": _sym(f"ours{i}") for i in range(5)} |
| 1386 | theirs_extra = {f"f.py::theirs{i}": _sym(f"theirs{i}") for i in range(5)} |
| 1387 | # Make shared symbols conflict |
| 1388 | ours_syms = { |
| 1389 | **{f"f.py::shared{i}": _sym(f"shared{i}", content_id=f"Y{i}") for i in range(5)}, |
| 1390 | **ours_extra, |
| 1391 | } |
| 1392 | theirs_syms = { |
| 1393 | **{f"f.py::shared{i}": _sym(f"shared{i}", content_id=f"Z{i}") for i in range(5)}, |
| 1394 | **theirs_extra, |
| 1395 | } |
| 1396 | _, out = _run_plan_merge( |
| 1397 | repo, mock_ours_syms=ours_syms, mock_theirs_syms=theirs_syms, mock_base_syms=base, |
| 1398 | ) |
| 1399 | data = json.loads(out) |
| 1400 | assert data["total_symbols"] == 15 |
| 1401 | assert data["conflicts"] == 5 |
| 1402 | assert data["clean"] == 10 |
| 1403 | |
| 1404 | |
| 1405 | # ───────────────────────────────────────────────────────────────────────────── |
| 1406 | # E2E tests — same-commit and multi-type plans |
| 1407 | # ───────────────────────────────────────────────────────────────────────────── |
| 1408 | |
| 1409 | |
| 1410 | class TestPlanMergeE2E: |
| 1411 | """E2E-style integration tests using mock commits that simulate real scenarios.""" |
| 1412 | |
| 1413 | def test_same_symbols_on_both_branches_zero_conflicts(self, repo: pathlib.Path) -> None: |
| 1414 | """Identical symbol trees → no conflicts, all clean.""" |
| 1415 | syms = {f"f.py::fn{i}": _sym(f"fn{i}") for i in range(20)} |
| 1416 | _, out = _run_plan_merge( |
| 1417 | repo, mock_ours_syms=syms, mock_theirs_syms=syms, mock_base_syms=syms, |
| 1418 | ) |
| 1419 | data = json.loads(out) |
| 1420 | assert data["conflicts"] == 0 |
| 1421 | assert data["total_symbols"] == 20 |
| 1422 | assert data["conflicts_by_type"] == {} |
| 1423 | |
| 1424 | def test_disjoint_changes_no_conflicts(self, repo: pathlib.Path) -> None: |
| 1425 | """Ours and theirs each modify different symbols → all no_conflict.""" |
| 1426 | base = {f"f.py::fn{i}": _sym(f"fn{i}", content_id=f"BASE{i}") for i in range(10)} |
| 1427 | ours = dict(base) |
| 1428 | theirs = dict(base) |
| 1429 | # Ours modifies 0–4, theirs modifies 5–9 |
| 1430 | for i in range(5): |
| 1431 | ours[f"f.py::fn{i}"] = _sym(f"fn{i}", content_id=f"OUR{i}") |
| 1432 | for i in range(5, 10): |
| 1433 | theirs[f"f.py::fn{i}"] = _sym(f"fn{i}", content_id=f"THEIR{i}") |
| 1434 | _, out = _run_plan_merge( |
| 1435 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1436 | ) |
| 1437 | data = json.loads(out) |
| 1438 | assert data["conflicts"] == 0 |
| 1439 | |
| 1440 | def test_one_new_symbol_on_each_branch_no_conflict(self, repo: pathlib.Path) -> None: |
| 1441 | """Each branch adds a different new symbol → no overlap → no conflict.""" |
| 1442 | base: SymbolTree = {} |
| 1443 | ours = {"f.py::fn_ours": _sym("fn_ours")} |
| 1444 | theirs = {"f.py::fn_theirs": _sym("fn_theirs")} |
| 1445 | _, out = _run_plan_merge( |
| 1446 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1447 | ) |
| 1448 | data = json.loads(out) |
| 1449 | assert data["conflicts"] == 0 |
| 1450 | |
| 1451 | def test_both_add_same_symbol_different_content_conflict(self, repo: pathlib.Path) -> None: |
| 1452 | """Both branches add the same address with different content → conflict.""" |
| 1453 | base: SymbolTree = {} |
| 1454 | ours = {"f.py::fn_new": _sym("fn_new", content_id="OUR", body_hash="BH1")} |
| 1455 | theirs = {"f.py::fn_new": _sym("fn_new", content_id="THEIR", body_hash="BH2")} |
| 1456 | _, out = _run_plan_merge( |
| 1457 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1458 | ) |
| 1459 | data = json.loads(out) |
| 1460 | assert data["conflicts"] == 1 |
| 1461 | |
| 1462 | def test_both_add_same_symbol_same_content_no_conflict(self, repo: pathlib.Path) -> None: |
| 1463 | """Both branches add the same symbol with identical content → no conflict.""" |
| 1464 | base: SymbolTree = {} |
| 1465 | sym = _sym("fn_new", content_id="SAME") |
| 1466 | _, out = _run_plan_merge( |
| 1467 | repo, mock_ours_syms={"f.py::fn_new": sym}, |
| 1468 | mock_theirs_syms={"f.py::fn_new": sym}, |
| 1469 | mock_base_syms=base, |
| 1470 | ) |
| 1471 | data = json.loads(out) |
| 1472 | assert data["conflicts"] == 0 |
| 1473 | |
| 1474 | def test_both_delete_same_symbol_no_conflict(self, repo: pathlib.Path) -> None: |
| 1475 | """Both branches delete the same symbol → no conflict.""" |
| 1476 | base = {"f.py::fn": _sym("fn")} |
| 1477 | ours: SymbolTree = {} |
| 1478 | theirs: SymbolTree = {} |
| 1479 | _, out = _run_plan_merge( |
| 1480 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1481 | ) |
| 1482 | data = json.loads(out) |
| 1483 | assert data["conflicts"] == 0 |
| 1484 | |
| 1485 | def test_rename_edit_and_overlap_in_same_plan(self, repo: pathlib.Path) -> None: |
| 1486 | """A plan with both rename_edit and symbol_edit_overlap appears correctly in conflicts_by_type.""" |
| 1487 | body = "UNIQUEBODY12345" |
| 1488 | # Symbol A: ours renames it, theirs modifies original → rename_edit |
| 1489 | base_a = {"f.py::fn_a_old": {**_sym("fn_a_old", body_hash=body), "name": "fn_a_old"}} |
| 1490 | ours_a = {"f.py::fn_a_new": {**_sym("fn_a_new", body_hash=body), "name": "fn_a_new"}} |
| 1491 | theirs_a = {"f.py::fn_a_old": {**_sym("fn_a_old", content_id="Z", body_hash="DIFF"), "name": "fn_a_old"}} |
| 1492 | # Symbol B: both change differently → symbol_edit_overlap |
| 1493 | base_b = {"f.py::fn_b": _sym("fn_b", content_id="X", body_hash="BH0", signature_id="SIG")} |
| 1494 | ours_b = {"f.py::fn_b": _sym("fn_b", content_id="Y", body_hash="BH1", signature_id="SIG")} |
| 1495 | theirs_b = {"f.py::fn_b": _sym("fn_b", content_id="Z2", body_hash="BH2", signature_id="SIG")} |
| 1496 | |
| 1497 | base = {**base_a, **base_b} |
| 1498 | ours = {**ours_a, **ours_b} |
| 1499 | theirs = {**theirs_a, **theirs_b} |
| 1500 | |
| 1501 | _, out = _run_plan_merge( |
| 1502 | repo, mock_ours_syms=ours, mock_theirs_syms=theirs, mock_base_syms=base, |
| 1503 | ) |
| 1504 | data = json.loads(out) |
| 1505 | cbt = data["conflicts_by_type"] |
| 1506 | assert cbt.get("rename_edit", 0) >= 1 |
| 1507 | assert cbt.get("symbol_edit_overlap", 0) >= 1 |
| 1508 | assert sum(cbt.values()) == data["conflicts"] |
| 1509 | |
| 1510 | |
| 1511 | class TestRegisterFlags: |
| 1512 | def _parse(self, *args: str) -> "argparse.Namespace": |
| 1513 | import argparse |
| 1514 | from muse.cli.commands.plan_merge import register |
| 1515 | p = argparse.ArgumentParser() |
| 1516 | subs = p.add_subparsers() |
| 1517 | register(subs) |
| 1518 | return p.parse_args(["plan-merge", "HEAD", "main", *args]) |
| 1519 | |
| 1520 | def test_json_short_flag(self) -> None: |
| 1521 | args = self._parse("-j") |
| 1522 | assert args.json_out is True |
| 1523 | |
| 1524 | def test_json_long_flag(self) -> None: |
| 1525 | args = self._parse("--json") |
| 1526 | assert args.json_out is True |
| 1527 | |
| 1528 | def test_default_no_json(self) -> None: |
| 1529 | args = self._parse() |
| 1530 | assert args.json_out is False |
File History
1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d
feat(pack): delta-encode snapshots in MPackBundle wire format
Sonnet 4.6
minor
⚠
122 days ago