cgcardona / muse public
test_muse_form.py python
384 lines 12.7 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse form`` -- CLI interface, flag parsing, and stub output format.
2
3 All CLI-level tests use ``typer.testing.CliRunner`` against the full ``muse``
4 app so that argument parsing, flag handling, and exit codes are exercised
5 end-to-end.
6
7 Async core tests call the internal async functions directly with an in-memory
8 SQLite session (the stub does not query the DB; the session satisfies the
9 signature contract only).
10 """
11 from __future__ import annotations
12
13 import json
14 import os
15 import pathlib
16 import uuid
17 from collections.abc import AsyncGenerator
18
19 import pytest
20 import pytest_asyncio
21 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
22 from sqlalchemy.pool import StaticPool
23 from typer.testing import CliRunner
24
25 from maestro.db.database import Base
26 import maestro.muse_cli.models # noqa: F401 -- registers MuseCli* with Base.metadata
27 from maestro.muse_cli.app import cli
28 from maestro.muse_cli.commands.form import (
29 FormAnalysisResult,
30 FormHistoryEntry,
31 FormSection,
32 ROLE_LABELS,
33 _VALID_ROLES,
34 _form_detect_async,
35 _form_history_async,
36 _form_set_async,
37 _render_form_text,
38 _render_history_text,
39 _render_map_text,
40 _sections_to_form_string,
41 _stub_form_sections,
42 )
43 from maestro.muse_cli.errors import ExitCode
44
45 runner = CliRunner()
46
47 # ---------------------------------------------------------------------------
48 # Fixtures
49 # ---------------------------------------------------------------------------
50
51
52 def _init_muse_repo(root: pathlib.Path, branch: str = "main") -> str:
53 """Create a minimal .muse/ layout with one commit ref."""
54 rid = str(uuid.uuid4())
55 muse = root / ".muse"
56 (muse / "refs" / "heads").mkdir(parents=True)
57 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
58 (muse / "HEAD").write_text(f"refs/heads/{branch}")
59 (muse / "refs" / "heads" / branch).write_text(
60 "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2"
61 )
62 return rid
63
64
65 @pytest_asyncio.fixture
66 async def db_session() -> AsyncGenerator[AsyncSession, None]:
67 """In-memory SQLite session (stub form analysis does not query it)."""
68 engine = create_async_engine(
69 "sqlite+aiosqlite:///:memory:",
70 connect_args={"check_same_thread": False},
71 poolclass=StaticPool,
72 )
73 async with engine.begin() as conn:
74 await conn.run_sync(Base.metadata.create_all)
75 factory = async_sessionmaker(bind=engine, expire_on_commit=False)
76 async with factory() as session:
77 yield session
78 async with engine.begin() as conn:
79 await conn.run_sync(Base.metadata.drop_all)
80 await engine.dispose()
81
82
83 # ---------------------------------------------------------------------------
84 # Unit -- stub data
85 # ---------------------------------------------------------------------------
86
87
88 def test_stub_form_sections_returns_sections() -> None:
89 """Stub produces at least one section."""
90 sections = _stub_form_sections()
91 assert len(sections) > 0
92
93
94 def test_stub_form_sections_have_valid_structure() -> None:
95 """Every stub section has label, role, and a sequential index."""
96 sections = _stub_form_sections()
97 for i, sec in enumerate(sections):
98 assert isinstance(sec["label"], str)
99 assert isinstance(sec["role"], str)
100 assert sec["index"] == i
101
102
103 def test_stub_form_sections_include_verse_and_chorus_roles() -> None:
104 """Stub includes verse and chorus roles -- the minimal pop form."""
105 sections = _stub_form_sections()
106 roles = {s["role"] for s in sections}
107 assert "verse" in roles
108 assert "chorus" in roles
109
110
111 def test_sections_to_form_string_pipe_separated() -> None:
112 """_sections_to_form_string produces a pipe-separated label sequence."""
113 sections = [
114 FormSection(label="intro", role="intro", index=0),
115 FormSection(label="A", role="verse", index=1),
116 FormSection(label="B", role="chorus", index=2),
117 ]
118 result = _sections_to_form_string(sections)
119 assert result == "intro | A | B"
120
121
122 def test_role_labels_constant_is_non_empty() -> None:
123 """ROLE_LABELS contains the standard section vocabulary."""
124 assert "verse" in ROLE_LABELS
125 assert "chorus" in ROLE_LABELS
126 assert "bridge" in ROLE_LABELS
127
128
129 def test_valid_roles_frozenset_matches_role_labels() -> None:
130 """_VALID_ROLES is the frozenset of ROLE_LABELS."""
131 assert _VALID_ROLES == frozenset(ROLE_LABELS)
132
133
134 # ---------------------------------------------------------------------------
135 # Unit -- renderers
136 # ---------------------------------------------------------------------------
137
138
139 def test_render_form_text_contains_commit_and_form() -> None:
140 """_render_form_text includes the commit ref and form string."""
141 result = FormAnalysisResult(
142 commit="a1b2c3d4",
143 branch="main",
144 form_string="intro | A | B | outro",
145 sections=[
146 FormSection(label="intro", role="intro", index=0),
147 FormSection(label="A", role="verse", index=1),
148 FormSection(label="B", role="chorus", index=2),
149 FormSection(label="outro", role="outro", index=3),
150 ],
151 source="stub",
152 )
153 text = _render_form_text(result)
154 assert "a1b2c3d4" in text
155 assert "intro | A | B | outro" in text
156 assert "Sections:" in text
157
158
159 def test_render_map_text_contains_commit() -> None:
160 """_render_map_text includes the commit ref."""
161 result = FormAnalysisResult(
162 commit="deadbeef",
163 branch="main",
164 form_string="A | B | A",
165 sections=[
166 FormSection(label="A", role="verse", index=0),
167 FormSection(label="B", role="chorus", index=1),
168 FormSection(label="A", role="verse", index=2),
169 ],
170 source="stub",
171 )
172 text = _render_map_text(result)
173 assert "deadbeef" in text
174 assert "A" in text
175 assert "B" in text
176
177
178 def test_render_history_text_formats_entries() -> None:
179 """_render_history_text shows position, commit, and form string."""
180 entry = FormHistoryEntry(
181 position=1,
182 result=FormAnalysisResult(
183 commit="abc123",
184 branch="main",
185 form_string="A | B | A",
186 sections=[],
187 source="stub",
188 ),
189 )
190 text = _render_history_text([entry])
191 assert "#1" in text
192 assert "abc123" in text
193 assert "A | B | A" in text
194
195
196 def test_render_history_text_empty_returns_placeholder() -> None:
197 """_render_history_text with no entries returns a placeholder string."""
198 text = _render_history_text([])
199 assert "no form history" in text.lower()
200
201
202 # ---------------------------------------------------------------------------
203 # Async core -- _form_detect_async (test_form_detects_verse_chorus_structure)
204 # ---------------------------------------------------------------------------
205
206
207 @pytest.mark.anyio
208 async def test_form_detects_verse_chorus_structure(
209 tmp_path: pathlib.Path,
210 db_session: AsyncSession,
211 ) -> None:
212 """_form_detect_async returns a result with verse and chorus in the stub form."""
213 _init_muse_repo(tmp_path)
214 result = await _form_detect_async(
215 root=tmp_path, session=db_session, commit=None
216 )
217 assert result["source"] == "stub"
218 roles = {s["role"] for s in result["sections"]}
219 assert "verse" in roles
220 assert "chorus" in roles
221 assert " | " in result["form_string"]
222
223
224 @pytest.mark.anyio
225 async def test_form_detect_resolves_commit_ref(
226 tmp_path: pathlib.Path,
227 db_session: AsyncSession,
228 ) -> None:
229 """_form_detect_async includes the abbreviated commit SHA in the result."""
230 _init_muse_repo(tmp_path)
231 result = await _form_detect_async(
232 root=tmp_path, session=db_session, commit=None
233 )
234 assert result["commit"] == "a1b2c3d4"
235 assert result["branch"] == "main"
236
237
238 @pytest.mark.anyio
239 async def test_form_detect_explicit_commit_ref(
240 tmp_path: pathlib.Path,
241 db_session: AsyncSession,
242 ) -> None:
243 """An explicit commit ref is preserved in the result unchanged."""
244 _init_muse_repo(tmp_path)
245 result = await _form_detect_async(
246 root=tmp_path, session=db_session, commit="deadbeef"
247 )
248 assert result["commit"] == "deadbeef"
249
250
251 @pytest.mark.anyio
252 async def test_form_detect_json_serializable(
253 tmp_path: pathlib.Path,
254 db_session: AsyncSession,
255 ) -> None:
256 """FormAnalysisResult returned by detect is fully JSON-serialisable."""
257 _init_muse_repo(tmp_path)
258 result = await _form_detect_async(
259 root=tmp_path, session=db_session, commit=None
260 )
261 serialised = json.dumps(dict(result))
262 parsed = json.loads(serialised)
263 assert parsed["form_string"] == result["form_string"]
264 assert isinstance(parsed["sections"], list)
265
266
267 # ---------------------------------------------------------------------------
268 # Async core -- _form_set_async (test_form_set_stores_annotation)
269 # ---------------------------------------------------------------------------
270
271
272 @pytest.mark.anyio
273 async def test_form_set_stores_annotation(
274 tmp_path: pathlib.Path,
275 db_session: AsyncSession,
276 ) -> None:
277 """_form_set_async persists the annotation to .muse/form_annotation.json."""
278 _init_muse_repo(tmp_path)
279 result = await _form_set_async(
280 root=tmp_path, session=db_session, form_value="verse-chorus-verse-chorus"
281 )
282 assert result["source"] == "annotation"
283 annotation_path = tmp_path / ".muse" / "form_annotation.json"
284 assert annotation_path.exists()
285 data = json.loads(annotation_path.read_text())
286 assert data["source"] == "annotation"
287 assert "verse" in data["form_string"].lower()
288 assert "chorus" in data["form_string"].lower()
289
290
291 @pytest.mark.anyio
292 async def test_form_set_pipe_separated_form(
293 tmp_path: pathlib.Path,
294 db_session: AsyncSession,
295 ) -> None:
296 """_form_set_async parses pipe-separated form strings correctly."""
297 _init_muse_repo(tmp_path)
298 result = await _form_set_async(
299 root=tmp_path,
300 session=db_session,
301 form_value="Intro | A | B | A | Bridge | A | Outro",
302 )
303 assert result["source"] == "annotation"
304 labels = [s["label"] for s in result["sections"]]
305 assert labels == ["Intro", "A", "B", "A", "Bridge", "A", "Outro"]
306
307
308 @pytest.mark.anyio
309 async def test_form_set_aaba_shorthand(
310 tmp_path: pathlib.Path,
311 db_session: AsyncSession,
312 ) -> None:
313 """_form_set_async handles space-separated shorthand like 'A A B A'."""
314 _init_muse_repo(tmp_path)
315 result = await _form_set_async(
316 root=tmp_path, session=db_session, form_value="A A B A"
317 )
318 labels = [s["label"] for s in result["sections"]]
319 assert labels == ["A", "A", "B", "A"]
320 assert result["form_string"] == "A | A | B | A"
321
322
323 # ---------------------------------------------------------------------------
324 # Async core -- _form_history_async (test_form_history_shows_restructure)
325 # ---------------------------------------------------------------------------
326
327
328 @pytest.mark.anyio
329 async def test_form_history_shows_restructure(
330 tmp_path: pathlib.Path,
331 db_session: AsyncSession,
332 ) -> None:
333 """_form_history_async returns at least one entry with form data."""
334 _init_muse_repo(tmp_path)
335 entries = await _form_history_async(root=tmp_path, session=db_session)
336 assert len(entries) >= 1
337 first = entries[0]
338 assert first["position"] == 1
339 assert " | " in first["result"]["form_string"]
340
341
342 @pytest.mark.anyio
343 async def test_form_history_entry_has_commit_and_branch(
344 tmp_path: pathlib.Path,
345 db_session: AsyncSession,
346 ) -> None:
347 """History entries carry commit and branch metadata."""
348 _init_muse_repo(tmp_path)
349 entries = await _form_history_async(root=tmp_path, session=db_session)
350 first = entries[0]
351 assert first["result"]["commit"] != ""
352 assert first["result"]["branch"] == "main"
353
354
355 # ---------------------------------------------------------------------------
356 # CLI integration -- CliRunner
357 # ---------------------------------------------------------------------------
358
359
360 def test_cli_form_outside_repo_exits_2(tmp_path: pathlib.Path) -> None:
361 """``muse form`` exits 2 when invoked outside a Muse repository."""
362 prev = os.getcwd()
363 try:
364 os.chdir(tmp_path)
365 result = runner.invoke(cli, ["form"], catch_exceptions=False)
366 finally:
367 os.chdir(prev)
368 assert result.exit_code == int(ExitCode.REPO_NOT_FOUND)
369 assert "not a muse repository" in result.output.lower()
370
371
372 def test_cli_form_help_lists_flags() -> None:
373 """``muse form --help`` shows all documented flags."""
374 result = runner.invoke(cli, ["form", "--help"])
375 assert result.exit_code == 0
376 for flag in ("--set", "--detect", "--map", "--history", "--json"):
377 assert flag in result.output, f"Flag '{flag}' not found in help"
378
379
380 def test_cli_form_appears_in_muse_help() -> None:
381 """``muse --help`` lists the form subcommand."""
382 result = runner.invoke(cli, ["--help"])
383 assert result.exit_code == 0
384 assert "form" in result.output