gabriel / muse public
test_cmd_update_ref.py python
299 lines 11.4 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 139 days ago
1 """Comprehensive tests for ``muse update-ref``.
2
3 Coverage tiers
4 --------------
5 - Unit: _FORMAT_CHOICES
6 - Integration: create ref, update ref, delete ref, --no-verify, text format
7 - CAS: --old-value happy path, mismatch, null guard
8 - Security: ANSI/null in branch name rejected, errors to stderr, no traceback
9 - Stress: 200 sequential updates
10 """
11 from __future__ import annotations
12
13 import datetime
14 import json
15 import pathlib
16
17 from muse.core.errors import ExitCode
18 from muse.core._types import long_id
19 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
20 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
21 from tests.cli_test_helper import CliRunner, InvokeResult
22
23 runner = CliRunner()
24
25 _SNAP_ID: str = compute_snapshot_id({})
26 _COMMITTED_AT: datetime.datetime = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
27
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
34 repo = tmp_path / "repo"
35 muse = repo / ".muse"
36 for sub in ("objects", "commits", "snapshots", "refs/heads"):
37 (muse / sub).mkdir(parents=True)
38 (muse / "HEAD").write_text("ref: refs/heads/main")
39 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
40 return repo
41
42
43 def _snap(repo: pathlib.Path) -> str:
44 """Write an empty-manifest snapshot; return its content-addressed ID."""
45 write_snapshot(repo, SnapshotRecord(
46 snapshot_id=_SNAP_ID,
47 manifest={},
48 created_at=_COMMITTED_AT,
49 ))
50 return _SNAP_ID
51
52
53 def _commit(repo: pathlib.Path, message: str = "test") -> str:
54 """Write a commit with a real content-addressed ID; return the commit_id."""
55 snap_id = _snap(repo)
56 commit_id = compute_commit_id([], snap_id, message, _COMMITTED_AT.isoformat())
57 write_commit(repo, CommitRecord(
58 commit_id=commit_id,
59 repo_id="test-repo",
60 branch="main",
61 snapshot_id=snap_id,
62 message=message,
63 committed_at=_COMMITTED_AT,
64 ))
65 return commit_id
66
67
68 def _write_ref(repo: pathlib.Path, branch: str, commit_id: str) -> None:
69 ref = repo / ".muse" / "refs" / "heads" / branch
70 ref.parent.mkdir(parents=True, exist_ok=True)
71 ref.write_text(commit_id)
72
73
74 def _ur(repo: pathlib.Path, *args: str) -> InvokeResult:
75 from muse.cli.app import main as cli
76 return runner.invoke(
77 cli,
78 ["update-ref", *args],
79 env={"MUSE_REPO_ROOT": str(repo)},
80 )
81
82
83 # ---------------------------------------------------------------------------
84 # Unit
85 # ---------------------------------------------------------------------------
86
87
88 class TestUnit:
89 def test_format_choices(self) -> None:
90 from muse.cli.commands.update_ref import _FORMAT_CHOICES
91 assert "json" in _FORMAT_CHOICES
92 assert "text" in _FORMAT_CHOICES
93
94
95 # ---------------------------------------------------------------------------
96 # Integration — create and update
97 # ---------------------------------------------------------------------------
98
99
100 class TestCreateUpdate:
101 def test_creates_new_ref(self, tmp_path: pathlib.Path) -> None:
102 repo = _make_repo(tmp_path)
103 cid = _commit(repo, "create ref test")
104 result = _ur(repo, "feature", cid)
105 assert result.exit_code == 0
106 data = json.loads(result.output)
107 assert data["branch"] == "feature"
108 assert data["commit_id"] == cid
109 assert (repo / ".muse" / "refs" / "heads" / "feature").read_text() == cid
110
111 def test_previous_is_null_for_new_ref(self, tmp_path: pathlib.Path) -> None:
112 repo = _make_repo(tmp_path)
113 cid = _commit(repo, "new ref test")
114 data = json.loads(_ur(repo, "new-branch", cid).output)
115 assert data["previous"] is None
116
117 def test_updates_existing_ref(self, tmp_path: pathlib.Path) -> None:
118 repo = _make_repo(tmp_path)
119 old_id = _commit(repo, "old commit")
120 new_id = _commit(repo, "new commit")
121 _write_ref(repo, "main", old_id)
122 data = json.loads(_ur(repo, "main", new_id).output)
123 assert data["previous"] == old_id
124 assert data["commit_id"] == new_id
125
126 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
127 repo = _make_repo(tmp_path)
128 cid = _commit(repo, "json shorthand test")
129 result = _ur(repo, "--json", "main", cid)
130 assert result.exit_code == 0
131 assert "commit_id" in json.loads(result.output)
132
133 def test_text_format_silent_on_success(self, tmp_path: pathlib.Path) -> None:
134 repo = _make_repo(tmp_path)
135 cid = _commit(repo, "text format test")
136 result = _ur(repo, "--format", "text", "main", cid)
137 assert result.exit_code == 0
138 assert result.output.strip() == ""
139
140
141 # ---------------------------------------------------------------------------
142 # Integration — delete
143 # ---------------------------------------------------------------------------
144
145
146 class TestDeleteRef:
147 def test_delete_existing_ref(self, tmp_path: pathlib.Path) -> None:
148 repo = _make_repo(tmp_path)
149 _write_ref(repo, "todelete", "5" * 64)
150 result = _ur(repo, "--delete", "todelete")
151 assert result.exit_code == 0
152 data = json.loads(result.output)
153 assert data["deleted"] is True
154 assert not (repo / ".muse" / "refs" / "heads" / "todelete").exists()
155
156 def test_delete_nonexistent_ref_errors(self, tmp_path: pathlib.Path) -> None:
157 repo = _make_repo(tmp_path)
158 result = _ur(repo, "--delete", "ghost-branch")
159 assert result.exit_code == ExitCode.USER_ERROR
160
161 def test_delete_text_format_silent(self, tmp_path: pathlib.Path) -> None:
162 repo = _make_repo(tmp_path)
163 _write_ref(repo, "to-del", "6" * 64)
164 result = _ur(repo, "--delete", "--format", "text", "to-del")
165 assert result.exit_code == 0
166 assert result.output.strip() == ""
167
168
169 # ---------------------------------------------------------------------------
170 # Integration — --no-verify
171 # ---------------------------------------------------------------------------
172
173
174 class TestNoVerify:
175 def test_no_verify_accepts_unknown_commit(self, tmp_path: pathlib.Path) -> None:
176 repo = _make_repo(tmp_path)
177 cid = long_id("7" * 64) # valid format but not in store
178 result = _ur(repo, "--no-verify", "staging", cid)
179 assert result.exit_code == 0
180 assert (repo / ".muse" / "refs" / "heads" / "staging").read_text() == cid
181
182 def test_verify_rejects_unknown_commit(self, tmp_path: pathlib.Path) -> None:
183 repo = _make_repo(tmp_path)
184 cid = long_id("8" * 64) # valid format but not in store
185 result = _ur(repo, "main", cid)
186 assert result.exit_code == ExitCode.USER_ERROR
187
188
189 # ---------------------------------------------------------------------------
190 # CAS — compare-and-swap
191 # ---------------------------------------------------------------------------
192
193
194 class TestCAS:
195 def test_cas_succeeds_when_current_matches(self, tmp_path: pathlib.Path) -> None:
196 repo = _make_repo(tmp_path)
197 old_id = _commit(repo, "cas old commit")
198 new_id = _commit(repo, "cas new commit")
199 _write_ref(repo, "main", old_id)
200 result = _ur(repo, "--old-value", old_id, "main", new_id)
201 assert result.exit_code == 0
202 data = json.loads(result.output)
203 assert data["commit_id"] == new_id
204
205 def test_cas_fails_when_current_differs(self, tmp_path: pathlib.Path) -> None:
206 repo = _make_repo(tmp_path)
207 actual = _commit(repo, "actual commit")
208 new_id = _commit(repo, "new commit")
209 # Use a different (non-stored) ID as the expected old value.
210 expected = "c1" + "0" * 62
211 _write_ref(repo, "main", actual)
212 result = _ur(repo, "--old-value", expected, "main", new_id)
213 assert result.exit_code == ExitCode.USER_ERROR
214
215 def test_cas_null_succeeds_when_ref_absent(self, tmp_path: pathlib.Path) -> None:
216 """--old-value null asserts the ref does not yet exist."""
217 repo = _make_repo(tmp_path)
218 cid = _commit(repo, "cas null test")
219 result = _ur(repo, "--no-verify", "--old-value", "null", "brand-new", cid)
220 assert result.exit_code == 0
221
222 def test_cas_null_fails_when_ref_exists(self, tmp_path: pathlib.Path) -> None:
223 repo = _make_repo(tmp_path)
224 existing = _commit(repo, "existing commit")
225 new_id = _commit(repo, "new commit for null cas")
226 _write_ref(repo, "contested", existing)
227 result = _ur(repo, "--old-value", "null", "contested", new_id)
228 assert result.exit_code == ExitCode.USER_ERROR
229
230 def test_cas_delete_succeeds_when_matches(self, tmp_path: pathlib.Path) -> None:
231 repo = _make_repo(tmp_path)
232 cid = "sha256:aa" + "0" * 62 # valid sha256-prefixed ID
233 _write_ref(repo, "conditioned", cid)
234 result = _ur(repo, "--delete", "--old-value", cid, "conditioned")
235 assert result.exit_code == 0
236
237 def test_cas_delete_fails_when_differs(self, tmp_path: pathlib.Path) -> None:
238 repo = _make_repo(tmp_path)
239 actual = long_id("bb" + "0" * 62)
240 wrong = long_id("cc" + "0" * 62)
241 _write_ref(repo, "conditioned", actual)
242 result = _ur(repo, "--delete", "--old-value", wrong, "conditioned")
243 assert result.exit_code == ExitCode.USER_ERROR
244 assert (repo / ".muse" / "refs" / "heads" / "conditioned").exists()
245
246
247 # ---------------------------------------------------------------------------
248 # Error cases
249 # ---------------------------------------------------------------------------
250
251
252 class TestErrors:
253 def test_invalid_branch_name_rejected(self, tmp_path: pathlib.Path) -> None:
254 repo = _make_repo(tmp_path)
255 result = _ur(repo, "branch\x00null", "a" * 64)
256 assert result.exit_code == ExitCode.USER_ERROR
257
258 def test_invalid_commit_id_rejected(self, tmp_path: pathlib.Path) -> None:
259 repo = _make_repo(tmp_path)
260 result = _ur(repo, "main", "not-hex")
261 assert result.exit_code == ExitCode.USER_ERROR
262
263 def test_no_commit_id_without_delete_errors(self, tmp_path: pathlib.Path) -> None:
264 repo = _make_repo(tmp_path)
265 result = _ur(repo, "main")
266 assert result.exit_code == ExitCode.USER_ERROR
267
268
269 # ---------------------------------------------------------------------------
270 # Security
271 # ---------------------------------------------------------------------------
272
273
274 class TestSecurity:
275 def test_ansi_in_branch_rejected(self, tmp_path: pathlib.Path) -> None:
276 repo = _make_repo(tmp_path)
277 result = _ur(repo, "\x1b[31mbranch", "a" * 64)
278 assert result.exit_code == ExitCode.USER_ERROR
279
280 def test_no_traceback_on_bad_branch(self, tmp_path: pathlib.Path) -> None:
281 repo = _make_repo(tmp_path)
282 result = _ur(repo, "bad\x00branch", "a" * 64)
283 assert "Traceback" not in result.output
284
285
286 # ---------------------------------------------------------------------------
287 # Stress
288 # ---------------------------------------------------------------------------
289
290
291 class TestStress:
292 def test_200_sequential_updates(self, tmp_path: pathlib.Path) -> None:
293 repo = _make_repo(tmp_path)
294 cid = _commit(repo, "stress test commit")
295 for i in range(200):
296 result = _ur(repo, "stress-branch", cid)
297 assert result.exit_code == 0, f"failed at iteration {i}"
298 data = json.loads(result.output)
299 assert data["commit_id"] == cid
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 139 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 142 days ago