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