test_cli_fetch_push.py
python
| 1 | """Tests for muse fetch, push, pull, and ls-remote CLI commands. |
| 2 | |
| 3 | All network calls are mocked — no real HTTP traffic occurs. |
| 4 | """ |
| 5 | |
| 6 | from __future__ import annotations |
| 7 | |
| 8 | import base64 |
| 9 | import datetime |
| 10 | import hashlib |
| 11 | import json |
| 12 | import pathlib |
| 13 | import unittest.mock |
| 14 | |
| 15 | import pytest |
| 16 | from typer.testing import CliRunner |
| 17 | |
| 18 | from muse._version import __version__ |
| 19 | from muse.cli.app import cli |
| 20 | from muse.cli.config import get_remote_head, get_upstream |
| 21 | from muse.core.object_store import write_object |
| 22 | from muse.core.pack import PackBundle, RemoteInfo |
| 23 | from muse.core.store import ( |
| 24 | CommitRecord, |
| 25 | SnapshotRecord, |
| 26 | get_head_commit_id, |
| 27 | read_commit, |
| 28 | write_commit, |
| 29 | write_snapshot, |
| 30 | ) |
| 31 | from muse.core.transport import TransportError |
| 32 | |
| 33 | runner = CliRunner() |
| 34 | |
| 35 | |
| 36 | # --------------------------------------------------------------------------- |
| 37 | # Fixture helpers |
| 38 | # --------------------------------------------------------------------------- |
| 39 | |
| 40 | |
| 41 | def _sha(content: bytes) -> str: |
| 42 | return hashlib.sha256(content).hexdigest() |
| 43 | |
| 44 | |
| 45 | @pytest.fixture |
| 46 | def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 47 | """Fully-initialised .muse/ repo with one commit on main.""" |
| 48 | muse_dir = tmp_path / ".muse" |
| 49 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 50 | (muse_dir / "objects").mkdir() |
| 51 | (muse_dir / "commits").mkdir() |
| 52 | (muse_dir / "snapshots").mkdir() |
| 53 | (muse_dir / "repo.json").write_text( |
| 54 | json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "midi"}) |
| 55 | ) |
| 56 | (muse_dir / "HEAD").write_text("ref: refs/heads/main\n") |
| 57 | |
| 58 | # Write one object + snapshot + commit so there is something to push. |
| 59 | content = b"hello" |
| 60 | oid = _sha(content) |
| 61 | write_object(tmp_path, oid, content) |
| 62 | snap = SnapshotRecord(snapshot_id="s" * 64, manifest={"file.txt": oid}) |
| 63 | write_snapshot(tmp_path, snap) |
| 64 | commit = CommitRecord( |
| 65 | commit_id="commit1", |
| 66 | repo_id="test-repo", |
| 67 | branch="main", |
| 68 | snapshot_id="s" * 64, |
| 69 | message="initial", |
| 70 | committed_at=datetime.datetime.now(datetime.timezone.utc), |
| 71 | ) |
| 72 | write_commit(tmp_path, commit) |
| 73 | (muse_dir / "refs" / "heads" / "main").write_text("commit1") |
| 74 | (muse_dir / "config.toml").write_text( |
| 75 | '[remotes.origin]\nurl = "https://hub.example.com/repos/r1"\n' |
| 76 | ) |
| 77 | |
| 78 | monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) |
| 79 | monkeypatch.chdir(tmp_path) |
| 80 | return tmp_path |
| 81 | |
| 82 | |
| 83 | def _make_remote_info( |
| 84 | branch_heads: dict[str, str] | None = None, |
| 85 | ) -> RemoteInfo: |
| 86 | return RemoteInfo( |
| 87 | repo_id="remote-repo", |
| 88 | domain="midi", |
| 89 | default_branch="main", |
| 90 | branch_heads=branch_heads or {"main": "remote_commit1"}, |
| 91 | ) |
| 92 | |
| 93 | |
| 94 | def _make_bundle(commit_id: str = "remote_commit1") -> PackBundle: |
| 95 | content = b"remote content" |
| 96 | oid = _sha(content) |
| 97 | return PackBundle( |
| 98 | commits=[ |
| 99 | { |
| 100 | "commit_id": commit_id, |
| 101 | "repo_id": "test-repo", |
| 102 | "branch": "main", |
| 103 | "snapshot_id": "remote_snap1", |
| 104 | "message": "remote", |
| 105 | "committed_at": "2026-01-01T00:00:00+00:00", |
| 106 | "parent_commit_id": None, |
| 107 | "parent2_commit_id": None, |
| 108 | "author": "remote", |
| 109 | "metadata": {}, |
| 110 | "structured_delta": None, |
| 111 | "sem_ver_bump": "none", |
| 112 | "breaking_changes": [], |
| 113 | "agent_id": "", |
| 114 | "model_id": "", |
| 115 | "toolchain_id": "", |
| 116 | "prompt_hash": "", |
| 117 | "signature": "", |
| 118 | "signer_key_id": "", |
| 119 | "format_version": 5, |
| 120 | "reviewed_by": [], |
| 121 | "test_runs": 0, |
| 122 | } |
| 123 | ], |
| 124 | snapshots=[ |
| 125 | { |
| 126 | "snapshot_id": "remote_snap1", |
| 127 | "manifest": {"remote.txt": oid}, |
| 128 | "created_at": "2026-01-01T00:00:00+00:00", |
| 129 | } |
| 130 | ], |
| 131 | objects=[ |
| 132 | ObjectPayload(object_id=oid, content_b64=base64.b64encode(content).decode()) |
| 133 | ], |
| 134 | branch_heads={"main": commit_id}, |
| 135 | ) |
| 136 | |
| 137 | |
| 138 | # Import ObjectPayload here to avoid circular issues in the fixture above. |
| 139 | from muse.core.pack import ObjectPayload # noqa: E402 |
| 140 | |
| 141 | |
| 142 | # --------------------------------------------------------------------------- |
| 143 | # muse fetch |
| 144 | # --------------------------------------------------------------------------- |
| 145 | |
| 146 | |
| 147 | class TestFetch: |
| 148 | def test_fetch_updates_tracking_head(self, repo: pathlib.Path) -> None: |
| 149 | info = _make_remote_info({"main": "remote_commit1"}) |
| 150 | bundle = _make_bundle("remote_commit1") |
| 151 | transport_mock = unittest.mock.MagicMock() |
| 152 | transport_mock.fetch_remote_info.return_value = info |
| 153 | transport_mock.fetch_pack.return_value = bundle |
| 154 | |
| 155 | with unittest.mock.patch( |
| 156 | "muse.cli.commands.fetch.make_transport", return_value=transport_mock |
| 157 | ): |
| 158 | result = runner.invoke(cli, ["fetch", "origin"]) |
| 159 | |
| 160 | assert result.exit_code == 0 |
| 161 | assert "Fetched" in result.output |
| 162 | tracking = get_remote_head("origin", "main", repo) |
| 163 | assert tracking == "remote_commit1" |
| 164 | |
| 165 | def test_fetch_no_remote_configured_fails( |
| 166 | self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch |
| 167 | ) -> None: |
| 168 | result = runner.invoke(cli, ["fetch", "nonexistent"]) |
| 169 | assert result.exit_code != 0 |
| 170 | assert "not configured" in result.output |
| 171 | |
| 172 | def test_fetch_branch_not_on_remote_fails(self, repo: pathlib.Path) -> None: |
| 173 | info = _make_remote_info({"main": "abc"}) |
| 174 | transport_mock = unittest.mock.MagicMock() |
| 175 | transport_mock.fetch_remote_info.return_value = info |
| 176 | |
| 177 | with unittest.mock.patch( |
| 178 | "muse.cli.commands.fetch.make_transport", return_value=transport_mock |
| 179 | ): |
| 180 | # Options must precede positional args in add_typer groups. |
| 181 | result = runner.invoke(cli, ["fetch", "--branch", "nonexistent", "origin"]) |
| 182 | |
| 183 | assert result.exit_code != 0 |
| 184 | assert "does not exist on remote" in result.output |
| 185 | |
| 186 | def test_fetch_transport_error_propagates(self, repo: pathlib.Path) -> None: |
| 187 | transport_mock = unittest.mock.MagicMock() |
| 188 | transport_mock.fetch_remote_info.side_effect = TransportError("timeout", 0) |
| 189 | |
| 190 | with unittest.mock.patch( |
| 191 | "muse.cli.commands.fetch.make_transport", return_value=transport_mock |
| 192 | ): |
| 193 | result = runner.invoke(cli, ["fetch", "origin"]) |
| 194 | |
| 195 | assert result.exit_code != 0 |
| 196 | assert "Cannot reach remote" in result.output |
| 197 | |
| 198 | |
| 199 | # --------------------------------------------------------------------------- |
| 200 | # muse push |
| 201 | # --------------------------------------------------------------------------- |
| 202 | |
| 203 | |
| 204 | class TestPush: |
| 205 | def test_push_sends_commits(self, repo: pathlib.Path) -> None: |
| 206 | push_result = {"ok": True, "message": "ok", "branch_heads": {"main": "commit1"}} |
| 207 | transport_mock = unittest.mock.MagicMock() |
| 208 | transport_mock.push_pack.return_value = push_result |
| 209 | |
| 210 | with unittest.mock.patch( |
| 211 | "muse.cli.commands.push.make_transport", return_value=transport_mock |
| 212 | ): |
| 213 | result = runner.invoke(cli, ["push", "origin"]) |
| 214 | |
| 215 | assert result.exit_code == 0 |
| 216 | assert "Pushed" in result.output |
| 217 | transport_mock.push_pack.assert_called_once() |
| 218 | |
| 219 | def test_push_no_remote_configured_fails(self, repo: pathlib.Path) -> None: |
| 220 | result = runner.invoke(cli, ["push", "nonexistent"]) |
| 221 | assert result.exit_code != 0 |
| 222 | assert "not configured" in result.output |
| 223 | |
| 224 | def test_push_set_upstream_records_tracking(self, repo: pathlib.Path) -> None: |
| 225 | push_result = {"ok": True, "message": "", "branch_heads": {"main": "commit1"}} |
| 226 | transport_mock = unittest.mock.MagicMock() |
| 227 | transport_mock.push_pack.return_value = push_result |
| 228 | |
| 229 | with unittest.mock.patch( |
| 230 | "muse.cli.commands.push.make_transport", return_value=transport_mock |
| 231 | ): |
| 232 | # Options must precede positional args in add_typer groups. |
| 233 | result = runner.invoke(cli, ["push", "-u", "origin"]) |
| 234 | |
| 235 | assert result.exit_code == 0 |
| 236 | assert get_upstream("main", repo) == "origin" |
| 237 | |
| 238 | def test_push_conflict_409_shows_helpful_message(self, repo: pathlib.Path) -> None: |
| 239 | transport_mock = unittest.mock.MagicMock() |
| 240 | transport_mock.push_pack.side_effect = TransportError("non-fast-forward", 409) |
| 241 | |
| 242 | with unittest.mock.patch( |
| 243 | "muse.cli.commands.push.make_transport", return_value=transport_mock |
| 244 | ): |
| 245 | result = runner.invoke(cli, ["push", "origin"]) |
| 246 | |
| 247 | assert result.exit_code != 0 |
| 248 | assert "diverged" in result.output |
| 249 | |
| 250 | def test_push_already_up_to_date(self, repo: pathlib.Path) -> None: |
| 251 | # Set tracking head to the same commit as local HEAD. |
| 252 | (repo / ".muse" / "remotes" / "origin").mkdir(parents=True) |
| 253 | (repo / ".muse" / "remotes" / "origin" / "main").write_text("commit1") |
| 254 | |
| 255 | transport_mock = unittest.mock.MagicMock() |
| 256 | with unittest.mock.patch( |
| 257 | "muse.cli.commands.push.make_transport", return_value=transport_mock |
| 258 | ): |
| 259 | result = runner.invoke(cli, ["push", "origin"]) |
| 260 | |
| 261 | assert result.exit_code == 0 |
| 262 | assert "up to date" in result.output |
| 263 | transport_mock.push_pack.assert_not_called() |
| 264 | |
| 265 | def test_push_force_flag_passed_to_transport(self, repo: pathlib.Path) -> None: |
| 266 | push_result = {"ok": True, "message": "", "branch_heads": {"main": "commit1"}} |
| 267 | transport_mock = unittest.mock.MagicMock() |
| 268 | transport_mock.push_pack.return_value = push_result |
| 269 | |
| 270 | with unittest.mock.patch( |
| 271 | "muse.cli.commands.push.make_transport", return_value=transport_mock |
| 272 | ): |
| 273 | result = runner.invoke(cli, ["push", "--force", "origin"]) |
| 274 | |
| 275 | assert result.exit_code == 0 |
| 276 | call_kwargs = transport_mock.push_pack.call_args |
| 277 | assert call_kwargs[0][4] is True # force=True positional arg |
| 278 | |
| 279 | |
| 280 | # --------------------------------------------------------------------------- |
| 281 | # muse ls-remote |
| 282 | # --------------------------------------------------------------------------- |
| 283 | |
| 284 | |
| 285 | class TestLsRemote: |
| 286 | def test_ls_remote_prints_branches(self, repo: pathlib.Path) -> None: |
| 287 | info = _make_remote_info({"main": "abc123", "dev": "def456"}) |
| 288 | transport_mock = unittest.mock.MagicMock() |
| 289 | transport_mock.fetch_remote_info.return_value = info |
| 290 | |
| 291 | with unittest.mock.patch( |
| 292 | "muse.cli.commands.plumbing.ls_remote.HttpTransport", |
| 293 | return_value=transport_mock, |
| 294 | ): |
| 295 | result = runner.invoke(cli, ["plumbing", "ls-remote", "origin"]) |
| 296 | |
| 297 | assert result.exit_code == 0 |
| 298 | assert "abc123" in result.output |
| 299 | assert "main" in result.output |
| 300 | |
| 301 | def test_ls_remote_json_output(self, repo: pathlib.Path) -> None: |
| 302 | info = _make_remote_info({"main": "abc123"}) |
| 303 | transport_mock = unittest.mock.MagicMock() |
| 304 | transport_mock.fetch_remote_info.return_value = info |
| 305 | |
| 306 | with unittest.mock.patch( |
| 307 | "muse.cli.commands.plumbing.ls_remote.HttpTransport", |
| 308 | return_value=transport_mock, |
| 309 | ): |
| 310 | result = runner.invoke(cli, ["plumbing", "ls-remote", "--format", "json", "origin"]) |
| 311 | |
| 312 | assert result.exit_code == 0 |
| 313 | data = json.loads(result.output) |
| 314 | assert data["branches"]["main"] == "abc123" |
| 315 | assert "repo_id" in data |
| 316 | |
| 317 | def test_ls_remote_unknown_name_fails(self, repo: pathlib.Path) -> None: |
| 318 | result = runner.invoke(cli, ["plumbing", "ls-remote", "ghost"]) |
| 319 | assert result.exit_code != 0 |
| 320 | |
| 321 | def test_ls_remote_bare_url_accepted(self, repo: pathlib.Path) -> None: |
| 322 | info = _make_remote_info({"main": "abc123"}) |
| 323 | transport_mock = unittest.mock.MagicMock() |
| 324 | transport_mock.fetch_remote_info.return_value = info |
| 325 | |
| 326 | with unittest.mock.patch( |
| 327 | "muse.cli.commands.plumbing.ls_remote.HttpTransport", |
| 328 | return_value=transport_mock, |
| 329 | ): |
| 330 | result = runner.invoke( |
| 331 | cli, ["plumbing", "ls-remote", "https://hub.example.com/repos/r1"] |
| 332 | ) |
| 333 | |
| 334 | assert result.exit_code == 0 |
| 335 | assert "abc123" in result.output |