gabriel / muse public
test_checkout_integrity.py python
465 lines 22.5 KB
Raw
sha256:f81d1ddcc1eec1eaccc667f22067476a0cf138860691611738c0c799df24a1db fixing more failing tests Human 42 days ago
1 """Tests for checkout data-integrity guarantees.
2
3 Coverage tiers
4 --------------
5 Unit — read_checkout_head / checkout_head_path helpers.
6 Integration — pre-flight aborts cleanly (zero mutations), CHECKOUT_HEAD
7 marker lifecycle, status detection of interrupted checkout.
8 End-to-end — full CLI round-trips: missing object, simulated mid-flight
9 kill (marker left behind), recovery via retry checkout.
10 Stress — rapid branch switching never leaves a stale marker.
11
12 Key invariants under test
13 -------------------------
14 1. Pre-flight: if any restore object is absent from the store, checkout
15 prints an error and does NOT modify any file on disk.
16 2. CHECKOUT_HEAD written before first mutation; removed on success.
17 3. ``muse status`` (text + JSON) detects a stale marker and warns loudly.
18 4. A successful checkout on retry clears the stale marker.
19 5. Interrupted checkout leaves HEAD pointing to the *old* branch.
20 """
21
22 from __future__ import annotations
23
24 import json
25 import os
26 import pathlib
27 import shutil
28
29 import pytest
30
31 from tests.cli_test_helper import CliRunner, InvokeResult
32 from muse.core._types import MsgpackDict
33 from muse.core.store import get_head_commit_id, read_current_branch
34 from muse.core.object_store import has_object
35 from muse.cli.commands.checkout import checkout_head_path, read_checkout_head
36
37 runner = CliRunner()
38
39 # ──────────────────────────────────────────────────────────────────────────────
40 # Helpers
41 # ──────────────────────────────────────────────────────────────────────────────
42
43
44 def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult:
45 saved = os.getcwd()
46 try:
47 os.chdir(repo)
48 return runner.invoke(None, args)
49 finally:
50 os.chdir(saved)
51
52
53 def _commit(repo: pathlib.Path, msg: str = "commit") -> InvokeResult:
54 return _invoke(repo, ["commit", "-m", msg])
55
56
57 def _checkout(repo: pathlib.Path, branch: str, *flags: str) -> InvokeResult:
58 return _invoke(repo, ["checkout", *flags, branch])
59
60
61 def _status_json(repo: pathlib.Path) -> MsgpackDict:
62 r = _invoke(repo, ["status", "--json"])
63 return json.loads(r.output)
64
65
66 def _status_text(repo: pathlib.Path) -> tuple[str, str]:
67 """Return (stdout, stderr) of muse status."""
68 r = _invoke(repo, ["status"])
69 return r.output, (r.stderr or "")
70
71
72 # ──────────────────────────────────────────────────────────────────────────────
73 # Fixtures
74 # ──────────────────────────────────────────────────────────────────────────────
75
76
77 @pytest.fixture()
78 def two_branch_repo(tmp_path: pathlib.Path) -> pathlib.Path:
79 """Repo with main and feat branches, each with distinct files.
80
81 main: a.py
82 feat: a.py (unchanged) + b.py (feat-only)
83 """
84 saved = os.getcwd()
85 try:
86 os.chdir(tmp_path)
87 runner.invoke(None, ["init"])
88 finally:
89 os.chdir(saved)
90 (tmp_path / "a.py").write_text("x = 1\n")
91 _commit(tmp_path, "main initial")
92 _invoke(tmp_path, ["checkout", "-b", "feat"])
93 (tmp_path / "b.py").write_text("y = 2\n")
94 _commit(tmp_path, "feat adds b.py")
95 _checkout(tmp_path, "main")
96 return tmp_path
97
98
99 # ──────────────────────────────────────────────────────────────────────────────
100 # Unit: helper functions
101 # ──────────────────────────────────────────────────────────────────────────────
102
103
104 class TestCheckoutHeadHelpers:
105 def test_checkout_head_path_points_into_muse_dir(self, two_branch_repo: pathlib.Path) -> None:
106 p = checkout_head_path(two_branch_repo)
107 assert p.parent == two_branch_repo / ".muse"
108 assert p.name == "CHECKOUT_HEAD"
109
110 def test_read_checkout_head_returns_none_when_absent(self, two_branch_repo: pathlib.Path) -> None:
111 assert read_checkout_head(two_branch_repo) is None
112
113 def test_read_checkout_head_returns_content_when_present(self, two_branch_repo: pathlib.Path) -> None:
114 marker = checkout_head_path(two_branch_repo)
115 marker.write_text("feat\n", encoding="utf-8")
116 assert read_checkout_head(two_branch_repo) == "feat"
117 marker.unlink()
118
119 def test_read_checkout_head_strips_trailing_newline(self, two_branch_repo: pathlib.Path) -> None:
120 marker = checkout_head_path(two_branch_repo)
121 marker.write_text("feat\n\n", encoding="utf-8")
122 assert read_checkout_head(two_branch_repo) == "feat"
123 marker.unlink()
124
125 def test_read_checkout_head_empty_file_returns_none(self, two_branch_repo: pathlib.Path) -> None:
126 marker = checkout_head_path(two_branch_repo)
127 marker.write_text("", encoding="utf-8")
128 assert read_checkout_head(two_branch_repo) is None
129 marker.unlink()
130
131
132 # ──────────────────────────────────────────────────────────────────────────────
133 # Integration: pre-flight object existence check
134 # ──────────────────────────────────────────────────────────────────────────────
135
136
137 class TestPreflightObjectCheck:
138 """Pre-flight: abort before any mutation if a restore object is missing."""
139
140 def _sabotage_object(self, repo: pathlib.Path, rel_path: str) -> pathlib.Path:
141 """Remove the object backing *rel_path* from the store; return its path."""
142 from muse.core.store import get_head_commit_id, read_commit, read_snapshot
143 from muse.core.object_store import _object_path_with_fallback
144
145 # Switch to feat first so its snapshot is current, then read its manifest
146 _checkout(repo, "feat")
147 branch = read_current_branch(repo)
148 commit_id = get_head_commit_id(repo, branch)
149 assert commit_id
150 commit = read_commit(repo, commit_id)
151 assert commit
152 snap = read_snapshot(repo, commit.snapshot_id)
153 assert snap
154 obj_id = snap.manifest[rel_path]
155 obj_path = _object_path_with_fallback(repo, obj_id)
156 assert obj_path.exists(), f"Object for {rel_path} not in store"
157 # Back on main before sabotaging
158 _checkout(repo, "main")
159 # Now remove the object
160 obj_path.unlink()
161 return obj_path
162
163 def test_missing_object_exits_nonzero(self, two_branch_repo: pathlib.Path) -> None:
164 self._sabotage_object(two_branch_repo, "b.py")
165 result = _checkout(two_branch_repo, "feat")
166 assert result.exit_code != 0
167
168 def test_missing_object_error_on_stderr(self, two_branch_repo: pathlib.Path) -> None:
169 self._sabotage_object(two_branch_repo, "b.py")
170 result = _checkout(two_branch_repo, "feat")
171 err = result.stderr or result.output
172 assert "missing" in err.lower() or "object" in err.lower()
173
174 def test_missing_object_working_tree_unchanged(self, two_branch_repo: pathlib.Path) -> None:
175 """Zero mutations — b.py must not appear on main after a failed checkout."""
176 self._sabotage_object(two_branch_repo, "b.py")
177 _checkout(two_branch_repo, "feat")
178 # Working tree must not contain b.py — no partial mutations
179 assert not (two_branch_repo / "b.py").exists()
180
181 def test_missing_object_a_py_unchanged(self, two_branch_repo: pathlib.Path) -> None:
182 """Files that would be kept unchanged must also remain untouched."""
183 self._sabotage_object(two_branch_repo, "b.py")
184 _checkout(two_branch_repo, "feat")
185 # a.py was already on main and unchanged; must still be present
186 assert (two_branch_repo / "a.py").exists()
187
188 def test_missing_object_head_stays_on_old_branch(self, two_branch_repo: pathlib.Path) -> None:
189 """HEAD must not advance when pre-flight aborts."""
190 self._sabotage_object(two_branch_repo, "b.py")
191 _checkout(two_branch_repo, "feat")
192 assert read_current_branch(two_branch_repo) == "main"
193
194 def test_missing_object_no_checkout_head_marker(self, two_branch_repo: pathlib.Path) -> None:
195 """Pre-flight abort fires before any marker is written."""
196 self._sabotage_object(two_branch_repo, "b.py")
197 _checkout(two_branch_repo, "feat")
198 # Marker must NOT be present — pre-flight aborted before any mutation
199 assert read_checkout_head(two_branch_repo) is None
200
201 def test_stderr_names_the_missing_file(self, two_branch_repo: pathlib.Path) -> None:
202 self._sabotage_object(two_branch_repo, "b.py")
203 result = _checkout(two_branch_repo, "feat")
204 err = result.stderr or result.output
205 assert "b.py" in err
206
207 def test_stderr_says_working_tree_not_modified(self, two_branch_repo: pathlib.Path) -> None:
208 self._sabotage_object(two_branch_repo, "b.py")
209 result = _checkout(two_branch_repo, "feat")
210 err = result.stderr or result.output
211 assert "NOT modified" in err or "not modified" in err.lower()
212
213
214 # ──────────────────────────────────────────────────────────────────────────────
215 # Integration: CHECKOUT_HEAD marker lifecycle
216 # ──────────────────────────────────────────────────────────────────────────────
217
218
219 class TestCheckoutHeadMarker:
220 """CHECKOUT_HEAD written before mutations, cleared after success."""
221
222 def test_marker_absent_after_clean_checkout(self, two_branch_repo: pathlib.Path) -> None:
223 _checkout(two_branch_repo, "feat")
224 assert read_checkout_head(two_branch_repo) is None
225
226 def test_marker_absent_after_round_trip(self, two_branch_repo: pathlib.Path) -> None:
227 _checkout(two_branch_repo, "feat")
228 _checkout(two_branch_repo, "main")
229 assert read_checkout_head(two_branch_repo) is None
230
231 def test_stale_marker_survives_failed_checkout(self, two_branch_repo: pathlib.Path) -> None:
232 """Manually plant a stale marker; it must persist (not auto-cleared)."""
233 marker = checkout_head_path(two_branch_repo)
234 marker.write_text("feat\n", encoding="utf-8")
235 # Successful checkout to main should clear it
236 _checkout(two_branch_repo, "main")
237 # main is already current, but checkout still fires the snapshot path
238 # which should clear the marker on success
239 # (Already on main — 'already_on' path does NOT call _checkout_snapshot,
240 # so the marker is unaffected. Plant while NOT on main.)
241 # Reset: switch to feat first, plant marker, switch back
242 _checkout(two_branch_repo, "feat")
243 marker.write_text("main\n", encoding="utf-8")
244 _checkout(two_branch_repo, "main")
245 assert read_checkout_head(two_branch_repo) is None
246
247 def test_simulated_interrupted_marker_persists(self, two_branch_repo: pathlib.Path) -> None:
248 """Simulate kill mid-checkout: plant marker, verify it stays."""
249 marker = checkout_head_path(two_branch_repo)
250 marker.write_text("feat\n", encoding="utf-8")
251 # Do not run checkout — marker lingers
252 assert read_checkout_head(two_branch_repo) == "feat"
253 marker.unlink()
254
255 def test_retry_checkout_clears_stale_marker(self, two_branch_repo: pathlib.Path) -> None:
256 """Retry of the interrupted checkout must clear the marker."""
257 # Simulate interrupted checkout of feat (marker left behind,
258 # b.py not restored)
259 marker = checkout_head_path(two_branch_repo)
260 marker.write_text("feat\n", encoding="utf-8")
261 # Retry — should succeed and clear marker
262 result = _checkout(two_branch_repo, "feat")
263 assert result.exit_code == 0
264 assert read_checkout_head(two_branch_repo) is None
265
266 def test_marker_records_target_branch_name(self, two_branch_repo: pathlib.Path) -> None:
267 """After a successful checkout the marker is gone; its content was the target."""
268 # We can't easily inspect the marker mid-flight without mocking,
269 # but we can write it ourselves and verify read_checkout_head returns it.
270 marker = checkout_head_path(two_branch_repo)
271 marker.write_text("feat\n")
272 assert read_checkout_head(two_branch_repo) == "feat"
273 marker.unlink()
274
275
276 # ──────────────────────────────────────────────────────────────────────────────
277 # Integration: muse status detects interrupted checkout
278 # ──────────────────────────────────────────────────────────────────────────────
279
280
281 class TestStatusDetectsInterruptedCheckout:
282 """muse status warns loudly when CHECKOUT_HEAD exists."""
283
284 def test_json_checkout_interrupted_false_normally(self, two_branch_repo: pathlib.Path) -> None:
285 data = _status_json(two_branch_repo)
286 assert data["checkout_interrupted"] is False
287
288 def test_json_checkout_target_null_normally(self, two_branch_repo: pathlib.Path) -> None:
289 data = _status_json(two_branch_repo)
290 assert data["checkout_target"] is None
291
292 def test_json_checkout_interrupted_true_with_marker(self, two_branch_repo: pathlib.Path) -> None:
293 marker = checkout_head_path(two_branch_repo)
294 marker.write_text("feat\n")
295 try:
296 data = _status_json(two_branch_repo)
297 assert data["checkout_interrupted"] is True
298 finally:
299 marker.unlink(missing_ok=True)
300
301 def test_json_checkout_target_set_with_marker(self, two_branch_repo: pathlib.Path) -> None:
302 marker = checkout_head_path(two_branch_repo)
303 marker.write_text("feat\n")
304 try:
305 data = _status_json(two_branch_repo)
306 assert data["checkout_target"] == "feat"
307 finally:
308 marker.unlink(missing_ok=True)
309
310 def test_json_keys_always_present(self, two_branch_repo: pathlib.Path) -> None:
311 data = _status_json(two_branch_repo)
312 assert "checkout_interrupted" in data
313 assert "checkout_target" in data
314
315 def test_text_status_warns_on_interrupted_checkout(self, two_branch_repo: pathlib.Path) -> None:
316 marker = checkout_head_path(two_branch_repo)
317 marker.write_text("feat\n")
318 try:
319 _, stderr = _status_text(two_branch_repo)
320 assert "CHECKOUT INTERRUPTED" in stderr or "CHECKOUT INTERRUPTED" in stderr.upper()
321 finally:
322 marker.unlink(missing_ok=True)
323
324 def test_text_status_names_target_branch(self, two_branch_repo: pathlib.Path) -> None:
325 marker = checkout_head_path(two_branch_repo)
326 marker.write_text("feat\n")
327 try:
328 _, stderr = _status_text(two_branch_repo)
329 assert "feat" in stderr
330 finally:
331 marker.unlink(missing_ok=True)
332
333 def test_text_status_mentions_retry_command(self, two_branch_repo: pathlib.Path) -> None:
334 marker = checkout_head_path(two_branch_repo)
335 marker.write_text("feat\n")
336 try:
337 _, stderr = _status_text(two_branch_repo)
338 assert "muse checkout" in stderr
339 finally:
340 marker.unlink(missing_ok=True)
341
342 def test_text_status_clean_no_warning(self, two_branch_repo: pathlib.Path) -> None:
343 _, stderr = _status_text(two_branch_repo)
344 assert "CHECKOUT INTERRUPTED" not in stderr
345
346
347 # ──────────────────────────────────────────────────────────────────────────────
348 # End-to-end: full recovery flow
349 # ──────────────────────────────────────────────────────────────────────────────
350
351
352 class TestCheckoutIntegrityE2E:
353 """Full end-to-end scenarios for the checkout integrity system."""
354
355 def test_successful_checkout_leaves_clean_status(self, two_branch_repo: pathlib.Path) -> None:
356 _checkout(two_branch_repo, "feat")
357 data = _status_json(two_branch_repo)
358 assert data["checkout_interrupted"] is False
359 assert data["checkout_target"] is None
360 assert data["clean"] is True
361
362 def test_interrupted_checkout_then_status_then_retry(self, two_branch_repo: pathlib.Path) -> None:
363 """Full recovery flow: interrupt → status warns → retry → clean."""
364 # Simulate interruption: plant marker and delete b.py as if
365 # the checkout partially ran (deleted files but didn't restore yet)
366 marker = checkout_head_path(two_branch_repo)
367 marker.write_text("feat\n")
368
369 # status should warn
370 data = _status_json(two_branch_repo)
371 assert data["checkout_interrupted"] is True
372
373 # retry the checkout — should succeed and clean up
374 result = _checkout(two_branch_repo, "feat")
375 assert result.exit_code == 0
376
377 # marker gone, status clean
378 assert read_checkout_head(two_branch_repo) is None
379 data2 = _status_json(two_branch_repo)
380 assert data2["checkout_interrupted"] is False
381 assert data2["checkout_target"] is None
382
383 def test_interrupted_checkout_head_unchanged(self, two_branch_repo: pathlib.Path) -> None:
384 """HEAD must still point to the old branch after a simulated interruption."""
385 # On main, plant a marker pretending we were switching to feat
386 marker = checkout_head_path(two_branch_repo)
387 marker.write_text("feat\n")
388 # HEAD has not changed — we only wrote the marker, didn't run checkout
389 assert read_current_branch(two_branch_repo) == "main"
390 marker.unlink()
391
392 def test_b_py_present_after_recovery_checkout(self, two_branch_repo: pathlib.Path) -> None:
393 """After successful retry checkout to feat, feat-only file must exist."""
394 # First do a clean checkout to feat to confirm it works
395 result = _checkout(two_branch_repo, "feat")
396 assert result.exit_code == 0
397 assert (two_branch_repo / "b.py").exists()
398
399 def test_b_py_absent_after_checkout_back_to_main(self, two_branch_repo: pathlib.Path) -> None:
400 """After checking back out to main, feat-only file must be gone."""
401 _checkout(two_branch_repo, "feat")
402 result = _checkout(two_branch_repo, "main")
403 assert result.exit_code == 0
404 assert not (two_branch_repo / "b.py").exists()
405
406 def test_preflight_then_retry_with_restored_object(
407 self, two_branch_repo: pathlib.Path
408 ) -> None:
409 """Remove an object, verify clean abort, restore object, verify checkout succeeds."""
410 from muse.core.store import get_head_commit_id, read_commit, read_snapshot
411 from muse.core.object_store import _object_path_with_fallback
412
413 # Read the object ID for b.py from the feat snapshot
414 _checkout(two_branch_repo, "feat")
415 commit_id = get_head_commit_id(two_branch_repo, "feat")
416 commit = read_commit(two_branch_repo, commit_id)
417 snap = read_snapshot(two_branch_repo, commit.snapshot_id)
418 obj_id = snap.manifest["b.py"]
419 obj_path = _object_path_with_fallback(two_branch_repo, obj_id)
420 # Save contents before sabotage
421 saved = obj_path.read_bytes()
422
423 _checkout(two_branch_repo, "main")
424 obj_path.unlink()
425
426 # Pre-flight fails cleanly
427 result = _checkout(two_branch_repo, "feat")
428 assert result.exit_code != 0
429 assert not (two_branch_repo / "b.py").exists()
430 assert read_current_branch(two_branch_repo) == "main"
431
432 # Restore the object (simulating a fetch)
433 obj_path.write_bytes(saved)
434
435 # Retry succeeds
436 result2 = _checkout(two_branch_repo, "feat")
437 assert result2.exit_code == 0
438 assert (two_branch_repo / "b.py").exists()
439 assert read_current_branch(two_branch_repo) == "feat"
440
441
442 # ──────────────────────────────────────────────────────────────────────────────
443 # Stress: rapid switching never leaves a stale marker
444 # ──────────────────────────────────────────────────────────────────────────────
445
446
447 class TestCheckoutIntegrityStress:
448 def test_rapid_switching_no_stale_marker(self, two_branch_repo: pathlib.Path) -> None:
449 """Switching branches 50 times must never leave CHECKOUT_HEAD behind."""
450 for i in range(25):
451 r1 = _checkout(two_branch_repo, "feat")
452 assert r1.exit_code == 0, f"iter {i}: checkout feat failed"
453 assert read_checkout_head(two_branch_repo) is None, f"iter {i}: marker after feat checkout"
454 r2 = _checkout(two_branch_repo, "main")
455 assert r2.exit_code == 0, f"iter {i}: checkout main failed"
456 assert read_checkout_head(two_branch_repo) is None, f"iter {i}: marker after main checkout"
457
458 def test_status_json_schema_stable_across_branches(self, two_branch_repo: pathlib.Path) -> None:
459 """checkout_interrupted and checkout_target always present in JSON output."""
460 required = {"checkout_interrupted", "checkout_target"}
461 for branch in ("feat", "main", "feat", "main"):
462 _checkout(two_branch_repo, branch)
463 data = _status_json(two_branch_repo)
464 missing = required - data.keys()
465 assert not missing, f"Missing keys after checkout to {branch}: {missing}"
File History 1 commit
sha256:f81d1ddcc1eec1eaccc667f22067476a0cf138860691611738c0c799df24a1db fixing more failing tests Human 42 days ago