"""Phase 1 — RL_01–RL_05: reflog coverage for ``muse pull`` and ``muse shelf``. TDD: tests are written before the implementation. Each class maps to one or more deliverables from issue #47, Phase 1. Coverage tiers -------------- RL_01 pull (fast-forward, bootstrap, three-way merge) calls append_reflog RL_02 shelf pop calls append_reflog RL_03 shelf apply calls append_reflog RL_04 append_reflog failure is non-fatal in pull, shelf pop, shelf apply RL_05 integration: fast-forward pull writes real entry to HEAD log on disk integration: shelf pop writes real entry to HEAD log on disk """ from __future__ import annotations import datetime import json import pathlib from unittest.mock import MagicMock, call, patch import pytest from muse.core.paths import muse_dir, logs_dir from muse.core.types import blob_id from tests.cli_test_helper import CliRunner runner = CliRunner() cli = None # argparse migration stub — CliRunner.invoke() requires this as first arg # --------------------------------------------------------------------------- # Shared repo helpers # --------------------------------------------------------------------------- def _make_repo( tmp_path: pathlib.Path, branch: str = "main", *, with_commit: bool = True, ) -> tuple[pathlib.Path, str | None]: """Create a minimal Muse repo. Returns (root, head_commit_id | None).""" from muse._version import __version__ from muse.core.commits import CommitRecord, write_commit from muse.core.ids import hash_commit, hash_snapshot from muse.core.object_store import write_object from muse.core.snapshots import SnapshotRecord, write_snapshot dot = muse_dir(tmp_path) for sub in ("refs/heads", "objects", "commits", "snapshots", "logs/refs/heads", "logs"): (dot / sub).mkdir(parents=True, exist_ok=True) (dot / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (dot / "HEAD").write_text(f"ref: refs/heads/{branch}\n") (dot / "config.toml").write_text( '[remotes.origin]\nurl = "https://hub.example.com/r"\n' ) if not with_commit: return tmp_path, None blob = b"hello\n" oid = blob_id(blob) write_object(tmp_path, oid, blob) snap_id = hash_snapshot({"a.txt": oid}) write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest={"a.txt": oid})) ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) cid = hash_commit(parent_ids=[], snapshot_id=snap_id, message="init", committed_at_iso=ts.isoformat()) write_commit(tmp_path, CommitRecord( commit_id=cid, branch=branch, snapshot_id=snap_id, message="init", committed_at=ts, )) (dot / "refs" / "heads" / branch).write_text(cid) return tmp_path, cid def _make_next_commit( root: pathlib.Path, parent_id: str | None = None, branch: str = "main", ) -> str: """Write a commit to the store (does NOT advance the branch ref). Returns the new commit ID. The caller controls when (and if) the branch ref is advanced — this lets us simulate 'commit exists in store but remote hasn't been pulled yet'. Pass ``parent_id=None`` for an initial commit (no parent). """ from muse.core.commits import CommitRecord, write_commit from muse.core.ids import hash_commit, hash_snapshot from muse.core.object_store import write_object from muse.core.snapshots import SnapshotRecord, write_snapshot blob = b"world\n" oid = blob_id(blob) write_object(root, oid, blob) snap_id = hash_snapshot({"a.txt": oid}) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest={"a.txt": oid})) ts = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc) parent_ids = [parent_id] if parent_id else [] cid = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message="update", committed_at_iso=ts.isoformat(), ) write_commit(root, CommitRecord( commit_id=cid, branch=branch, snapshot_id=snap_id, message="update", committed_at=ts, parent_commit_id=parent_id if parent_id else None, )) return cid def _env(root: pathlib.Path) -> dict[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _remote_info(branch_heads: dict[str, str]) -> dict: return { "repo_id": "test-repo", "domain": "code", "default_branch": "main", "branch_heads": branch_heads, } def _read_head_log(root: pathlib.Path) -> list[str]: """Return non-empty lines from the HEAD reflog, newest-first.""" from muse.core.reflog import read_reflog entries = read_reflog(root, branch=None, limit=100) return [e.operation for e in entries] def _read_branch_log(root: pathlib.Path, branch: str = "main") -> list[str]: """Return operation strings from the branch reflog, newest-first.""" from muse.core.reflog import read_reflog entries = read_reflog(root, branch=branch, limit=100) return [e.operation for e in entries] # --------------------------------------------------------------------------- # RL_01 — pull writes a reflog entry (mocked transport) # --------------------------------------------------------------------------- class TestPullReflogFastForward: """RL_01 fast-forward path: write_branch_ref advances to theirs_commit_id.""" def test_calls_append_reflog_on_fast_forward(self, tmp_path: pathlib.Path) -> None: root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog") as mock_rl: r = runner.invoke(cli, ["pull", "--json"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_called_once() call_kwargs = mock_rl.call_args assert call_kwargs[0][0] == root # repo_root assert call_kwargs[0][1] == "main" # branch assert call_kwargs[1]["old_id"] == commit_a assert call_kwargs[1]["new_id"] == commit_b assert "pull" in call_kwargs[1]["operation"] def test_operation_string_includes_remote_and_branch(self, tmp_path: pathlib.Path) -> None: root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog") as mock_rl: runner.invoke(cli, ["pull", "origin", "main"], env=_env(root)) op = mock_rl.call_args[1]["operation"] assert "origin" in op assert "main" in op class TestPullReflogBootstrap: """RL_01 bootstrap path: ours_commit_id is None (empty repo).""" def test_calls_append_reflog_on_bootstrap(self, tmp_path: pathlib.Path) -> None: """Bootstrap pull (no local commits yet) must write a reflog entry.""" root, _ = _make_repo(tmp_path, with_commit=False) # Create the remote commit in the store so read_commit succeeds. # No parent — this is the initial commit on the remote. commit_b = _make_next_commit(root, parent_id=None) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog") as mock_rl: r = runner.invoke(cli, ["pull", "--json"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_called_once() call_kwargs = mock_rl.call_args # old_id must be None (no previous commit) assert call_kwargs[1]["old_id"] is None assert call_kwargs[1]["new_id"] == commit_b class TestPullReflogMerge: """RL_01 three-way merge path: diverged branches produce a merge commit.""" def test_calls_append_reflog_on_three_way_merge(self, tmp_path: pathlib.Path) -> None: """Three-way merge must call append_reflog with the new merge commit ID.""" from muse.core.commits import CommitRecord, write_commit from muse.core.ids import hash_commit, hash_snapshot from muse.core.object_store import write_object from muse.core.snapshots import SnapshotRecord, write_snapshot root, commit_a = _make_repo(tmp_path) assert commit_a is not None # Diverged remote commit (shares parent A, different content). blob_r = b"remote\n" oid_r = blob_id(blob_r) write_object(root, oid_r, blob_r) snap_r = hash_snapshot({"b.txt": oid_r}) write_snapshot(root, SnapshotRecord(snapshot_id=snap_r, manifest={"b.txt": oid_r})) ts_r = datetime.datetime(2026, 1, 3, tzinfo=datetime.timezone.utc) commit_remote = hash_commit( parent_ids=[commit_a], snapshot_id=snap_r, message="remote change", committed_at_iso=ts_r.isoformat(), ) write_commit(root, CommitRecord( commit_id=commit_remote, branch="main", snapshot_id=snap_r, message="remote change", committed_at=ts_r, parent_commit_id=commit_a, )) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_remote}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog") as mock_rl: r = runner.invoke(cli, ["pull", "--json"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_called_once() call_kwargs = mock_rl.call_args assert call_kwargs[1]["old_id"] == commit_a # new_id is a freshly-created merge commit — just verify it's not the old head assert call_kwargs[1]["new_id"] != commit_a assert "pull" in call_kwargs[1]["operation"] class TestPullReflogNotCalled: """append_reflog must NOT be called when the branch pointer does not move.""" def test_no_reflog_on_dry_run(self, tmp_path: pathlib.Path) -> None: root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog") as mock_rl: r = runner.invoke(cli, ["pull", "--dry-run"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_not_called() def test_no_reflog_on_no_merge(self, tmp_path: pathlib.Path) -> None: """--no-merge fetches but does not advance the local branch.""" root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog") as mock_rl: r = runner.invoke(cli, ["pull", "--no-merge"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_not_called() def test_no_reflog_when_already_up_to_date(self, tmp_path: pathlib.Path) -> None: root, commit_a = _make_repo(tmp_path) assert commit_a is not None with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=commit_a): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_a}) with patch("muse.cli.commands.pull.append_reflog") as mock_rl: runner.invoke(cli, ["pull"], env=_env(root)) mock_rl.assert_not_called() # --------------------------------------------------------------------------- # RL_04 — pull reflog failure is non-fatal # --------------------------------------------------------------------------- class TestPullReflogNonFatal: """RL_04: an exception from append_reflog must never abort pull.""" def test_ff_pull_completes_despite_reflog_error(self, tmp_path: pathlib.Path) -> None: root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) def _explode(*args, **kwargs): raise OSError("disk full") with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.append_reflog", side_effect=_explode): r = runner.invoke(cli, ["pull", "--json"], env=_env(root)) # Pull must succeed even though reflog write exploded. assert r.exit_code == 0, r.output # Branch pointer must have advanced. from muse.core.refs import get_head_commit_id assert get_head_commit_id(root, "main") == commit_b # --------------------------------------------------------------------------- # RL_02 — shelf pop writes a reflog entry # --------------------------------------------------------------------------- def _make_shelf_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: """Create a repo with a commit, save a shelf entry, return (root, entry_name).""" root, commit_a = _make_repo(tmp_path) assert commit_a is not None # Write a file to disk so shelf save has something to capture. (root / "work.txt").write_text("in progress") # Save a shelf entry via the CLI. r = runner.invoke(cli, ["shelf", "save", "my-work", "--json"], env=_env(root)) assert r.exit_code == 0, f"shelf save failed: {r.output}" return root, "my-work" class TestShelfPopReflog: """RL_02: shelf pop writes a reflog entry.""" def test_calls_append_reflog_on_pop(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) with patch("muse.cli.commands.shelf.append_reflog") as mock_rl: r = runner.invoke(cli, ["shelf", "pop", entry_name, "--json"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_called_once() def test_operation_string_contains_name(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) with patch("muse.cli.commands.shelf.append_reflog") as mock_rl: runner.invoke(cli, ["shelf", "pop", entry_name], env=_env(root)) op = mock_rl.call_args[1]["operation"] assert entry_name in op assert "shelf-pop" in op def test_reflog_branch_is_current_branch(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) with patch("muse.cli.commands.shelf.append_reflog") as mock_rl: runner.invoke(cli, ["shelf", "pop", entry_name], env=_env(root)) branch_arg = mock_rl.call_args[0][1] assert branch_arg == "main" def test_new_id_is_current_head(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) from muse.core.refs import get_head_commit_id head = get_head_commit_id(root, "main") with patch("muse.cli.commands.shelf.append_reflog") as mock_rl: runner.invoke(cli, ["shelf", "pop", entry_name], env=_env(root)) new_id = mock_rl.call_args[1]["new_id"] assert new_id == head # --------------------------------------------------------------------------- # RL_03 — shelf apply writes a reflog entry # --------------------------------------------------------------------------- class TestShelfApplyReflog: """RL_03: shelf apply writes a reflog entry (entry is kept in the shelf).""" def test_calls_append_reflog_on_apply(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) with patch("muse.cli.commands.shelf.append_reflog") as mock_rl: r = runner.invoke(cli, ["shelf", "apply", entry_name, "--json"], env=_env(root)) assert r.exit_code == 0, r.output mock_rl.assert_called_once() def test_operation_string_contains_name(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) with patch("muse.cli.commands.shelf.append_reflog") as mock_rl: runner.invoke(cli, ["shelf", "apply", entry_name], env=_env(root)) op = mock_rl.call_args[1]["operation"] assert entry_name in op assert "shelf-apply" in op def test_entry_survives_after_apply(self, tmp_path: pathlib.Path) -> None: """Unlike pop, apply keeps the shelf entry.""" root, entry_name = _make_shelf_repo(tmp_path) with patch("muse.cli.commands.shelf.append_reflog"): runner.invoke(cli, ["shelf", "apply", entry_name], env=_env(root)) r = runner.invoke(cli, ["shelf", "list", "--json"], env=_env(root)) data = json.loads(r.output) names = [e["name"] for e in data.get("entries", [])] assert entry_name in names, "apply must not remove the shelf entry" # --------------------------------------------------------------------------- # RL_04 — shelf reflog failure is non-fatal # --------------------------------------------------------------------------- class TestShelfReflogNonFatal: """RL_04: an exception from append_reflog must never abort shelf pop or apply.""" def test_shelf_pop_completes_despite_reflog_error(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) (root / "work.txt").unlink(missing_ok=True) # reset working tree def _explode(*args, **kwargs): raise OSError("read-only filesystem") with patch("muse.cli.commands.shelf.append_reflog", side_effect=_explode): r = runner.invoke(cli, ["shelf", "pop", entry_name, "--json"], env=_env(root)) assert r.exit_code == 0, r.output # File must still have been restored. assert (root / "work.txt").exists(), "shelf pop must restore files even if reflog fails" def test_shelf_apply_completes_despite_reflog_error(self, tmp_path: pathlib.Path) -> None: root, entry_name = _make_shelf_repo(tmp_path) (root / "work.txt").unlink(missing_ok=True) def _explode(*args, **kwargs): raise PermissionError("no write access to .muse/logs") with patch("muse.cli.commands.shelf.append_reflog", side_effect=_explode): r = runner.invoke(cli, ["shelf", "apply", entry_name, "--json"], env=_env(root)) assert r.exit_code == 0, r.output assert (root / "work.txt").exists(), "shelf apply must restore files even if reflog fails" # --------------------------------------------------------------------------- # RL_05 — integration: real log entries written to disk # --------------------------------------------------------------------------- class TestReflogIntegration: """RL_05: verify that the actual .muse/logs/ files are updated.""" def test_ff_pull_writes_head_log_entry(self, tmp_path: pathlib.Path) -> None: """Fast-forward pull must append a line to .muse/logs/HEAD.""" root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): r = runner.invoke(cli, ["pull", "--json"], env=_env(root)) assert r.exit_code == 0, r.output ops = _read_head_log(root) assert ops, "HEAD log must have at least one entry after pull" assert any("pull" in op for op in ops), f"No pull entry in HEAD log; found: {ops}" def test_ff_pull_writes_branch_log_entry(self, tmp_path: pathlib.Path) -> None: """Fast-forward pull must also append to the branch-specific log.""" root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): runner.invoke(cli, ["pull"], env=_env(root)) ops = _read_branch_log(root, "main") assert any("pull" in op for op in ops), f"No pull entry in main log; found: {ops}" def test_ff_pull_old_and_new_ids_in_log(self, tmp_path: pathlib.Path) -> None: """The log entry must reference the correct old and new commit IDs.""" from muse.core.reflog import read_reflog root, commit_a = _make_repo(tmp_path) assert commit_a is not None commit_b = _make_next_commit(root, commit_a) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.get_remote_head", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _remote_info({"main": commit_b}) mt.return_value.fetch_mpack.return_value = { "commits": [], "snapshots": [], "blobs_received": 0 } with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "blobs_written": 0, "blobs_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): runner.invoke(cli, ["pull"], env=_env(root)) entries = read_reflog(root, branch=None, limit=10) assert entries, "HEAD log must have entries" entry = entries[0] # old_id and new_id are stored as bare hex; commit_a and commit_b are sha256:-prefixed assert commit_a.removeprefix("sha256:") in entry.old_id or entry.old_id == commit_a assert commit_b.removeprefix("sha256:") in entry.new_id or entry.new_id == commit_b def test_shelf_pop_writes_head_log_entry(self, tmp_path: pathlib.Path) -> None: """shelf pop must write an entry to .muse/logs/HEAD.""" root, entry_name = _make_shelf_repo(tmp_path) r = runner.invoke(cli, ["shelf", "pop", entry_name, "--json"], env=_env(root)) assert r.exit_code == 0, r.output ops = _read_head_log(root) assert any("shelf-pop" in op for op in ops), f"No shelf-pop entry in HEAD log; found: {ops}" def test_shelf_apply_writes_head_log_entry(self, tmp_path: pathlib.Path) -> None: """shelf apply must write an entry to .muse/logs/HEAD.""" root, entry_name = _make_shelf_repo(tmp_path) r = runner.invoke(cli, ["shelf", "apply", entry_name, "--json"], env=_env(root)) assert r.exit_code == 0, r.output ops = _read_head_log(root) assert any("shelf-apply" in op for op in ops), f"No shelf-apply entry in HEAD log; found: {ops}" def test_shelf_pop_log_references_entry_name(self, tmp_path: pathlib.Path) -> None: """The shelf-pop log entry must name the shelf entry that was restored.""" from muse.core.reflog import read_reflog root, entry_name = _make_shelf_repo(tmp_path) runner.invoke(cli, ["shelf", "pop", entry_name], env=_env(root)) entries = read_reflog(root, branch=None, limit=10) pop_entries = [e for e in entries if "shelf-pop" in e.operation] assert pop_entries, "No shelf-pop entry found in HEAD log" assert entry_name in pop_entries[0].operation