test_muse_grep.py
python
| 1 | """Tests for muse grep — pattern search across Muse VCS commits. |
| 2 | |
| 3 | Verifies: |
| 4 | - Pattern matching against commit messages (case-insensitive). |
| 5 | - Pattern matching against branch names. |
| 6 | - Non-matching commits are excluded. |
| 7 | - --commits flag produces one commit ID per line. |
| 8 | - --json flag produces valid JSON array. |
| 9 | - --track / --section / --rhythm-invariant emit future-work warnings. |
| 10 | - Empty history produces a graceful no-commits message. |
| 11 | - Multiple matches across a chain are all returned. |
| 12 | - Boundary seal (AST). |
| 13 | """ |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import ast |
| 17 | import dataclasses |
| 18 | import json |
| 19 | import pathlib |
| 20 | import textwrap |
| 21 | from collections.abc import AsyncGenerator |
| 22 | from datetime import datetime, timezone |
| 23 | from unittest.mock import patch |
| 24 | |
| 25 | import pytest |
| 26 | from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine |
| 27 | |
| 28 | from maestro.db.database import Base |
| 29 | from maestro.muse_cli.commands.grep_cmd import ( |
| 30 | GrepMatch, |
| 31 | _grep_async, |
| 32 | _load_all_commits, |
| 33 | _match_commit, |
| 34 | _render_matches, |
| 35 | ) |
| 36 | from maestro.muse_cli.models import MuseCliCommit, MuseCliSnapshot |
| 37 | |
| 38 | |
| 39 | # --------------------------------------------------------------------------- |
| 40 | # Fixtures |
| 41 | # --------------------------------------------------------------------------- |
| 42 | |
| 43 | |
| 44 | @pytest.fixture |
| 45 | async def async_session() -> AsyncGenerator[AsyncSession, None]: |
| 46 | """In-memory SQLite async session — creates all tables before each test.""" |
| 47 | engine = create_async_engine("sqlite+aiosqlite:///:memory:") |
| 48 | async with engine.begin() as conn: |
| 49 | await conn.run_sync(Base.metadata.create_all) |
| 50 | factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) |
| 51 | async with factory() as session: |
| 52 | yield session |
| 53 | await engine.dispose() |
| 54 | |
| 55 | |
| 56 | def _utc(year: int = 2026, month: int = 1, day: int = 1) -> datetime: |
| 57 | return datetime(year, month, day, tzinfo=timezone.utc) |
| 58 | |
| 59 | |
| 60 | def _snapshot(session: AsyncSession, snap_id: str) -> MuseCliSnapshot: |
| 61 | s = MuseCliSnapshot(snapshot_id=snap_id, manifest={}) |
| 62 | session.add(s) |
| 63 | return s |
| 64 | |
| 65 | |
| 66 | def _commit( |
| 67 | session: AsyncSession, |
| 68 | *, |
| 69 | commit_id: str, |
| 70 | repo_id: str = "repo-1", |
| 71 | branch: str = "main", |
| 72 | message: str = "test commit", |
| 73 | parent_id: str | None = None, |
| 74 | snap_id: str = "snap-0000", |
| 75 | ts: datetime | None = None, |
| 76 | ) -> MuseCliCommit: |
| 77 | c = MuseCliCommit( |
| 78 | commit_id=commit_id, |
| 79 | repo_id=repo_id, |
| 80 | branch=branch, |
| 81 | parent_commit_id=parent_id, |
| 82 | parent2_commit_id=None, |
| 83 | snapshot_id=snap_id, |
| 84 | message=message, |
| 85 | author="test-user", |
| 86 | committed_at=ts or _utc(), |
| 87 | ) |
| 88 | session.add(c) |
| 89 | return c |
| 90 | |
| 91 | |
| 92 | # --------------------------------------------------------------------------- |
| 93 | # Unit tests: _match_commit |
| 94 | # --------------------------------------------------------------------------- |
| 95 | |
| 96 | |
| 97 | def _make_commit_obj( |
| 98 | *, |
| 99 | commit_id: str = "abc12345" * 8, |
| 100 | branch: str = "main", |
| 101 | message: str = "boom bap groove", |
| 102 | ) -> MuseCliCommit: |
| 103 | """Build a MuseCliCommit using its normal constructor (no DB session needed). |
| 104 | |
| 105 | SQLAlchemy ORM models can be instantiated without a session by using the |
| 106 | regular constructor. The instance is transient (not associated with any |
| 107 | session) which is sufficient for testing ``_match_commit``. |
| 108 | """ |
| 109 | return MuseCliCommit( |
| 110 | commit_id=commit_id, |
| 111 | repo_id="repo-1", |
| 112 | branch=branch, |
| 113 | parent_commit_id=None, |
| 114 | parent2_commit_id=None, |
| 115 | snapshot_id="snap-0000" + "0" * 60, |
| 116 | message=message, |
| 117 | author="test-user", |
| 118 | committed_at=_utc(), |
| 119 | ) |
| 120 | |
| 121 | |
| 122 | def test_match_commit_finds_pattern_in_message() -> None: |
| 123 | """Pattern matched in message → GrepMatch with source='message'.""" |
| 124 | c = _make_commit_obj(message="add C4 E4 G4 riff to chorus") |
| 125 | result = _match_commit( |
| 126 | c, |
| 127 | "C4 E4 G4", |
| 128 | track=None, |
| 129 | section=None, |
| 130 | transposition_invariant=True, |
| 131 | rhythm_invariant=False, |
| 132 | ) |
| 133 | assert result is not None |
| 134 | assert result.match_source == "message" |
| 135 | assert result.commit_id == c.commit_id |
| 136 | |
| 137 | |
| 138 | def test_match_commit_case_insensitive() -> None: |
| 139 | """Pattern matching is case-insensitive.""" |
| 140 | c = _make_commit_obj(message="Added CM7 chord voicing") |
| 141 | result = _match_commit( |
| 142 | c, |
| 143 | "cm7", |
| 144 | track=None, |
| 145 | section=None, |
| 146 | transposition_invariant=True, |
| 147 | rhythm_invariant=False, |
| 148 | ) |
| 149 | assert result is not None |
| 150 | |
| 151 | |
| 152 | def test_match_commit_finds_pattern_in_branch() -> None: |
| 153 | """Pattern matched in branch name → GrepMatch with source='branch'.""" |
| 154 | c = _make_commit_obj(branch="feature/pentatonic-scale", message="initial commit") |
| 155 | result = _match_commit( |
| 156 | c, |
| 157 | "pentatonic", |
| 158 | track=None, |
| 159 | section=None, |
| 160 | transposition_invariant=True, |
| 161 | rhythm_invariant=False, |
| 162 | ) |
| 163 | assert result is not None |
| 164 | assert result.match_source == "branch" |
| 165 | |
| 166 | |
| 167 | def test_match_commit_no_match_returns_none() -> None: |
| 168 | """Commit with no pattern occurrence → None.""" |
| 169 | c = _make_commit_obj(message="unrelated commit", branch="main") |
| 170 | result = _match_commit( |
| 171 | c, |
| 172 | "Am7", |
| 173 | track=None, |
| 174 | section=None, |
| 175 | transposition_invariant=True, |
| 176 | rhythm_invariant=False, |
| 177 | ) |
| 178 | assert result is None |
| 179 | |
| 180 | |
| 181 | def test_match_commit_message_takes_priority_over_branch() -> None: |
| 182 | """When message matches, source is 'message' even if branch would also match.""" |
| 183 | c = _make_commit_obj(message="groove pattern", branch="groove-branch") |
| 184 | result = _match_commit( |
| 185 | c, |
| 186 | "groove", |
| 187 | track=None, |
| 188 | section=None, |
| 189 | transposition_invariant=True, |
| 190 | rhythm_invariant=False, |
| 191 | ) |
| 192 | assert result is not None |
| 193 | assert result.match_source == "message" |
| 194 | |
| 195 | |
| 196 | # --------------------------------------------------------------------------- |
| 197 | # Integration tests: _load_all_commits + _grep_async (with real in-memory DB) |
| 198 | # --------------------------------------------------------------------------- |
| 199 | |
| 200 | |
| 201 | @pytest.mark.anyio |
| 202 | async def test_load_all_commits_walks_chain(async_session: AsyncSession) -> None: |
| 203 | """_load_all_commits returns all commits in newest-first order.""" |
| 204 | snap_id = "snap-aaaa" + "0" * 55 |
| 205 | _snapshot(async_session, snap_id[:64]) |
| 206 | c1 = _commit(async_session, commit_id="aaa" + "0" * 61, snap_id=snap_id[:64], message="first") |
| 207 | c2 = _commit( |
| 208 | async_session, |
| 209 | commit_id="bbb" + "0" * 61, |
| 210 | snap_id=snap_id[:64], |
| 211 | parent_id=c1.commit_id, |
| 212 | message="second", |
| 213 | ts=_utc(day=2), |
| 214 | ) |
| 215 | await async_session.commit() |
| 216 | |
| 217 | commits = await _load_all_commits(async_session, head_commit_id=c2.commit_id, limit=100) |
| 218 | assert len(commits) == 2 |
| 219 | assert commits[0].commit_id == c2.commit_id # newest first |
| 220 | assert commits[1].commit_id == c1.commit_id |
| 221 | |
| 222 | |
| 223 | @pytest.mark.anyio |
| 224 | async def test_grep_async_matches_message( |
| 225 | async_session: AsyncSession, tmp_path: pathlib.Path |
| 226 | ) -> None: |
| 227 | """_grep_async finds commits whose messages contain the pattern.""" |
| 228 | # Set up a minimal .muse repo structure |
| 229 | muse_dir = tmp_path / ".muse" |
| 230 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 231 | head_ref = "refs/heads/main" |
| 232 | (muse_dir / "HEAD").write_text(head_ref) |
| 233 | |
| 234 | snap_id = "s" * 64 |
| 235 | _snapshot(async_session, snap_id) |
| 236 | c1 = _commit( |
| 237 | async_session, |
| 238 | commit_id="c" * 64, |
| 239 | snap_id=snap_id, |
| 240 | message="add pentatonic riff", |
| 241 | ts=_utc(day=2), |
| 242 | ) |
| 243 | await async_session.commit() |
| 244 | (muse_dir / head_ref).write_text(c1.commit_id) |
| 245 | |
| 246 | matches = await _grep_async( |
| 247 | root=tmp_path, |
| 248 | session=async_session, |
| 249 | pattern="pentatonic", |
| 250 | track=None, |
| 251 | section=None, |
| 252 | transposition_invariant=True, |
| 253 | rhythm_invariant=False, |
| 254 | show_commits=False, |
| 255 | output_json=False, |
| 256 | ) |
| 257 | assert len(matches) == 1 |
| 258 | assert matches[0].match_source == "message" |
| 259 | assert matches[0].commit_id == c1.commit_id |
| 260 | |
| 261 | |
| 262 | @pytest.mark.anyio |
| 263 | async def test_grep_async_no_matches( |
| 264 | async_session: AsyncSession, tmp_path: pathlib.Path |
| 265 | ) -> None: |
| 266 | """_grep_async returns empty list when pattern is not found.""" |
| 267 | muse_dir = tmp_path / ".muse" |
| 268 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 269 | (muse_dir / "HEAD").write_text("refs/heads/main") |
| 270 | |
| 271 | snap_id = "s" * 64 |
| 272 | _snapshot(async_session, snap_id) |
| 273 | c1 = _commit( |
| 274 | async_session, |
| 275 | commit_id="c" * 64, |
| 276 | snap_id=snap_id, |
| 277 | message="unrelated commit", |
| 278 | ) |
| 279 | await async_session.commit() |
| 280 | (muse_dir / "refs" / "heads" / "main").write_text(c1.commit_id) |
| 281 | |
| 282 | matches = await _grep_async( |
| 283 | root=tmp_path, |
| 284 | session=async_session, |
| 285 | pattern="Cm7", |
| 286 | track=None, |
| 287 | section=None, |
| 288 | transposition_invariant=True, |
| 289 | rhythm_invariant=False, |
| 290 | show_commits=False, |
| 291 | output_json=False, |
| 292 | ) |
| 293 | assert matches == [] |
| 294 | |
| 295 | |
| 296 | @pytest.mark.anyio |
| 297 | async def test_grep_async_empty_history( |
| 298 | async_session: AsyncSession, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str] |
| 299 | ) -> None: |
| 300 | """_grep_async handles branches with no commits gracefully.""" |
| 301 | muse_dir = tmp_path / ".muse" |
| 302 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 303 | (muse_dir / "HEAD").write_text("refs/heads/main") |
| 304 | # No HEAD ref file → no commits |
| 305 | |
| 306 | matches = await _grep_async( |
| 307 | root=tmp_path, |
| 308 | session=async_session, |
| 309 | pattern="anything", |
| 310 | track=None, |
| 311 | section=None, |
| 312 | transposition_invariant=True, |
| 313 | rhythm_invariant=False, |
| 314 | show_commits=False, |
| 315 | output_json=False, |
| 316 | ) |
| 317 | assert matches == [] |
| 318 | |
| 319 | |
| 320 | @pytest.mark.anyio |
| 321 | async def test_grep_async_multiple_matches_across_chain( |
| 322 | async_session: AsyncSession, tmp_path: pathlib.Path |
| 323 | ) -> None: |
| 324 | """All matching commits across a long chain are returned.""" |
| 325 | muse_dir = tmp_path / ".muse" |
| 326 | (muse_dir / "refs" / "heads").mkdir(parents=True) |
| 327 | (muse_dir / "HEAD").write_text("refs/heads/main") |
| 328 | |
| 329 | snap_id = "s" * 64 |
| 330 | _snapshot(async_session, snap_id) |
| 331 | |
| 332 | # 3-commit chain; c1 and c3 match "groove", c2 does not |
| 333 | c1 = _commit(async_session, commit_id="1" * 64, snap_id=snap_id, message="groove intro", ts=_utc(day=1)) |
| 334 | c2 = _commit(async_session, commit_id="2" * 64, snap_id=snap_id, message="bridge section", parent_id=c1.commit_id, ts=_utc(day=2)) |
| 335 | c3 = _commit(async_session, commit_id="3" * 64, snap_id=snap_id, message="add groove variation", parent_id=c2.commit_id, ts=_utc(day=3)) |
| 336 | await async_session.commit() |
| 337 | (muse_dir / "refs" / "heads" / "main").write_text(c3.commit_id) |
| 338 | |
| 339 | matches = await _grep_async( |
| 340 | root=tmp_path, |
| 341 | session=async_session, |
| 342 | pattern="groove", |
| 343 | track=None, |
| 344 | section=None, |
| 345 | transposition_invariant=True, |
| 346 | rhythm_invariant=False, |
| 347 | show_commits=False, |
| 348 | output_json=False, |
| 349 | ) |
| 350 | assert len(matches) == 2 |
| 351 | commit_ids = {m.commit_id for m in matches} |
| 352 | assert c1.commit_id in commit_ids |
| 353 | assert c3.commit_id in commit_ids |
| 354 | assert c2.commit_id not in commit_ids |
| 355 | |
| 356 | |
| 357 | # --------------------------------------------------------------------------- |
| 358 | # Output rendering tests |
| 359 | # --------------------------------------------------------------------------- |
| 360 | |
| 361 | |
| 362 | def test_render_matches_json_output(capsys: pytest.CaptureFixture[str]) -> None: |
| 363 | """--json flag produces a valid JSON array of match dicts.""" |
| 364 | matches = [ |
| 365 | GrepMatch( |
| 366 | commit_id="abc" * 21 + "a", |
| 367 | branch="main", |
| 368 | message="pentatonic solo", |
| 369 | committed_at="2026-01-01T00:00:00+00:00", |
| 370 | match_source="message", |
| 371 | ) |
| 372 | ] |
| 373 | _render_matches(matches, pattern="pentatonic", show_commits=False, output_json=True) |
| 374 | captured = capsys.readouterr() |
| 375 | parsed = json.loads(captured.out) |
| 376 | assert isinstance(parsed, list) |
| 377 | assert len(parsed) == 1 |
| 378 | assert parsed[0]["match_source"] == "message" |
| 379 | assert parsed[0]["branch"] == "main" |
| 380 | |
| 381 | |
| 382 | def test_render_matches_commits_flag(capsys: pytest.CaptureFixture[str]) -> None: |
| 383 | """--commits flag outputs one commit ID per line.""" |
| 384 | commit_ids = ["a" * 64, "b" * 64] |
| 385 | matches = [ |
| 386 | GrepMatch(commit_id=cid, branch="main", message="msg", committed_at="2026-01-01T00:00:00+00:00", match_source="message") |
| 387 | for cid in commit_ids |
| 388 | ] |
| 389 | _render_matches(matches, pattern="msg", show_commits=True, output_json=False) |
| 390 | captured = capsys.readouterr() |
| 391 | lines = captured.out.strip().splitlines() |
| 392 | assert lines == commit_ids |
| 393 | |
| 394 | |
| 395 | def test_render_matches_default_human_output(capsys: pytest.CaptureFixture[str]) -> None: |
| 396 | """Default output includes commit ID, branch, date, match source, and message.""" |
| 397 | matches = [ |
| 398 | GrepMatch( |
| 399 | commit_id="d" * 64, |
| 400 | branch="feature/groove", |
| 401 | message="add groove pattern", |
| 402 | committed_at="2026-02-01T00:00:00+00:00", |
| 403 | match_source="message", |
| 404 | ) |
| 405 | ] |
| 406 | _render_matches(matches, pattern="groove", show_commits=False, output_json=False) |
| 407 | captured = capsys.readouterr() |
| 408 | assert "groove" in captured.out |
| 409 | assert "feature/groove" in captured.out |
| 410 | assert "message" in captured.out |
| 411 | |
| 412 | |
| 413 | def test_render_matches_no_matches_message(capsys: pytest.CaptureFixture[str]) -> None: |
| 414 | """When no matches found, a descriptive message is printed.""" |
| 415 | _render_matches([], pattern="Cm7", show_commits=False, output_json=False) |
| 416 | captured = capsys.readouterr() |
| 417 | assert "Cm7" in captured.out |
| 418 | assert "No commits" in captured.out |
| 419 | |
| 420 | |
| 421 | def test_render_matches_json_empty_list(capsys: pytest.CaptureFixture[str]) -> None: |
| 422 | """--json with no matches outputs an empty JSON array.""" |
| 423 | _render_matches([], pattern="nothing", show_commits=False, output_json=True) |
| 424 | captured = capsys.readouterr() |
| 425 | parsed = json.loads(captured.out) |
| 426 | assert parsed == [] |
| 427 | |
| 428 | |
| 429 | # --------------------------------------------------------------------------- |
| 430 | # GrepMatch dataclass integrity |
| 431 | # --------------------------------------------------------------------------- |
| 432 | |
| 433 | |
| 434 | def test_grep_match_asdict_roundtrip() -> None: |
| 435 | """GrepMatch is a plain dataclass — asdict() should be lossless.""" |
| 436 | m = GrepMatch( |
| 437 | commit_id="x" * 64, |
| 438 | branch="main", |
| 439 | message="test pattern", |
| 440 | committed_at="2026-01-01T00:00:00+00:00", |
| 441 | match_source="message", |
| 442 | ) |
| 443 | d = dataclasses.asdict(m) |
| 444 | assert d["commit_id"] == "x" * 64 |
| 445 | assert d["branch"] == "main" |
| 446 | assert d["match_source"] == "message" |
| 447 | |
| 448 | |
| 449 | # --------------------------------------------------------------------------- |
| 450 | # Boundary seal — AST checks |
| 451 | # --------------------------------------------------------------------------- |
| 452 | |
| 453 | |
| 454 | def test_grep_cmd_module_has_future_annotations() -> None: |
| 455 | """grep_cmd.py must start with 'from __future__ import annotations'.""" |
| 456 | src = pathlib.Path(__file__).parent.parent / "maestro" / "muse_cli" / "commands" / "grep_cmd.py" |
| 457 | tree = ast.parse(src.read_text()) |
| 458 | first_import = next( |
| 459 | (n for n in ast.walk(tree) if isinstance(n, ast.ImportFrom)), |
| 460 | None, |
| 461 | ) |
| 462 | assert first_import is not None |
| 463 | assert first_import.module == "__future__" |
| 464 | names = [a.name for a in first_import.names] |
| 465 | assert "annotations" in names |
| 466 | |
| 467 | |
| 468 | def test_grep_cmd_no_print_statements() -> None: |
| 469 | """grep_cmd.py must not use print() — only logging and typer.echo.""" |
| 470 | src = pathlib.Path(__file__).parent.parent / "maestro" / "muse_cli" / "commands" / "grep_cmd.py" |
| 471 | tree = ast.parse(src.read_text()) |
| 472 | print_calls = [ |
| 473 | n for n in ast.walk(tree) |
| 474 | if isinstance(n, ast.Call) |
| 475 | and isinstance(n.func, ast.Name) |
| 476 | and n.func.id == "print" |
| 477 | ] |
| 478 | assert print_calls == [], "grep_cmd.py must not contain print() calls" |