test_core_query_engine.py
python
| 1 | """Tests for the generic query engine in muse/core/query_engine.py.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import datetime |
| 5 | import pathlib |
| 6 | import tempfile |
| 7 | |
| 8 | import pytest |
| 9 | |
| 10 | from muse.core.query_engine import QueryMatch, format_matches, walk_history |
| 11 | from muse.core.store import CommitRecord, write_commit |
| 12 | |
| 13 | |
| 14 | # --------------------------------------------------------------------------- |
| 15 | # Helpers |
| 16 | # --------------------------------------------------------------------------- |
| 17 | |
| 18 | |
| 19 | def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 20 | """Set up a minimal .muse/ structure for query_engine tests.""" |
| 21 | muse = tmp_path / ".muse" |
| 22 | muse.mkdir() |
| 23 | (muse / "repo.json").write_text('{"repo_id":"test-repo"}') |
| 24 | (muse / "HEAD").write_text("refs/heads/main") |
| 25 | (muse / "commits").mkdir() |
| 26 | (muse / "snapshots").mkdir() |
| 27 | (muse / "refs" / "heads").mkdir(parents=True) |
| 28 | return tmp_path |
| 29 | |
| 30 | |
| 31 | def _write_commit(root: pathlib.Path, commit_id: str, parent_id: str | None = None) -> CommitRecord: |
| 32 | record = CommitRecord( |
| 33 | commit_id=commit_id, |
| 34 | repo_id="test-repo", |
| 35 | branch="main", |
| 36 | snapshot_id="snap-" + commit_id, |
| 37 | message=f"commit {commit_id}", |
| 38 | committed_at=datetime.datetime.now(datetime.timezone.utc), |
| 39 | parent_commit_id=parent_id, |
| 40 | author="test-author", |
| 41 | ) |
| 42 | write_commit(root, record) |
| 43 | return record |
| 44 | |
| 45 | |
| 46 | # --------------------------------------------------------------------------- |
| 47 | # walk_history |
| 48 | # --------------------------------------------------------------------------- |
| 49 | |
| 50 | |
| 51 | class TestWalkHistory: |
| 52 | def test_empty_branch_returns_empty(self) -> None: |
| 53 | with tempfile.TemporaryDirectory() as tmp: |
| 54 | root = _make_repo(pathlib.Path(tmp)) |
| 55 | results = walk_history(root, "main", lambda c, m, r: []) |
| 56 | assert results == [] |
| 57 | |
| 58 | def test_single_commit_visited(self) -> None: |
| 59 | with tempfile.TemporaryDirectory() as tmp: |
| 60 | root = _make_repo(pathlib.Path(tmp)) |
| 61 | _write_commit(root, "aaa111") |
| 62 | (root / ".muse" / "refs" / "heads" / "main").write_text("aaa111") |
| 63 | |
| 64 | visited: list[str] = [] |
| 65 | |
| 66 | def evaluator(commit: CommitRecord, manifest: dict[str, str], r: pathlib.Path) -> list[QueryMatch]: |
| 67 | visited.append(commit.commit_id) |
| 68 | return [] |
| 69 | |
| 70 | walk_history(root, "main", evaluator) |
| 71 | assert visited == ["aaa111"] |
| 72 | |
| 73 | def test_chain_walked_newest_first(self) -> None: |
| 74 | with tempfile.TemporaryDirectory() as tmp: |
| 75 | root = _make_repo(pathlib.Path(tmp)) |
| 76 | _write_commit(root, "aaa111") |
| 77 | _write_commit(root, "bbb222", parent_id="aaa111") |
| 78 | (root / ".muse" / "refs" / "heads" / "main").write_text("bbb222") |
| 79 | |
| 80 | visited: list[str] = [] |
| 81 | |
| 82 | def evaluator(commit: CommitRecord, manifest: dict[str, str], r: pathlib.Path) -> list[QueryMatch]: |
| 83 | visited.append(commit.commit_id) |
| 84 | return [] |
| 85 | |
| 86 | walk_history(root, "main", evaluator) |
| 87 | assert visited == ["bbb222", "aaa111"] |
| 88 | |
| 89 | def test_matches_collected(self) -> None: |
| 90 | with tempfile.TemporaryDirectory() as tmp: |
| 91 | root = _make_repo(pathlib.Path(tmp)) |
| 92 | _write_commit(root, "ccc333") |
| 93 | (root / ".muse" / "refs" / "heads" / "main").write_text("ccc333") |
| 94 | |
| 95 | def evaluator(commit: CommitRecord, manifest: dict[str, str], r: pathlib.Path) -> list[QueryMatch]: |
| 96 | return [QueryMatch( |
| 97 | commit_id=commit.commit_id, |
| 98 | author=commit.author, |
| 99 | committed_at=commit.committed_at.isoformat(), |
| 100 | branch=commit.branch, |
| 101 | detail="test match", |
| 102 | extra={}, |
| 103 | )] |
| 104 | |
| 105 | results = walk_history(root, "main", evaluator) |
| 106 | assert len(results) == 1 |
| 107 | assert results[0]["detail"] == "test match" |
| 108 | |
| 109 | def test_max_commits_limits_walk(self) -> None: |
| 110 | with tempfile.TemporaryDirectory() as tmp: |
| 111 | root = _make_repo(pathlib.Path(tmp)) |
| 112 | ids = [f"commit{i:03d}" for i in range(10)] |
| 113 | for i, cid in enumerate(ids): |
| 114 | parent = ids[i - 1] if i > 0 else None |
| 115 | _write_commit(root, cid, parent_id=parent) |
| 116 | (root / ".muse" / "refs" / "heads" / "main").write_text(ids[-1]) |
| 117 | |
| 118 | visited: list[str] = [] |
| 119 | |
| 120 | def evaluator(commit: CommitRecord, manifest: dict[str, str], r: pathlib.Path) -> list[QueryMatch]: |
| 121 | visited.append(commit.commit_id) |
| 122 | return [] |
| 123 | |
| 124 | walk_history(root, "main", evaluator, max_commits=3) |
| 125 | assert len(visited) == 3 |
| 126 | |
| 127 | def test_head_commit_id_override(self) -> None: |
| 128 | with tempfile.TemporaryDirectory() as tmp: |
| 129 | root = _make_repo(pathlib.Path(tmp)) |
| 130 | _write_commit(root, "aaa111") |
| 131 | _write_commit(root, "bbb222", parent_id="aaa111") |
| 132 | # HEAD points to bbb222 but we override to aaa111. |
| 133 | (root / ".muse" / "refs" / "heads" / "main").write_text("bbb222") |
| 134 | |
| 135 | visited: list[str] = [] |
| 136 | |
| 137 | def evaluator(commit: CommitRecord, manifest: dict[str, str], r: pathlib.Path) -> list[QueryMatch]: |
| 138 | visited.append(commit.commit_id) |
| 139 | return [] |
| 140 | |
| 141 | walk_history(root, "main", evaluator, head_commit_id="aaa111") |
| 142 | assert visited == ["aaa111"] |
| 143 | |
| 144 | |
| 145 | # --------------------------------------------------------------------------- |
| 146 | # format_matches |
| 147 | # --------------------------------------------------------------------------- |
| 148 | |
| 149 | |
| 150 | class TestFormatMatches: |
| 151 | def test_empty_returns_no_matches(self) -> None: |
| 152 | assert "No matches" in format_matches([]) |
| 153 | |
| 154 | def test_single_match_formatted(self) -> None: |
| 155 | m = QueryMatch( |
| 156 | commit_id="abc12345", |
| 157 | author="gabriel", |
| 158 | committed_at="2026-03-18T12:00:00+00:00", |
| 159 | branch="main", |
| 160 | detail="my_function (added)", |
| 161 | extra={}, |
| 162 | ) |
| 163 | out = format_matches([m]) |
| 164 | assert "abc12345"[:8] in out |
| 165 | assert "gabriel" in out |
| 166 | assert "my_function (added)" in out |
| 167 | |
| 168 | def test_agent_id_shown_when_present(self) -> None: |
| 169 | m = QueryMatch( |
| 170 | commit_id="abc12345", |
| 171 | author="bot", |
| 172 | committed_at="2026-03-18T12:00:00+00:00", |
| 173 | branch="main", |
| 174 | detail="something", |
| 175 | extra={}, |
| 176 | agent_id="claude-v4", |
| 177 | ) |
| 178 | out = format_matches([m]) |
| 179 | assert "claude-v4" in out |
| 180 | |
| 181 | def test_max_results_capped(self) -> None: |
| 182 | matches = [ |
| 183 | QueryMatch( |
| 184 | commit_id=f"commit{i:04d}", |
| 185 | author="x", |
| 186 | committed_at="2026-01-01T00:00:00+00:00", |
| 187 | branch="main", |
| 188 | detail=f"match {i}", |
| 189 | extra={}, |
| 190 | ) |
| 191 | for i in range(100) |
| 192 | ] |
| 193 | out = format_matches(matches, max_results=5) |
| 194 | assert "95 more" in out |