gabriel / muse public
test_phase6_checkout_interruption.py python
188 lines 7.0 KB
Raw
sha256:51116ec824246acde6abf729e6ba854c223dc5173eff31a645520208023b0652 refactor(bridge): comprehensive spec sweep — close all issu… Sonnet 4.6 minor ⚠ breaking 28 days ago
1 """Phase 6 — Checkout interruption recovery.
2
3 Invariant: if the process is killed between the working-tree update and the
4 HEAD pointer write, muse status must detect the interrupted state and warn
5 the user. Without this, the user sees target-branch files but HEAD says
6 old branch — they might commit the wrong thing.
7
8 Fix: _checkout_snapshot now accepts clear_marker=False so callers can keep
9 CHECKOUT_HEAD alive through write_head_branch / write_head_commit.
10
11 Timeline for a clean checkout:
12 1. write CHECKOUT_HEAD (target branch name)
13 2. mutate working tree (delete removed, restore added/modified)
14 3. call domain plugin.apply()
15 4. write HEAD (write_head_branch or write_head_commit)
16 5. unlink CHECKOUT_HEAD
17
18 If killed between 2 and 4: CHECKOUT_HEAD exists, status shows interrupt.
19 If killed between 4 and 5: CHECKOUT_HEAD exists, status still shows interrupt.
20 (next muse checkout will clear it)
21
22 Testing tiers
23 -------------
24 Unit CHECKOUT_HEAD present after working-tree update, cleared after HEAD write
25 Integration muse status reports checkout_interrupted=True when marker exists
26 Data after a simulated crash (marker left), status correctly identifies target
27 E2E successful checkout clears CHECKOUT_HEAD after write_head_branch
28 """
29
30 from __future__ import annotations
31
32 import json
33 import pathlib
34 from collections.abc import Mapping
35 from unittest.mock import patch
36
37 import pytest
38
39 from tests.cli_test_helper import CliRunner
40 from muse.core.paths import checkout_head_path
41
42 runner = CliRunner()
43
44
45 def _run(repo: pathlib.Path, *args: str, expect_failure: bool = False) -> "CliRunner.Result":
46 r = runner.invoke(None, list(args), cwd=repo)
47 if not expect_failure:
48 assert r.exit_code == 0, f"muse {' '.join(args)} failed:\n{r.output}"
49 return r
50
51
52 def _checkout_head_path(repo: pathlib.Path) -> pathlib.Path:
53 return checkout_head_path(repo)
54
55
56 def _status(repo: pathlib.Path) -> Mapping[str, object]:
57 r = runner.invoke(None, ["status", "--json"], cwd=repo)
58 return json.loads(r.output)
59
60
61 def _setup_two_branches(repo: pathlib.Path) -> None:
62 """Create main with one file, branch-b with a different file."""
63 f = repo / "file.py"
64 f.write_text("main content\n")
65 _run(repo, "code", "add", "file.py")
66 _run(repo, "commit", "-m", "main commit")
67
68 _run(repo, "checkout", "-b", "branch-b")
69 f.write_text("branch-b content\n")
70 _run(repo, "code", "add", "file.py")
71 _run(repo, "commit", "-m", "branch-b commit")
72
73 _run(repo, "checkout", "main")
74
75
76 # ---------------------------------------------------------------------------
77 # Unit — CHECKOUT_HEAD lifecycle
78 # ---------------------------------------------------------------------------
79
80 class TestCheckoutHeadLifecycle:
81 def test_checkout_head_absent_after_successful_checkout(
82 self, muse_repo: pathlib.Path
83 ) -> None:
84 """CHECKOUT_HEAD must not exist after a successful branch switch."""
85 _setup_two_branches(muse_repo)
86
87 _run(muse_repo, "checkout", "branch-b")
88
89 assert not _checkout_head_path(muse_repo).exists(), (
90 "CHECKOUT_HEAD exists after a successful checkout — marker not cleared"
91 )
92
93 def test_checkout_head_present_during_head_write(
94 self, muse_repo: pathlib.Path
95 ) -> None:
96 """CHECKOUT_HEAD must still exist when write_head_branch is called.
97
98 This is the key invariant: if the process is killed just before
99 write_head_branch, the marker must be present so muse status detects it.
100 """
101 _setup_two_branches(muse_repo)
102
103 marker_state_during_write: list[bool] = []
104
105 real_write_head = __import__(
106 "muse.core.store", fromlist=["write_head_branch"]
107 ).write_head_branch
108
109 def _spy_write_head(root: pathlib.Path, branch: str) -> None:
110 marker_state_during_write.append(_checkout_head_path(root).exists())
111 return real_write_head(root, branch)
112
113 with patch("muse.cli.commands.checkout.write_head_branch", side_effect=_spy_write_head):
114 _run(muse_repo, "checkout", "branch-b")
115
116 assert marker_state_during_write, "write_head_branch was never called"
117 assert marker_state_during_write[0], (
118 "CHECKOUT_HEAD was already cleared before write_head_branch — "
119 "interrupt in that window is undetectable"
120 )
121
122
123 # ---------------------------------------------------------------------------
124 # Integration — muse status detects planted CHECKOUT_HEAD
125 # ---------------------------------------------------------------------------
126
127 class TestStatusDetectsInterrupt:
128 def test_status_shows_checkout_interrupted_when_marker_exists(
129 self, muse_repo: pathlib.Path
130 ) -> None:
131 """If CHECKOUT_HEAD exists, status must report checkout_interrupted=True."""
132 _setup_two_branches(muse_repo)
133
134 # Plant the marker (simulates crash during checkout to branch-b)
135 _checkout_head_path(muse_repo).write_text("branch-b\n")
136
137 status = _status(muse_repo)
138
139 assert status["checkout_interrupted"] is True, (
140 "status did not detect CHECKOUT_HEAD — checkout_interrupted should be True"
141 )
142
143 def test_status_shows_checkout_target(self, muse_repo: pathlib.Path) -> None:
144 """checkout_target must match the branch name written in CHECKOUT_HEAD."""
145 _setup_two_branches(muse_repo)
146
147 _checkout_head_path(muse_repo).write_text("branch-b\n")
148
149 status = _status(muse_repo)
150
151 assert status.get("checkout_target") == "branch-b", (
152 f"checkout_target should be 'branch-b', got {status.get('checkout_target')!r}"
153 )
154
155 def test_status_clean_after_marker_removed(self, muse_repo: pathlib.Path) -> None:
156 """After removing a planted CHECKOUT_HEAD, status shows no interrupt."""
157 _setup_two_branches(muse_repo)
158
159 _checkout_head_path(muse_repo).write_text("branch-b\n")
160 assert _status(muse_repo)["checkout_interrupted"] is True
161
162 _checkout_head_path(muse_repo).unlink()
163 assert _status(muse_repo)["checkout_interrupted"] is False
164
165
166 # ---------------------------------------------------------------------------
167 # Data — crash recovery: re-checkout fixes the state
168 # ---------------------------------------------------------------------------
169
170 class TestCrashRecovery:
171 def test_re_checkout_after_interrupted_state_succeeds(
172 self, muse_repo: pathlib.Path
173 ) -> None:
174 """Re-running muse checkout on a repo with CHECKOUT_HEAD clears it."""
175 _setup_two_branches(muse_repo)
176
177 # Simulate partial checkout: working tree at branch-b, HEAD at main,
178 # CHECKOUT_HEAD present
179 (muse_repo / "file.py").write_text("branch-b content\n")
180 _checkout_head_path(muse_repo).write_text("branch-b\n")
181
182 # Re-checkout to branch-b — needs --force because tree is dirty
183 _run(muse_repo, "checkout", "--force", "branch-b")
184
185 assert not _checkout_head_path(muse_repo).exists(), (
186 "CHECKOUT_HEAD not cleared after re-checkout"
187 )
188 assert _status(muse_repo)["checkout_interrupted"] is False
File History 1 commit
sha256:51116ec824246acde6abf729e6ba854c223dc5173eff31a645520208023b0652 refactor(bridge): comprehensive spec sweep — close all issu… Sonnet 4.6 minor ⚠ 28 days ago