cgcardona / muse public
test_read_tree.py python
483 lines 15.4 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse read-tree`` command.
2
3 Exercises ``_read_tree_async`` directly (async core) and indirectly
4 through the object store integration with ``_commit_async``.
5
6 All async tests use ``@pytest.mark.anyio``.
7 The ``muse_cli_db_session`` fixture is defined in tests/muse_cli/conftest.py.
8 """
9 from __future__ import annotations
10
11 import json
12 import pathlib
13 import uuid
14
15 import pytest
16 import typer
17 from sqlalchemy.ext.asyncio import AsyncSession
18
19 from maestro.muse_cli.commands.commit import _commit_async
20 from maestro.muse_cli.commands.read_tree import (
21 ReadTreeResult,
22 _read_tree_async,
23 _resolve_snapshot,
24 )
25 from maestro.muse_cli.errors import ExitCode
26 from maestro.muse_cli.object_store import has_object, object_path, write_object
27 from maestro.muse_cli.snapshot import compute_snapshot_id, hash_file
28
29
30 # ---------------------------------------------------------------------------
31 # Test helpers
32 # ---------------------------------------------------------------------------
33
34
35 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
36 """Create a minimal ``.muse/`` layout for testing."""
37 rid = repo_id or str(uuid.uuid4())
38 muse = root / ".muse"
39 (muse / "refs" / "heads").mkdir(parents=True)
40 (muse / "repo.json").write_text(
41 json.dumps({"repo_id": rid, "schema_version": "1"})
42 )
43 (muse / "HEAD").write_text("refs/heads/main")
44 (muse / "refs" / "heads" / "main").write_text("")
45 return rid
46
47
48 def _populate_workdir(
49 root: pathlib.Path,
50 files: dict[str, bytes] | None = None,
51 ) -> None:
52 """Create muse-work/ with the given files (default: two MIDI stubs)."""
53 workdir = root / "muse-work"
54 workdir.mkdir(exist_ok=True)
55 if files is None:
56 files = {"beat.mid": b"MIDI-BEAT", "lead.mid": b"MIDI-LEAD"}
57 for name, content in files.items():
58 target = workdir / name
59 target.parent.mkdir(parents=True, exist_ok=True)
60 target.write_bytes(content)
61
62
63 # ---------------------------------------------------------------------------
64 # object_store unit tests
65 # ---------------------------------------------------------------------------
66
67
68 class TestObjectStore:
69 """Unit tests for the local content-addressed object store."""
70
71 def test_write_and_read_object(self, tmp_path: pathlib.Path) -> None:
72 _init_muse_repo(tmp_path)
73 content = b"hello muse"
74 oid = "a" * 64
75 write_object(tmp_path, oid, content)
76 dest = object_path(tmp_path, oid)
77 assert dest.exists()
78 assert dest.read_bytes() == content
79
80 def test_write_object_idempotent(self, tmp_path: pathlib.Path) -> None:
81 _init_muse_repo(tmp_path)
82 content = b"original"
83 oid = "b" * 64
84 assert write_object(tmp_path, oid, content) is True
85 # Second write with different bytes — should be skipped (content-addressed).
86 assert write_object(tmp_path, oid, b"different") is False
87 dest = object_path(tmp_path, oid)
88 assert dest.read_bytes() == content # Original bytes preserved.
89
90 def test_has_object_false_before_write(self, tmp_path: pathlib.Path) -> None:
91 _init_muse_repo(tmp_path)
92 assert not has_object(tmp_path, "c" * 64)
93
94 def test_has_object_true_after_write(self, tmp_path: pathlib.Path) -> None:
95 _init_muse_repo(tmp_path)
96 oid = "d" * 64
97 write_object(tmp_path, oid, b"data")
98 assert has_object(tmp_path, oid)
99
100
101 # ---------------------------------------------------------------------------
102 # commit stores objects in local store (regression)
103 # ---------------------------------------------------------------------------
104
105
106 @pytest.mark.anyio
107 async def test_commit_writes_objects_to_local_store(
108 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
109 ) -> None:
110 """muse commit must persist file bytes to .muse/objects/ so read-tree works.
111
112 Regression: before this feature, commit only wrote object metadata to
113 the DB without persisting bytes on disk.
114 """
115 _init_muse_repo(tmp_path)
116 _populate_workdir(tmp_path, {"beat.mid": b"BEAT-BYTES", "lead.mid": b"LEAD-BYTES"})
117
118 await _commit_async(
119 message="store objects test",
120 root=tmp_path,
121 session=muse_cli_db_session,
122 )
123
124 workdir = tmp_path / "muse-work"
125 for filename in ["beat.mid", "lead.mid"]:
126 file_path = workdir / filename
127 oid = hash_file(file_path)
128 assert has_object(tmp_path, oid), f"Object for {filename} not in local store"
129
130
131 # ---------------------------------------------------------------------------
132 # _resolve_snapshot tests
133 # ---------------------------------------------------------------------------
134
135
136 @pytest.mark.anyio
137 async def test_resolve_snapshot_full_id(
138 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
139 ) -> None:
140 """Full 64-char snapshot ID resolves correctly."""
141 _init_muse_repo(tmp_path)
142 _populate_workdir(tmp_path)
143
144 await _commit_async(
145 message="resolve full id",
146 root=tmp_path,
147 session=muse_cli_db_session,
148 )
149
150 workdir = tmp_path / "muse-work"
151 from maestro.muse_cli.snapshot import build_snapshot_manifest
152 manifest = build_snapshot_manifest(workdir)
153 snap_id = compute_snapshot_id(manifest)
154
155 snapshot = await _resolve_snapshot(muse_cli_db_session, snap_id)
156 assert snapshot is not None
157 assert snapshot.snapshot_id == snap_id
158
159
160 @pytest.mark.anyio
161 async def test_resolve_snapshot_prefix(
162 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
163 ) -> None:
164 """Abbreviated 8-char prefix resolves to the correct snapshot."""
165 _init_muse_repo(tmp_path)
166 _populate_workdir(tmp_path)
167
168 await _commit_async(
169 message="resolve prefix",
170 root=tmp_path,
171 session=muse_cli_db_session,
172 )
173
174 workdir = tmp_path / "muse-work"
175 from maestro.muse_cli.snapshot import build_snapshot_manifest
176 manifest = build_snapshot_manifest(workdir)
177 snap_id = compute_snapshot_id(manifest)
178
179 snapshot = await _resolve_snapshot(muse_cli_db_session, snap_id[:8])
180 assert snapshot is not None
181 assert snapshot.snapshot_id == snap_id
182
183
184 @pytest.mark.anyio
185 async def test_resolve_snapshot_unknown_returns_none(
186 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
187 ) -> None:
188 """Unknown snapshot ID returns None."""
189 snapshot = await _resolve_snapshot(muse_cli_db_session, "0" * 64)
190 assert snapshot is None
191
192
193 # ---------------------------------------------------------------------------
194 # _read_tree_async — basic population
195 # ---------------------------------------------------------------------------
196
197
198 @pytest.mark.anyio
199 async def test_read_tree_populates_workdir(
200 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
201 ) -> None:
202 """read-tree writes correct file content to muse-work/."""
203 _init_muse_repo(tmp_path)
204 file_content = b"PIANO-RIFF"
205 _populate_workdir(tmp_path, {"piano.mid": file_content})
206
207 await _commit_async(
208 message="piano riff",
209 root=tmp_path,
210 session=muse_cli_db_session,
211 )
212
213 from maestro.muse_cli.snapshot import build_snapshot_manifest
214 manifest = build_snapshot_manifest(tmp_path / "muse-work")
215 snap_id = compute_snapshot_id(manifest)
216
217 # Simulate a cleared working directory.
218 (tmp_path / "muse-work" / "piano.mid").unlink()
219
220 result = await _read_tree_async(
221 snapshot_id=snap_id,
222 root=tmp_path,
223 session=muse_cli_db_session,
224 )
225
226 assert isinstance(result, ReadTreeResult)
227 assert result.snapshot_id == snap_id
228 assert "piano.mid" in result.files_written
229 assert not result.dry_run
230 assert not result.reset
231
232 restored = tmp_path / "muse-work" / "piano.mid"
233 assert restored.exists()
234 assert restored.read_bytes() == file_content
235
236
237 @pytest.mark.anyio
238 async def test_read_tree_populates_nested_paths(
239 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
240 ) -> None:
241 """read-tree creates parent directories for nested file paths."""
242 _init_muse_repo(tmp_path)
243 _populate_workdir(tmp_path, {"tracks/drums/kick.mid": b"KICK-DATA"})
244
245 await _commit_async(
246 message="nested",
247 root=tmp_path,
248 session=muse_cli_db_session,
249 )
250
251 from maestro.muse_cli.snapshot import build_snapshot_manifest
252 manifest = build_snapshot_manifest(tmp_path / "muse-work")
253 snap_id = compute_snapshot_id(manifest)
254
255 (tmp_path / "muse-work" / "tracks" / "drums" / "kick.mid").unlink()
256
257 result = await _read_tree_async(
258 snapshot_id=snap_id,
259 root=tmp_path,
260 session=muse_cli_db_session,
261 )
262
263 assert "tracks/drums/kick.mid" in result.files_written
264 restored = tmp_path / "muse-work" / "tracks" / "drums" / "kick.mid"
265 assert restored.exists()
266 assert restored.read_bytes() == b"KICK-DATA"
267
268
269 # ---------------------------------------------------------------------------
270 # _read_tree_async — dry-run
271 # ---------------------------------------------------------------------------
272
273
274 @pytest.mark.anyio
275 async def test_read_tree_dry_run_does_not_write(
276 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
277 ) -> None:
278 """--dry-run must not write files to muse-work/."""
279 _init_muse_repo(tmp_path)
280 _populate_workdir(tmp_path, {"bass.mid": b"BASS-GROOVE"})
281
282 await _commit_async(
283 message="dry-run test",
284 root=tmp_path,
285 session=muse_cli_db_session,
286 )
287
288 from maestro.muse_cli.snapshot import build_snapshot_manifest
289 manifest = build_snapshot_manifest(tmp_path / "muse-work")
290 snap_id = compute_snapshot_id(manifest)
291
292 original_file = tmp_path / "muse-work" / "bass.mid"
293 original_file.unlink() # Remove so we can detect if it's restored.
294
295 result = await _read_tree_async(
296 snapshot_id=snap_id,
297 root=tmp_path,
298 session=muse_cli_db_session,
299 dry_run=True,
300 )
301
302 assert result.dry_run
303 assert "bass.mid" in result.files_written
304 # File must NOT have been written.
305 assert not original_file.exists()
306
307
308 # ---------------------------------------------------------------------------
309 # _read_tree_async — reset flag
310 # ---------------------------------------------------------------------------
311
312
313 @pytest.mark.anyio
314 async def test_read_tree_reset_clears_workdir_first(
315 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
316 ) -> None:
317 """--reset removes files not in the snapshot before populating."""
318 _init_muse_repo(tmp_path)
319 _populate_workdir(tmp_path, {"main.mid": b"MAIN"})
320
321 await _commit_async(
322 message="reset snapshot",
323 root=tmp_path,
324 session=muse_cli_db_session,
325 )
326
327 from maestro.muse_cli.snapshot import build_snapshot_manifest
328 manifest = build_snapshot_manifest(tmp_path / "muse-work")
329 snap_id = compute_snapshot_id(manifest)
330
331 # Add an extra file that is NOT in the snapshot.
332 extra = tmp_path / "muse-work" / "scratch.mid"
333 extra.write_bytes(b"SCRATCH")
334
335 result = await _read_tree_async(
336 snapshot_id=snap_id,
337 root=tmp_path,
338 session=muse_cli_db_session,
339 reset=True,
340 )
341
342 assert result.reset
343 assert "main.mid" in result.files_written
344 # Extra file must be gone after --reset.
345 assert not extra.exists()
346 # Snapshot file must be present.
347 assert (tmp_path / "muse-work" / "main.mid").exists()
348
349
350 @pytest.mark.anyio
351 async def test_read_tree_without_reset_leaves_extra_files(
352 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
353 ) -> None:
354 """Without --reset, files not in the snapshot are left untouched."""
355 _init_muse_repo(tmp_path)
356 _populate_workdir(tmp_path, {"main.mid": b"MAIN"})
357
358 await _commit_async(
359 message="no-reset snapshot",
360 root=tmp_path,
361 session=muse_cli_db_session,
362 )
363
364 from maestro.muse_cli.snapshot import build_snapshot_manifest
365 manifest = build_snapshot_manifest(tmp_path / "muse-work")
366 snap_id = compute_snapshot_id(manifest)
367
368 extra = tmp_path / "muse-work" / "extra.mid"
369 extra.write_bytes(b"EXTRA")
370
371 await _read_tree_async(
372 snapshot_id=snap_id,
373 root=tmp_path,
374 session=muse_cli_db_session,
375 reset=False,
376 )
377
378 # Extra file must still be present.
379 assert extra.exists()
380 assert extra.read_bytes() == b"EXTRA"
381
382
383 # ---------------------------------------------------------------------------
384 # _read_tree_async — error cases
385 # ---------------------------------------------------------------------------
386
387
388 @pytest.mark.anyio
389 async def test_read_tree_unknown_snapshot_id_exits(
390 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
391 ) -> None:
392 """Unknown snapshot ID produces USER_ERROR exit."""
393 _init_muse_repo(tmp_path)
394
395 with pytest.raises(typer.Exit) as exc_info:
396 await _read_tree_async(
397 snapshot_id="0" * 64,
398 root=tmp_path,
399 session=muse_cli_db_session,
400 )
401 assert exc_info.value.exit_code == ExitCode.USER_ERROR
402
403
404 @pytest.mark.anyio
405 async def test_read_tree_short_id_exits(
406 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
407 ) -> None:
408 """Snapshot ID shorter than 4 chars produces USER_ERROR exit."""
409 _init_muse_repo(tmp_path)
410
411 with pytest.raises(typer.Exit) as exc_info:
412 await _read_tree_async(
413 snapshot_id="ab",
414 root=tmp_path,
415 session=muse_cli_db_session,
416 )
417 assert exc_info.value.exit_code == ExitCode.USER_ERROR
418
419
420 @pytest.mark.anyio
421 async def test_read_tree_missing_objects_exits(
422 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
423 ) -> None:
424 """Snapshot with objects missing from local store exits with USER_ERROR.
425
426 This simulates a snapshot whose objects were never committed locally
427 (e.g., the snapshot was pulled from a remote but never locally committed).
428 """
429 from maestro.muse_cli.models import MuseCliSnapshot
430 from maestro.muse_cli.snapshot import compute_snapshot_id
431
432 _init_muse_repo(tmp_path)
433
434 # Manually insert a snapshot that references objects NOT in the store.
435 fake_manifest = {"ghost.mid": "a" * 64}
436 fake_snap_id = compute_snapshot_id(fake_manifest)
437 snap = MuseCliSnapshot(snapshot_id=fake_snap_id, manifest=fake_manifest)
438 muse_cli_db_session.add(snap)
439 await muse_cli_db_session.flush()
440
441 with pytest.raises(typer.Exit) as exc_info:
442 await _read_tree_async(
443 snapshot_id=fake_snap_id,
444 root=tmp_path,
445 session=muse_cli_db_session,
446 )
447 assert exc_info.value.exit_code == ExitCode.USER_ERROR
448
449
450 # ---------------------------------------------------------------------------
451 # Abbreviated snapshot ID resolves correctly end-to-end
452 # ---------------------------------------------------------------------------
453
454
455 @pytest.mark.anyio
456 async def test_read_tree_abbreviated_snapshot_id(
457 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
458 ) -> None:
459 """read-tree accepts abbreviated (≥ 4 char) snapshot IDs."""
460 _init_muse_repo(tmp_path)
461 _populate_workdir(tmp_path, {"keys.mid": b"KEYS-LOOP"})
462
463 await _commit_async(
464 message="abbreviated id test",
465 root=tmp_path,
466 session=muse_cli_db_session,
467 )
468
469 from maestro.muse_cli.snapshot import build_snapshot_manifest
470 manifest = build_snapshot_manifest(tmp_path / "muse-work")
471 snap_id = compute_snapshot_id(manifest)
472
473 (tmp_path / "muse-work" / "keys.mid").unlink()
474
475 result = await _read_tree_async(
476 snapshot_id=snap_id[:8], # 8-char abbreviated ID
477 root=tmp_path,
478 session=muse_cli_db_session,
479 )
480
481 assert result.snapshot_id == snap_id
482 assert "keys.mid" in result.files_written
483 assert (tmp_path / "muse-work" / "keys.mid").read_bytes() == b"KEYS-LOOP"