"""Comprehensive tests for ``muse plumbing update-ref``. Coverage tiers -------------- - Unit: _FORMAT_CHOICES - Integration: create ref, update ref, delete ref, --no-verify, text format - CAS: --old-value happy path, mismatch, null guard - Security: ANSI/null in branch name rejected, errors to stderr, no traceback - Stress: 200 sequential updates """ from __future__ import annotations import datetime import json import pathlib from muse.core.errors import ExitCode from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() _SNAP_ID: str = compute_snapshot_id({}) _COMMITTED_AT: datetime.datetime = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: repo = tmp_path / "repo" muse = repo / ".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": "test-repo", "domain": "code"})) return repo def _snap(repo: pathlib.Path) -> str: """Write an empty-manifest snapshot; return its content-addressed ID.""" write_snapshot(repo, SnapshotRecord( snapshot_id=_SNAP_ID, manifest={}, created_at=_COMMITTED_AT, )) return _SNAP_ID def _commit(repo: pathlib.Path, message: str = "test") -> str: """Write a commit with a real content-addressed ID; return the commit_id.""" snap_id = _snap(repo) commit_id = compute_commit_id([], snap_id, message, _COMMITTED_AT.isoformat()) write_commit(repo, CommitRecord( commit_id=commit_id, repo_id="test-repo", branch="main", snapshot_id=snap_id, message=message, committed_at=_COMMITTED_AT, )) return commit_id def _write_ref(repo: pathlib.Path, branch: str, commit_id: str) -> None: ref = repo / ".muse" / "refs" / "heads" / branch ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(commit_id) def _ur(repo: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke( cli, ["update-ref", *args], env={"MUSE_REPO_ROOT": str(repo)}, ) # --------------------------------------------------------------------------- # Unit # --------------------------------------------------------------------------- class TestUnit: def test_format_choices(self) -> None: from muse.cli.commands.plumbing.update_ref import _FORMAT_CHOICES assert "json" in _FORMAT_CHOICES assert "text" in _FORMAT_CHOICES # --------------------------------------------------------------------------- # Integration — create and update # --------------------------------------------------------------------------- class TestCreateUpdate: def test_creates_new_ref(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, "create ref test") result = _ur(repo, "feature", cid) assert result.exit_code == 0 data = json.loads(result.output) assert data["branch"] == "feature" assert data["commit_id"] == cid assert (repo / ".muse" / "refs" / "heads" / "feature").read_text() == cid def test_previous_is_null_for_new_ref(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, "new ref test") data = json.loads(_ur(repo, "new-branch", cid).output) assert data["previous"] is None def test_updates_existing_ref(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) old_id = _commit(repo, "old commit") new_id = _commit(repo, "new commit") _write_ref(repo, "main", old_id) data = json.loads(_ur(repo, "main", new_id).output) assert data["previous"] == old_id assert data["commit_id"] == new_id def test_json_shorthand(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, "json shorthand test") result = _ur(repo, "--json", "main", cid) assert result.exit_code == 0 assert "commit_id" in json.loads(result.output) def test_text_format_silent_on_success(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, "text format test") result = _ur(repo, "--format", "text", "main", cid) assert result.exit_code == 0 assert result.output.strip() == "" # --------------------------------------------------------------------------- # Integration — delete # --------------------------------------------------------------------------- class TestDeleteRef: def test_delete_existing_ref(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_ref(repo, "todelete", "5" * 64) result = _ur(repo, "--delete", "todelete") assert result.exit_code == 0 data = json.loads(result.output) assert data["deleted"] is True assert not (repo / ".muse" / "refs" / "heads" / "todelete").exists() def test_delete_nonexistent_ref_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ur(repo, "--delete", "ghost-branch") assert result.exit_code == ExitCode.USER_ERROR def test_delete_text_format_silent(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) _write_ref(repo, "to-del", "6" * 64) result = _ur(repo, "--delete", "--format", "text", "to-del") assert result.exit_code == 0 assert result.output.strip() == "" # --------------------------------------------------------------------------- # Integration — --no-verify # --------------------------------------------------------------------------- class TestNoVerify: def test_no_verify_accepts_unknown_commit(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = "7" * 64 # not in store result = _ur(repo, "--no-verify", "staging", cid) assert result.exit_code == 0 assert (repo / ".muse" / "refs" / "heads" / "staging").read_text() == cid def test_verify_rejects_unknown_commit(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = "8" * 64 # not in store result = _ur(repo, "main", cid) assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # CAS — compare-and-swap # --------------------------------------------------------------------------- class TestCAS: def test_cas_succeeds_when_current_matches(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) old_id = _commit(repo, "cas old commit") new_id = _commit(repo, "cas new commit") _write_ref(repo, "main", old_id) result = _ur(repo, "--old-value", old_id, "main", new_id) assert result.exit_code == 0 data = json.loads(result.output) assert data["commit_id"] == new_id def test_cas_fails_when_current_differs(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) actual = _commit(repo, "actual commit") new_id = _commit(repo, "new commit") # Use a different (non-stored) ID as the expected old value. expected = "c1" + "0" * 62 _write_ref(repo, "main", actual) result = _ur(repo, "--old-value", expected, "main", new_id) assert result.exit_code == ExitCode.USER_ERROR def test_cas_null_succeeds_when_ref_absent(self, tmp_path: pathlib.Path) -> None: """--old-value null asserts the ref does not yet exist.""" repo = _make_repo(tmp_path) cid = _commit(repo, "cas null test") result = _ur(repo, "--no-verify", "--old-value", "null", "brand-new", cid) assert result.exit_code == 0 def test_cas_null_fails_when_ref_exists(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) existing = _commit(repo, "existing commit") new_id = _commit(repo, "new commit for null cas") _write_ref(repo, "contested", existing) result = _ur(repo, "--old-value", "null", "contested", new_id) assert result.exit_code == ExitCode.USER_ERROR def test_cas_delete_succeeds_when_matches(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = "aa" + "0" * 62 _write_ref(repo, "conditioned", cid) result = _ur(repo, "--delete", "--old-value", cid, "conditioned") assert result.exit_code == 0 def test_cas_delete_fails_when_differs(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) actual = "bb" + "0" * 62 wrong = "cc" + "0" * 62 _write_ref(repo, "conditioned", actual) result = _ur(repo, "--delete", "--old-value", wrong, "conditioned") assert result.exit_code == ExitCode.USER_ERROR assert (repo / ".muse" / "refs" / "heads" / "conditioned").exists() # --------------------------------------------------------------------------- # Error cases # --------------------------------------------------------------------------- class TestErrors: def test_invalid_branch_name_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ur(repo, "branch\x00null", "a" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_invalid_commit_id_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ur(repo, "main", "not-hex") assert result.exit_code == ExitCode.USER_ERROR def test_no_commit_id_without_delete_errors(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ur(repo, "main") assert result.exit_code == ExitCode.USER_ERROR # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_ansi_in_branch_rejected(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ur(repo, "\x1b[31mbranch", "a" * 64) assert result.exit_code == ExitCode.USER_ERROR def test_no_traceback_on_bad_branch(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) result = _ur(repo, "bad\x00branch", "a" * 64) assert "Traceback" not in result.output # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_200_sequential_updates(self, tmp_path: pathlib.Path) -> None: repo = _make_repo(tmp_path) cid = _commit(repo, "stress test commit") for i in range(200): result = _ur(repo, "stress-branch", cid) assert result.exit_code == 0, f"failed at iteration {i}" data = json.loads(result.output) assert data["commit_id"] == cid