cgcardona / muse public
test_revert.py python
485 lines 15.6 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse revert`` — safe undo via forward commit.
2
3 Exercises:
4 - ``test_muse_revert_creates_undo_commit`` — regression: full revert creates a
5 new commit pointing to the parent's snapshot.
6 - ``test_muse_revert_no_commit_stages_only`` — --no-commit writes to
7 muse-work/ without creating a DB commit.
8 - ``test_muse_revert_scoped_by_track`` — --track limits which paths are
9 reverted; other paths remain at HEAD state.
10 - ``test_muse_revert_blocked_during_merge`` — blocked when a merge is in
11 progress (conflict_paths non-empty).
12 - ``test_muse_revert_root_commit`` — reverting the root commit produces an
13 empty snapshot.
14 - ``test_muse_revert_noop_when_already_reverted`` — reverting a commit that
15 is already at its parent's state emits a noop.
16 - ``compute_revert_manifest_*`` — pure-function unit tests for the manifest
17 computation.
18
19 All async tests use ``@pytest.mark.anyio``.
20
21 """
22 from __future__ import annotations
23
24 import json
25 import pathlib
26 import uuid
27
28 import pytest
29 import pytest_asyncio
30 from sqlalchemy.ext.asyncio import AsyncSession
31
32 from maestro.muse_cli.commands.commit import _commit_async
33 from maestro.muse_cli.merge_engine import write_merge_state
34 from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot
35 from maestro.services.muse_revert import (
36 RevertResult,
37 _revert_async,
38 compute_revert_manifest,
39 )
40
41
42 # ---------------------------------------------------------------------------
43 # Repo / workdir helpers (shared with test_commit.py pattern)
44 # ---------------------------------------------------------------------------
45
46
47 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
48 """Create a minimal .muse/ layout."""
49 rid = repo_id or str(uuid.uuid4())
50 muse = root / ".muse"
51 (muse / "refs" / "heads").mkdir(parents=True)
52 (muse / "repo.json").write_text(
53 json.dumps({"repo_id": rid, "schema_version": "1"})
54 )
55 (muse / "HEAD").write_text("refs/heads/main")
56 (muse / "refs" / "heads" / "main").write_text("")
57 return rid
58
59
60 def _populate_workdir(
61 root: pathlib.Path, files: dict[str, bytes] | None = None
62 ) -> None:
63 """Create muse-work/ with the specified files."""
64 workdir = root / "muse-work"
65 workdir.mkdir(exist_ok=True)
66 if files is None:
67 files = {"beat.mid": b"MIDI-DATA", "lead.mp3": b"MP3-DATA"}
68 for name, content in files.items():
69 path = workdir / name
70 path.parent.mkdir(parents=True, exist_ok=True)
71 path.write_bytes(content)
72
73
74 # ---------------------------------------------------------------------------
75 # Unit tests — pure functions
76 # ---------------------------------------------------------------------------
77
78
79 def test_compute_revert_manifest_full_revert() -> None:
80 """Full (unscoped) revert returns the parent manifest verbatim."""
81 parent = {"beat.mid": "aaa", "lead.mp3": "bbb"}
82 head = {"beat.mid": "ccc", "lead.mp3": "bbb", "synth.mid": "ddd"}
83
84 result, scoped = compute_revert_manifest(parent_manifest=parent, head_manifest=head)
85
86 assert result == parent
87 assert scoped == ()
88
89
90 def test_compute_revert_manifest_scoped_by_track() -> None:
91 """--track reverts only paths under tracks/<track>/."""
92 parent = {
93 "tracks/drums/beat.mid": "old-drums",
94 "tracks/bass/bass.mid": "old-bass",
95 "sections/verse/lead.mid": "verse-lead",
96 }
97 head = {
98 "tracks/drums/beat.mid": "new-drums",
99 "tracks/bass/bass.mid": "new-bass",
100 "sections/verse/lead.mid": "verse-lead",
101 "tracks/drums/fill.mid": "head-only-fill",
102 }
103
104 result, scoped = compute_revert_manifest(
105 parent_manifest=parent, head_manifest=head, track="drums"
106 )
107
108 # drums paths should come from parent or be removed if head-only
109 assert result["tracks/drums/beat.mid"] == "old-drums"
110 # fill.mid exists only in head (under drums) → should be removed
111 assert "tracks/drums/fill.mid" not in result
112 # bass and section paths remain at HEAD
113 assert result["tracks/bass/bass.mid"] == "new-bass"
114 assert result["sections/verse/lead.mid"] == "verse-lead"
115 # Scoped paths should include drums paths from both manifests
116 assert "tracks/drums/beat.mid" in scoped
117 assert "tracks/drums/fill.mid" in scoped
118 assert "tracks/bass/bass.mid" not in scoped
119
120
121 def test_compute_revert_manifest_scoped_by_section() -> None:
122 """--section reverts only paths under sections/<section>/."""
123 parent = {"sections/chorus/chords.mid": "old-chords"}
124 head = {
125 "sections/chorus/chords.mid": "new-chords",
126 "sections/verse/lead.mid": "verse-lead",
127 }
128
129 result, scoped = compute_revert_manifest(
130 parent_manifest=parent, head_manifest=head, section="chorus"
131 )
132
133 assert result["sections/chorus/chords.mid"] == "old-chords"
134 assert result["sections/verse/lead.mid"] == "verse-lead"
135 assert "sections/chorus/chords.mid" in scoped
136 assert "sections/verse/lead.mid" not in scoped
137
138
139 def test_compute_revert_manifest_empty_parent() -> None:
140 """Reverting the root commit (no parent) produces an empty manifest."""
141 result, scoped = compute_revert_manifest(
142 parent_manifest={},
143 head_manifest={"beat.mid": "aaa"},
144 )
145 assert result == {}
146 assert scoped == ()
147
148
149 # ---------------------------------------------------------------------------
150 # Integration tests — async DB
151 # ---------------------------------------------------------------------------
152
153
154 @pytest.mark.anyio
155 async def test_muse_revert_creates_undo_commit(
156 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
157 ) -> None:
158 """Regression: muse revert <commit> creates a new commit on the parent's snapshot."""
159 _init_muse_repo(tmp_path)
160
161 # Commit A — initial state
162 _populate_workdir(tmp_path, {"beat.mid": b"take-1"})
163 commit_a_id = await _commit_async(
164 message="initial take",
165 root=tmp_path,
166 session=muse_cli_db_session,
167 )
168
169 # Commit B — a bad arrangement
170 _populate_workdir(tmp_path, {"beat.mid": b"bad-take"})
171 commit_b_id = await _commit_async(
172 message="bad drum arrangement",
173 root=tmp_path,
174 session=muse_cli_db_session,
175 )
176
177 # Revert B → should create commit C pointing to A's snapshot
178 result = await _revert_async(
179 commit_ref=commit_b_id,
180 root=tmp_path,
181 session=muse_cli_db_session,
182 )
183
184 assert not result.noop
185 assert not result.no_commit
186 assert result.commit_id != ""
187 assert result.message == f"Revert 'bad drum arrangement'"
188 assert result.target_commit_id == commit_b_id
189 assert result.parent_commit_id == commit_a_id
190
191 # The new commit should point to A's snapshot
192 commit_a_row = await muse_cli_db_session.get(MuseCliCommit, commit_a_id)
193 commit_c_row = await muse_cli_db_session.get(MuseCliCommit, result.commit_id)
194 assert commit_c_row is not None
195 assert commit_c_row.snapshot_id == commit_a_row.snapshot_id # type: ignore[union-attr]
196 assert commit_c_row.parent_commit_id == commit_b_id
197
198 # HEAD ref file updated to new commit
199 head_ref = (tmp_path / ".muse" / "HEAD").read_text().strip()
200 ref_path = tmp_path / ".muse" / pathlib.Path(head_ref)
201 assert ref_path.read_text().strip() == result.commit_id
202
203
204 @pytest.mark.anyio
205 async def test_muse_revert_no_commit_stages_only(
206 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
207 ) -> None:
208 """--no-commit: no new commit row is created; muse-work/ deletions are applied."""
209 _init_muse_repo(tmp_path)
210
211 # Commit A — initial state
212 _populate_workdir(tmp_path, {"beat.mid": b"take-1"})
213 commit_a_id = await _commit_async(
214 message="initial take",
215 root=tmp_path,
216 session=muse_cli_db_session,
217 )
218
219 # Commit B — adds an extra file
220 _populate_workdir(tmp_path, {"beat.mid": b"take-1", "extra.mid": b"EXTRA"})
221 commit_b_id = await _commit_async(
222 message="added extra track",
223 root=tmp_path,
224 session=muse_cli_db_session,
225 )
226
227 # Revert B with --no-commit
228 result = await _revert_async(
229 commit_ref=commit_b_id,
230 root=tmp_path,
231 session=muse_cli_db_session,
232 no_commit=True,
233 )
234
235 assert result.no_commit is True
236 assert result.commit_id == "" # no commit created
237 # extra.mid should have been deleted from muse-work/
238 assert "extra.mid" in result.paths_deleted
239 assert not (tmp_path / "muse-work" / "extra.mid").exists()
240
241 # No new commit in DB
242 from sqlalchemy.future import select
243
244 commit_count = (
245 await muse_cli_db_session.execute(select(MuseCliCommit))
246 ).scalars().all()
247 assert len(commit_count) == 2 # only A and B, not a C
248
249
250 @pytest.mark.anyio
251 async def test_muse_revert_scoped_by_track(
252 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
253 ) -> None:
254 """--track reverts only the specified track; other paths stay at HEAD."""
255 _init_muse_repo(tmp_path)
256
257 # Commit A — initial state for both drums and bass
258 _populate_workdir(
259 tmp_path,
260 {
261 "tracks/drums/beat.mid": b"drums-v1",
262 "tracks/bass/bass.mid": b"bass-v1",
263 },
264 )
265 commit_a_id = await _commit_async(
266 message="initial arrangement",
267 root=tmp_path,
268 session=muse_cli_db_session,
269 )
270
271 # Commit B — new drums, same bass
272 _populate_workdir(
273 tmp_path,
274 {
275 "tracks/drums/beat.mid": b"drums-v2",
276 "tracks/bass/bass.mid": b"bass-v1",
277 },
278 )
279 commit_b_id = await _commit_async(
280 message="updated drums only",
281 root=tmp_path,
282 session=muse_cli_db_session,
283 )
284
285 # Revert B --track drums
286 result = await _revert_async(
287 commit_ref=commit_b_id,
288 root=tmp_path,
289 session=muse_cli_db_session,
290 track="drums",
291 )
292
293 assert not result.noop
294 assert result.commit_id != ""
295 assert "tracks/drums/beat.mid" in result.scoped_paths
296 assert "tracks/bass/bass.mid" not in result.scoped_paths
297
298 # The revert commit's snapshot: drums should be at v1, bass still at v1 (same)
299 commit_c_row = await muse_cli_db_session.get(MuseCliCommit, result.commit_id)
300 assert commit_c_row is not None
301 snap_row = await muse_cli_db_session.get(MuseCliSnapshot, commit_c_row.snapshot_id)
302 assert snap_row is not None
303 manifest: dict[str, str] = dict(snap_row.manifest)
304
305 # Drums path should come from A's snapshot (v1)
306 commit_a_snap_row = await muse_cli_db_session.get(MuseCliCommit, commit_a_id)
307 a_snap = await muse_cli_db_session.get(
308 MuseCliSnapshot, commit_a_snap_row.snapshot_id # type: ignore[union-attr]
309 )
310 assert a_snap is not None
311 a_manifest: dict[str, str] = dict(a_snap.manifest)
312 assert manifest.get("tracks/drums/beat.mid") == a_manifest.get(
313 "tracks/drums/beat.mid"
314 )
315
316
317 @pytest.mark.anyio
318 async def test_muse_revert_blocked_during_merge(
319 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
320 ) -> None:
321 """muse revert is blocked when a merge is in-progress with conflicts."""
322 _init_muse_repo(tmp_path)
323 _populate_workdir(tmp_path, {"beat.mid": b"take-1"})
324 commit_a_id = await _commit_async(
325 message="initial take",
326 root=tmp_path,
327 session=muse_cli_db_session,
328 )
329
330 # Simulate an in-progress merge with conflicts
331 write_merge_state(
332 tmp_path,
333 base_commit="base-abc",
334 ours_commit="ours-def",
335 theirs_commit="theirs-ghi",
336 conflict_paths=["beat.mid"],
337 other_branch="feature/bad-drums",
338 )
339
340 import typer
341
342 with pytest.raises(typer.Exit) as exc_info:
343 await _revert_async(
344 commit_ref=commit_a_id,
345 root=tmp_path,
346 session=muse_cli_db_session,
347 )
348
349 from maestro.muse_cli.errors import ExitCode
350
351 assert exc_info.value.exit_code == ExitCode.USER_ERROR
352
353
354 @pytest.mark.anyio
355 async def test_muse_revert_root_commit_produces_empty_snapshot(
356 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
357 ) -> None:
358 """Reverting the root commit (no parent) creates a commit with an empty snapshot."""
359 _init_muse_repo(tmp_path)
360 _populate_workdir(tmp_path, {"beat.mid": b"take-1"})
361 root_commit_id = await _commit_async(
362 message="root commit",
363 root=tmp_path,
364 session=muse_cli_db_session,
365 )
366
367 result = await _revert_async(
368 commit_ref=root_commit_id,
369 root=tmp_path,
370 session=muse_cli_db_session,
371 )
372
373 assert not result.noop
374 assert result.commit_id != ""
375 assert result.parent_commit_id == "" # root has no parent
376
377 snap_row = await muse_cli_db_session.get(MuseCliSnapshot, result.revert_snapshot_id)
378 assert snap_row is not None
379 assert snap_row.manifest == {}
380
381
382 @pytest.mark.anyio
383 async def test_muse_revert_abbreviated_commit_ref(
384 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
385 ) -> None:
386 """muse revert accepts an abbreviated commit SHA (prefix match)."""
387 _init_muse_repo(tmp_path)
388 _populate_workdir(tmp_path, {"beat.mid": b"v1"})
389 commit_a_id = await _commit_async(
390 message="v1",
391 root=tmp_path,
392 session=muse_cli_db_session,
393 )
394 _populate_workdir(tmp_path, {"beat.mid": b"v2"})
395 commit_b_id = await _commit_async(
396 message="v2",
397 root=tmp_path,
398 session=muse_cli_db_session,
399 )
400
401 # Use abbreviated prefix (first 8 chars)
402 result = await _revert_async(
403 commit_ref=commit_b_id[:8],
404 root=tmp_path,
405 session=muse_cli_db_session,
406 )
407
408 assert result.target_commit_id == commit_b_id
409 assert result.commit_id != ""
410
411
412 @pytest.mark.anyio
413 async def test_muse_revert_noop_when_already_reverted(
414 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
415 ) -> None:
416 """Reverting the same commit twice is a noop on the second attempt.
417
418 After the first revert, HEAD's snapshot equals the target commit's parent
419 snapshot. A second revert of the same target would produce the identical
420 snapshot → noop.
421 """
422 _init_muse_repo(tmp_path)
423
424 # Commit A — initial state
425 _populate_workdir(tmp_path, {"beat.mid": b"v1"})
426 commit_a_id = await _commit_async(
427 message="initial take",
428 root=tmp_path,
429 session=muse_cli_db_session,
430 )
431
432 # Commit B — changed state
433 _populate_workdir(tmp_path, {"beat.mid": b"v2"})
434 commit_b_id = await _commit_async(
435 message="updated arrangement",
436 root=tmp_path,
437 session=muse_cli_db_session,
438 )
439
440 # First revert: B → creates commit C whose snapshot == A's snapshot
441 first = await _revert_async(
442 commit_ref=commit_b_id,
443 root=tmp_path,
444 session=muse_cli_db_session,
445 )
446 assert not first.noop
447 assert first.parent_commit_id == commit_a_id
448
449 # Second revert of the same B: HEAD is now C (snapshot == A's snapshot).
450 # Reverting B again would produce the same snapshot as HEAD → noop.
451 second = await _revert_async(
452 commit_ref=commit_b_id,
453 root=tmp_path,
454 session=muse_cli_db_session,
455 )
456 assert second.noop is True
457 assert second.commit_id == ""
458 assert second.target_commit_id == commit_b_id
459
460
461 @pytest.mark.anyio
462 async def test_muse_revert_unknown_commit_raises_exit(
463 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
464 ) -> None:
465 """muse revert with an unknown commit ID exits with USER_ERROR."""
466 _init_muse_repo(tmp_path)
467 _populate_workdir(tmp_path)
468 await _commit_async(
469 message="initial",
470 root=tmp_path,
471 session=muse_cli_db_session,
472 )
473
474 import typer
475
476 with pytest.raises(typer.Exit) as exc_info:
477 await _revert_async(
478 commit_ref="deadbeef",
479 root=tmp_path,
480 session=muse_cli_db_session,
481 )
482
483 from maestro.muse_cli.errors import ExitCode
484
485 assert exc_info.value.exit_code == ExitCode.USER_ERROR