"""TDD — muse#43: push refspec syntax `local:remote` for cross-branch pushes. Phase 1 (RS_01-RS_05): parse_push_refspec unit tests — pure function, no I/O. Phase 2 (RS_06-RS_10): muse push integration tests. """ from __future__ import annotations import datetime import json import pathlib from unittest.mock import MagicMock, patch import pytest from muse._version import __version__ from muse.core.mpack import RemoteInfo from muse.core.object_store import write_object from muse.core.paths import heads_dir, muse_dir from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import CommitRecord, write_commit from muse.core.snapshots import SnapshotRecord, write_snapshot from muse.core.types import Manifest, blob_id from muse.cli.commands.push import parse_push_refspec from tests.cli_test_helper import CliRunner cli = None runner = CliRunner() # --------------------------------------------------------------------------- # Phase 1 — RS_01-RS_05: parse_push_refspec unit tests # --------------------------------------------------------------------------- class TestParsePushRefspec: def test_rs01_bare_name(self) -> None: assert parse_push_refspec("dev") == ("dev", "dev", False) def test_rs02_local_colon_remote(self) -> None: assert parse_push_refspec("dev:main") == ("dev", "main", False) def test_rs03_force_prefix_with_colon(self) -> None: assert parse_push_refspec("+dev:main") == ("dev", "main", True) def test_rs04_force_prefix_bare_name(self) -> None: assert parse_push_refspec("+dev") == ("dev", "dev", True) def test_rs05_empty_remote_side_raises(self) -> None: with pytest.raises(ValueError): parse_push_refspec("dev:") def test_rs05_empty_local_side_raises(self) -> None: with pytest.raises(ValueError): parse_push_refspec(":main") def test_rs05_too_many_colons_raises(self) -> None: with pytest.raises(ValueError): parse_push_refspec("dev:main:extra") def test_rs05_error_messages_are_clear(self) -> None: with pytest.raises(ValueError, match="dev:"): parse_push_refspec("dev:") with pytest.raises(ValueError, match=":main"): parse_push_refspec(":main") # --------------------------------------------------------------------------- # Phase 2 — RS_06-RS_10: integration tests # --------------------------------------------------------------------------- def _bare_repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: muse = muse_dir(tmp_path) for d in ("commits", "snapshots", "objects", "refs/heads", "remotes"): (muse / d).mkdir(parents=True, exist_ok=True) (muse / "HEAD").write_text("ref: refs/heads/main\n") (muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (muse / "config.toml").write_text('[remotes.origin]\nurl = "https://hub.example.com/r"\n') monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) return tmp_path def _make_commit( root: pathlib.Path, label: str, branch: str = "main", parent_id: str | None = None, content: bytes | None = None, ) -> CommitRecord: raw = content if content is not None else f"content-{label}".encode() oid = blob_id(raw) write_object(root, oid, raw) manifest: Manifest = {"file.txt": oid} snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) parent_ids = [parent_id] if parent_id else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=snap_id, message=f"commit {label}", committed_at_iso=committed_at.isoformat(), ) commit = CommitRecord( commit_id=cid, branch=branch, snapshot_id=snap_id, message=f"commit {label}", committed_at=committed_at, parent_commit_id=parent_id, ) write_commit(root, commit) return commit def _mpack_push_transport(local_head: str, remote_branch: str = "main") -> MagicMock: """Transport mock that makes _push_mpack succeed via the new urllib path.""" transport = MagicMock() transport.fetch_remote_info.return_value = RemoteInfo( domain="code", default_branch="main", branch_heads={}, ) mock_req = MagicMock() mock_req.headers = {"Authorization": "MSign stub", "Content-Type": "application/x-msgpack"} transport._build_request.return_value = mock_req transport.push_mpack_presign.return_value = { "upload_url": "https://minio.example.com/put?sig=x", "mpack_key": "sha256:fake", } transport.push_mpack_put.return_value = None transport.push_mpack_unpack.return_value = { "job_id": "j", "head": local_head, "branch": remote_branch, "blobs_in_mpack": 0, "commits_in_mpack": 0, } return transport class TestRefspecSameNameUnchanged: def test_rs06_dev_colon_dev_behaves_like_plain_dev( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: root = _bare_repo(tmp_path, monkeypatch) commit = _make_commit(root, "rs06", branch="dev") (heads_dir(root) / "dev").write_text(commit.commit_id) transport = _mpack_push_transport(commit.commit_id, remote_branch="dev") with patch("muse.cli.commands.push.make_transport", return_value=transport): result = runner.invoke( cli, ["push", "origin", "dev:dev", "--json"], catch_exceptions=False ) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["status"] == "pushed" assert data["branch"] == "dev" _, _, kwargs = transport.push_mpack_unpack.mock_calls[0] assert kwargs["branch"] == "dev" class TestRefspecCrossBranch: def test_rs07_dev_colon_main_advances_remote_main_not_dev( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Local dev's commits are read, but the remote ref advanced is main.""" root = _bare_repo(tmp_path, monkeypatch) commit = _make_commit(root, "rs07", branch="dev") (heads_dir(root) / "dev").write_text(commit.commit_id) transport = _mpack_push_transport(commit.commit_id, remote_branch="main") with patch("muse.cli.commands.push.make_transport", return_value=transport): result = runner.invoke( cli, ["push", "origin", "dev:main", "--json"], catch_exceptions=False ) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["status"] == "pushed" assert data["branch"] == "main" assert data["head"] == commit.commit_id # The commit walk / unpack call targeted "main" on the remote, but # the commit itself is local dev's tip. _, args, kwargs = transport.push_mpack_unpack.mock_calls[0] assert kwargs["branch"] == "main" assert kwargs["head"] == commit.commit_id class TestRefspecForcePrefix: def test_rs08_force_prefix_equivalent_to_force_flag( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: root = _bare_repo(tmp_path, monkeypatch) commit = _make_commit(root, "rs08", branch="dev") (heads_dir(root) / "dev").write_text(commit.commit_id) transport = _mpack_push_transport(commit.commit_id, remote_branch="main") with patch("muse.cli.commands.push.make_transport", return_value=transport): result = runner.invoke( cli, ["push", "origin", "+dev:main", "--json"], catch_exceptions=False ) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["force"] is True _, _, kwargs = transport.push_mpack_unpack.mock_calls[0] assert kwargs["force"] is True class TestRefspecNewRemoteBranch: def test_rs09_pushing_to_nonexistent_remote_branch_creates_it( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """No special handling needed — the server already supports creating a new branch by name; this just confirms the refspec's remote name reaches the server unchanged.""" root = _bare_repo(tmp_path, monkeypatch) commit = _make_commit(root, "rs09", branch="dev") (heads_dir(root) / "dev").write_text(commit.commit_id) transport = _mpack_push_transport(commit.commit_id, remote_branch="nonexistent") with patch("muse.cli.commands.push.make_transport", return_value=transport): result = runner.invoke( cli, ["push", "origin", "dev:nonexistent", "--json"], catch_exceptions=False ) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["branch"] == "nonexistent" _, _, kwargs = transport.push_mpack_unpack.mock_calls[0] assert kwargs["branch"] == "nonexistent" class TestRefspecInvalid: def test_rs10_trailing_colon_exits_nonzero_before_network_call( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: root = _bare_repo(tmp_path, monkeypatch) commit = _make_commit(root, "rs10", branch="dev") (heads_dir(root) / "dev").write_text(commit.commit_id) transport = MagicMock() transport.fetch_remote_info.side_effect = AssertionError( "invalid refspec must be rejected before any network call" ) with patch("muse.cli.commands.push.make_transport", return_value=transport): result = runner.invoke( cli, ["push", "origin", "dev:", "--json"], catch_exceptions=False ) assert result.exit_code != 0 data = json.loads(result.output) assert data["error"] == "invalid_refspec" transport.fetch_remote_info.assert_not_called()