|
1
|
# Copyright CONFLICT LLC 2026 (weareconflict.com) |
|
2
|
""" |
|
3
|
Test coverage mapping — link test functions to production code via TESTS edges. |
|
4
|
|
|
5
|
Finds test functions (name starts with test_), resolves the production |
|
6
|
symbol they exercise via: |
|
7
|
1. Existing CALLS edges from the test function to non-test symbols |
|
8
|
2. Name heuristics: test_foo → foo, test_foo_bar → foo / foo_bar |
|
9
|
|
|
10
|
Creates TESTS edges in the graph for discovered links. |
|
11
|
""" |
|
12
|
|
|
13
|
from __future__ import annotations |
|
14
|
|
|
15
|
from dataclasses import dataclass, field |
|
16
|
from typing import Any |
|
17
|
|
|
18
|
from navegador.graph import GraphStore |
|
19
|
|
|
20
|
# All functions starting with test_ |
|
21
|
_TEST_FUNCTIONS_QUERY = """ |
|
22
|
MATCH (fn) |
|
23
|
WHERE (fn:Function OR fn:Method) AND fn.name STARTS WITH 'test_' |
|
24
|
RETURN fn.name AS name, coalesce(fn.file_path, '') AS file_path, |
|
25
|
fn.line_start AS line_start |
|
26
|
ORDER BY fn.file_path, fn.name |
|
27
|
""" |
|
28
|
|
|
29
|
# Functions directly called by a test function |
|
30
|
_CALLS_FROM_TEST = """ |
|
31
|
MATCH (test {name: $test_name})-[:CALLS]->(callee) |
|
32
|
WHERE NOT callee.name STARTS WITH 'test_' |
|
33
|
AND ($file_path = '' OR test.file_path = $file_path) |
|
34
|
RETURN labels(callee)[0] AS type, callee.name AS name, |
|
35
|
coalesce(callee.file_path, '') AS file_path |
|
36
|
""" |
|
37
|
|
|
38
|
# Lookup a production symbol by name (not a test function) |
|
39
|
_FIND_PRODUCTION_SYMBOL = """ |
|
40
|
MATCH (n) |
|
41
|
WHERE n.name = $name AND NOT n.name STARTS WITH 'test_' |
|
42
|
AND (n:Function OR n:Method OR n:Class) |
|
43
|
RETURN labels(n)[0] AS type, n.name AS name, |
|
44
|
coalesce(n.file_path, '') AS file_path |
|
45
|
LIMIT 1 |
|
46
|
""" |
|
47
|
|
|
48
|
# Create a TESTS edge |
|
49
|
_CREATE_TESTS_EDGE = """ |
|
50
|
MATCH (test), (prod) |
|
51
|
WHERE (test.name = $test_name AND (test.file_path = $test_file OR $test_file = '')) |
|
52
|
AND (prod.name = $prod_name AND (prod.file_path = $prod_file OR $prod_file = '')) |
|
53
|
MERGE (test)-[r:TESTS]->(prod) |
|
54
|
""" |
|
55
|
|
|
56
|
|
|
57
|
@dataclass |
|
58
|
class TestLink: |
|
59
|
"""A resolved link between a test function and a production symbol.""" |
|
60
|
|
|
61
|
test_name: str |
|
62
|
test_file: str |
|
63
|
prod_name: str |
|
64
|
prod_file: str |
|
65
|
prod_type: str |
|
66
|
source: str # "calls" | "heuristic" |
|
67
|
|
|
68
|
|
|
69
|
@dataclass |
|
70
|
class TestMapResult: |
|
71
|
"""Result of running test coverage mapping.""" |
|
72
|
|
|
73
|
links: list[TestLink] = field(default_factory=list) |
|
74
|
unmatched_tests: list[dict[str, Any]] = field(default_factory=list) |
|
75
|
edges_created: int = 0 |
|
76
|
|
|
77
|
def to_dict(self) -> dict[str, Any]: |
|
78
|
return { |
|
79
|
"links": [ |
|
80
|
{ |
|
81
|
"test_name": lnk.test_name, |
|
82
|
"test_file": lnk.test_file, |
|
83
|
"prod_name": lnk.prod_name, |
|
84
|
"prod_file": lnk.prod_file, |
|
85
|
"prod_type": lnk.prod_type, |
|
86
|
"source": lnk.source, |
|
87
|
} |
|
88
|
for lnk in self.links |
|
89
|
], |
|
90
|
"unmatched_tests": self.unmatched_tests, |
|
91
|
"edges_created": self.edges_created, |
|
92
|
"summary": { |
|
93
|
"matched": len(self.links), |
|
94
|
"unmatched": len(self.unmatched_tests), |
|
95
|
"edges_created": self.edges_created, |
|
96
|
}, |
|
97
|
} |
|
98
|
|
|
99
|
|
|
100
|
class TestMapper: |
|
101
|
""" |
|
102
|
Map test functions to production code and persist TESTS edges. |
|
103
|
|
|
104
|
Usage:: |
|
105
|
|
|
106
|
store = GraphStore.sqlite() |
|
107
|
mapper = TestMapper(store) |
|
108
|
result = mapper.map_tests() |
|
109
|
print(result.links) |
|
110
|
""" |
|
111
|
|
|
112
|
def __init__(self, store: GraphStore) -> None: |
|
113
|
self.store = store |
|
114
|
|
|
115
|
def map_tests(self) -> TestMapResult: |
|
116
|
""" |
|
117
|
Discover test → production mappings and write TESTS edges. |
|
118
|
|
|
119
|
Strategy per test function: |
|
120
|
1. Follow existing CALLS edges to non-test symbols (direct call evidence). |
|
121
|
2. Apply name heuristics: strip test_ prefix and look for matching symbol. |
|
122
|
|
|
123
|
Returns: |
|
124
|
TestMapResult with links, unmatched_tests, and edges_created count. |
|
125
|
""" |
|
126
|
test_fns = self._get_test_functions() |
|
127
|
if not test_fns: |
|
128
|
return TestMapResult() |
|
129
|
|
|
130
|
links: list[TestLink] = [] |
|
131
|
unmatched: list[dict[str, Any]] = [] |
|
132
|
edges_created = 0 |
|
133
|
|
|
134
|
for test in test_fns: |
|
135
|
test_name = test["name"] |
|
136
|
test_file = test["file_path"] |
|
137
|
|
|
138
|
resolved = self._resolve_via_calls(test_name, test_file) |
|
139
|
if not resolved: |
|
140
|
resolved = self._resolve_via_heuristic(test_name) |
|
141
|
|
|
142
|
if resolved: |
|
143
|
prod_type, prod_name, prod_file = resolved |
|
144
|
link = TestLink( |
|
145
|
test_name=test_name, |
|
146
|
test_file=test_file, |
|
147
|
prod_name=prod_name, |
|
148
|
prod_file=prod_file, |
|
149
|
prod_type=prod_type, |
|
150
|
source=( |
|
151
|
"calls" if self._resolve_via_calls(test_name, test_file) else "heuristic" |
|
152
|
), |
|
153
|
) |
|
154
|
links.append(link) |
|
155
|
# Persist the TESTS edge |
|
156
|
try: |
|
157
|
self.store.query( |
|
158
|
_CREATE_TESTS_EDGE, |
|
159
|
{ |
|
160
|
"test_name": test_name, |
|
161
|
"test_file": test_file, |
|
162
|
"prod_name": prod_name, |
|
163
|
"prod_file": prod_file, |
|
164
|
}, |
|
165
|
) |
|
166
|
edges_created += 1 |
|
167
|
except Exception: |
|
168
|
pass |
|
169
|
else: |
|
170
|
unmatched.append(test) |
|
171
|
|
|
172
|
return TestMapResult( |
|
173
|
links=links, |
|
174
|
unmatched_tests=unmatched, |
|
175
|
edges_created=edges_created, |
|
176
|
) |
|
177
|
|
|
178
|
def _get_test_functions(self) -> list[dict[str, Any]]: |
|
179
|
try: |
|
180
|
result = self.store.query(_TEST_FUNCTIONS_QUERY) |
|
181
|
rows = result.result_set or [] |
|
182
|
except Exception: |
|
183
|
return [] |
|
184
|
|
|
185
|
return [ |
|
186
|
{"name": row[0] or "", "file_path": row[1] or "", "line_start": row[2]} for row in rows |
|
187
|
] |
|
188
|
|
|
189
|
def _resolve_via_calls(self, test_name: str, test_file: str) -> tuple[str, str, str] | None: |
|
190
|
"""Return (type, name, file_path) of the first non-test callee, or None.""" |
|
191
|
try: |
|
192
|
result = self.store.query( |
|
193
|
_CALLS_FROM_TEST, {"test_name": test_name, "file_path": test_file} |
|
194
|
) |
|
195
|
rows = result.result_set or [] |
|
196
|
except Exception: |
|
197
|
return None |
|
198
|
|
|
199
|
if rows: |
|
200
|
row = rows[0] |
|
201
|
return (row[0] or "Function", row[1] or "", row[2] or "") |
|
202
|
return None |
|
203
|
|
|
204
|
def _resolve_via_heuristic(self, test_name: str) -> tuple[str, str, str] | None: |
|
205
|
""" |
|
206
|
Strip test_ prefix and try increasingly shorter name suffixes. |
|
207
|
|
|
208
|
test_validate_token → validate_token, then validate |
|
209
|
""" |
|
210
|
if not test_name.startswith("test_"): |
|
211
|
return None |
|
212
|
|
|
213
|
stripped = test_name[len("test_") :] |
|
214
|
parts = stripped.split("_") |
|
215
|
|
|
216
|
# Try full stripped name first, then progressively shorter prefixes |
|
217
|
candidates = [] |
|
218
|
for i in range(len(parts), 0, -1): |
|
219
|
candidates.append("_".join(parts[:i])) |
|
220
|
|
|
221
|
for candidate in candidates: |
|
222
|
try: |
|
223
|
result = self.store.query(_FIND_PRODUCTION_SYMBOL, {"name": candidate}) |
|
224
|
rows = result.result_set or [] |
|
225
|
except Exception: |
|
226
|
continue |
|
227
|
|
|
228
|
if rows: |
|
229
|
row = rows[0] |
|
230
|
return (row[0] or "Function", row[1] or "", row[2] or "") |
|
231
|
|
|
232
|
return None |
|
233
|
|