"""Tests for muse clone CLI command.""" from __future__ import annotations import base64 import json import pathlib import unittest.mock import pytest from tests.cli_test_helper import CliRunner cli = None # argparse migration — CliRunner ignores this arg from muse.cli.config import get_remote, get_upstream from muse.core.mpack import ObjectPayload, MPack, RemoteInfo from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import read_commit from muse.core.snapshots import read_snapshot from muse.core.transport import TransportError from muse.core.types import Manifest, blob_id, long_id from muse.core.paths import head_path, muse_dir, repo_json_path runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_remote_info( domain: str = "midi", default_branch: str = "main", branch_heads: Manifest | None = None, repo_id: str = "remote-repo-id", ) -> RemoteInfo: return RemoteInfo( repo_id=repo_id, domain=domain, default_branch=default_branch, branch_heads={"main": long_id("c" * 64)} if branch_heads is None else branch_heads, ) def _make_bundle(branch: str = "main", message: str = "initial") -> MPack: """Build a MPack whose commit/snapshot IDs are content-addressed. Uses ``hash_commit`` and ``hash_snapshot`` so that ``read_commit``/``read_snapshot`` pass ``_verify_commit_id``/ ``_verify_snapshot_id`` on the receiving end. """ content = b"cloned content" oid = blob_id(content) committed_at = "2026-01-01T00:00:00+00:00" snap_id = hash_snapshot({"hello.txt": oid}) cid = hash_commit( parent_ids=[], snapshot_id=snap_id, message=message, committed_at_iso=committed_at, author="alice", ) return MPack( commits=[ { "commit_id": cid, "repo_id": "remote-repo-id", "branch": branch, "snapshot_id": snap_id, "message": message, "committed_at": committed_at, "parent_commit_id": None, "parent2_commit_id": None, "author": "alice", "metadata": {}, "structured_delta": None, "sem_ver_bump": "none", "breaking_changes": [], "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": "", "signer_key_id": "", "reviewed_by": [], "test_runs": 0, } ], snapshots=[ { "snapshot_id": snap_id, "delta_add": {"hello.txt": oid}, "delta_remove": [], "parent_snapshot_id": None, "created_at": committed_at, } ], objects=[ ObjectPayload(object_id=oid, content=content) ], branch_heads={branch: cid}, ) def _mock_transport(info: RemoteInfo, mpack: MPack) -> unittest.mock.MagicMock: mock = unittest.mock.MagicMock() mock.fetch_remote_info.return_value = info mock.fetch_pack.return_value = mpack _fetch_result = { "repo_id": info["repo_id"], "domain": info["domain"], "default_branch": info["default_branch"], "branch_heads": info["branch_heads"], "commits": mpack["commits"], "snapshots": mpack["snapshots"], "objects_received": len(mpack["objects"]), } mock.fetch_mpack.return_value = _fetch_result return mock # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestClone: def test_clone_creates_muse_dir( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info() mpack = _make_bundle() mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): result = runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "my-repo"] ) assert result.exit_code == 0, result.output assert (tmp_path / "my-repo" / ".muse").is_dir() def test_clone_sets_origin_remote( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info() mpack = _make_bundle() mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "my-repo"] ) origin = get_remote("origin", tmp_path / "my-repo") assert origin == "https://hub.example.com/repos/r1" def test_clone_sets_upstream_tracking( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info() mpack = _make_bundle() mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "my-repo"] ) upstream = get_upstream("main", tmp_path / "my-repo") assert upstream == "origin" def test_clone_writes_commits_and_snapshots( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) mpack = _make_bundle() cid = mpack["commits"][0]["commit_id"] info = _make_remote_info(branch_heads={"main": cid}) mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "dest"] ) dest = tmp_path / "dest" cid = mpack["commits"][0]["commit_id"] snap_id = mpack["snapshots"][0]["snapshot_id"] assert isinstance(cid, str) and isinstance(snap_id, str) assert read_commit(dest, cid) is not None assert read_snapshot(dest, snap_id) is not None def test_clone_propagates_domain( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info(domain="code") mpack = _make_bundle() mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "dest"] ) dest = tmp_path / "dest" repo_meta = json.loads((repo_json_path(dest)).read_text()) assert repo_meta["domain"] == "code" def test_clone_uses_remote_repo_id( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info(repo_id="the-real-repo-id") mpack = _make_bundle() mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "dest"] ) dest = tmp_path / "dest" repo_meta = json.loads((repo_json_path(dest)).read_text()) assert repo_meta["repo_id"] == "the-real-repo-id" def test_clone_infers_directory_name_from_url( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info() mpack = _make_bundle() mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke(cli, ["clone", "https://hub.example.com/repos/my-project"]) assert (tmp_path / "my-project" / ".muse").is_dir() def test_clone_transport_error_fails_cleanly( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) mock = unittest.mock.MagicMock() mock.fetch_remote_info.side_effect = TransportError("connection refused", 0) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): result = runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "dest"] ) assert result.exit_code != 0 assert "Cannot reach remote" in result.stderr def test_clone_existing_repo_fails( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) # Pre-create a .muse directory. muse_dir(tmp_path / "dest").mkdir(parents=True) mock = unittest.mock.MagicMock() with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): result = runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "dest"] ) assert result.exit_code != 0 assert "already a Muse repository" in result.stderr def test_clone_empty_repo_fails( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info(branch_heads={}) # no branches mock = unittest.mock.MagicMock() mock.fetch_remote_info.return_value = info with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): result = runner.invoke( cli, ["clone", "https://hub.example.com/repos/r1", "dest"] ) assert result.exit_code != 0 assert "empty" in result.stderr.lower() or "no branches" in result.stderr.lower() def test_clone_branch_flag_selects_branch( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.chdir(tmp_path) info = _make_remote_info( default_branch="main", branch_heads={"main": long_id("c" * 64), "dev": long_id("d" * 64)}, ) mpack = _make_bundle(branch="dev", message="initial on dev") mock = _mock_transport(info, mpack) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): # Options must precede positional args in add_typer groups. result = runner.invoke( cli, ["clone", "--branch", "dev", "https://hub.example.com/repos/r1", "dest"], ) assert result.exit_code == 0, result.output dest = tmp_path / "dest" head_ref = (head_path(dest)).read_text().strip() assert "dev" in head_ref