cgcardona / muse public
test_rev_parse.py python
505 lines 14.9 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse rev-parse``.
2
3 All async tests call ``_rev_parse_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`` for realistic parent
6 chain data.
7
8 Covers all revision expression types:
9 - HEAD
10 - <branch>
11 - <commit_id> (full and prefix)
12 - HEAD~N
13 - <branch>~N
14
15 And all flags:
16 - --short
17 - --verify
18 - --abbrev-ref
19 """
20 from __future__ import annotations
21
22 import json
23 import os
24 import pathlib
25 import uuid
26
27 import pytest
28 import typer
29 from sqlalchemy.ext.asyncio import AsyncSession
30
31 from maestro.muse_cli.commands.commit import _commit_async
32 from maestro.muse_cli.commands.rev_parse import (
33 RevParseResult,
34 _rev_parse_async,
35 resolve_revision,
36 )
37 from maestro.muse_cli.errors import ExitCode
38
39
40 # ---------------------------------------------------------------------------
41 # Helpers
42 # ---------------------------------------------------------------------------
43
44
45 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
46 """Initialise a minimal .muse/ directory and return the repo_id."""
47 rid = repo_id or str(uuid.uuid4())
48 muse = root / ".muse"
49 (muse / "refs" / "heads").mkdir(parents=True)
50 (muse / "repo.json").write_text(
51 json.dumps({"repo_id": rid, "schema_version": "1"})
52 )
53 (muse / "HEAD").write_text("refs/heads/main")
54 (muse / "refs" / "heads" / "main").write_text("")
55 return rid
56
57
58 def _write_workdir(root: pathlib.Path, files: dict[str, bytes]) -> None:
59 workdir = root / "muse-work"
60 workdir.mkdir(exist_ok=True)
61 for name, content in files.items():
62 (workdir / name).write_bytes(content)
63
64
65 async def _make_commits(
66 root: pathlib.Path,
67 session: AsyncSession,
68 messages: list[str],
69 file_seed: int = 0,
70 ) -> list[str]:
71 """Seed N commits on main, returning their commit_ids oldest-first."""
72 commit_ids: list[str] = []
73 for i, msg in enumerate(messages):
74 _write_workdir(root, {f"track_{file_seed + i}.mid": f"MIDI-{file_seed + i}".encode()})
75 cid = await _commit_async(message=msg, root=root, session=session)
76 commit_ids.append(cid)
77 return commit_ids
78
79
80 # ---------------------------------------------------------------------------
81 # test_rev_parse_HEAD_returns_head_commit
82 # ---------------------------------------------------------------------------
83
84
85 @pytest.mark.anyio
86 async def test_rev_parse_HEAD_returns_head_commit(
87 tmp_path: pathlib.Path,
88 muse_cli_db_session: AsyncSession,
89 capsys: pytest.CaptureFixture[str],
90 ) -> None:
91 """``muse rev-parse HEAD`` prints the most recent commit ID."""
92 _init_muse_repo(tmp_path)
93 cids = await _make_commits(tmp_path, muse_cli_db_session, ["take 1", "take 2"])
94
95 capsys.readouterr()
96 await _rev_parse_async(
97 root=tmp_path,
98 session=muse_cli_db_session,
99 revision="HEAD",
100 short=False,
101 verify=False,
102 abbrev_ref=False,
103 )
104 out = capsys.readouterr().out.strip()
105 assert out == cids[-1] # newest commit
106
107
108 # ---------------------------------------------------------------------------
109 # test_rev_parse_branch_name_returns_tip
110 # ---------------------------------------------------------------------------
111
112
113 @pytest.mark.anyio
114 async def test_rev_parse_branch_name_returns_tip(
115 tmp_path: pathlib.Path,
116 muse_cli_db_session: AsyncSession,
117 capsys: pytest.CaptureFixture[str],
118 ) -> None:
119 """``muse rev-parse main`` resolves the tip of the named branch."""
120 _init_muse_repo(tmp_path)
121 cids = await _make_commits(tmp_path, muse_cli_db_session, ["alpha", "beta"])
122
123 capsys.readouterr()
124 await _rev_parse_async(
125 root=tmp_path,
126 session=muse_cli_db_session,
127 revision="main",
128 short=False,
129 verify=False,
130 abbrev_ref=False,
131 )
132 out = capsys.readouterr().out.strip()
133 assert out == cids[-1]
134
135
136 # ---------------------------------------------------------------------------
137 # test_rev_parse_full_commit_id
138 # ---------------------------------------------------------------------------
139
140
141 @pytest.mark.anyio
142 async def test_rev_parse_full_commit_id(
143 tmp_path: pathlib.Path,
144 muse_cli_db_session: AsyncSession,
145 capsys: pytest.CaptureFixture[str],
146 ) -> None:
147 """``muse rev-parse <full_id>`` echoes the same commit ID back."""
148 _init_muse_repo(tmp_path)
149 cids = await _make_commits(tmp_path, muse_cli_db_session, ["v1", "v2", "v3"])
150 target = cids[1] # middle commit
151
152 capsys.readouterr()
153 await _rev_parse_async(
154 root=tmp_path,
155 session=muse_cli_db_session,
156 revision=target,
157 short=False,
158 verify=False,
159 abbrev_ref=False,
160 )
161 out = capsys.readouterr().out.strip()
162 assert out == target
163
164
165 # ---------------------------------------------------------------------------
166 # test_rev_parse_prefix_commit_id
167 # ---------------------------------------------------------------------------
168
169
170 @pytest.mark.anyio
171 async def test_rev_parse_prefix_commit_id(
172 tmp_path: pathlib.Path,
173 muse_cli_db_session: AsyncSession,
174 capsys: pytest.CaptureFixture[str],
175 ) -> None:
176 """An 8-char prefix is resolved to the full commit ID."""
177 _init_muse_repo(tmp_path)
178 cids = await _make_commits(tmp_path, muse_cli_db_session, ["groove"])
179 prefix = cids[0][:8]
180
181 capsys.readouterr()
182 await _rev_parse_async(
183 root=tmp_path,
184 session=muse_cli_db_session,
185 revision=prefix,
186 short=False,
187 verify=False,
188 abbrev_ref=False,
189 )
190 out = capsys.readouterr().out.strip()
191 assert out == cids[0]
192
193
194 # ---------------------------------------------------------------------------
195 # test_rev_parse_HEAD_tilde_1
196 # ---------------------------------------------------------------------------
197
198
199 @pytest.mark.anyio
200 async def test_rev_parse_HEAD_tilde_1(
201 tmp_path: pathlib.Path,
202 muse_cli_db_session: AsyncSession,
203 capsys: pytest.CaptureFixture[str],
204 ) -> None:
205 """``HEAD~1`` resolves to the parent of the current HEAD."""
206 _init_muse_repo(tmp_path)
207 cids = await _make_commits(tmp_path, muse_cli_db_session, ["a", "b", "c"])
208
209 capsys.readouterr()
210 await _rev_parse_async(
211 root=tmp_path,
212 session=muse_cli_db_session,
213 revision="HEAD~1",
214 short=False,
215 verify=False,
216 abbrev_ref=False,
217 )
218 out = capsys.readouterr().out.strip()
219 assert out == cids[1] # one step back from cids[2]
220
221
222 # ---------------------------------------------------------------------------
223 # test_rev_parse_HEAD_tilde_2
224 # ---------------------------------------------------------------------------
225
226
227 @pytest.mark.anyio
228 async def test_rev_parse_HEAD_tilde_2(
229 tmp_path: pathlib.Path,
230 muse_cli_db_session: AsyncSession,
231 capsys: pytest.CaptureFixture[str],
232 ) -> None:
233 """``HEAD~2`` walks two parents back from HEAD."""
234 _init_muse_repo(tmp_path)
235 cids = await _make_commits(tmp_path, muse_cli_db_session, ["x", "y", "z"])
236
237 capsys.readouterr()
238 await _rev_parse_async(
239 root=tmp_path,
240 session=muse_cli_db_session,
241 revision="HEAD~2",
242 short=False,
243 verify=False,
244 abbrev_ref=False,
245 )
246 out = capsys.readouterr().out.strip()
247 assert out == cids[0] # two steps back from cids[2]
248
249
250 # ---------------------------------------------------------------------------
251 # test_rev_parse_branch_tilde
252 # ---------------------------------------------------------------------------
253
254
255 @pytest.mark.anyio
256 async def test_rev_parse_branch_tilde(
257 tmp_path: pathlib.Path,
258 muse_cli_db_session: AsyncSession,
259 capsys: pytest.CaptureFixture[str],
260 ) -> None:
261 """``main~1`` walks one parent back from the main branch tip."""
262 _init_muse_repo(tmp_path)
263 cids = await _make_commits(tmp_path, muse_cli_db_session, ["first", "second"])
264
265 capsys.readouterr()
266 await _rev_parse_async(
267 root=tmp_path,
268 session=muse_cli_db_session,
269 revision="main~1",
270 short=False,
271 verify=False,
272 abbrev_ref=False,
273 )
274 out = capsys.readouterr().out.strip()
275 assert out == cids[0]
276
277
278 # ---------------------------------------------------------------------------
279 # test_rev_parse_short_flag
280 # ---------------------------------------------------------------------------
281
282
283 @pytest.mark.anyio
284 async def test_rev_parse_short_flag(
285 tmp_path: pathlib.Path,
286 muse_cli_db_session: AsyncSession,
287 capsys: pytest.CaptureFixture[str],
288 ) -> None:
289 """``--short`` outputs only the first 8 characters of the commit ID."""
290 _init_muse_repo(tmp_path)
291 cids = await _make_commits(tmp_path, muse_cli_db_session, ["take"])
292
293 capsys.readouterr()
294 await _rev_parse_async(
295 root=tmp_path,
296 session=muse_cli_db_session,
297 revision="HEAD",
298 short=True,
299 verify=False,
300 abbrev_ref=False,
301 )
302 out = capsys.readouterr().out.strip()
303 assert out == cids[0][:8]
304 assert len(out) == 8
305
306
307 # ---------------------------------------------------------------------------
308 # test_rev_parse_abbrev_ref_HEAD
309 # ---------------------------------------------------------------------------
310
311
312 @pytest.mark.anyio
313 async def test_rev_parse_abbrev_ref_HEAD(
314 tmp_path: pathlib.Path,
315 muse_cli_db_session: AsyncSession,
316 capsys: pytest.CaptureFixture[str],
317 ) -> None:
318 """``--abbrev-ref HEAD`` prints the current branch name, not the commit ID."""
319 _init_muse_repo(tmp_path)
320 await _make_commits(tmp_path, muse_cli_db_session, ["beat"])
321
322 capsys.readouterr()
323 await _rev_parse_async(
324 root=tmp_path,
325 session=muse_cli_db_session,
326 revision="HEAD",
327 short=False,
328 verify=False,
329 abbrev_ref=True,
330 )
331 out = capsys.readouterr().out.strip()
332 assert out == "main"
333
334
335 # ---------------------------------------------------------------------------
336 # test_rev_parse_verify_fails_on_unknown_ref
337 # ---------------------------------------------------------------------------
338
339
340 @pytest.mark.anyio
341 async def test_rev_parse_verify_fails_on_unknown_ref(
342 tmp_path: pathlib.Path,
343 muse_cli_db_session: AsyncSession,
344 capsys: pytest.CaptureFixture[str],
345 ) -> None:
346 """``--verify`` exits USER_ERROR when the revision does not resolve."""
347 _init_muse_repo(tmp_path)
348 # No commits — nothing to resolve
349
350 with pytest.raises(typer.Exit) as exc_info:
351 await _rev_parse_async(
352 root=tmp_path,
353 session=muse_cli_db_session,
354 revision="nonexistent",
355 short=False,
356 verify=True,
357 abbrev_ref=False,
358 )
359
360 assert exc_info.value.exit_code == ExitCode.USER_ERROR
361
362
363 # ---------------------------------------------------------------------------
364 # test_rev_parse_no_verify_prints_nothing_on_miss
365 # ---------------------------------------------------------------------------
366
367
368 @pytest.mark.anyio
369 async def test_rev_parse_no_verify_prints_nothing_on_miss(
370 tmp_path: pathlib.Path,
371 muse_cli_db_session: AsyncSession,
372 capsys: pytest.CaptureFixture[str],
373 ) -> None:
374 """Without ``--verify``, an unresolvable ref prints nothing and exits 0."""
375 _init_muse_repo(tmp_path)
376
377 # Should not raise
378 await _rev_parse_async(
379 root=tmp_path,
380 session=muse_cli_db_session,
381 revision="deadbeef",
382 short=False,
383 verify=False,
384 abbrev_ref=False,
385 )
386 out = capsys.readouterr().out.strip()
387 assert out == ""
388
389
390 # ---------------------------------------------------------------------------
391 # test_rev_parse_tilde_beyond_root_returns_nothing
392 # ---------------------------------------------------------------------------
393
394
395 @pytest.mark.anyio
396 async def test_rev_parse_tilde_beyond_root_returns_nothing(
397 tmp_path: pathlib.Path,
398 muse_cli_db_session: AsyncSession,
399 capsys: pytest.CaptureFixture[str],
400 ) -> None:
401 """``HEAD~10`` on a 2-commit chain prints nothing (no --verify)."""
402 _init_muse_repo(tmp_path)
403 await _make_commits(tmp_path, muse_cli_db_session, ["one", "two"])
404
405 capsys.readouterr() # discard commit output from _make_commits
406 await _rev_parse_async(
407 root=tmp_path,
408 session=muse_cli_db_session,
409 revision="HEAD~10",
410 short=False,
411 verify=False,
412 abbrev_ref=False,
413 )
414 out = capsys.readouterr().out.strip()
415 assert out == ""
416
417
418 # ---------------------------------------------------------------------------
419 # test_rev_parse_head_tilde_zero_equals_HEAD
420 # ---------------------------------------------------------------------------
421
422
423 @pytest.mark.anyio
424 async def test_rev_parse_head_tilde_zero_equals_HEAD(
425 tmp_path: pathlib.Path,
426 muse_cli_db_session: AsyncSession,
427 capsys: pytest.CaptureFixture[str],
428 ) -> None:
429 """``HEAD~0`` resolves to the same commit as ``HEAD``."""
430 _init_muse_repo(tmp_path)
431 cids = await _make_commits(tmp_path, muse_cli_db_session, ["r1", "r2"])
432
433 capsys.readouterr()
434 await _rev_parse_async(
435 root=tmp_path,
436 session=muse_cli_db_session,
437 revision="HEAD~0",
438 short=False,
439 verify=False,
440 abbrev_ref=False,
441 )
442 out_tilde = capsys.readouterr().out.strip()
443
444 capsys.readouterr()
445 await _rev_parse_async(
446 root=tmp_path,
447 session=muse_cli_db_session,
448 revision="HEAD",
449 short=False,
450 verify=False,
451 abbrev_ref=False,
452 )
453 out_head = capsys.readouterr().out.strip()
454
455 assert out_tilde == out_head == cids[-1]
456
457
458 # ---------------------------------------------------------------------------
459 # test_resolve_revision_returns_RevParseResult_type
460 # ---------------------------------------------------------------------------
461
462
463 @pytest.mark.anyio
464 async def test_resolve_revision_returns_RevParseResult_type(
465 tmp_path: pathlib.Path,
466 muse_cli_db_session: AsyncSession,
467 ) -> None:
468 """``resolve_revision`` returns a ``RevParseResult`` with correct fields."""
469 repo_id = _init_muse_repo(tmp_path)
470 cids = await _make_commits(tmp_path, muse_cli_db_session, ["init"])
471
472 result = await resolve_revision(
473 session=muse_cli_db_session,
474 repo_id=repo_id,
475 current_branch="main",
476 muse_dir=tmp_path / ".muse",
477 revision_expr="HEAD",
478 )
479
480 assert result is not None
481 assert isinstance(result, RevParseResult)
482 assert result.commit_id == cids[0]
483 assert result.branch == "main"
484 assert result.revision_expr == "HEAD"
485
486
487 # ---------------------------------------------------------------------------
488 # test_rev_parse_outside_repo_exits_2
489 # ---------------------------------------------------------------------------
490
491
492 def test_rev_parse_outside_repo_exits_2(tmp_path: pathlib.Path) -> None:
493 """``muse rev-parse`` outside a .muse/ directory exits with code 2."""
494 from typer.testing import CliRunner
495 from maestro.muse_cli.app import cli
496
497 runner = CliRunner()
498 prev = os.getcwd()
499 try:
500 os.chdir(tmp_path)
501 result = runner.invoke(cli, ["rev-parse", "HEAD"], catch_exceptions=False)
502 finally:
503 os.chdir(prev)
504
505 assert result.exit_code == ExitCode.REPO_NOT_FOUND