cgcardona / muse public
test_muse_similarity.py python
518 lines 17.2 KB
Raw
12901c5a Initial extraction from tellurstori/maestro cgcardona <gabriel@tellurstori.com> 4d ago
1 """Tests for ``muse similarity`` — CLI interface, flag parsing, and stub output.
2
3 All CLI-level tests use ``typer.testing.CliRunner`` against the full ``muse``
4 app so argument parsing, flag handling, and exit codes are exercised end-to-end.
5
6 Async core tests call ``_similarity_async`` directly with an in-memory SQLite
7 session. The stub does not query the DB, so the session is injected only to
8 satisfy the signature contract.
9 """
10 from __future__ import annotations
11
12 import json
13 import os
14 import pathlib
15 import uuid
16 from collections.abc import AsyncGenerator
17
18 import pytest
19 import pytest_asyncio
20 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
21 from sqlalchemy.pool import StaticPool
22 from typer.testing import CliRunner
23
24 from maestro.db.database import Base
25 import maestro.muse_cli.models # noqa: F401 — registers MuseCli* with Base.metadata
26 from maestro.muse_cli.app import cli
27 from maestro.muse_cli.commands.similarity import (
28 DIMENSION_NAMES,
29 DimensionScore,
30 SimilarityResult,
31 _ALL_DIMENSIONS,
32 _bar,
33 _max_divergence_dimension,
34 _overall_label,
35 _similarity_async,
36 _stub_dimension_scores,
37 _weighted_overall,
38 build_similarity_result,
39 render_similarity_json,
40 render_similarity_text,
41 )
42 from maestro.muse_cli.errors import ExitCode
43
44 runner = CliRunner()
45
46 # ---------------------------------------------------------------------------
47 # Fixtures
48 # ---------------------------------------------------------------------------
49
50
51 def _init_muse_repo(root: pathlib.Path, branch: str = "main") -> str:
52 """Create a minimal .muse/ layout required by _similarity_async."""
53 rid = str(uuid.uuid4())
54 muse = root / ".muse"
55 (muse / "refs" / "heads").mkdir(parents=True)
56 (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"}))
57 (muse / "HEAD").write_text(f"refs/heads/{branch}")
58 (muse / "refs" / "heads" / branch).write_text("a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2")
59 return rid
60
61
62 @pytest_asyncio.fixture
63 async def db_session() -> AsyncGenerator[AsyncSession, None]:
64 """In-memory SQLite session — stub similarity does not query the DB."""
65 engine = create_async_engine(
66 "sqlite+aiosqlite:///:memory:",
67 connect_args={"check_same_thread": False},
68 poolclass=StaticPool,
69 )
70 async with engine.begin() as conn:
71 await conn.run_sync(Base.metadata.create_all)
72 factory = async_sessionmaker(bind=engine, expire_on_commit=False)
73 async with factory() as session:
74 yield session
75 async with engine.begin() as conn:
76 await conn.run_sync(Base.metadata.drop_all)
77 await engine.dispose()
78
79
80 # ---------------------------------------------------------------------------
81 # Unit — helpers
82 # ---------------------------------------------------------------------------
83
84
85 def test_bar_full_score() -> None:
86 """A score of 1.0 produces a fully filled bar."""
87 bar = _bar(1.0, width=10)
88 assert bar == "\u2588" * 10
89
90
91 def test_bar_zero_score() -> None:
92 """A score of 0.0 produces an empty bar."""
93 bar = _bar(0.0, width=10)
94 assert bar == "\u2591" * 10
95
96
97 def test_bar_half_score() -> None:
98 """A score of 0.5 produces a half-filled bar."""
99 bar = _bar(0.5, width=10)
100 assert bar.count("\u2588") == 5
101 assert bar.count("\u2591") == 5
102 assert len(bar) == 10
103
104
105 def test_overall_label_nearly_identical() -> None:
106 assert _overall_label(0.95) == "Nearly identical — minimal change"
107
108
109 def test_overall_label_significantly_different() -> None:
110 assert _overall_label(0.45) == "Significantly different — major rework"
111
112
113 def test_overall_label_completely_different() -> None:
114 assert _overall_label(0.0) == "Completely different — new direction"
115
116
117 def test_max_divergence_returns_lowest() -> None:
118 scores = [
119 DimensionScore(dimension="harmonic", score=0.45, note=""),
120 DimensionScore(dimension="rhythmic", score=0.89, note=""),
121 DimensionScore(dimension="dynamic", score=0.55, note=""),
122 ]
123 assert _max_divergence_dimension(scores) == "harmonic"
124
125
126 def test_max_divergence_empty_returns_empty() -> None:
127 assert _max_divergence_dimension([]) == ""
128
129
130 def test_weighted_overall_all_dimensions() -> None:
131 """Weighted average across all five dimensions is within [0, 1]."""
132 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
133 overall = _weighted_overall(scores)
134 assert 0.0 <= overall <= 1.0
135
136
137 def test_weighted_overall_empty() -> None:
138 assert _weighted_overall([]) == 0.0
139
140
141 def test_weighted_overall_single_dimension() -> None:
142 scores = [DimensionScore(dimension="harmonic", score=0.6, note="")]
143 result = _weighted_overall(scores)
144 assert result == 0.6
145
146
147 def test_stub_dimension_scores_all() -> None:
148 """Stub returns all five dimensions in DIMENSION_NAMES order."""
149 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
150 assert len(scores) == 5
151 assert [s["dimension"] for s in scores] == list(DIMENSION_NAMES)
152
153
154 def test_stub_dimension_scores_subset() -> None:
155 """Subset filter returns only the requested dimensions."""
156 subset = frozenset({"harmonic", "rhythmic"})
157 scores = _stub_dimension_scores(subset)
158 dims = {s["dimension"] for s in scores}
159 assert dims == subset
160
161
162 def test_stub_dimension_scores_in_range() -> None:
163 """Every stub score is in [0.0, 1.0]."""
164 for s in _stub_dimension_scores(_ALL_DIMENSIONS):
165 assert 0.0 <= s["score"] <= 1.0
166
167
168 # ---------------------------------------------------------------------------
169 # Unit — build_similarity_result
170 # ---------------------------------------------------------------------------
171
172
173 def test_build_similarity_result_structure() -> None:
174 """build_similarity_result returns a fully-populated SimilarityResult."""
175 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
176 result = build_similarity_result("HEAD~10", "HEAD", scores)
177 assert result["commit_a"] == "HEAD~10"
178 assert result["commit_b"] == "HEAD"
179 assert isinstance(result["dimensions"], list)
180 assert 0.0 <= result["overall"] <= 1.0
181 assert isinstance(result["label"], str)
182 assert result["max_divergence"] in _ALL_DIMENSIONS
183
184
185 def test_build_similarity_result_max_divergence_is_lowest() -> None:
186 """max_divergence should match the dimension with the lowest score."""
187 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
188 result = build_similarity_result("a", "b", scores)
189 min_score = min(s["score"] for s in scores)
190 lowest_dim = next(s["dimension"] for s in scores if s["score"] == min_score)
191 assert result["max_divergence"] == lowest_dim
192
193
194 # ---------------------------------------------------------------------------
195 # Unit — renderers
196 # ---------------------------------------------------------------------------
197
198
199 def test_render_similarity_text_contains_commit_refs() -> None:
200 """Text output includes both commit refs in the header."""
201 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
202 result = build_similarity_result("HEAD~10", "HEAD", scores)
203 text = render_similarity_text(result)
204 assert "HEAD~10" in text
205 assert "HEAD" in text
206
207
208 def test_render_similarity_text_contains_all_dimensions() -> None:
209 """Text output mentions all five dimension names."""
210 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
211 result = build_similarity_result("a", "b", scores)
212 text = render_similarity_text(result)
213 for dim in DIMENSION_NAMES:
214 assert dim in text.lower()
215
216
217 def test_render_similarity_text_contains_overall() -> None:
218 """Text output includes the overall score and label."""
219 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
220 result = build_similarity_result("a", "b", scores)
221 text = render_similarity_text(result)
222 assert "Overall" in text
223 assert result["label"] in text
224
225
226 def test_render_similarity_json_is_valid() -> None:
227 """JSON output is parseable and contains the expected top-level keys."""
228 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
229 result = build_similarity_result("HEAD~5", "HEAD", scores)
230 raw = render_similarity_json(result)
231 payload = json.loads(raw)
232 assert payload["commit_a"] == "HEAD~5"
233 assert payload["commit_b"] == "HEAD"
234 assert isinstance(payload["dimensions"], list)
235 assert isinstance(payload["overall"], float)
236 assert isinstance(payload["label"], str)
237 assert "max_divergence" in payload
238
239
240 def test_render_similarity_json_dimensions_structure() -> None:
241 """Each dimension entry in JSON has dimension, score, and note fields."""
242 scores = _stub_dimension_scores(_ALL_DIMENSIONS)
243 result = build_similarity_result("a", "b", scores)
244 payload = json.loads(render_similarity_json(result))
245 for entry in payload["dimensions"]:
246 assert "dimension" in entry
247 assert "score" in entry
248 assert "note" in entry
249
250
251 # ---------------------------------------------------------------------------
252 # Regression: test_muse_similarity_returns_per_dimension_scores
253 # ---------------------------------------------------------------------------
254
255
256 @pytest.mark.anyio
257 async def test_muse_similarity_returns_per_dimension_scores(
258 tmp_path: pathlib.Path,
259 db_session: AsyncSession,
260 capsys: pytest.CaptureFixture[str],
261 ) -> None:
262 """_similarity_async with no filters produces all five dimension scores."""
263 _init_muse_repo(tmp_path)
264 exit_code = await _similarity_async(
265 root=tmp_path,
266 session=db_session,
267 commit_a="HEAD~10",
268 commit_b="HEAD",
269 dimensions=_ALL_DIMENSIONS,
270 section=None,
271 track=None,
272 threshold=None,
273 as_json=False,
274 )
275 assert exit_code == int(ExitCode.SUCCESS)
276 out = capsys.readouterr().out
277 assert "HEAD~10" in out
278 for dim in DIMENSION_NAMES:
279 assert dim in out.lower()
280 assert "Overall" in out
281
282
283 # ---------------------------------------------------------------------------
284 # Unit: test_muse_similarity_dimensions_flag_filters_output
285 # ---------------------------------------------------------------------------
286
287
288 @pytest.mark.anyio
289 async def test_muse_similarity_dimensions_flag_filters_output(
290 tmp_path: pathlib.Path,
291 db_session: AsyncSession,
292 capsys: pytest.CaptureFixture[str],
293 ) -> None:
294 """--dimensions harmonic,rhythmic shows only those two dimensions."""
295 _init_muse_repo(tmp_path)
296 subset = frozenset({"harmonic", "rhythmic"})
297 exit_code = await _similarity_async(
298 root=tmp_path,
299 session=db_session,
300 commit_a="a1b2c3d4",
301 commit_b="e5f6a7b8",
302 dimensions=subset,
303 section=None,
304 track=None,
305 threshold=None,
306 as_json=False,
307 )
308 assert exit_code == int(ExitCode.SUCCESS)
309 out = capsys.readouterr().out
310 assert "harmonic" in out.lower()
311 assert "rhythmic" in out.lower()
312 assert "melodic" not in out.lower()
313 assert "structural" not in out.lower()
314 assert "dynamic" not in out.lower()
315
316
317 # ---------------------------------------------------------------------------
318 # Unit: test_muse_similarity_threshold_exits_nonzero_when_below
319 # ---------------------------------------------------------------------------
320
321
322 @pytest.mark.anyio
323 async def test_muse_similarity_threshold_exits_nonzero_when_below(
324 tmp_path: pathlib.Path,
325 db_session: AsyncSession,
326 capsys: pytest.CaptureFixture[str],
327 ) -> None:
328 """--threshold exits 1 when overall similarity is below the threshold."""
329 _init_muse_repo(tmp_path)
330 # Stub overall is ~0.65; threshold of 0.99 forces a non-zero exit.
331 exit_code = await _similarity_async(
332 root=tmp_path,
333 session=db_session,
334 commit_a="a",
335 commit_b="b",
336 dimensions=_ALL_DIMENSIONS,
337 section=None,
338 track=None,
339 threshold=0.99,
340 as_json=False,
341 )
342 assert exit_code == 1
343
344
345 @pytest.mark.anyio
346 async def test_muse_similarity_threshold_exits_zero_when_above(
347 tmp_path: pathlib.Path,
348 db_session: AsyncSession,
349 capsys: pytest.CaptureFixture[str],
350 ) -> None:
351 """--threshold exits 0 when overall similarity meets or exceeds threshold."""
352 _init_muse_repo(tmp_path)
353 # Stub overall is ~0.65; threshold of 0.5 should pass.
354 exit_code = await _similarity_async(
355 root=tmp_path,
356 session=db_session,
357 commit_a="a",
358 commit_b="b",
359 dimensions=_ALL_DIMENSIONS,
360 section=None,
361 track=None,
362 threshold=0.5,
363 as_json=False,
364 )
365 assert exit_code == int(ExitCode.SUCCESS)
366
367
368 # ---------------------------------------------------------------------------
369 # Unit: test_muse_similarity_json_output
370 # ---------------------------------------------------------------------------
371
372
373 @pytest.mark.anyio
374 async def test_muse_similarity_json_output(
375 tmp_path: pathlib.Path,
376 db_session: AsyncSession,
377 capsys: pytest.CaptureFixture[str],
378 ) -> None:
379 """--json produces valid JSON with expected top-level fields."""
380 _init_muse_repo(tmp_path)
381 exit_code = await _similarity_async(
382 root=tmp_path,
383 session=db_session,
384 commit_a="HEAD~5",
385 commit_b="HEAD",
386 dimensions=_ALL_DIMENSIONS,
387 section=None,
388 track=None,
389 threshold=None,
390 as_json=True,
391 )
392 assert exit_code == int(ExitCode.SUCCESS)
393 raw = capsys.readouterr().out
394 payload = json.loads(raw)
395 assert payload["commit_a"] == "HEAD~5"
396 assert payload["commit_b"] == "HEAD"
397 assert len(payload["dimensions"]) == 5
398 assert 0.0 <= payload["overall"] <= 1.0
399
400
401 @pytest.mark.anyio
402 async def test_muse_similarity_section_flag_warns(
403 tmp_path: pathlib.Path,
404 db_session: AsyncSession,
405 capsys: pytest.CaptureFixture[str],
406 ) -> None:
407 """--section emits a stub warning but still produces output."""
408 _init_muse_repo(tmp_path)
409 exit_code = await _similarity_async(
410 root=tmp_path,
411 session=db_session,
412 commit_a="a",
413 commit_b="b",
414 dimensions=_ALL_DIMENSIONS,
415 section="verse",
416 track=None,
417 threshold=None,
418 as_json=False,
419 )
420 assert exit_code == int(ExitCode.SUCCESS)
421 out = capsys.readouterr().out
422 assert "section" in out.lower()
423
424
425 @pytest.mark.anyio
426 async def test_muse_similarity_track_flag_warns(
427 tmp_path: pathlib.Path,
428 db_session: AsyncSession,
429 capsys: pytest.CaptureFixture[str],
430 ) -> None:
431 """--track emits a stub warning but still produces output."""
432 _init_muse_repo(tmp_path)
433 exit_code = await _similarity_async(
434 root=tmp_path,
435 session=db_session,
436 commit_a="a",
437 commit_b="b",
438 dimensions=_ALL_DIMENSIONS,
439 section=None,
440 track="bass",
441 threshold=None,
442 as_json=False,
443 )
444 assert exit_code == int(ExitCode.SUCCESS)
445 out = capsys.readouterr().out
446 assert "track" in out.lower()
447
448
449 # ---------------------------------------------------------------------------
450 # CLI integration — CliRunner
451 # ---------------------------------------------------------------------------
452
453
454 def test_cli_similarity_outside_repo_exits_2(tmp_path: pathlib.Path) -> None:
455 """``muse similarity`` exits 2 when not inside a Muse repository."""
456 prev = os.getcwd()
457 try:
458 os.chdir(tmp_path)
459 result = runner.invoke(cli, ["similarity", "HEAD~1", "HEAD"], catch_exceptions=False)
460 finally:
461 os.chdir(prev)
462 assert result.exit_code == int(ExitCode.REPO_NOT_FOUND)
463 assert "not a muse repository" in result.output.lower()
464
465
466 def test_cli_similarity_help_lists_flags() -> None:
467 """``muse similarity --help`` shows all documented flags."""
468 result = runner.invoke(cli, ["similarity", "--help"])
469 assert result.exit_code == 0
470 for flag in ("--dimensions", "--section", "--track", "--json", "--threshold"):
471 assert flag in result.output, f"Flag '{flag}' not found in help output"
472
473
474 def test_cli_similarity_appears_in_muse_help() -> None:
475 """``muse --help`` lists the similarity subcommand."""
476 result = runner.invoke(cli, ["--help"])
477 assert result.exit_code == 0
478 assert "similarity" in result.output
479
480
481 def test_cli_similarity_invalid_dimension_exits_1(tmp_path: pathlib.Path) -> None:
482 """``muse similarity --dimensions badvalue`` exits USER_ERROR before repo detection.
483
484 Options must precede positional arguments to satisfy Click/Typer parsing
485 in nested callback groups (known behavior with invoke_without_command=True).
486 Flag validation runs before require_repo(), so no .muse/ dir is needed.
487 """
488 prev = os.getcwd()
489 try:
490 os.chdir(tmp_path)
491 result = runner.invoke(
492 cli,
493 ["similarity", "--dimensions", "badvalue", "HEAD~1", "HEAD"],
494 catch_exceptions=False,
495 )
496 finally:
497 os.chdir(prev)
498 assert result.exit_code == int(ExitCode.USER_ERROR)
499
500
501 def test_cli_similarity_threshold_out_of_range_exits_1(tmp_path: pathlib.Path) -> None:
502 """``muse similarity --threshold 2.0`` exits USER_ERROR before repo detection.
503
504 Options must precede positional arguments to satisfy Click/Typer parsing
505 in nested callback groups (known behavior with invoke_without_command=True).
506 Flag validation runs before require_repo(), so no .muse/ dir is needed.
507 """
508 prev = os.getcwd()
509 try:
510 os.chdir(tmp_path)
511 result = runner.invoke(
512 cli,
513 ["similarity", "--threshold", "2.0", "HEAD~1", "HEAD"],
514 catch_exceptions=False,
515 )
516 finally:
517 os.chdir(prev)
518 assert result.exit_code == int(ExitCode.USER_ERROR)