cgcardona / muse public
test_checkout.py python
412 lines 13.6 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse checkout``.
2
3 All async tests call ``_checkout_async`` directly with an in-memory SQLite
4 session and a ``tmp_path`` repo root — no real Postgres or running process
5 required. Commits are seeded via ``_commit_async`` so the two commands are
6 tested as an integrated pair.
7 """
8 from __future__ import annotations
9
10 import json
11 import pathlib
12 import uuid
13
14 import pytest
15 import typer
16 from sqlalchemy.ext.asyncio import AsyncSession
17
18 from maestro.muse_cli.commands.checkout import _checkout_async
19 from maestro.muse_cli.commands.commit import _commit_async
20 from maestro.muse_cli.commands.log import _log_async
21 from maestro.muse_cli.errors import ExitCode
22
23
24 # ---------------------------------------------------------------------------
25 # Helpers (same pattern as test_log.py / test_commit.py)
26 # ---------------------------------------------------------------------------
27
28
29 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
30 rid = repo_id or str(uuid.uuid4())
31 muse = root / ".muse"
32 (muse / "refs" / "heads").mkdir(parents=True)
33 (muse / "repo.json").write_text(
34 json.dumps({"repo_id": rid, "schema_version": "1"})
35 )
36 (muse / "HEAD").write_text("refs/heads/main")
37 (muse / "refs" / "heads" / "main").write_text("")
38 return rid
39
40
41 def _write_workdir(root: pathlib.Path, files: dict[str, bytes]) -> None:
42 workdir = root / "muse-work"
43 workdir.mkdir(exist_ok=True)
44 for name, content in files.items():
45 (workdir / name).write_bytes(content)
46
47
48 async def _make_commit(
49 root: pathlib.Path,
50 session: AsyncSession,
51 message: str,
52 filename: str = "track.mid",
53 content: bytes = b"MIDI-0",
54 ) -> str:
55 _write_workdir(root, {filename: content})
56 return await _commit_async(message=message, root=root, session=session)
57
58
59 async def _do_checkout(
60 root: pathlib.Path,
61 session: AsyncSession,
62 branch_name: str,
63 *,
64 create: bool = False,
65 force: bool = False,
66 ) -> None:
67 await _checkout_async(
68 branch_name=branch_name,
69 create=create,
70 force=force,
71 root=root,
72 session=session,
73 )
74
75
76 # ---------------------------------------------------------------------------
77 # test_checkout_switches_head
78 # ---------------------------------------------------------------------------
79
80
81 @pytest.mark.anyio
82 async def test_checkout_switches_head(
83 tmp_path: pathlib.Path,
84 muse_cli_db_session: AsyncSession,
85 ) -> None:
86 """``.muse/HEAD`` is updated when switching to an existing branch."""
87 _init_muse_repo(tmp_path)
88
89 # Create a second branch manually (as if already committed there)
90 (tmp_path / ".muse" / "refs" / "heads" / "experiment").write_text("")
91
92 await _do_checkout(tmp_path, muse_cli_db_session, "experiment")
93
94 head = (tmp_path / ".muse" / "HEAD").read_text().strip()
95 assert head == "refs/heads/experiment"
96
97
98 # ---------------------------------------------------------------------------
99 # test_checkout_b_creates_branch
100 # ---------------------------------------------------------------------------
101
102
103 @pytest.mark.anyio
104 async def test_checkout_b_creates_branch(
105 tmp_path: pathlib.Path,
106 muse_cli_db_session: AsyncSession,
107 ) -> None:
108 """-b creates ``.muse/refs/heads/<branch>`` and switches HEAD."""
109 _init_muse_repo(tmp_path)
110 await _make_commit(tmp_path, muse_cli_db_session, "initial commit")
111
112 await _do_checkout(tmp_path, muse_cli_db_session, "feature", create=True)
113
114 ref_file = tmp_path / ".muse" / "refs" / "heads" / "feature"
115 assert ref_file.exists(), "ref file for new branch should exist"
116 head = (tmp_path / ".muse" / "HEAD").read_text().strip()
117 assert head == "refs/heads/feature"
118
119
120 # ---------------------------------------------------------------------------
121 # test_checkout_b_forks_at_current_head_commit
122 # ---------------------------------------------------------------------------
123
124
125 @pytest.mark.anyio
126 async def test_checkout_b_forks_at_current_head_commit(
127 tmp_path: pathlib.Path,
128 muse_cli_db_session: AsyncSession,
129 ) -> None:
130 """New branch ref points to the same commit as main's HEAD."""
131 _init_muse_repo(tmp_path)
132 cid = await _make_commit(tmp_path, muse_cli_db_session, "initial")
133
134 # main ref should now have the commit id
135 main_commit = (tmp_path / ".muse" / "refs" / "heads" / "main").read_text().strip()
136 assert main_commit == cid
137
138 await _do_checkout(tmp_path, muse_cli_db_session, "neo-soul", create=True)
139
140 new_branch_commit = (
141 tmp_path / ".muse" / "refs" / "heads" / "neo-soul"
142 ).read_text().strip()
143 assert new_branch_commit == cid, "new branch should fork from current HEAD commit"
144
145
146 # ---------------------------------------------------------------------------
147 # test_checkout_nonexistent_without_b_exits_1
148 # ---------------------------------------------------------------------------
149
150
151 @pytest.mark.anyio
152 async def test_checkout_nonexistent_without_b_exits_1(
153 tmp_path: pathlib.Path,
154 muse_cli_db_session: AsyncSession,
155 capsys: pytest.CaptureFixture[str],
156 ) -> None:
157 """Checking out a non-existent branch without -b exits 1 with a hint."""
158 _init_muse_repo(tmp_path)
159
160 with pytest.raises(typer.Exit) as exc_info:
161 await _do_checkout(tmp_path, muse_cli_db_session, "ghost-branch")
162
163 assert exc_info.value.exit_code == ExitCode.USER_ERROR
164 out = capsys.readouterr().out
165 assert "does not exist" in out
166 assert "-b" in out
167
168
169 # ---------------------------------------------------------------------------
170 # test_checkout_dirty_workdir_blocked
171 # ---------------------------------------------------------------------------
172
173
174 @pytest.mark.anyio
175 async def test_checkout_dirty_workdir_blocked(
176 tmp_path: pathlib.Path,
177 muse_cli_db_session: AsyncSession,
178 capsys: pytest.CaptureFixture[str],
179 ) -> None:
180 """Checkout is blocked when muse-work/ has uncommitted changes."""
181 _init_muse_repo(tmp_path)
182 await _make_commit(tmp_path, muse_cli_db_session, "initial commit")
183
184 # Create target branch
185 (tmp_path / ".muse" / "refs" / "heads" / "other").write_text("")
186
187 # Dirty the working tree
188 _write_workdir(tmp_path, {"new_track.mid": b"MIDI-DIRTY"})
189
190 with pytest.raises(typer.Exit) as exc_info:
191 await _do_checkout(tmp_path, muse_cli_db_session, "other")
192
193 assert exc_info.value.exit_code == ExitCode.USER_ERROR
194 out = capsys.readouterr().out
195 assert "Uncommitted changes" in out
196
197
198 # ---------------------------------------------------------------------------
199 # test_checkout_force_ignores_dirty
200 # ---------------------------------------------------------------------------
201
202
203 @pytest.mark.anyio
204 async def test_checkout_force_ignores_dirty(
205 tmp_path: pathlib.Path,
206 muse_cli_db_session: AsyncSession,
207 capsys: pytest.CaptureFixture[str],
208 ) -> None:
209 """--force succeeds even when muse-work/ has uncommitted changes."""
210 _init_muse_repo(tmp_path)
211 await _make_commit(tmp_path, muse_cli_db_session, "initial commit")
212
213 # Create target branch
214 (tmp_path / ".muse" / "refs" / "heads" / "other").write_text("")
215
216 # Dirty the working tree
217 _write_workdir(tmp_path, {"new_track.mid": b"MIDI-DIRTY"})
218
219 # Force checkout should succeed (no exception)
220 capsys.readouterr()
221 await _do_checkout(tmp_path, muse_cli_db_session, "other", force=True)
222
223 head = (tmp_path / ".muse" / "HEAD").read_text().strip()
224 assert head == "refs/heads/other"
225 out = capsys.readouterr().out
226 assert "Switched to branch" in out
227
228
229 # ---------------------------------------------------------------------------
230 # test_checkout_branches_diverge
231 # ---------------------------------------------------------------------------
232
233
234 @pytest.mark.anyio
235 async def test_checkout_branches_diverge(
236 tmp_path: pathlib.Path,
237 muse_cli_db_session: AsyncSession,
238 capsys: pytest.CaptureFixture[str],
239 ) -> None:
240 """After branching, ``muse log`` shows different histories per branch."""
241 _init_muse_repo(tmp_path)
242
243 # Commit on main
244 await _make_commit(tmp_path, muse_cli_db_session, "main: initial beat", filename="beat.mid", content=b"MIDI-MAIN-1")
245
246 # Create and switch to experiment branch
247 await _do_checkout(tmp_path, muse_cli_db_session, "experiment", create=True)
248
249 assert (tmp_path / ".muse" / "HEAD").read_text().strip() == "refs/heads/experiment"
250
251 # Commit on experiment branch (unique file)
252 _write_workdir(tmp_path, {"experiment.mid": b"MIDI-EXP-1"})
253 cid_exp = await _commit_async(
254 message="experiment: neo-soul take 1",
255 root=tmp_path,
256 session=muse_cli_db_session,
257 )
258
259 # Check experiment log
260 capsys.readouterr()
261 await _log_async(root=tmp_path, session=muse_cli_db_session, limit=100, graph=False)
262 exp_log = capsys.readouterr().out
263
264 # Switch back to main
265 await _do_checkout(tmp_path, muse_cli_db_session, "main")
266
267 assert (tmp_path / ".muse" / "HEAD").read_text().strip() == "refs/heads/main"
268
269 # Commit something else on main
270 _write_workdir(tmp_path, {"mainv2.mid": b"MIDI-MAIN-2"})
271 cid_main2 = await _commit_async(
272 message="main: second arrangement",
273 root=tmp_path,
274 session=muse_cli_db_session,
275 )
276
277 # Check main log
278 capsys.readouterr()
279 await _log_async(root=tmp_path, session=muse_cli_db_session, limit=100, graph=False)
280 main_log = capsys.readouterr().out
281
282 # Main log should have "second arrangement" but NOT the experiment commit
283 assert "second arrangement" in main_log
284 assert "neo-soul" not in main_log
285
286 # Experiment log should have "neo-soul" but NOT "second arrangement"
287 assert "neo-soul" in exp_log
288 assert "second arrangement" not in exp_log
289
290 # The experiment commit id should appear in experiment log, main's second in main log
291 assert cid_exp[:8] in exp_log
292 assert cid_main2[:8] in main_log
293
294
295 # ---------------------------------------------------------------------------
296 # test_checkout_already_on_branch
297 # ---------------------------------------------------------------------------
298
299
300 @pytest.mark.anyio
301 async def test_checkout_already_on_branch(
302 tmp_path: pathlib.Path,
303 muse_cli_db_session: AsyncSession,
304 capsys: pytest.CaptureFixture[str],
305 ) -> None:
306 """Checking out the current branch exits 0 with an 'Already on' message."""
307 _init_muse_repo(tmp_path)
308
309 with pytest.raises(typer.Exit) as exc_info:
310 await _do_checkout(tmp_path, muse_cli_db_session, "main")
311
312 assert exc_info.value.exit_code == ExitCode.SUCCESS
313 out = capsys.readouterr().out
314 assert "Already on" in out
315
316
317 # ---------------------------------------------------------------------------
318 # test_checkout_b_fails_if_branch_exists
319 # ---------------------------------------------------------------------------
320
321
322 @pytest.mark.anyio
323 async def test_checkout_b_fails_if_branch_exists(
324 tmp_path: pathlib.Path,
325 muse_cli_db_session: AsyncSession,
326 capsys: pytest.CaptureFixture[str],
327 ) -> None:
328 """-b exits 1 when the target branch already exists."""
329 _init_muse_repo(tmp_path)
330 # Create branch manually
331 (tmp_path / ".muse" / "refs" / "heads" / "existing").write_text("abc123")
332
333 with pytest.raises(typer.Exit) as exc_info:
334 await _do_checkout(tmp_path, muse_cli_db_session, "existing", create=True)
335
336 assert exc_info.value.exit_code == ExitCode.USER_ERROR
337 out = capsys.readouterr().out
338 assert "already exists" in out
339
340
341 # ---------------------------------------------------------------------------
342 # test_checkout_outside_repo_exits_2 (via Typer CLI runner)
343 # ---------------------------------------------------------------------------
344
345
346 def test_checkout_outside_repo_exits_2(tmp_path: pathlib.Path) -> None:
347 """``muse checkout`` exits 2 when no ``.muse/`` directory exists."""
348 import os
349
350 from typer.testing import CliRunner
351
352 from maestro.muse_cli.app import cli
353
354 runner = CliRunner()
355 prev = os.getcwd()
356 try:
357 os.chdir(tmp_path)
358 result = runner.invoke(cli, ["checkout", "main"])
359 assert result.exit_code == int(ExitCode.REPO_NOT_FOUND), (
360 f"Expected exit 2, got {result.exit_code}: {result.output}"
361 )
362 assert "Not a Muse repository" in result.output
363 finally:
364 os.chdir(prev)
365
366
367 # ---------------------------------------------------------------------------
368 # test_checkout_invalid_branch_name_exits_1
369 # ---------------------------------------------------------------------------
370
371
372 @pytest.mark.anyio
373 async def test_checkout_invalid_branch_name_exits_1(
374 tmp_path: pathlib.Path,
375 muse_cli_db_session: AsyncSession,
376 capsys: pytest.CaptureFixture[str],
377 ) -> None:
378 """A branch name with illegal characters exits 1."""
379 _init_muse_repo(tmp_path)
380
381 with pytest.raises(typer.Exit) as exc_info:
382 await _do_checkout(tmp_path, muse_cli_db_session, "bad name!")
383
384 assert exc_info.value.exit_code == ExitCode.USER_ERROR
385 out = capsys.readouterr().out
386 assert "Invalid branch name" in out
387
388
389 # ---------------------------------------------------------------------------
390 # test_checkout_clean_tree_not_blocked
391 # ---------------------------------------------------------------------------
392
393
394 @pytest.mark.anyio
395 async def test_checkout_clean_tree_not_blocked(
396 tmp_path: pathlib.Path,
397 muse_cli_db_session: AsyncSession,
398 capsys: pytest.CaptureFixture[str],
399 ) -> None:
400 """Checkout succeeds when muse-work/ matches the last commit snapshot."""
401 _init_muse_repo(tmp_path)
402 await _make_commit(tmp_path, muse_cli_db_session, "committed state")
403
404 # Create target branch
405 (tmp_path / ".muse" / "refs" / "heads" / "clean-target").write_text("")
406
407 # Do NOT modify muse-work/ — tree is clean
408 capsys.readouterr()
409 await _do_checkout(tmp_path, muse_cli_db_session, "clean-target")
410
411 head = (tmp_path / ".muse" / "HEAD").read_text().strip()
412 assert head == "refs/heads/clean-target"