test_cmd_merge_hardening.py
python
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
133 days ago
| 1 | """Hardening tests for ``muse merge`` — security, schema, error routing. |
| 2 | |
| 3 | These tests cover the 9 issues fixed in the security/correctness/agent-UX |
| 4 | audit. They are intentionally distinct from the existing test_cmd_merge.py |
| 5 | and test_cmd_merge_dry_run.py suites, which cover the core merge algorithm. |
| 6 | |
| 7 | Coverage tiers |
| 8 | -------------- |
| 9 | Unit — parser flags, dead-code removal, _use_color, _semver_from_op_log. |
| 10 | Integration — error routing to stderr, JSON schema stability across all statuses. |
| 11 | End-to-end — full CLI: security, branch-name sanitization, abort, strategy JSON. |
| 12 | Security — ANSI injection in branch names, commit messages, conflict paths. |
| 13 | Stress — large merges, concurrent repos, abort+re-merge cycles. |
| 14 | """ |
| 15 | |
| 16 | from __future__ import annotations |
| 17 | |
| 18 | import json |
| 19 | import os |
| 20 | import pathlib |
| 21 | import subprocess |
| 22 | import threading |
| 23 | import time |
| 24 | from typing import TYPE_CHECKING |
| 25 | |
| 26 | import pytest |
| 27 | |
| 28 | from tests.cli_test_helper import CliRunner, InvokeResult |
| 29 | from muse.core.store import get_head_commit_id, read_commit, read_current_branch |
| 30 | |
| 31 | if TYPE_CHECKING: |
| 32 | import argparse |
| 33 | |
| 34 | runner = CliRunner() |
| 35 | |
| 36 | # ────────────────────────────────────────────────────────────────────────────── |
| 37 | # Helpers |
| 38 | # ────────────────────────────────────────────────────────────────────────────── |
| 39 | |
| 40 | JSON_REQUIRED_KEYS = { |
| 41 | "status", "commit_id", "branch", "current_branch", |
| 42 | "base_commit_id", "conflicts", "files_changed", "semver_impact", |
| 43 | "strategy", "dry_run", |
| 44 | } |
| 45 | |
| 46 | |
| 47 | def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult: |
| 48 | saved = os.getcwd() |
| 49 | try: |
| 50 | os.chdir(repo) |
| 51 | return runner.invoke(None, args) |
| 52 | finally: |
| 53 | os.chdir(saved) |
| 54 | |
| 55 | |
| 56 | def _merge(repo: pathlib.Path, *extra: str) -> InvokeResult: |
| 57 | return _invoke(repo, ["merge", *extra]) |
| 58 | |
| 59 | |
| 60 | def _commit(repo: pathlib.Path, *extra: str) -> InvokeResult: |
| 61 | return _invoke(repo, ["commit", *extra]) |
| 62 | |
| 63 | |
| 64 | @pytest.fixture() |
| 65 | def repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 66 | """Initialised repo with one commit on ``main``.""" |
| 67 | saved = os.getcwd() |
| 68 | try: |
| 69 | os.chdir(tmp_path) |
| 70 | runner.invoke(None, ["init"]) |
| 71 | finally: |
| 72 | os.chdir(saved) |
| 73 | (tmp_path / "a.py").write_text("x = 1\n") |
| 74 | _commit(tmp_path, "-m", "initial") |
| 75 | return tmp_path |
| 76 | |
| 77 | |
| 78 | @pytest.fixture() |
| 79 | def ff_repo(repo: pathlib.Path) -> pathlib.Path: |
| 80 | """Repo where ``feat`` is strictly ahead of ``main`` → fast-forward.""" |
| 81 | _invoke(repo, ["branch", "feat"]) |
| 82 | _invoke(repo, ["checkout", "feat"]) |
| 83 | (repo / "b.py").write_text("y = 2\n") |
| 84 | _commit(repo, "-m", "feat add b") |
| 85 | _invoke(repo, ["checkout", "main"]) |
| 86 | return repo |
| 87 | |
| 88 | |
| 89 | @pytest.fixture() |
| 90 | def three_way_repo(repo: pathlib.Path) -> pathlib.Path: |
| 91 | """Repo requiring a clean three-way merge (both sides diverged).""" |
| 92 | _invoke(repo, ["branch", "feat"]) |
| 93 | _invoke(repo, ["checkout", "feat"]) |
| 94 | (repo / "b.py").write_text("y = 2\n") |
| 95 | _commit(repo, "-m", "feat add b") |
| 96 | _invoke(repo, ["checkout", "main"]) |
| 97 | (repo / "c.py").write_text("z = 3\n") |
| 98 | _commit(repo, "-m", "main add c") |
| 99 | return repo |
| 100 | |
| 101 | |
| 102 | @pytest.fixture() |
| 103 | def conflict_repo(repo: pathlib.Path) -> pathlib.Path: |
| 104 | """Repo where both sides modified the same file — conflict.""" |
| 105 | _invoke(repo, ["branch", "feat"]) |
| 106 | _invoke(repo, ["checkout", "feat"]) |
| 107 | (repo / "a.py").write_text("x = 999\n") |
| 108 | _commit(repo, "-m", "feat modify a") |
| 109 | _invoke(repo, ["checkout", "main"]) |
| 110 | (repo / "a.py").write_text("x = 42\n") |
| 111 | _commit(repo, "-m", "main modify a") |
| 112 | return repo |
| 113 | |
| 114 | |
| 115 | # ────────────────────────────────────────────────────────────────────────────── |
| 116 | # Unit — parser flags |
| 117 | # ────────────────────────────────────────────────────────────────────────────── |
| 118 | |
| 119 | |
| 120 | class TestRegisterFlags: |
| 121 | def _parse(self, *args: str) -> "argparse.Namespace": |
| 122 | import argparse |
| 123 | |
| 124 | from muse.cli.commands.merge import register |
| 125 | |
| 126 | p = argparse.ArgumentParser() |
| 127 | sub = p.add_subparsers() |
| 128 | register(sub) |
| 129 | return p.parse_args(["merge", *args]) |
| 130 | |
| 131 | def test_default_json_is_false(self) -> None: |
| 132 | ns = self._parse("feat") |
| 133 | assert ns.json_out is False |
| 134 | |
| 135 | def test_json_flag_sets_json_out(self) -> None: |
| 136 | ns = self._parse("feat", "--json") |
| 137 | assert ns.json_out is True |
| 138 | |
| 139 | def test_dry_run_default_false(self) -> None: |
| 140 | ns = self._parse("feat") |
| 141 | assert ns.dry_run is False |
| 142 | |
| 143 | def test_dry_run_flag(self) -> None: |
| 144 | ns = self._parse("feat", "--dry-run") |
| 145 | assert ns.dry_run is True |
| 146 | |
| 147 | def test_strategy_default_none(self) -> None: |
| 148 | ns = self._parse("feat") |
| 149 | assert ns.strategy is None |
| 150 | |
| 151 | def test_strategy_ours(self) -> None: |
| 152 | ns = self._parse("feat", "--strategy", "ours") |
| 153 | assert ns.strategy == "ours" |
| 154 | |
| 155 | def test_strategy_theirs(self) -> None: |
| 156 | ns = self._parse("feat", "--strategy", "theirs") |
| 157 | assert ns.strategy == "theirs" |
| 158 | |
| 159 | def test_no_ff_default_false(self) -> None: |
| 160 | ns = self._parse("feat") |
| 161 | assert ns.no_ff is False |
| 162 | |
| 163 | def test_abort_default_false(self) -> None: |
| 164 | ns = self._parse() |
| 165 | assert ns.abort is False |
| 166 | |
| 167 | def test_harmony_autoupdate_default_true(self) -> None: |
| 168 | ns = self._parse("feat") |
| 169 | assert ns.harmony_autoupdate is True |
| 170 | |
| 171 | def test_no_harmony_autoupdate(self) -> None: |
| 172 | ns = self._parse("feat", "--no-harmony-autoupdate") |
| 173 | assert ns.harmony_autoupdate is False |
| 174 | |
| 175 | |
| 176 | # ────────────────────────────────────────────────────────────────────────────── |
| 177 | # Unit — dead-code removal |
| 178 | # ────────────────────────────────────────────────────────────────────────────── |
| 179 | |
| 180 | |
| 181 | class TestDeadCodeRemoved: |
| 182 | def test_read_branch_wrapper_removed(self) -> None: |
| 183 | import muse.cli.commands.merge as m |
| 184 | |
| 185 | assert not hasattr(m, "_read_branch"), ( |
| 186 | "_read_branch was a dead one-liner wrapper and must be deleted" |
| 187 | ) |
| 188 | |
| 189 | def test_restore_from_manifest_wrapper_removed(self) -> None: |
| 190 | import muse.cli.commands.merge as m |
| 191 | |
| 192 | assert not hasattr(m, "_restore_from_manifest"), ( |
| 193 | "_restore_from_manifest was a dead one-liner wrapper and must be deleted" |
| 194 | ) |
| 195 | |
| 196 | |
| 197 | # ────────────────────────────────────────────────────────────────────────────── |
| 198 | # Unit — _use_color |
| 199 | # ────────────────────────────────────────────────────────────────────────────── |
| 200 | |
| 201 | |
| 202 | class TestUseColor: |
| 203 | def test_no_color_env_disables_color(self, monkeypatch: pytest.MonkeyPatch) -> None: |
| 204 | from muse.core.terminal import use_color |
| 205 | |
| 206 | monkeypatch.setenv("NO_COLOR", "1") |
| 207 | assert use_color() is False |
| 208 | |
| 209 | def test_term_dumb_disables_color(self, monkeypatch: pytest.MonkeyPatch) -> None: |
| 210 | from muse.core.terminal import use_color |
| 211 | |
| 212 | monkeypatch.setenv("TERM", "dumb") |
| 213 | assert use_color() is False |
| 214 | |
| 215 | def test_c_helper_respects_use_color(self, monkeypatch: pytest.MonkeyPatch) -> None: |
| 216 | from muse.cli.commands.merge import _c, _GREEN |
| 217 | |
| 218 | monkeypatch.setenv("NO_COLOR", "1") |
| 219 | result = _c("hello", _GREEN) |
| 220 | assert "\x1b[" not in result |
| 221 | assert result == "hello" |
| 222 | |
| 223 | |
| 224 | # ────────────────────────────────────────────────────────────────────────────── |
| 225 | # Unit — _semver_from_op_log |
| 226 | # ────────────────────────────────────────────────────────────────────────────── |
| 227 | |
| 228 | |
| 229 | class TestSemverFromOpLog: |
| 230 | """Verify _semver_from_op_log with an empty list (the only type-safe call site). |
| 231 | Non-empty behaviour is exercised via dry-run integration tests below, |
| 232 | which receive semver_impact in the JSON output after a real plugin diff.""" |
| 233 | |
| 234 | def test_empty_returns_empty(self) -> None: |
| 235 | from muse.cli.commands.merge import _semver_from_op_log |
| 236 | |
| 237 | assert _semver_from_op_log([]) == "" |
| 238 | |
| 239 | def test_semver_impact_present_in_dry_run_json( |
| 240 | self, three_way_repo: pathlib.Path |
| 241 | ) -> None: |
| 242 | """semver_impact is always a str in the dry-run JSON schema.""" |
| 243 | result = _merge(three_way_repo, "feat", "--dry-run", "--json") |
| 244 | data = json.loads(result.output) |
| 245 | assert "semver_impact" in data |
| 246 | assert isinstance(data["semver_impact"], str) |
| 247 | |
| 248 | def test_semver_impact_present_in_live_merge_json( |
| 249 | self, three_way_repo: pathlib.Path |
| 250 | ) -> None: |
| 251 | result = _merge(three_way_repo, "feat", "--json") |
| 252 | assert result.exit_code == 0 |
| 253 | data = json.loads(result.output) |
| 254 | assert isinstance(data["semver_impact"], str) |
| 255 | |
| 256 | |
| 257 | # ────────────────────────────────────────────────────────────────────────────── |
| 258 | # Integration — error routing to stderr |
| 259 | # ────────────────────────────────────────────────────────────────────────────── |
| 260 | |
| 261 | |
| 262 | class TestErrorRouting: |
| 263 | def test_merge_itself_error_to_stderr(self, repo: pathlib.Path) -> None: |
| 264 | result = _merge(repo, "main") |
| 265 | assert result.exit_code == 1 |
| 266 | assert "Cannot merge" in (result.stderr or "") |
| 267 | assert "Cannot merge" not in result.output.replace(result.stderr or "", "") |
| 268 | |
| 269 | def test_no_branch_arg_error_to_stderr(self, repo: pathlib.Path) -> None: |
| 270 | result = _merge(repo) |
| 271 | assert result.exit_code == 1 |
| 272 | assert "Usage" in (result.stderr or "") |
| 273 | |
| 274 | def test_nonexistent_branch_error_to_stderr(self, repo: pathlib.Path) -> None: |
| 275 | result = _merge(repo, "ghost-branch") |
| 276 | assert result.exit_code == 1 |
| 277 | assert "no commits" in (result.stderr or "").lower() |
| 278 | |
| 279 | def test_unrecognized_flag_exits_nonzero(self, repo: pathlib.Path) -> None: |
| 280 | result = _merge(repo, "main", "--no-such-flag") |
| 281 | assert result.exit_code != 0 |
| 282 | |
| 283 | def test_conflict_error_to_stderr(self, conflict_repo: pathlib.Path) -> None: |
| 284 | result = _merge(conflict_repo, "feat") |
| 285 | assert result.exit_code == 1 |
| 286 | assert "conflict" in (result.stderr or "").lower() |
| 287 | |
| 288 | def test_abort_no_merge_error_to_stderr(self, repo: pathlib.Path) -> None: |
| 289 | result = _merge(repo, "--abort") |
| 290 | assert result.exit_code == 1 |
| 291 | assert "No merge in progress" in (result.stderr or "") |
| 292 | |
| 293 | |
| 294 | # ────────────────────────────────────────────────────────────────────────────── |
| 295 | # Integration — JSON schema stability |
| 296 | # ────────────────────────────────────────────────────────────────────────────── |
| 297 | |
| 298 | |
| 299 | class TestJsonSchema: |
| 300 | def test_up_to_date_has_all_keys(self, ff_repo: pathlib.Path) -> None: |
| 301 | """First merge puts us at up-to-date; merge again to get up_to_date status.""" |
| 302 | _merge(ff_repo, "feat") |
| 303 | result = _merge(ff_repo, "feat", "--json") |
| 304 | data = json.loads(result.output) |
| 305 | assert data["status"] == "up_to_date" |
| 306 | missing = JSON_REQUIRED_KEYS - set(data) |
| 307 | assert not missing, f"Missing keys in up_to_date JSON: {missing}" |
| 308 | |
| 309 | def test_fast_forward_has_all_keys(self, ff_repo: pathlib.Path) -> None: |
| 310 | result = _merge(ff_repo, "feat", "--json") |
| 311 | data = json.loads(result.output) |
| 312 | assert data["status"] == "fast_forward" |
| 313 | missing = JSON_REQUIRED_KEYS - set(data) |
| 314 | assert not missing, f"Missing keys in fast_forward JSON: {missing}" |
| 315 | |
| 316 | def test_three_way_merged_has_all_keys(self, three_way_repo: pathlib.Path) -> None: |
| 317 | result = _merge(three_way_repo, "feat", "--json") |
| 318 | assert result.exit_code == 0 |
| 319 | data = json.loads(result.output) |
| 320 | assert data["status"] == "merged" |
| 321 | missing = JSON_REQUIRED_KEYS - set(data) |
| 322 | assert not missing, f"Missing keys in three-way merged JSON: {missing}" |
| 323 | |
| 324 | def test_conflict_has_all_keys(self, conflict_repo: pathlib.Path) -> None: |
| 325 | result = _merge(conflict_repo, "feat", "--json") |
| 326 | assert result.exit_code == 1 |
| 327 | data = json.loads(result.output) |
| 328 | assert data["status"] == "conflict" |
| 329 | # conflict status uses same schema (minus commit_id which is null) |
| 330 | core_keys = JSON_REQUIRED_KEYS - {"symbol_conflicts"} |
| 331 | missing = core_keys - set(data) |
| 332 | assert not missing, f"Missing keys in conflict JSON: {missing}" |
| 333 | |
| 334 | def test_dry_run_merged_has_all_keys(self, three_way_repo: pathlib.Path) -> None: |
| 335 | result = _merge(three_way_repo, "feat", "--dry-run", "--json") |
| 336 | assert result.exit_code == 0 |
| 337 | data = json.loads(result.output) |
| 338 | assert data["dry_run"] is True |
| 339 | missing = JSON_REQUIRED_KEYS - set(data) |
| 340 | assert not missing, f"Missing keys in dry-run merged JSON: {missing}" |
| 341 | |
| 342 | def test_strategy_ours_has_all_keys(self, conflict_repo: pathlib.Path) -> None: |
| 343 | result = _merge(conflict_repo, "feat", "--strategy", "ours", "--json") |
| 344 | assert result.exit_code == 0 |
| 345 | data = json.loads(result.output) |
| 346 | assert data["strategy"] == "ours" |
| 347 | missing = JSON_REQUIRED_KEYS - set(data) |
| 348 | assert not missing, f"Missing keys in strategy=ours JSON: {missing}" |
| 349 | |
| 350 | def test_strategy_theirs_has_all_keys(self, conflict_repo: pathlib.Path) -> None: |
| 351 | result = _merge(conflict_repo, "feat", "--strategy", "theirs", "--json") |
| 352 | assert result.exit_code == 0 |
| 353 | data = json.loads(result.output) |
| 354 | assert data["strategy"] == "theirs" |
| 355 | missing = JSON_REQUIRED_KEYS - set(data) |
| 356 | assert not missing, f"Missing keys in strategy=theirs JSON: {missing}" |
| 357 | |
| 358 | def test_fast_forward_has_base_commit_id(self, ff_repo: pathlib.Path) -> None: |
| 359 | result = _merge(ff_repo, "feat", "--json") |
| 360 | data = json.loads(result.output) |
| 361 | assert "base_commit_id" in data |
| 362 | assert data["base_commit_id"] is not None # FF always has a base |
| 363 | |
| 364 | def test_three_way_has_files_changed(self, three_way_repo: pathlib.Path) -> None: |
| 365 | result = _merge(three_way_repo, "feat", "--json") |
| 366 | assert result.exit_code == 0 |
| 367 | data = json.loads(result.output) |
| 368 | fc = data["files_changed"] |
| 369 | assert "added" in fc and "modified" in fc and "deleted" in fc |
| 370 | |
| 371 | def test_three_way_has_semver_impact(self, three_way_repo: pathlib.Path) -> None: |
| 372 | result = _merge(three_way_repo, "feat", "--json") |
| 373 | assert result.exit_code == 0 |
| 374 | data = json.loads(result.output) |
| 375 | assert "semver_impact" in data |
| 376 | assert isinstance(data["semver_impact"], str) |
| 377 | |
| 378 | def test_three_way_has_strategy_null(self, three_way_repo: pathlib.Path) -> None: |
| 379 | result = _merge(three_way_repo, "feat", "--json") |
| 380 | assert result.exit_code == 0 |
| 381 | data = json.loads(result.output) |
| 382 | assert data["strategy"] is None |
| 383 | |
| 384 | def test_three_way_has_dry_run_false(self, three_way_repo: pathlib.Path) -> None: |
| 385 | result = _merge(three_way_repo, "feat", "--json") |
| 386 | assert result.exit_code == 0 |
| 387 | data = json.loads(result.output) |
| 388 | assert data["dry_run"] is False |
| 389 | |
| 390 | def test_dry_run_commit_id_is_null(self, three_way_repo: pathlib.Path) -> None: |
| 391 | result = _merge(three_way_repo, "feat", "--dry-run", "--json") |
| 392 | data = json.loads(result.output) |
| 393 | assert data["commit_id"] is None |
| 394 | |
| 395 | def test_live_merge_commit_id_is_sha(self, three_way_repo: pathlib.Path) -> None: |
| 396 | result = _merge(three_way_repo, "feat", "--json") |
| 397 | data = json.loads(result.output) |
| 398 | assert data["commit_id"] is not None |
| 399 | assert data["commit_id"].startswith("sha256:") # canonical OID |
| 400 | |
| 401 | def test_files_changed_correct_for_ff(self, ff_repo: pathlib.Path) -> None: |
| 402 | result = _merge(ff_repo, "feat", "--json") |
| 403 | data = json.loads(result.output) |
| 404 | fc = data["files_changed"] |
| 405 | assert fc["added"] == 1 # b.py added on feat |
| 406 | assert fc["modified"] == 0 |
| 407 | assert fc["deleted"] == 0 |
| 408 | |
| 409 | |
| 410 | # ────────────────────────────────────────────────────────────────────────────── |
| 411 | # Integration — abort |
| 412 | # ────────────────────────────────────────────────────────────────────────────── |
| 413 | |
| 414 | |
| 415 | class TestAbort: |
| 416 | def test_abort_no_merge_exits_1(self, repo: pathlib.Path) -> None: |
| 417 | result = _merge(repo, "--abort") |
| 418 | assert result.exit_code == 1 |
| 419 | |
| 420 | def test_abort_no_merge_json(self, repo: pathlib.Path) -> None: |
| 421 | result = _merge(repo, "--abort", "--json") |
| 422 | data = json.loads(result.output) |
| 423 | assert data["error"] == "no_merge_in_progress" |
| 424 | |
| 425 | def test_abort_restores_working_tree(self, conflict_repo: pathlib.Path) -> None: |
| 426 | original = (conflict_repo / "a.py").read_text() |
| 427 | _merge(conflict_repo, "feat") # leaves conflict state |
| 428 | # Verify conflict state was created |
| 429 | assert (conflict_repo / ".muse" / "MERGE_STATE.json").exists() |
| 430 | result = _merge(conflict_repo, "--abort") |
| 431 | assert result.exit_code == 0 |
| 432 | # MERGE_STATE should be gone |
| 433 | assert not (conflict_repo / ".muse" / "MERGE_STATE.json").exists() |
| 434 | |
| 435 | def test_abort_json_has_status(self, conflict_repo: pathlib.Path) -> None: |
| 436 | _merge(conflict_repo, "feat") |
| 437 | result = _merge(conflict_repo, "--abort", "--json") |
| 438 | assert result.exit_code == 0 |
| 439 | data = json.loads(result.output) |
| 440 | assert data["status"] == "aborted" |
| 441 | assert "restored_to" in data |
| 442 | |
| 443 | def test_abort_uses_read_merge_state(self, conflict_repo: pathlib.Path) -> None: |
| 444 | """_run_abort must use read_merge_state() not raw json.loads().""" |
| 445 | import inspect |
| 446 | from muse.cli.commands.merge import _run_abort |
| 447 | |
| 448 | src = inspect.getsource(_run_abort) |
| 449 | assert "json.loads" not in src, ( |
| 450 | "_run_abort must use read_merge_state() instead of raw json.loads() " |
| 451 | "to benefit from schema validation" |
| 452 | ) |
| 453 | assert "read_merge_state" in src |
| 454 | |
| 455 | |
| 456 | # ────────────────────────────────────────────────────────────────────────────── |
| 457 | # Security — ANSI injection |
| 458 | # ────────────────────────────────────────────────────────────────────────────── |
| 459 | |
| 460 | |
| 461 | class TestSecurityAnsi: |
| 462 | ESC = "\x1b[" |
| 463 | |
| 464 | def test_error_routing_no_ansi_in_stdout(self, repo: pathlib.Path) -> None: |
| 465 | """Error messages for invalid operations must not bleed ANSI into stdout.""" |
| 466 | result = _merge(repo, "main") # merge into itself |
| 467 | assert self.ESC not in result.output.replace(result.stderr or "", "") |
| 468 | |
| 469 | def test_ansi_in_strategy_arg_sanitized_in_stderr(self, repo: pathlib.Path) -> None: |
| 470 | result = _merge(repo, "main", "--strategy", f"{self.ESC}31mxml{self.ESC}0m") |
| 471 | assert self.ESC not in (result.stderr or "") |
| 472 | |
| 473 | def test_conflict_paths_sanitized_in_json(self, conflict_repo: pathlib.Path) -> None: |
| 474 | """Any ANSI that leaked into conflict_paths must be sanitized in JSON output.""" |
| 475 | result = _merge(conflict_repo, "feat", "--json") |
| 476 | data = json.loads(result.output) |
| 477 | for path in data.get("conflicts", []): |
| 478 | assert self.ESC not in path |
| 479 | |
| 480 | def test_merge_message_sanitized_in_commit(self, three_way_repo: pathlib.Path) -> None: |
| 481 | """Branch name embedded in merge commit message must be ANSI-clean.""" |
| 482 | result = _merge(three_way_repo, "feat") |
| 483 | assert result.exit_code == 0 |
| 484 | cid = get_head_commit_id(three_way_repo, "main") |
| 485 | assert cid is not None |
| 486 | commit = read_commit(three_way_repo, cid) |
| 487 | assert commit is not None |
| 488 | assert self.ESC not in commit.message |
| 489 | |
| 490 | def test_applied_strategies_sanitized(self, three_way_repo: pathlib.Path) -> None: |
| 491 | """Any applied_strategies entries are sanitized before printing.""" |
| 492 | result = _merge(three_way_repo, "feat") |
| 493 | assert self.ESC not in result.output |
| 494 | |
| 495 | |
| 496 | # ────────────────────────────────────────────────────────────────────────────── |
| 497 | # Integration — strategy shortcuts |
| 498 | # ────────────────────────────────────────────────────────────────────────────── |
| 499 | |
| 500 | |
| 501 | class TestStrategy: |
| 502 | def test_strategy_ours_resolves_conflict(self, conflict_repo: pathlib.Path) -> None: |
| 503 | result = _merge(conflict_repo, "feat", "--strategy", "ours") |
| 504 | assert result.exit_code == 0 |
| 505 | |
| 506 | def test_strategy_ours_keeps_our_content(self, conflict_repo: pathlib.Path) -> None: |
| 507 | _merge(conflict_repo, "feat", "--strategy", "ours") |
| 508 | content = (conflict_repo / "a.py").read_text() |
| 509 | assert "42" in content # main's version |
| 510 | |
| 511 | def test_strategy_theirs_keeps_their_content(self, conflict_repo: pathlib.Path) -> None: |
| 512 | _merge(conflict_repo, "feat", "--strategy", "theirs") |
| 513 | content = (conflict_repo / "a.py").read_text() |
| 514 | assert "999" in content # feat's version |
| 515 | |
| 516 | def test_strategy_ours_creates_merge_commit(self, conflict_repo: pathlib.Path) -> None: |
| 517 | before = get_head_commit_id(conflict_repo, "main") |
| 518 | _merge(conflict_repo, "feat", "--strategy", "ours") |
| 519 | after = get_head_commit_id(conflict_repo, "main") |
| 520 | assert after != before |
| 521 | |
| 522 | def test_strategy_json_has_correct_strategy_field(self, conflict_repo: pathlib.Path) -> None: |
| 523 | result = _merge(conflict_repo, "feat", "--strategy", "ours", "--json") |
| 524 | data = json.loads(result.output) |
| 525 | assert data["strategy"] == "ours" |
| 526 | |
| 527 | def test_strategy_json_has_files_changed(self, conflict_repo: pathlib.Path) -> None: |
| 528 | result = _merge(conflict_repo, "feat", "--strategy", "ours", "--json") |
| 529 | data = json.loads(result.output) |
| 530 | assert "files_changed" in data |
| 531 | |
| 532 | def test_strategy_dry_run_does_not_write(self, conflict_repo: pathlib.Path) -> None: |
| 533 | before = get_head_commit_id(conflict_repo, "main") |
| 534 | # --dry-run bypasses strategy shortcuts; simulates three-way instead |
| 535 | result = _merge(conflict_repo, "feat", "--strategy", "ours", "--dry-run") |
| 536 | after = get_head_commit_id(conflict_repo, "main") |
| 537 | assert before == after |
| 538 | |
| 539 | |
| 540 | # ────────────────────────────────────────────────────────────────────────────── |
| 541 | # Stress |
| 542 | # ────────────────────────────────────────────────────────────────────────────── |
| 543 | |
| 544 | |
| 545 | @pytest.mark.slow |
| 546 | class TestStress: |
| 547 | def test_merge_100_file_branch_fast(self, repo: pathlib.Path) -> None: |
| 548 | """Merging 100 new files must complete in under 5s.""" |
| 549 | _invoke(repo, ["branch", "big-feat"]) |
| 550 | _invoke(repo, ["checkout", "big-feat"]) |
| 551 | for i in range(100): |
| 552 | (repo / f"f{i:03d}.py").write_text(f"x={i}\n") |
| 553 | _commit(repo, "-m", "add 100 files") |
| 554 | _invoke(repo, ["checkout", "main"]) |
| 555 | (repo / "main_extra.py").write_text("m=1\n") |
| 556 | _commit(repo, "-m", "main diverges") |
| 557 | |
| 558 | t0 = time.perf_counter() |
| 559 | result = _merge(repo, "big-feat") |
| 560 | elapsed = (time.perf_counter() - t0) * 1000 |
| 561 | assert result.exit_code == 0 |
| 562 | assert elapsed < 5000, f"100-file merge took {elapsed:.0f}ms (limit 5s)" |
| 563 | |
| 564 | def test_dry_run_100_file_branch_fast(self, repo: pathlib.Path) -> None: |
| 565 | """Dry-run of a 100-file merge must complete in under 3s.""" |
| 566 | _invoke(repo, ["branch", "big-feat"]) |
| 567 | _invoke(repo, ["checkout", "big-feat"]) |
| 568 | for i in range(100): |
| 569 | (repo / f"g{i:03d}.py").write_text(f"y={i}\n") |
| 570 | _commit(repo, "-m", "add 100 files") |
| 571 | _invoke(repo, ["checkout", "main"]) |
| 572 | (repo / "main_extra2.py").write_text("m=2\n") |
| 573 | _commit(repo, "-m", "main diverges") |
| 574 | |
| 575 | t0 = time.perf_counter() |
| 576 | result = _merge(repo, "big-feat", "--dry-run") |
| 577 | elapsed = (time.perf_counter() - t0) * 1000 |
| 578 | assert result.exit_code == 0 |
| 579 | assert elapsed < 3000, f"100-file dry-run took {elapsed:.0f}ms (limit 3s)" |
| 580 | |
| 581 | def test_abort_cycle_10_times(self, conflict_repo: pathlib.Path) -> None: |
| 582 | """Abort should cleanly reset MERGE_STATE each time.""" |
| 583 | for i in range(10): |
| 584 | r_merge = _merge(conflict_repo, "feat") |
| 585 | assert r_merge.exit_code == 1 # conflict |
| 586 | assert (conflict_repo / ".muse" / "MERGE_STATE.json").exists() |
| 587 | r_abort = _merge(conflict_repo, "--abort") |
| 588 | assert r_abort.exit_code == 0 |
| 589 | assert not (conflict_repo / ".muse" / "MERGE_STATE.json").exists() |
| 590 | |
| 591 | def test_concurrent_merges_separate_repos(self, tmp_path: pathlib.Path) -> None: |
| 592 | """Multiple repos merging concurrently must not interfere.""" |
| 593 | errors: list[str] = [] |
| 594 | |
| 595 | def do_merge(idx: int) -> None: |
| 596 | repo_dir = tmp_path / f"repo_{idx}" |
| 597 | repo_dir.mkdir() |
| 598 | subprocess.run(["muse", "init"], cwd=str(repo_dir), capture_output=True) |
| 599 | (repo_dir / "a.py").write_text(f"x={idx}\n") |
| 600 | subprocess.run( |
| 601 | ["muse", "commit", "-m", f"base{idx}"], |
| 602 | cwd=str(repo_dir), capture_output=True, |
| 603 | ) |
| 604 | subprocess.run( |
| 605 | ["muse", "branch", "feat"], cwd=str(repo_dir), capture_output=True |
| 606 | ) |
| 607 | subprocess.run( |
| 608 | ["muse", "checkout", "feat"], cwd=str(repo_dir), capture_output=True |
| 609 | ) |
| 610 | (repo_dir / "b.py").write_text(f"y={idx}\n") |
| 611 | subprocess.run( |
| 612 | ["muse", "commit", "-m", f"feat{idx}"], |
| 613 | cwd=str(repo_dir), capture_output=True, |
| 614 | ) |
| 615 | subprocess.run( |
| 616 | ["muse", "checkout", "main"], cwd=str(repo_dir), capture_output=True |
| 617 | ) |
| 618 | r = subprocess.run( |
| 619 | ["muse", "merge", "feat", "--json"], |
| 620 | cwd=str(repo_dir), capture_output=True, text=True, |
| 621 | ) |
| 622 | if r.returncode != 0: |
| 623 | errors.append(f"repo_{idx}: exit={r.returncode}") |
| 624 | return |
| 625 | try: |
| 626 | data = json.loads(r.stdout) |
| 627 | if data["status"] not in ("fast_forward", "merged"): |
| 628 | errors.append(f"repo_{idx}: unexpected status {data['status']}") |
| 629 | except Exception as e: |
| 630 | errors.append(f"repo_{idx}: {e}") |
| 631 | |
| 632 | threads = [threading.Thread(target=do_merge, args=(i,)) for i in range(6)] |
| 633 | for t in threads: |
| 634 | t.start() |
| 635 | for t in threads: |
| 636 | t.join() |
| 637 | assert not errors, "Concurrent merge errors:\n" + "\n".join(errors) |
File History
3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
133 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c
docs: docstring sprint for-each-ref→hotspots — idiomatic ru…
Sonnet 4.6
patch
139 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9
fix(cursorignore): remove git-ism (.git/worktrees)
Human
minor
⚠
142 days ago