gabriel / muse public
test_code_query.py python
171 lines 5.8 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago
1 """Tests for the code domain query evaluator."""
2
3 import pathlib
4
5 import pytest
6
7 from muse.domain import SemVerBump
8 from muse.plugins.code._code_query import (
9 AndExpr,
10 Comparison,
11 OrExpr,
12 build_evaluator,
13 _parse_query,
14 )
15
16
17 # ---------------------------------------------------------------------------
18 # Parser
19 # ---------------------------------------------------------------------------
20
21
22 class TestParseQuery:
23 def test_simple_equality(self) -> None:
24 q = _parse_query("author == 'gabriel'")
25 assert isinstance(q, OrExpr)
26 assert len(q.clauses) == 1
27 and_expr = q.clauses[0]
28 assert isinstance(and_expr, AndExpr)
29 cmp = and_expr.clauses[0]
30 assert cmp.field == "author"
31 assert cmp.op == "=="
32 assert cmp.value == "gabriel"
33
34 def test_and_expression(self) -> None:
35 q = _parse_query("author == 'x' and language == 'Python'")
36 assert len(q.clauses[0].clauses) == 2
37
38 def test_or_expression(self) -> None:
39 q = _parse_query("author == 'x' or author == 'y'")
40 assert len(q.clauses) == 2
41
42 def test_contains_operator(self) -> None:
43 q = _parse_query("agent_id contains claude")
44 cmp = q.clauses[0].clauses[0]
45 assert cmp.op == "contains"
46 assert cmp.value == "claude"
47
48 def test_startswith_operator(self) -> None:
49 q = _parse_query("symbol startswith my_")
50 cmp = q.clauses[0].clauses[0]
51 assert cmp.op == "startswith"
52
53 def test_unknown_field_raises(self) -> None:
54 with pytest.raises(ValueError, match="Unknown field"):
55 _parse_query("nonexistent == 'x'")
56
57 def test_double_quoted_string(self) -> None:
58 q = _parse_query('author == "gabriel"')
59 cmp = q.clauses[0].clauses[0]
60 assert cmp.value == "gabriel"
61
62
63 # ---------------------------------------------------------------------------
64 # build_evaluator + evaluator logic
65 # ---------------------------------------------------------------------------
66
67
68 import datetime
69 from muse.core.query_engine import QueryMatch
70 from muse.core.commits import CommitRecord
71 from muse.domain import StructuredDelta, InsertOp
72
73
74 def _make_commit(
75 author: str = "alice",
76 agent_id: str = "",
77 model_id: str = "",
78 branch: str = "main",
79 sem_ver_bump: SemVerBump = "none",
80 delta: StructuredDelta | None = None,
81 ) -> CommitRecord:
82 return CommitRecord(
83 commit_id="abc1234",
84 branch=branch,
85 snapshot_id="s" * 64,
86 message="test commit",
87 committed_at=datetime.datetime.now(datetime.timezone.utc),
88 author=author,
89 agent_id=agent_id,
90 model_id=model_id,
91 sem_ver_bump=sem_ver_bump,
92 structured_delta=delta,
93 )
94
95
96 class TestBuildEvaluator:
97 def test_author_match(self) -> None:
98 evaluator = build_evaluator("author == 'alice'")
99 commit = _make_commit(author="alice")
100 results = evaluator(commit, {}, pathlib.Path("."))
101 assert len(results) == 1
102
103 def test_author_no_match(self) -> None:
104 evaluator = build_evaluator("author == 'bob'")
105 commit = _make_commit(author="alice")
106 results = evaluator(commit, {}, pathlib.Path("."))
107 assert results == []
108
109 def test_agent_id_contains(self) -> None:
110 evaluator = build_evaluator("agent_id contains claude")
111 commit = _make_commit(agent_id="claude-v4")
112 results = evaluator(commit, {}, pathlib.Path("."))
113 assert len(results) == 1
114
115 def test_sem_ver_bump_match(self) -> None:
116 evaluator = build_evaluator("sem_ver_bump == 'major'")
117 commit = _make_commit(sem_ver_bump="major")
118 results = evaluator(commit, {}, pathlib.Path("."))
119 assert len(results) == 1
120
121 def test_and_both_must_match(self) -> None:
122 evaluator = build_evaluator("author == 'alice' and agent_id == 'bot'")
123 # Only author matches, not agent_id.
124 commit = _make_commit(author="alice", agent_id="human")
125 results = evaluator(commit, {}, pathlib.Path("."))
126 assert results == []
127
128 def test_or_one_match_sufficient(self) -> None:
129 evaluator = build_evaluator("author == 'alice' or author == 'bob'")
130 commit_alice = _make_commit(author="alice")
131 commit_bob = _make_commit(author="bob")
132 assert len(evaluator(commit_alice, {}, pathlib.Path("."))) >= 1
133 assert len(evaluator(commit_bob, {}, pathlib.Path("."))) >= 1
134
135 def test_branch_match(self) -> None:
136 evaluator = build_evaluator("branch == 'dev'")
137 commit = _make_commit(branch="dev")
138 assert len(evaluator(commit, {}, pathlib.Path("."))) >= 1
139
140 def test_symbol_match_from_delta(self) -> None:
141 op = InsertOp(
142 op="insert",
143 address="src/utils.py::my_func",
144 position=None,
145 content_id="hash1",
146 content_summary="added my_func",
147 )
148 delta = StructuredDelta(domain="code", ops=[op], summary="1 symbol added")
149 commit = _make_commit(delta=delta)
150 evaluator = build_evaluator("symbol == 'my_func'")
151 results = evaluator(commit, {}, pathlib.Path("."))
152 assert len(results) >= 1
153 assert any("my_func" in r.get("detail", "") for r in results)
154
155 def test_change_added_match(self) -> None:
156 op = InsertOp(
157 op="insert",
158 address="src/foo.py::bar",
159 position=None,
160 content_id="h1",
161 content_summary="bar added",
162 )
163 delta = StructuredDelta(domain="code", ops=[op], summary="added bar")
164 commit = _make_commit(delta=delta)
165 evaluator = build_evaluator("change == 'added'")
166 results = evaluator(commit, {}, pathlib.Path("."))
167 assert len(results) >= 1
168
169 def test_invalid_query_raises(self) -> None:
170 with pytest.raises(ValueError):
171 build_evaluator("badfield == 'x'")
File History 1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 6 days ago