"""muse#22 — fetch.prune config option; muse pull prunes stale tracking refs. Background: when a merge proposal is merged on the hub with "source branch deleted", `.muse/remotes//` for that branch was never cleaned up by a subsequent `muse pull` — only an explicit `muse fetch --prune` removed it. `muse fetch --prune` itself already existed; this suite covers the two things that didn't: 1. A `fetch.prune` config switch (repo + global, repo wins) so pruning can be made the default without passing `--prune` on every invocation, mirroring git's own `fetch.prune` exactly (the user's own proposed solution on this ticket). 2. `muse pull` actually pruning at all — it had its own independent fetch logic that never called the pruning routine `muse fetch --prune` uses. """ from __future__ import annotations import datetime import json import pathlib from unittest.mock import MagicMock, patch import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import Manifest, blob_id from muse.core.paths import muse_dir, remotes_dir cli = None runner = CliRunner() REMOTE_ID = "a" * 64 STALE_ID = "b" * 64 def _env(root: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(root)} def _write_remote_ref(tmp_path: pathlib.Path, remote: str, branch: str, commit_id: str) -> None: ref_file = remotes_dir(tmp_path) / remote / branch ref_file.parent.mkdir(parents=True, exist_ok=True) ref_file.write_text(commit_id) # --------------------------------------------------------------------------- # Unit — get_fetch_prune precedence # --------------------------------------------------------------------------- class TestGetFetchPrunePrecedence: def test_default_false_when_unset(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.config import get_fetch_prune fake_home = tmp_path / "fakehome" fake_home.mkdir() monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) repo = tmp_path / "repo" repo.mkdir() assert get_fetch_prune(repo) is False def test_repo_true_wins(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.config import get_fetch_prune, set_config_value fake_home = tmp_path / "fakehome" fake_home.mkdir() monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) repo = tmp_path / "repo" repo.mkdir() set_config_value("fetch.prune", "true", repo) assert get_fetch_prune(repo) is True def test_global_true_used_when_repo_unset(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.config import get_fetch_prune, set_config_value fake_home = tmp_path / "fakehome" fake_home.mkdir() monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) set_config_value("fetch.prune", "true", fake_home) repo = tmp_path / "repo" repo.mkdir() assert get_fetch_prune(repo) is True def test_repo_false_overrides_global_true(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.config import get_fetch_prune, set_config_value fake_home = tmp_path / "fakehome" fake_home.mkdir() monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) set_config_value("fetch.prune", "true", fake_home) repo = tmp_path / "repo" repo.mkdir() set_config_value("fetch.prune", "false", repo) assert get_fetch_prune(repo) is False # --------------------------------------------------------------------------- # Integration — muse config set/get --global # --------------------------------------------------------------------------- class TestConfigGlobalFlag: def test_set_global_writes_to_home_not_repo( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: fake_home = tmp_path / "fakehome" fake_home.mkdir() repo = tmp_path / "repo" repo.mkdir() monkeypatch.setenv("MUSE_REPO_ROOT", str(repo)) monkeypatch.chdir(repo) monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) r = runner.invoke(cli, ["config", "set", "--global", "fetch.prune", "true", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output assert not (repo / ".muse" / "config.toml").exists(), ( "--global must not write to the repo config" ) assert (fake_home / ".muse" / "config.toml").exists() assert "prune = true" in (fake_home / ".muse" / "config.toml").read_text() def test_get_global_reads_from_home( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: from muse.cli.config import set_config_value fake_home = tmp_path / "fakehome" fake_home.mkdir() repo = tmp_path / "repo" repo.mkdir() monkeypatch.setenv("MUSE_REPO_ROOT", str(repo)) monkeypatch.chdir(repo) monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) set_config_value("fetch.prune", "true", fake_home) r = runner.invoke(cli, ["config", "get", "--global", "fetch.prune", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["value"] == "true" def test_repo_set_does_not_leak_to_global_get( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: fake_home = tmp_path / "fakehome" fake_home.mkdir() repo = tmp_path / "repo" repo.mkdir() monkeypatch.setenv("MUSE_REPO_ROOT", str(repo)) monkeypatch.chdir(repo) monkeypatch.setattr(pathlib.Path, "home", lambda: fake_home) r = runner.invoke(cli, ["config", "set", "fetch.prune", "true", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output r = runner.invoke(cli, ["config", "get", "--global", "fetch.prune", "--json"], env=_env(repo)) assert r.exit_code != 0, "repo-level set must not be visible via --global get" # --------------------------------------------------------------------------- # Integration — muse fetch honours fetch.prune config # --------------------------------------------------------------------------- def _init_fetch_repo(tmp_path: pathlib.Path) -> None: dot_muse = muse_dir(tmp_path) for sub in ("objects", "commits", "snapshots", "remotes", "refs/heads"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads" / "main").write_text("") (dot_muse / "config.toml").write_text('[remotes.origin]\nurl = "http://localhost:19999"\n') (dot_muse / "repo.json").write_text('{"id": "test-repo-id"}') def _make_remote_info(branch_heads: Manifest) -> dict: return { "repo_id": "test-repo-id", "domain": "code", "default_branch": "main", "branch_heads": branch_heads, } class TestFetchHonoursPruneConfig: def test_config_true_prunes_without_explicit_flag( self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str] ) -> None: from muse.cli.config import set_config_value _init_fetch_repo(tmp_path) _write_remote_ref(tmp_path, "origin", "dead-branch", STALE_ID) set_config_value("fetch.prune", "true", tmp_path) info = _make_remote_info({"main": REMOTE_ID}) with patch("muse.cli.commands.fetch.get_remote", return_value="http://localhost:19999"): with patch("muse.cli.commands.fetch.get_signing_identity", return_value=None): with patch("muse.cli.commands.fetch.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info mt.return_value.fetch_mpack.return_value = { "repo_id": "test-repo-id", "domain": "code", "default_branch": "main", "branch_heads": {"main": REMOTE_ID}, "commits": [], "snapshots": [], "blobs": [], "blobs_received": 0, "shallow_commits": [], } r = runner.invoke(cli, ["fetch", "origin", "--json"], env=_env(tmp_path)) assert r.exit_code == 0, r.output data = json.loads([l for l in r.output.splitlines() if l.strip().startswith("{")][-1]) assert data["results"][0]["pruned"] == ["origin/dead-branch"], data def test_no_prune_flag_overrides_config_true( self, tmp_path: pathlib.Path ) -> None: from muse.cli.config import set_config_value _init_fetch_repo(tmp_path) _write_remote_ref(tmp_path, "origin", "dead-branch", STALE_ID) set_config_value("fetch.prune", "true", tmp_path) info = _make_remote_info({"main": REMOTE_ID}) with patch("muse.cli.commands.fetch.get_remote", return_value="http://localhost:19999"): with patch("muse.cli.commands.fetch.get_signing_identity", return_value=None): with patch("muse.cli.commands.fetch.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info mt.return_value.fetch_mpack.return_value = { "repo_id": "test-repo-id", "domain": "code", "default_branch": "main", "branch_heads": {"main": REMOTE_ID}, "commits": [], "snapshots": [], "blobs": [], "blobs_received": 0, "shallow_commits": [], } r = runner.invoke(cli, ["fetch", "origin", "--no-prune", "--json"], env=_env(tmp_path)) assert r.exit_code == 0, r.output data = json.loads([l for l in r.output.splitlines() if l.strip().startswith("{")][-1]) assert data["results"][0]["pruned"] == [] assert (remotes_dir(tmp_path) / "origin" / "dead-branch").exists() # --------------------------------------------------------------------------- # Integration — muse pull now prunes (it previously never did at all) # --------------------------------------------------------------------------- @pytest.fixture() def pull_repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal .muse/ repo with one commit on main, ready for muse pull.""" from muse.core.object_store import write_object from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import CommitRecord, write_commit from muse.core.snapshots import SnapshotRecord, write_snapshot dot_muse = muse_dir(tmp_path) for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"): (dot_muse / sub).mkdir(parents=True, exist_ok=True) (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"})) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "config.toml").write_text('[remotes.origin]\nurl = "https://hub.example.com/r"\n') blob = b"x = 1\n" oid = blob_id(blob) write_object(tmp_path, oid, blob) snap_id = hash_snapshot({"a.py": oid}) write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest={"a.py": oid})) ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) cid = hash_commit(parent_ids=[], snapshot_id=snap_id, message="base", committed_at_iso=ts.isoformat()) write_commit(tmp_path, CommitRecord( commit_id=cid, branch="main", snapshot_id=snap_id, message="base", committed_at=ts, )) (dot_muse / "refs" / "heads" / "main").write_text(cid) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) return tmp_path class TestPullPrunesStaleRefs: def test_pull_up_to_date_prunes_when_explicit_flag(self, pull_repo: pathlib.Path) -> None: """The exact muse#22 repro: pulling a branch that's already up to date must still prune a DIFFERENT branch's stale tracking ref.""" from muse.cli.commands.pull import _PullJson # noqa: F401 - import smoke cid = (muse_dir(pull_repo) / "refs" / "heads" / "main").read_text().strip() _write_remote_ref(pull_repo, "origin", "fix/proposal-closed-state-badge", STALE_ID) info = { "repo_id": "test-repo", "domain": "code", "default_branch": "main", "branch_heads": {"main": cid}, } with patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com/r"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info r = runner.invoke( cli, ["pull", "origin", "--prune", "--json"], env=_env(pull_repo), ) assert r.exit_code == 0, r.output data = json.loads([l for l in r.output.splitlines() if l.strip().startswith("{")][-1]) assert data["status"] == "up_to_date" assert data["pruned"] == ["origin/fix/proposal-closed-state-badge"] assert not (remotes_dir(pull_repo) / "origin" / "fix/proposal-closed-state-badge").exists() def test_pull_prunes_when_config_true_no_explicit_flag(self, pull_repo: pathlib.Path) -> None: from muse.cli.config import set_config_value set_config_value("fetch.prune", "true", pull_repo) cid = (muse_dir(pull_repo) / "refs" / "heads" / "main").read_text().strip() _write_remote_ref(pull_repo, "origin", "stale-branch", STALE_ID) info = { "repo_id": "test-repo", "domain": "code", "default_branch": "main", "branch_heads": {"main": cid}, } with patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com/r"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info r = runner.invoke(cli, ["pull", "origin", "--json"], env=_env(pull_repo)) assert r.exit_code == 0, r.output data = json.loads([l for l in r.output.splitlines() if l.strip().startswith("{")][-1]) assert data["pruned"] == ["origin/stale-branch"] def test_pull_does_not_prune_by_default(self, pull_repo: pathlib.Path) -> None: cid = (muse_dir(pull_repo) / "refs" / "heads" / "main").read_text().strip() _write_remote_ref(pull_repo, "origin", "stale-branch", STALE_ID) info = { "repo_id": "test-repo", "domain": "code", "default_branch": "main", "branch_heads": {"main": cid}, } with patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com/r"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info r = runner.invoke(cli, ["pull", "origin", "--json"], env=_env(pull_repo)) assert r.exit_code == 0, r.output data = json.loads([l for l in r.output.splitlines() if l.strip().startswith("{")][-1]) assert data["pruned"] == [] assert (remotes_dir(pull_repo) / "origin" / "stale-branch").exists() def test_pull_dry_run_prune_does_not_delete(self, pull_repo: pathlib.Path) -> None: cid = (muse_dir(pull_repo) / "refs" / "heads" / "main").read_text().strip() _write_remote_ref(pull_repo, "origin", "stale-branch", STALE_ID) info = { "repo_id": "test-repo", "domain": "code", "default_branch": "main", "branch_heads": {"main": cid}, } with patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com/r"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info r = runner.invoke( cli, ["pull", "origin", "--prune", "--dry-run", "--json"], env=_env(pull_repo), ) assert r.exit_code == 0, r.output data = json.loads([l for l in r.output.splitlines() if l.strip().startswith("{")][-1]) assert data["pruned"] == ["origin/stale-branch"], ( "dry-run must still report what WOULD be pruned" ) assert (remotes_dir(pull_repo) / "origin" / "stale-branch").exists(), ( "--dry-run must not actually delete the ref" )