test_plumbing_update_ref.py
python
| 1 | """Tests for ``muse plumbing update-ref``. |
| 2 | |
| 3 | Covers: normal update, new-branch creation, previous-commit tracking, |
| 4 | ``--delete`` mode, ``--no-verify`` bypass, commit-not-found error when |
| 5 | ``--verify`` is active, bad-commit-ID format, and I/O error recovery. |
| 6 | """ |
| 7 | |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import datetime |
| 11 | import hashlib |
| 12 | import json |
| 13 | import pathlib |
| 14 | |
| 15 | from typer.testing import CliRunner |
| 16 | |
| 17 | from muse.cli.app import cli |
| 18 | from muse.core.errors import ExitCode |
| 19 | from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot |
| 20 | |
| 21 | runner = CliRunner() |
| 22 | |
| 23 | |
| 24 | # --------------------------------------------------------------------------- |
| 25 | # Helpers |
| 26 | # --------------------------------------------------------------------------- |
| 27 | |
| 28 | |
| 29 | def _sha(tag: str) -> str: |
| 30 | return hashlib.sha256(tag.encode()).hexdigest() |
| 31 | |
| 32 | |
| 33 | def _init_repo(path: pathlib.Path) -> pathlib.Path: |
| 34 | muse = path / ".muse" |
| 35 | (muse / "commits").mkdir(parents=True) |
| 36 | (muse / "snapshots").mkdir(parents=True) |
| 37 | (muse / "objects").mkdir(parents=True) |
| 38 | (muse / "refs" / "heads").mkdir(parents=True) |
| 39 | (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") |
| 40 | (muse / "repo.json").write_text( |
| 41 | json.dumps({"repo_id": "test-repo", "domain": "midi"}), encoding="utf-8" |
| 42 | ) |
| 43 | return path |
| 44 | |
| 45 | |
| 46 | def _env(repo: pathlib.Path) -> dict[str, str]: |
| 47 | return {"MUSE_REPO_ROOT": str(repo)} |
| 48 | |
| 49 | |
| 50 | def _snap(repo: pathlib.Path) -> str: |
| 51 | sid = _sha("snap") |
| 52 | write_snapshot( |
| 53 | repo, |
| 54 | SnapshotRecord( |
| 55 | snapshot_id=sid, |
| 56 | manifest={}, |
| 57 | created_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), |
| 58 | ), |
| 59 | ) |
| 60 | return sid |
| 61 | |
| 62 | |
| 63 | def _commit( |
| 64 | repo: pathlib.Path, tag: str, sid: str, branch: str = "main", parent: str | None = None |
| 65 | ) -> str: |
| 66 | cid = _sha(tag) |
| 67 | write_commit( |
| 68 | repo, |
| 69 | CommitRecord( |
| 70 | commit_id=cid, |
| 71 | repo_id="test-repo", |
| 72 | branch=branch, |
| 73 | snapshot_id=sid, |
| 74 | message=tag, |
| 75 | committed_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), |
| 76 | author="tester", |
| 77 | parent_commit_id=parent, |
| 78 | ), |
| 79 | ) |
| 80 | return cid |
| 81 | |
| 82 | |
| 83 | def _set_branch(repo: pathlib.Path, branch: str, cid: str) -> None: |
| 84 | ref = repo / ".muse" / "refs" / "heads" / branch |
| 85 | ref.parent.mkdir(parents=True, exist_ok=True) |
| 86 | ref.write_text(cid, encoding="utf-8") |
| 87 | |
| 88 | |
| 89 | # --------------------------------------------------------------------------- |
| 90 | # Unit: format validation |
| 91 | # --------------------------------------------------------------------------- |
| 92 | |
| 93 | |
| 94 | class TestUpdateRefUnit: |
| 95 | def test_bad_commit_id_format_exits_user_error(self, tmp_path: pathlib.Path) -> None: |
| 96 | repo = _init_repo(tmp_path) |
| 97 | result = runner.invoke( |
| 98 | cli, ["plumbing", "update-ref", "main", "not-a-hex-id"], env=_env(repo) |
| 99 | ) |
| 100 | assert result.exit_code == ExitCode.USER_ERROR |
| 101 | |
| 102 | def test_missing_commit_id_without_delete_exits_user_error(self, tmp_path: pathlib.Path) -> None: |
| 103 | repo = _init_repo(tmp_path) |
| 104 | result = runner.invoke(cli, ["plumbing", "update-ref", "main"], env=_env(repo)) |
| 105 | assert result.exit_code == ExitCode.USER_ERROR |
| 106 | |
| 107 | def test_commit_not_in_store_exits_user_error_with_verify(self, tmp_path: pathlib.Path) -> None: |
| 108 | repo = _init_repo(tmp_path) |
| 109 | phantom = _sha("phantom") |
| 110 | result = runner.invoke( |
| 111 | cli, ["plumbing", "update-ref", "main", phantom], env=_env(repo) |
| 112 | ) |
| 113 | assert result.exit_code == ExitCode.USER_ERROR |
| 114 | assert "error" in json.loads(result.stdout) |
| 115 | |
| 116 | |
| 117 | # --------------------------------------------------------------------------- |
| 118 | # Integration: normal update |
| 119 | # --------------------------------------------------------------------------- |
| 120 | |
| 121 | |
| 122 | class TestUpdateRefNormal: |
| 123 | def test_update_creates_branch_ref_file(self, tmp_path: pathlib.Path) -> None: |
| 124 | repo = _init_repo(tmp_path) |
| 125 | sid = _snap(repo) |
| 126 | cid = _commit(repo, "c1", sid) |
| 127 | result = runner.invoke( |
| 128 | cli, ["plumbing", "update-ref", "main", cid], env=_env(repo) |
| 129 | ) |
| 130 | assert result.exit_code == 0, result.output |
| 131 | ref_file = repo / ".muse" / "refs" / "heads" / "main" |
| 132 | assert ref_file.read_text(encoding="utf-8") == cid |
| 133 | |
| 134 | def test_output_contains_branch_and_commit_id(self, tmp_path: pathlib.Path) -> None: |
| 135 | repo = _init_repo(tmp_path) |
| 136 | sid = _snap(repo) |
| 137 | cid = _commit(repo, "c2", sid) |
| 138 | result = runner.invoke(cli, ["plumbing", "update-ref", "main", cid], env=_env(repo)) |
| 139 | assert result.exit_code == 0 |
| 140 | data = json.loads(result.stdout) |
| 141 | assert data["branch"] == "main" |
| 142 | assert data["commit_id"] == cid |
| 143 | |
| 144 | def test_previous_field_is_null_for_new_branch(self, tmp_path: pathlib.Path) -> None: |
| 145 | repo = _init_repo(tmp_path) |
| 146 | sid = _snap(repo) |
| 147 | cid = _commit(repo, "c3", sid) |
| 148 | result = runner.invoke(cli, ["plumbing", "update-ref", "newbranch", cid], env=_env(repo)) |
| 149 | assert result.exit_code == 0 |
| 150 | assert json.loads(result.stdout)["previous"] is None |
| 151 | |
| 152 | def test_previous_field_reflects_old_commit(self, tmp_path: pathlib.Path) -> None: |
| 153 | repo = _init_repo(tmp_path) |
| 154 | sid = _snap(repo) |
| 155 | cid_old = _commit(repo, "old", sid) |
| 156 | cid_new = _commit(repo, "new", sid, parent=cid_old) |
| 157 | _set_branch(repo, "main", cid_old) |
| 158 | result = runner.invoke(cli, ["plumbing", "update-ref", "main", cid_new], env=_env(repo)) |
| 159 | assert result.exit_code == 0 |
| 160 | assert json.loads(result.stdout)["previous"] == cid_old |
| 161 | |
| 162 | def test_update_creates_nested_branch_dir(self, tmp_path: pathlib.Path) -> None: |
| 163 | """Branch names with slashes create subdirectories under refs/heads/.""" |
| 164 | repo = _init_repo(tmp_path) |
| 165 | sid = _snap(repo) |
| 166 | cid = _commit(repo, "feat-c", sid, branch="feat/new") |
| 167 | result = runner.invoke( |
| 168 | cli, ["plumbing", "update-ref", "feat/new", cid], env=_env(repo) |
| 169 | ) |
| 170 | assert result.exit_code == 0 |
| 171 | ref_file = repo / ".muse" / "refs" / "heads" / "feat" / "new" |
| 172 | assert ref_file.exists() |
| 173 | assert ref_file.read_text(encoding="utf-8") == cid |
| 174 | |
| 175 | |
| 176 | # --------------------------------------------------------------------------- |
| 177 | # Integration: --delete mode |
| 178 | # --------------------------------------------------------------------------- |
| 179 | |
| 180 | |
| 181 | class TestUpdateRefDelete: |
| 182 | def test_delete_removes_ref_file(self, tmp_path: pathlib.Path) -> None: |
| 183 | repo = _init_repo(tmp_path) |
| 184 | sid = _snap(repo) |
| 185 | cid = _commit(repo, "d1", sid) |
| 186 | _set_branch(repo, "to-delete", cid) |
| 187 | result = runner.invoke( |
| 188 | cli, ["plumbing", "update-ref", "--delete", "to-delete"], env=_env(repo) |
| 189 | ) |
| 190 | assert result.exit_code == 0 |
| 191 | assert not (repo / ".muse" / "refs" / "heads" / "to-delete").exists() |
| 192 | |
| 193 | def test_delete_output_has_deleted_true(self, tmp_path: pathlib.Path) -> None: |
| 194 | repo = _init_repo(tmp_path) |
| 195 | sid = _snap(repo) |
| 196 | cid = _commit(repo, "d2", sid) |
| 197 | _set_branch(repo, "bye", cid) |
| 198 | result = runner.invoke(cli, ["plumbing", "update-ref", "-d", "bye"], env=_env(repo)) |
| 199 | assert result.exit_code == 0 |
| 200 | assert json.loads(result.stdout)["deleted"] is True |
| 201 | |
| 202 | def test_delete_nonexistent_ref_exits_user_error(self, tmp_path: pathlib.Path) -> None: |
| 203 | repo = _init_repo(tmp_path) |
| 204 | result = runner.invoke( |
| 205 | cli, ["plumbing", "update-ref", "--delete", "does-not-exist"], env=_env(repo) |
| 206 | ) |
| 207 | assert result.exit_code == ExitCode.USER_ERROR |
| 208 | |
| 209 | |
| 210 | # --------------------------------------------------------------------------- |
| 211 | # Integration: --no-verify bypass |
| 212 | # --------------------------------------------------------------------------- |
| 213 | |
| 214 | |
| 215 | class TestUpdateRefNoVerify: |
| 216 | def test_no_verify_writes_without_commit_in_store(self, tmp_path: pathlib.Path) -> None: |
| 217 | repo = _init_repo(tmp_path) |
| 218 | phantom = _sha("not-in-store") |
| 219 | result = runner.invoke( |
| 220 | cli, ["plumbing", "update-ref", "--no-verify", "main", phantom], env=_env(repo) |
| 221 | ) |
| 222 | assert result.exit_code == 0 |
| 223 | ref_file = repo / ".muse" / "refs" / "heads" / "main" |
| 224 | assert ref_file.read_text(encoding="utf-8") == phantom |