cgcardona / muse public
test_write_tree.py python
381 lines 12.9 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse write-tree``.
2
3 ``_write_tree_async`` is exercised directly with an in-memory SQLite session
4 (via the ``muse_cli_db_session`` fixture) so no real Postgres instance is
5 needed. The Typer CLI runner covers the command surface (repo detection,
6 exit codes, flag handling).
7
8 All async tests use ``@pytest.mark.anyio``.
9 """
10 from __future__ import annotations
11
12 import json
13 import pathlib
14 import uuid
15
16 import pytest
17 import typer
18 from sqlalchemy.ext.asyncio import AsyncSession
19 from sqlalchemy.future import select
20
21 from maestro.muse_cli.commands.write_tree import _write_tree_async
22 from maestro.muse_cli.errors import ExitCode
23 from maestro.muse_cli.models import MuseCliCommit, MuseCliObject, MuseCliSnapshot
24 from maestro.muse_cli.snapshot import build_snapshot_manifest, compute_snapshot_id
25
26
27 # ---------------------------------------------------------------------------
28 # Helpers
29 # ---------------------------------------------------------------------------
30
31
32 def _init_muse_repo(root: pathlib.Path, repo_id: str | None = None) -> str:
33 """Create a minimal .muse/ layout without any commits."""
34 rid = repo_id or str(uuid.uuid4())
35 muse = root / ".muse"
36 (muse / "refs" / "heads").mkdir(parents=True)
37 (muse / "repo.json").write_text(
38 json.dumps({"repo_id": rid, "schema_version": "1"})
39 )
40 (muse / "HEAD").write_text("refs/heads/main")
41 (muse / "refs" / "heads" / "main").write_text("")
42 return rid
43
44
45 def _populate_workdir(
46 root: pathlib.Path,
47 files: dict[str, bytes] | None = None,
48 subdir: str | None = None,
49 ) -> None:
50 """Create muse-work/ with the given files (defaults to two sample files)."""
51 workdir = root / "muse-work"
52 if subdir:
53 workdir = workdir / subdir
54 workdir.mkdir(parents=True, exist_ok=True)
55 if files is None:
56 files = {"beat.mid": b"MIDI-DATA", "lead.mp3": b"MP3-DATA"}
57 for name, content in files.items():
58 (workdir / name).write_bytes(content)
59
60
61 # ---------------------------------------------------------------------------
62 # Core behaviour
63 # ---------------------------------------------------------------------------
64
65
66 @pytest.mark.anyio
67 async def test_write_tree_returns_snapshot_id(
68 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
69 ) -> None:
70 """_write_tree_async returns a 64-char hex snapshot_id."""
71 _init_muse_repo(tmp_path)
72 _populate_workdir(tmp_path)
73
74 snapshot_id = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
75
76 assert len(snapshot_id) == 64
77 assert all(c in "0123456789abcdef" for c in snapshot_id)
78
79
80 @pytest.mark.anyio
81 async def test_write_tree_idempotent_same_content_same_id(
82 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
83 ) -> None:
84 """Calling write-tree twice on identical content yields the same snapshot_id."""
85 _init_muse_repo(tmp_path)
86 _populate_workdir(tmp_path, {"track.mid": b"CONSTANT"})
87
88 id1 = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
89 id2 = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
90
91 assert id1 == id2
92
93
94 @pytest.mark.anyio
95 async def test_write_tree_different_content_different_id(
96 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
97 ) -> None:
98 """Changing a file produces a different snapshot_id."""
99 _init_muse_repo(tmp_path)
100 _populate_workdir(tmp_path, {"track.mid": b"VERSION1"})
101
102 id1 = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
103
104 (tmp_path / "muse-work" / "track.mid").write_bytes(b"VERSION2")
105
106 id2 = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
107
108 assert id1 != id2
109
110
111 @pytest.mark.anyio
112 async def test_write_tree_snapshot_id_matches_snapshot_module(
113 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
114 ) -> None:
115 """snapshot_id returned by _write_tree_async equals compute_snapshot_id(manifest)."""
116 _init_muse_repo(tmp_path)
117 _populate_workdir(tmp_path, {"a.mid": b"ALPHA", "b.mp3": b"BETA"})
118
119 snapshot_id = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
120
121 manifest = build_snapshot_manifest(tmp_path / "muse-work")
122 expected = compute_snapshot_id(manifest)
123
124 assert snapshot_id == expected
125
126
127 # ---------------------------------------------------------------------------
128 # DB persistence
129 # ---------------------------------------------------------------------------
130
131
132 @pytest.mark.anyio
133 async def test_write_tree_persists_snapshot_row(
134 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
135 ) -> None:
136 """A MuseCliSnapshot row is written to the DB."""
137 _init_muse_repo(tmp_path)
138 _populate_workdir(tmp_path, {"beat.mid": b"MIDI"})
139
140 snapshot_id = await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
141 await muse_cli_db_session.flush()
142
143 snap = await muse_cli_db_session.get(MuseCliSnapshot, snapshot_id)
144 assert snap is not None
145 assert "beat.mid" in snap.manifest
146
147
148 @pytest.mark.anyio
149 async def test_write_tree_persists_object_rows(
150 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
151 ) -> None:
152 """A MuseCliObject row is written for every unique file."""
153 _init_muse_repo(tmp_path)
154 _populate_workdir(tmp_path, {"drums.mid": b"DRUM-BYTES", "bass.mid": b"BASS-BYTES"})
155
156 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
157 await muse_cli_db_session.flush()
158
159 result = await muse_cli_db_session.execute(select(MuseCliObject))
160 objects = result.scalars().all()
161 assert len(objects) == 2
162
163
164 @pytest.mark.anyio
165 async def test_write_tree_objects_are_deduplicated(
166 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
167 ) -> None:
168 """Running write-tree twice does not create duplicate object rows."""
169 _init_muse_repo(tmp_path)
170 _populate_workdir(tmp_path, {"track.mid": b"SHARED-CONTENT"})
171
172 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
173 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
174 await muse_cli_db_session.flush()
175
176 result = await muse_cli_db_session.execute(select(MuseCliObject))
177 objects = result.scalars().all()
178 # Only one unique file → only one object row
179 assert len(objects) == 1
180
181
182 @pytest.mark.anyio
183 async def test_write_tree_does_not_create_commit(
184 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
185 ) -> None:
186 """write-tree must NOT create a MuseCliCommit row."""
187 _init_muse_repo(tmp_path)
188 _populate_workdir(tmp_path)
189
190 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
191 await muse_cli_db_session.flush()
192
193 result = await muse_cli_db_session.execute(select(MuseCliCommit))
194 commits = result.scalars().all()
195 assert len(commits) == 0, "write-tree must not create commit rows"
196
197
198 @pytest.mark.anyio
199 async def test_write_tree_does_not_modify_branch_ref(
200 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
201 ) -> None:
202 """The branch HEAD ref file must be unchanged after write-tree."""
203 _init_muse_repo(tmp_path)
204 _populate_workdir(tmp_path)
205
206 ref_path = tmp_path / ".muse" / "refs" / "heads" / "main"
207 before = ref_path.read_text()
208
209 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
210
211 after = ref_path.read_text()
212 assert before == after, "write-tree must not update the branch HEAD pointer"
213
214
215 # ---------------------------------------------------------------------------
216 # --prefix filter
217 # ---------------------------------------------------------------------------
218
219
220 @pytest.mark.anyio
221 async def test_write_tree_prefix_includes_matching_files_only(
222 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
223 ) -> None:
224 """--prefix restricts the snapshot to files under the given subdirectory."""
225 _init_muse_repo(tmp_path)
226 _populate_workdir(tmp_path, subdir="drums", files={"kick.mid": b"KICK"})
227 _populate_workdir(tmp_path, subdir="bass", files={"bassline.mid": b"BASS"})
228
229 snapshot_id = await _write_tree_async(
230 root=tmp_path,
231 session=muse_cli_db_session,
232 prefix="drums",
233 )
234 await muse_cli_db_session.flush()
235
236 snap = await muse_cli_db_session.get(MuseCliSnapshot, snapshot_id)
237 assert snap is not None
238 assert any("kick.mid" in k for k in snap.manifest.keys())
239 assert not any("bassline.mid" in k for k in snap.manifest.keys())
240
241
242 @pytest.mark.anyio
243 async def test_write_tree_prefix_with_trailing_slash(
244 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
245 ) -> None:
246 """--prefix 'drums/' and --prefix 'drums' produce the same snapshot_id."""
247 _init_muse_repo(tmp_path)
248 _populate_workdir(tmp_path, subdir="drums", files={"hi_hat.mid": b"HH"})
249 _populate_workdir(tmp_path, subdir="bass", files={"bassline.mid": b"BASS"})
250
251 id_without_slash = await _write_tree_async(
252 root=tmp_path, session=muse_cli_db_session, prefix="drums"
253 )
254 id_with_slash = await _write_tree_async(
255 root=tmp_path, session=muse_cli_db_session, prefix="drums/"
256 )
257
258 assert id_without_slash == id_with_slash
259
260
261 @pytest.mark.anyio
262 async def test_write_tree_prefix_no_match_exits_1_by_default(
263 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
264 ) -> None:
265 """--prefix that matches no files exits USER_ERROR when --missing-ok is absent."""
266 _init_muse_repo(tmp_path)
267 _populate_workdir(tmp_path, subdir="drums", files={"kick.mid": b"KICK"})
268
269 with pytest.raises(typer.Exit) as exc_info:
270 await _write_tree_async(
271 root=tmp_path,
272 session=muse_cli_db_session,
273 prefix="nonexistent-subdir",
274 )
275
276 assert exc_info.value.exit_code == ExitCode.USER_ERROR
277
278
279 @pytest.mark.anyio
280 async def test_write_tree_prefix_no_match_missing_ok_succeeds(
281 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
282 ) -> None:
283 """--prefix that matches no files + --missing-ok returns an empty snapshot_id."""
284 _init_muse_repo(tmp_path)
285 _populate_workdir(tmp_path, subdir="drums", files={"kick.mid": b"KICK"})
286
287 snapshot_id = await _write_tree_async(
288 root=tmp_path,
289 session=muse_cli_db_session,
290 prefix="nonexistent-subdir",
291 missing_ok=True,
292 )
293
294 # Empty manifest → deterministic empty snapshot_id
295 expected = compute_snapshot_id({})
296 assert snapshot_id == expected
297
298
299 # ---------------------------------------------------------------------------
300 # --missing-ok flag
301 # ---------------------------------------------------------------------------
302
303
304 @pytest.mark.anyio
305 async def test_write_tree_missing_workdir_exits_1_by_default(
306 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
307 ) -> None:
308 """Missing muse-work/ exits USER_ERROR (1) without --missing-ok."""
309 _init_muse_repo(tmp_path)
310 # Deliberately do NOT create muse-work/
311
312 with pytest.raises(typer.Exit) as exc_info:
313 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
314
315 assert exc_info.value.exit_code == ExitCode.USER_ERROR
316
317
318 @pytest.mark.anyio
319 async def test_write_tree_empty_workdir_exits_1_by_default(
320 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
321 ) -> None:
322 """Empty muse-work/ exits USER_ERROR (1) without --missing-ok."""
323 _init_muse_repo(tmp_path)
324 (tmp_path / "muse-work").mkdir()
325
326 with pytest.raises(typer.Exit) as exc_info:
327 await _write_tree_async(root=tmp_path, session=muse_cli_db_session)
328
329 assert exc_info.value.exit_code == ExitCode.USER_ERROR
330
331
332 @pytest.mark.anyio
333 async def test_write_tree_missing_workdir_missing_ok_returns_empty_snapshot(
334 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
335 ) -> None:
336 """With --missing-ok, absent muse-work/ returns the empty snapshot_id."""
337 _init_muse_repo(tmp_path)
338 # No muse-work/ directory
339
340 snapshot_id = await _write_tree_async(
341 root=tmp_path, session=muse_cli_db_session, missing_ok=True
342 )
343
344 expected = compute_snapshot_id({})
345 assert snapshot_id == expected
346
347
348 @pytest.mark.anyio
349 async def test_write_tree_empty_workdir_missing_ok_returns_empty_snapshot(
350 tmp_path: pathlib.Path, muse_cli_db_session: AsyncSession
351 ) -> None:
352 """With --missing-ok, an empty muse-work/ returns the empty snapshot_id."""
353 _init_muse_repo(tmp_path)
354 (tmp_path / "muse-work").mkdir()
355
356 snapshot_id = await _write_tree_async(
357 root=tmp_path, session=muse_cli_db_session, missing_ok=True
358 )
359
360 expected = compute_snapshot_id({})
361 assert snapshot_id == expected
362
363
364 # ---------------------------------------------------------------------------
365 # CLI surface (repo detection, output format)
366 # ---------------------------------------------------------------------------
367
368
369 def test_write_tree_outside_repo_exits_2(tmp_path: pathlib.Path) -> None:
370 """muse write-tree exits REPO_NOT_FOUND (2) when not inside a Muse repo."""
371 import os
372
373 from typer.testing import CliRunner
374
375 from maestro.muse_cli.app import cli
376
377 runner = CliRunner()
378 # Invoke without setting MUSE_REPO_ROOT so it falls back to cwd discovery.
379 env = {**os.environ, "MUSE_REPO_ROOT": str(tmp_path)}
380 result = runner.invoke(cli, ["write-tree"], env=env, catch_exceptions=False)
381 assert result.exit_code == ExitCode.REPO_NOT_FOUND