"""End-to-end integration tests for ``muse push local`` + ``muse pull`` round-trip. These tests exercise the full MPack protocol pipeline: - push.py pre-compression fix (raw bytes, encoding="raw") - LocalFileTransport.push_stream (direct filesystem write) - apply_mpack writing objects/commits/snapshots/refs to the remote - muse pull fetching commits and objects back from the remote All tests use the CliRunner from tests.cli_test_helper and LocalFileTransport (file:// URL) so no HTTP server is required. """ from __future__ import annotations from collections.abc import Mapping import datetime import json import pathlib import pytest from muse.core.compression import choose_compression from muse.core.object_store import object_path, write_object from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, commit_path, write_commit, write_snapshot from tests.cli_test_helper import CliRunner, InvokeResult from muse.core._types import blob_id runner = CliRunner() def _parse_json_output(result: InvokeResult) -> Mapping[str, object]: """Extract the JSON object from push/pull output. CliRunner combines stdout (JSON) and stderr (progress lines). Find the first line that parses as a JSON object. """ for line in result.output.splitlines(): line = line.strip() if line.startswith("{"): return json.loads(line) raise ValueError(f"No JSON line found in output:\n{result.output!r}") # --------------------------------------------------------------------------- # Repo setup helpers # --------------------------------------------------------------------------- def _make_repo(path: pathlib.Path, *, repo_id: str = "test-repo", domain: str = "code") -> pathlib.Path: """Create a minimal muse repo at *path* and return the root.""" muse = path / ".muse" for sub in ("objects", "commits", "snapshots", "refs/heads"): (muse / sub).mkdir(parents=True) (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "repo.json").write_text( json.dumps({"repo_id": repo_id, "domain": domain, "default_branch": "main"}) ) return path def _write_config_toml(repo: pathlib.Path, remotes: Mapping[str, str]) -> None: """Write .muse/config.toml with one [remotes.] section per entry.""" lines = ["[remotes]\n"] for name, url in remotes.items(): lines.append(f'[remotes.{name}]\n') lines.append(f'url = "{url}"\n') (repo / ".muse" / "config.toml").write_text("".join(lines)) def _snap(repo: pathlib.Path, manifest: Mapping[str, str] | None = None) -> str: m = manifest or {} snap_id = compute_snapshot_id(m) write_snapshot(repo, SnapshotRecord( snapshot_id=snap_id, manifest=m, created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), )) return snap_id def _commit( repo: pathlib.Path, snap_id: str, *, parent: str | None = None, message: str = "test commit", branch: str = "main", ) -> str: committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) parent_ids: list[str] = [parent] if parent else [] commit_id = compute_commit_id( repo_id="test-repo", parent_ids=parent_ids, snapshot_id=snap_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_commit(repo, CommitRecord( commit_id=commit_id, repo_id="test-repo", created_on_branch=branch, snapshot_id=snap_id, message=message, committed_at=committed_at, parent_commit_id=parent, )) return commit_id def _set_ref(repo: pathlib.Path, branch: str, commit_id: str) -> None: ref_dir = repo / ".muse" / "refs" / "heads" ref_dir.mkdir(parents=True, exist_ok=True) (ref_dir / branch).write_text(commit_id) def _push(src: pathlib.Path, remote: str = "local", branch: str = "main", *extra: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["push", remote, branch, "--json", *extra], env={"MUSE_REPO_ROOT": str(src)}, ) def _pull(dst: pathlib.Path, remote: str = "origin", branch: str = "main", *extra: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["pull", remote, branch, *extra], env={"MUSE_REPO_ROOT": str(dst)}, ) def _object_path(repo: pathlib.Path, oid: str) -> pathlib.Path: """Return the on-disk path for a content-addressed object.""" return object_path(repo, oid) def _branch_ref(repo: pathlib.Path, branch: str) -> str | None: ref = repo / ".muse" / "refs" / "heads" / branch return ref.read_text().strip() if ref.exists() else None # --------------------------------------------------------------------------- # Basic push round-trip # --------------------------------------------------------------------------- class TestPushBasic: def test_push_single_commit(self, tmp_path: pathlib.Path) -> None: """A single commit with no objects is pushed; dst has the commit and ref.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) cid = _commit(src, sid, message="initial") _set_ref(src, "main", cid) result = _push(src) assert result.exit_code == 0, result.output data = _parse_json_output(result) assert data["status"] == "pushed" assert data["branch"] == "main" assert data["commits_sent"] == 1 assert commit_path(dst, cid).exists(), "commit file missing from remote" assert _branch_ref(dst, "main") == cid, "remote branch ref not updated" def test_push_with_objects(self, tmp_path: pathlib.Path) -> None: """Objects referenced by the commit are transferred to the remote.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) content = b"hello muse e2e push" oid = blob_id(content) write_object(src, oid, content) sid = _snap(src, {"hello.txt": oid}) cid = _commit(src, sid, message="add hello.txt") _set_ref(src, "main", cid) result = _push(src) assert result.exit_code == 0, result.output data = _parse_json_output(result) assert data["status"] == "pushed" assert data["objects_sent"] == 1 assert _object_path(dst, oid).exists(), "object file missing from remote" def test_push_reports_compression_type(self, tmp_path: pathlib.Path) -> None: """Objects use choose_compression() — zstd when available, zlib fallback.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) content = b"compression selection test " * 100 oid = blob_id(content) write_object(src, oid, content) sid = _snap(src, {"data.bin": oid}) cid = _commit(src, sid) _set_ref(src, "main", cid) result = _push(src) assert result.exit_code == 0, result.output # The remote object should exist regardless of which algorithm was chosen. assert _object_path(dst, oid).exists() # Verify the correct algorithm would have been selected (no assertion on # the actual stored bytes — that is an implementation detail of transport). expected_algo = choose_compression() assert expected_algo in ("zstd", "zlib"), f"unexpected algorithm: {expected_algo}" def test_push_up_to_date(self, tmp_path: pathlib.Path) -> None: """Second push of the same HEAD reports up_to_date.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) cid = _commit(src, sid) _set_ref(src, "main", cid) r1 = _push(src) assert r1.exit_code == 0 r2 = _push(src) assert r2.exit_code == 0 data = _parse_json_output(r2) assert data["status"] == "up_to_date" assert data["commits_sent"] == 0 def test_push_multi_commit_chain(self, tmp_path: pathlib.Path) -> None: """A chain of commits is pushed in full on first push.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) c1 = _commit(src, sid, message="first") c2 = _commit(src, sid, parent=c1, message="second") c3 = _commit(src, sid, parent=c2, message="third") _set_ref(src, "main", c3) result = _push(src) assert result.exit_code == 0, result.output data = _parse_json_output(result) assert data["status"] == "pushed" assert data["commits_sent"] == 3 for cid in (c1, c2, c3): assert commit_path(dst, cid).exists(), f"commit {cid[:8]} missing" assert _branch_ref(dst, "main") == c3 def test_push_incremental_second_commit(self, tmp_path: pathlib.Path) -> None: """Second push sends only the new commit, not the already-transferred one.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) c1 = _commit(src, sid, message="initial") _set_ref(src, "main", c1) r1 = _push(src) assert r1.exit_code == 0 assert _parse_json_output(r1)["commits_sent"] == 1 c2 = _commit(src, sid, parent=c1, message="follow-up") _set_ref(src, "main", c2) r2 = _push(src) assert r2.exit_code == 0 data2 = _parse_json_output(r2) assert data2["status"] == "pushed" assert data2["commits_sent"] == 1 # only the new commit assert _branch_ref(dst, "main") == c2 # --------------------------------------------------------------------------- # Push error conditions # --------------------------------------------------------------------------- class TestPushErrors: def test_push_no_remote_configured(self, tmp_path: pathlib.Path) -> None: """Push to an unconfigured remote exits with error.""" src = _make_repo(tmp_path / "src") sid = _snap(src) cid = _commit(src, sid) _set_ref(src, "main", cid) result = _push(src, remote="nonexistent") assert result.exit_code != 0 data = _parse_json_output(result) assert data["error"] == "remote_not_configured" def test_push_no_commits(self, tmp_path: pathlib.Path) -> None: """Push with no commits on the branch exits with error.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) result = _push(src) assert result.exit_code != 0 data = _parse_json_output(result) assert "error" in data def test_push_dry_run(self, tmp_path: pathlib.Path) -> None: """--dry-run returns status dry_run without writing to remote.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) cid = _commit(src, sid, message="dry run test") _set_ref(src, "main", cid) result = _push(src, "local", "main", "--dry-run") assert result.exit_code == 0, result.output data = _parse_json_output(result) assert data["status"] == "dry_run" assert data["dry_run"] is True # Remote must not have been modified. assert _branch_ref(dst, "main") is None def test_push_diverged_rejected_without_force(self, tmp_path: pathlib.Path) -> None: """Push is rejected when the remote branch has diverged (non-fast-forward).""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) c1 = _commit(src, sid, message="shared root") _set_ref(src, "main", c1) # First push establishes c1 on the remote. r1 = _push(src) assert r1.exit_code == 0 # Advance the remote independently (simulate another push from elsewhere). c_remote = _commit(dst, sid, parent=c1, message="remote advancement") _set_ref(dst, "main", c_remote) # Advance local independently from c1. c_local = _commit(src, sid, parent=c1, message="local advancement") _set_ref(src, "main", c_local) result = _push(src) assert result.exit_code != 0 def test_push_force_overwrites_diverged(self, tmp_path: pathlib.Path) -> None: """--force allows pushing over a diverged remote branch.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) sid = _snap(src) c1 = _commit(src, sid, message="shared root") _set_ref(src, "main", c1) _push(src) # Diverge remote and local. c_remote = _commit(dst, sid, parent=c1, message="remote diverge") _set_ref(dst, "main", c_remote) c_local = _commit(src, sid, parent=c1, message="local diverge") _set_ref(src, "main", c_local) result = _push(src, "local", "main", "--force") assert result.exit_code == 0, result.output data = _parse_json_output(result) assert data["status"] == "pushed" assert _branch_ref(dst, "main") == c_local # --------------------------------------------------------------------------- # Object content integrity # --------------------------------------------------------------------------- class TestObjectIntegrity: def test_object_content_survives_push(self, tmp_path: pathlib.Path) -> None: """Object bytes at the remote match what was written to src.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) content = b"binary\x00\x01\x02data" * 512 oid = blob_id(content) write_object(src, oid, content) sid = _snap(src, {"bin.dat": oid}) cid = _commit(src, sid) _set_ref(src, "main", cid) result = _push(src) assert result.exit_code == 0, result.output from muse.core.object_store import read_object recovered = read_object(dst, oid) assert recovered == content, "object content mismatch after push" def test_multiple_objects_all_transferred(self, tmp_path: pathlib.Path) -> None: """All objects in the manifest are present on the remote after push.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) manifest: dict[str, str] = {} for i in range(5): content = f"file content {i}".encode() * 100 oid = blob_id(content) write_object(src, oid, content) manifest[f"file{i}.txt"] = oid sid = _snap(src, manifest) cid = _commit(src, sid) _set_ref(src, "main", cid) result = _push(src) assert result.exit_code == 0, result.output data = _parse_json_output(result) assert data["objects_sent"] == 5 from muse.core.object_store import read_object for oid in manifest.values(): assert read_object(dst, oid) is not None, f"object {oid[:16]} missing" def test_dedup_objects_not_resent(self, tmp_path: pathlib.Path) -> None: """Objects already on the remote are not re-transferred on subsequent push.""" src = _make_repo(tmp_path / "src") dst = _make_repo(tmp_path / "dst") _write_config_toml(src, {"local": dst.as_uri()}) content = b"shared object content" oid = blob_id(content) write_object(src, oid, content) sid = _snap(src, {"shared.txt": oid}) c1 = _commit(src, sid, message="first") _set_ref(src, "main", c1) r1 = _push(src) assert r1.exit_code == 0 assert _parse_json_output(r1)["objects_sent"] == 1 # Add a new commit with an EMPTY snapshot — no new objects to transfer. sid2 = _snap(src) c2 = _commit(src, sid2, parent=c1, message="second (empty snapshot)") _set_ref(src, "main", c2) r2 = _push(src) assert r2.exit_code == 0 data2 = _parse_json_output(r2) # The second commit's snapshot has no objects — nothing new to send. assert data2["objects_sent"] == 0 # --------------------------------------------------------------------------- # Pull round-trip (fetch_stream via LocalFileTransport) # --------------------------------------------------------------------------- class TestPullRoundTrip: def test_pull_fetches_commit_and_objects(self, tmp_path: pathlib.Path) -> None: """After push→pull, a third empty repo has the commit and object.""" src = _make_repo(tmp_path / "src") remote = _make_repo(tmp_path / "remote") _write_config_toml(src, {"local": remote.as_uri()}) content = b"pull round-trip content" oid = blob_id(content) write_object(src, oid, content) sid = _snap(src, {"payload.bin": oid}) cid = _commit(src, sid, message="push payload") _set_ref(src, "main", cid) push_result = _push(src) assert push_result.exit_code == 0, push_result.output # Set up a fresh dst that pulls from the same remote. dst = _make_repo(tmp_path / "dst") _write_config_toml(dst, {"origin": remote.as_uri()}) pull_result = _pull(dst) assert pull_result.exit_code == 0, pull_result.output assert commit_path(dst, cid).exists(), "commit missing after pull" assert _branch_ref(dst, "main") == cid from muse.core.object_store import read_object assert read_object(dst, oid) == content, "object content mismatch after pull"