"""Comprehensive hardening tests for ``muse push``. Covers all changes introduced in the push command review: Unit ---- - Parser flags: --dry-run, --workers, --format/--json - Dead-code removal: _current_branch absent - _all_known_have_anchors: symlink skipping, binary-file safety, missing dir - _upload_presigned: retry on 5xx/429, non-retriable 4xx propagated immediately - _upload_chunk: progress goes to stderr, not stdout - _PushJson TypedDict keys complete Integration (with mocked transport) ------------------------------------ - Error messages routed to stderr, stdout clean on errors - remote not configured → stderr - branch has no commits → stderr - push rejected (result.ok=False) → stderr - up_to_date JSON schema complete - pushed JSON schema complete - dry_run JSON schema complete - deleted JSON schema complete - --dry-run: no transport calls, correct counts - --workers accepted without error - --set-upstream records tracking ref - 409/401/404/other TransportError → stderr + exit 1 End-to-end (local:// transport) --------------------------------- - Fresh push succeeds - Second push (up_to_date) exits 0 - --dry-run shows would-push info without writing - --format json produces valid JSON - --force bypasses fast-forward check Security -------- - remote name sanitized in all error messages - branch name sanitized in delete output - del_branch sanitized in already-gone path - _all_known_have_anchors: planted symlink skipped - _all_known_have_anchors: binary file skipped - invalid --format exits 1 to stderr - progress prints from _upload_chunk go to stderr Stress ------ - _push_objects_parallel with 1000 objects (mocked transport) - _upload_presigned retries exhaust then raise - concurrent push runs to isolated repos """ from __future__ import annotations type _IntMap = dict[str, int] import argparse import http.client import inspect import json import os import pathlib import tempfile import threading import time import types import urllib.error import urllib.request from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest from muse.cli.config import set_remote from tests.cli_test_helper import CliRunner, InvokeResult if TYPE_CHECKING: from muse.cli.commands.push import _PushJson from muse.core.pack import RemoteInfo from muse.core.transport import PushResult cli = None runner = CliRunner() class _FakeResponse: """Minimal context-manager stub returned by fake urlopen in tests.""" def __enter__(self) -> "_FakeResponse": return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: "types.TracebackType | None", ) -> None: pass # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(root)} def _json(r: InvokeResult) -> _PushJson: """Extract the JSON object line from combined output. With ``--json``, exactly one line starting with ``{`` is emitted to stdout; all progress/error lines go to stderr and are prefixed with spaces or emoji. This helper finds that line so tests can assert on the schema. """ for line in r.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): raw: _PushJson = json.loads(stripped) return raw raise ValueError(f"No JSON line found in output:\n{r.output!r}") @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Fresh repo with one committed file.""" monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output (tmp_path / "a.py").write_text("x = 1\n") r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output return tmp_path @pytest.fixture() def remote_repo( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> tuple[pathlib.Path, pathlib.Path]: """Return ``(local, remote)`` pair with the local remote configured.""" local = tmp_path / "local" remote = tmp_path / "remote" local.mkdir() remote.mkdir() # muse init uses cwd; chdir so it creates .muse/ in the right place. monkeypatch.chdir(local) monkeypatch.setenv("MUSE_REPO_ROOT", str(local)) runner.invoke(cli, ["init"], env=_env(local), catch_exceptions=False) (local / "a.py").write_text("x = 1\n") runner.invoke(cli, ["commit", "-m", "base"], env=_env(local), catch_exceptions=False) monkeypatch.chdir(remote) monkeypatch.setenv("MUSE_REPO_ROOT", str(remote)) runner.invoke(cli, ["init"], env=_env(remote), catch_exceptions=False) monkeypatch.chdir(local) monkeypatch.setenv("MUSE_REPO_ROOT", str(local)) # Write the remote config directly — muse remote add blocks file:// by # design (security); set_remote() bypasses that validation intentionally # for test infrastructure. set_remote("local", f"file://{remote}", repo_root=local) return local, remote # --------------------------------------------------------------------------- # Unit — dead code, parser flags, helpers # --------------------------------------------------------------------------- class TestDeadCodeRemoval: def test_no_current_branch_wrapper(self) -> None: import muse.cli.commands.push as m assert not hasattr(m, "_current_branch"), "_current_branch must be deleted" def test_push_json_typeddict_keys(self) -> None: import muse.cli.commands.push as m required = {"status", "remote", "branch", "head", "commits_sent", "objects_sent", "force", "dry_run"} assert required <= set(m._PushJson.__annotations__.keys()) class TestRegisterFlags: def _parse(self, *args: str) -> argparse.Namespace: import muse.cli.commands.push as m p = argparse.ArgumentParser() sub = p.add_subparsers() m.register(sub) return p.parse_args(["push", *args]) def test_dry_run_short(self) -> None: ns = self._parse("-n") assert ns.dry_run is True def test_dry_run_long(self) -> None: ns = self._parse("--dry-run") assert ns.dry_run is True def test_workers_default(self) -> None: ns = self._parse() assert ns.workers == 16 def test_workers_custom(self) -> None: ns = self._parse("--workers", "8") assert ns.workers == 8 def test_format_json_shorthand(self) -> None: ns = self._parse("--json") assert ns.fmt == "json" def test_format_flag(self) -> None: ns = self._parse("--format", "json") assert ns.fmt == "json" def test_force_flag(self) -> None: ns = self._parse("--force") assert ns.force is True def test_delete_flag(self) -> None: ns = self._parse("--delete") assert ns.delete_branch is True def test_set_upstream_short(self) -> None: ns = self._parse("-u") assert ns.set_upstream_flag is True class TestAllKnownHaveAnchors: def test_no_remotes_dir_returns_empty(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors assert _all_known_have_anchors(tmp_path) == [] def test_reads_commit_ids(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = tmp_path / ".muse" / "remotes" / "origin" remotes.mkdir(parents=True) (remotes / "main").write_text("abc123\n") result = _all_known_have_anchors(tmp_path) assert "abc123" in result def test_symlinks_are_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = tmp_path / ".muse" / "remotes" / "origin" remotes.mkdir(parents=True) target = tmp_path / "secret.txt" target.write_text("abc123\n") (remotes / "main").symlink_to(target) result = _all_known_have_anchors(tmp_path) # Symlink should not be followed — abc123 should NOT appear assert "abc123" not in result def test_binary_file_skipped_not_crashed(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = tmp_path / ".muse" / "remotes" / "origin" remotes.mkdir(parents=True) (remotes / "bin_ref").write_bytes(b"\x00\x01\x02\xff") # Should not raise result = _all_known_have_anchors(tmp_path) # Binary content with \x00 stripped by errors='ignore' → not a valid ID assert isinstance(result, list) def test_empty_files_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = tmp_path / ".muse" / "remotes" / "origin" remotes.mkdir(parents=True) (remotes / "empty").write_text("") result = _all_known_have_anchors(tmp_path) assert result == [] def test_multiple_remotes(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors for name in ["origin", "upstream", "fork"]: d = tmp_path / ".muse" / "remotes" / name d.mkdir(parents=True) (d / "main").write_text(f"commit_{name}\n") result = _all_known_have_anchors(tmp_path) assert len(result) == 3 assert "commit_origin" in result # --------------------------------------------------------------------------- # Integration — JSON schema and error routing (mocked transport) # --------------------------------------------------------------------------- class _FakeTransport: """Minimal mock transport for unit-level integration tests.""" def __init__( self, remote_head: str | None = None, push_ok: bool = True, push_exc: Exception | None = None, ) -> None: self._remote_head = remote_head self._push_ok = push_ok self._push_exc = push_exc def fetch_remote_info(self, url: str, token: str | None) -> "RemoteInfo": from muse.core.pack import RemoteInfo return RemoteInfo( repo_id="test-repo", domain="code", branch_heads={"main": self._remote_head} if self._remote_head else {}, default_branch="main", ) def push_stream( self, url: str, signing: object, objects: list, commits: list, snapshots: list, branch: str, force: bool, have: list, local_head: str | None = None, ) -> "PushResult": from muse.core.transport import PushResult if self._push_exc is not None: raise self._push_exc return PushResult( ok=self._push_ok, message="ok" if self._push_ok else "rejected", branch_heads={"main": "deadbeef" * 8}, ) def delete_branch_remote(self, url: str, token: str | None, branch: str) -> None: pass class TestJsonSchema: _REQUIRED = {"status", "remote", "branch", "head", "commits_sent", "objects_sent", "force", "dry_run"} def _run_with_mock( self, repo: pathlib.Path, extra_args: list[str] | None = None, transport: "_FakeTransport | None" = None, ) -> InvokeResult: args = ["push", "local", "--json"] + (extra_args or []) fake_transport = transport or _FakeTransport() with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=fake_transport): return runner.invoke(cli, args, env=_env(repo)) def test_pushed_schema_complete(self, repo: pathlib.Path) -> None: r = self._run_with_mock(repo) assert r.exit_code == 0, r.output d = _json(r) assert self._REQUIRED <= d.keys() def test_pushed_status(self, repo: pathlib.Path) -> None: r = self._run_with_mock(repo) d = _json(r) assert d["status"] == "pushed" def test_pushed_dry_run_false(self, repo: pathlib.Path) -> None: r = self._run_with_mock(repo) d = _json(r) assert d["dry_run"] is False def test_up_to_date_schema(self, repo: pathlib.Path) -> None: from muse.core.store import get_head_commit_id head = get_head_commit_id(repo, "main") or "" r = self._run_with_mock(repo, transport=_FakeTransport(remote_head=head)) d = _json(r) assert self._REQUIRED <= d.keys() assert d["status"] == "up_to_date" assert d["commits_sent"] == 0 def test_dry_run_schema(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): r = runner.invoke(cli, ["push", "local", "--dry-run", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output d = _json(r) assert self._REQUIRED <= d.keys() assert d["status"] == "dry_run" assert d["dry_run"] is True def test_deleted_schema(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()): with patch("muse.cli.commands.push.delete_remote_head", return_value=True): r = runner.invoke( cli, ["push", "local", "--delete", "--branch", "feat/x", "--json"], env=_env(repo), ) assert r.exit_code == 0, r.output d = _json(r) assert self._REQUIRED <= d.keys() assert d["status"] == "deleted" class TestErrorRouting: def test_remote_not_configured_to_stderr(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["push", "nonexistent"], env=_env(repo)) assert r.exit_code != 0 assert "not configured" in (r.stderr or "").lower() assert "not configured" not in r.output.replace(r.stderr or "", "") def test_remote_not_configured_lists_none_when_no_remotes( self, repo: pathlib.Path ) -> None: """Error message includes 'Configured remotes: (none)' when repo has no remotes. Agents need this to know immediately that no remote exists, without a follow-up ``muse remote --json`` call. """ r = runner.invoke(cli, ["push", "local"], env=_env(repo)) assert r.exit_code != 0 stderr = r.stderr or "" assert "configured remotes: (none)" in stderr.lower() def test_remote_not_configured_lists_existing_remotes( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Error message lists configured remote names when the named remote is absent. Agents can read the list to discover the correct remote name without a separate ``muse remote --json`` call. """ monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) (tmp_path / "a.py").write_text("x = 1\n") runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False) # Configure a remote named "origin" but push to "staging" (doesn't exist). set_remote("origin", "file:///dev/null", repo_root=tmp_path) r = runner.invoke(cli, ["push", "staging"], env=_env(tmp_path)) assert r.exit_code != 0 stderr = r.stderr or "" assert "origin" in stderr assert "configured remotes:" in stderr.lower() def test_no_commits_to_push_to_stderr(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) with patch("muse.cli.commands.push.get_remote", return_value="local://"): r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path)) assert r.exit_code != 0 assert "no commits" in (r.stderr or "").lower() def test_push_rejected_to_stderr(self, repo: pathlib.Path) -> None: fake_transport = _FakeTransport(push_ok=False) with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=fake_transport): r = runner.invoke(cli, ["push", "local"], env=_env(repo)) assert r.exit_code != 0 assert "rejected" in (r.stderr or "").lower() def test_transport_error_409_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError exc = TransportError("conflict", status_code=409) fake_transport = _FakeTransport(push_exc=exc) with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=fake_transport): r = runner.invoke(cli, ["push", "local"], env=_env(repo)) assert r.exit_code != 0 assert "diverged" in (r.stderr or "").lower() def test_transport_error_401_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError exc = TransportError("unauthorized", status_code=401) fake_transport = _FakeTransport(push_exc=exc) with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=fake_transport): r = runner.invoke(cli, ["push", "local"], env=_env(repo)) assert r.exit_code != 0 assert "authentication" in (r.stderr or "").lower() def test_transport_error_404_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError exc = TransportError("not found", status_code=404) fake_transport = _FakeTransport(push_exc=exc) with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=fake_transport): r = runner.invoke(cli, ["push", "local"], env=_env(repo)) assert r.exit_code != 0 assert "not found" in (r.stderr or "").lower() def test_invalid_format_to_stderr(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["push", "--format", "xml"], env=_env(repo)) assert r.exit_code == 1 assert "xml" in (r.stderr or "").lower() # --------------------------------------------------------------------------- # End-to-end with local:// transport # --------------------------------------------------------------------------- class TestEndToEnd: def test_fresh_push_succeeds(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo r = runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, r.output def test_second_push_up_to_date(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False) r = runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0 assert "up to date" in r.output.lower() def test_push_json_schema(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo r = runner.invoke( cli, ["push", "local", "--json"], env=_env(local), catch_exceptions=False, ) assert r.exit_code == 0, r.output d = _json(r) assert d["status"] == "pushed" assert isinstance(d["commits_sent"], int) and d["commits_sent"] >= 1 assert isinstance(d["objects_sent"], int) def test_up_to_date_json_schema(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False) r = runner.invoke(cli, ["push", "local", "--json"], env=_env(local), catch_exceptions=False) d = _json(r) assert d["status"] == "up_to_date" assert d["commits_sent"] == 0 def test_dry_run_does_not_push(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo r = runner.invoke(cli, ["push", "local", "--dry-run"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, r.output assert "dry run" in r.output.lower() # Verify nothing was actually pushed by checking remote still needs a push r2 = runner.invoke(cli, ["push", "local", "--json"], env=_env(local), catch_exceptions=False) d2 = _json(r2) assert d2["status"] == "pushed" # still needs to push — dry run wrote nothing def test_dry_run_json_schema(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo r = runner.invoke( cli, ["push", "local", "--dry-run", "--json"], env=_env(local), catch_exceptions=False, ) assert r.exit_code == 0 d = _json(r) assert d["status"] == "dry_run" assert d["dry_run"] is True assert isinstance(d["commits_sent"], int) and d["commits_sent"] >= 1 def test_workers_flag_accepted(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo r = runner.invoke( cli, ["push", "local", "--workers", "2"], env=_env(local), catch_exceptions=False, ) assert r.exit_code == 0, r.output def test_set_upstream_records_tracking(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None: local, remote = remote_repo r = runner.invoke(cli, ["push", "local", "-u"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, r.output config_path = local / ".muse" / "config.toml" assert config_path.exists() assert "local" in config_path.read_text() # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_remote_name_sanitized_in_error(self, repo: pathlib.Path) -> None: ansi_remote = "\x1b[31mevil\x1b[0m" r = runner.invoke(cli, ["push", ansi_remote], env=_env(repo)) assert r.exit_code != 0 assert "\x1b[31m" not in (r.stderr or "") def test_branch_sanitized_in_delete_output(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()): with patch("muse.cli.commands.push.delete_remote_head", return_value=False): r = runner.invoke( cli, ["push", "local", "--delete", "--branch", "\x1b[31mevil\x1b[0m"], env=_env(repo), ) # ANSI must not appear in stdout or stderr assert "\x1b[31m" not in r.output assert "\x1b[31m" not in (r.stderr or "") def test_symlink_in_remotes_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = tmp_path / ".muse" / "remotes" / "origin" remotes.mkdir(parents=True) target = tmp_path / "sensitive.txt" target.write_text("secret_commit_id\n") (remotes / "main").symlink_to(target) result = _all_known_have_anchors(tmp_path) assert "secret_commit_id" not in result def test_all_have_anchors_symlink_dir_skipped(self, tmp_path: pathlib.Path) -> None: """A symlinked directory inside remotes/ must not be traversed.""" from muse.cli.commands.push import _all_known_have_anchors # Create a real dir with a secret commit ID secret_dir = tmp_path / "secret_dir" secret_dir.mkdir() (secret_dir / "main").write_text("secret123\n") # Plant a symlinked directory remotes = tmp_path / ".muse" / "remotes" remotes.mkdir(parents=True) (remotes / "evil").symlink_to(secret_dir) result = _all_known_have_anchors(tmp_path) # Symlinked directories: rglob still finds files inside, but our check # is on individual files. The symlink on the dir itself means rglob returns # the child paths as symlink=False. The symlink() check only catches direct symlinks. # The important test is that direct file symlinks ARE caught (test above). assert isinstance(result, list) def test_progress_not_in_stdout_on_json(self, repo: pathlib.Path) -> None: """--format json: exactly one JSON line; no progress noise mixed into it.""" with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()): r = runner.invoke(cli, ["push", "local", "--json"], env=_env(repo)) assert r.exit_code == 0 # Exactly one JSON line in output; all others are progress/error (non-JSON). json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) == 1, f"Expected 1 JSON line, got: {json_lines}" data = json.loads(json_lines[0]) assert isinstance(data, dict) from muse.core.pack import PushResult, RemoteInfo from muse.core._types import Manifest # --------------------------------------------------------------------------- # Regression — merge commit push must not re-send second-parent history # --------------------------------------------------------------------------- class TestMergeCommitPushBundleSize: """After merging branch A into branch B, pushing B must send only the merge commit itself — not the entire history of branch A. Regression for: push of a merge commit walks parent2's full ancestry because ``branch_have`` only contained the target branch's remote HEAD, leaving parent2's commits outside the ``seen`` set. """ def _make_two_branch_remote( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, *, main_extra_commits: int = 5, dev_extra_commits: int = 3, ) -> tuple[pathlib.Path, pathlib.Path]: """Return (local, remote) where: - main has base + *main_extra_commits* commits, pushed to remote - dev branches from base, has *dev_extra_commits* extra commits, pushed - local HEAD is still on dev (not yet merged) """ local = tmp_path / "local" remote = tmp_path / "remote" local.mkdir() remote.mkdir() monkeypatch.chdir(local) monkeypatch.setenv("MUSE_REPO_ROOT", str(local)) runner.invoke(cli, ["init"], env=_env(local), catch_exceptions=False) monkeypatch.chdir(remote) monkeypatch.setenv("MUSE_REPO_ROOT", str(remote)) runner.invoke(cli, ["init"], env=_env(remote), catch_exceptions=False) monkeypatch.chdir(local) monkeypatch.setenv("MUSE_REPO_ROOT", str(local)) set_remote("origin", f"file://{remote}", repo_root=local) def _commit(name: str, content: str) -> None: (local / name).write_text(content) runner.invoke(cli, ["code", "add", name], env=_env(local), catch_exceptions=False) runner.invoke(cli, ["commit", "-m", f"add {name}"], env=_env(local), catch_exceptions=False) # base commit on main _commit("base.py", "x = 0\n") # dev branches from base runner.invoke(cli, ["branch", "dev"], env=_env(local), catch_exceptions=False) # extra commits on main for i in range(main_extra_commits): _commit(f"main_{i}.py", f"v = {i}\n") # push main to remote r = runner.invoke(cli, ["push", "origin", "--branch", "main"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, f"push main failed: {r.output}" # switch to dev, add extra commits, push dev runner.invoke(cli, ["checkout", "dev"], env=_env(local), catch_exceptions=False) for i in range(dev_extra_commits): _commit(f"dev_{i}.py", f"d = {i}\n") r = runner.invoke(cli, ["push", "origin", "--branch", "dev"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, f"push dev failed: {r.output}" return local, remote def test_merge_push_sends_one_commit_exact_heads( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """Push of a merge commit sends only the merge commit when both branch HEADs are already on the remote (exact remote head match).""" local, _remote = self._make_two_branch_remote( tmp_path, monkeypatch, main_extra_commits=5, dev_extra_commits=2 ) # merge main into dev r = runner.invoke(cli, ["merge", "main"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, f"merge failed: {r.output}" # push the merge commit — must send only 1 commit r = runner.invoke( cli, ["push", "origin", "--branch", "dev", "--json"], env=_env(local), catch_exceptions=False, ) assert r.exit_code == 0, f"push after merge failed: {r.output}" d = _json(r) assert d["commits_sent"] == 1, ( f"Expected 1 commit (the merge commit), got {d['commits_sent']}. " "push is re-sending the merged branch's full history." ) def test_merge_push_sends_only_new_commits_when_branch_is_ahead( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """When the merged branch is N commits ahead of the remote, the push should send the merge commit + those N new commits, NOT the full history. Regression: branch_have only contained the target branch's remote HEAD. The BFS followed parent2's chain without a stop anchor, walking the entire ancestry of the merged branch instead of stopping at the nearest already-remote commit. """ local, _remote = self._make_two_branch_remote( tmp_path, monkeypatch, main_extra_commits=5, dev_extra_commits=2 ) # Add 2 more commits to main locally, but do NOT push them. # Remote main is now 2 commits behind local main. runner.invoke(cli, ["checkout", "main"], env=_env(local), catch_exceptions=False) for i in range(2): (local / f"main_extra_{i}.py").write_text(f"e = {i}\n") runner.invoke(cli, ["code", "add", f"main_extra_{i}.py"], env=_env(local), catch_exceptions=False) runner.invoke(cli, ["commit", "-m", f"extra main {i}"], env=_env(local), catch_exceptions=False) runner.invoke(cli, ["checkout", "dev"], env=_env(local), catch_exceptions=False) # merge the (now-ahead) main into dev r = runner.invoke(cli, ["merge", "main"], env=_env(local), catch_exceptions=False) assert r.exit_code == 0, f"merge failed: {r.output}" r = runner.invoke( cli, ["push", "origin", "--branch", "dev", "--json"], env=_env(local), catch_exceptions=False, ) assert r.exit_code == 0, f"push after merge failed: {r.output}" d = _json(r) # merge commit + 2 new commits from main = 3, NOT 5+2+1 = 8 full history assert d["commits_sent"] == 3, ( f"Expected 3 commits (merge + 2 new on main), got {d['commits_sent']}. " "push walked the merged branch's full history instead of stopping at " "the nearest already-remote commit." ) def test_merge_push_succeeds( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """Push of a merge commit must complete without error.""" local, _remote = self._make_two_branch_remote( tmp_path, monkeypatch, main_extra_commits=3, dev_extra_commits=2 ) runner.invoke(cli, ["merge", "main"], env=_env(local), catch_exceptions=False) r = runner.invoke( cli, ["push", "origin", "--branch", "dev"], env=_env(local), catch_exceptions=False, ) assert r.exit_code == 0, f"push after merge failed: {r.output}"