|
57ba118…
|
lmata
|
1 |
"""Tests for navegador.graph.export — text-based graph export and import.""" |
|
57ba118…
|
lmata
|
2 |
|
|
57ba118…
|
lmata
|
3 |
import json |
|
57ba118…
|
lmata
|
4 |
import tempfile |
|
57ba118…
|
lmata
|
5 |
from pathlib import Path |
|
57ba118…
|
lmata
|
6 |
from unittest.mock import MagicMock |
|
57ba118…
|
lmata
|
7 |
|
|
57ba118…
|
lmata
|
8 |
import pytest |
|
57ba118…
|
lmata
|
9 |
|
|
57ba118…
|
lmata
|
10 |
from navegador.graph.export import ( |
|
57ba118…
|
lmata
|
11 |
_export_edges, |
|
57ba118…
|
lmata
|
12 |
_export_nodes, |
|
57ba118…
|
lmata
|
13 |
_import_edge, |
|
57ba118…
|
lmata
|
14 |
_import_node, |
|
57ba118…
|
lmata
|
15 |
export_graph, |
|
57ba118…
|
lmata
|
16 |
import_graph, |
|
57ba118…
|
lmata
|
17 |
) |
|
57ba118…
|
lmata
|
18 |
|
|
57ba118…
|
lmata
|
19 |
|
|
57ba118…
|
lmata
|
20 |
def _mock_store(nodes=None, edges=None): |
|
57ba118…
|
lmata
|
21 |
store = MagicMock() |
|
57ba118…
|
lmata
|
22 |
|
|
57ba118…
|
lmata
|
23 |
def query_side_effect(cypher, params=None): |
|
57ba118…
|
lmata
|
24 |
result = MagicMock() |
|
57ba118…
|
lmata
|
25 |
if "labels(n)" in cypher and "properties" in cypher: |
|
57ba118…
|
lmata
|
26 |
result.result_set = nodes or [] |
|
57ba118…
|
lmata
|
27 |
elif "type(r)" in cypher: |
|
57ba118…
|
lmata
|
28 |
result.result_set = edges or [] |
|
57ba118…
|
lmata
|
29 |
elif "DETACH DELETE" in cypher: |
|
57ba118…
|
lmata
|
30 |
result.result_set = [] |
|
57ba118…
|
lmata
|
31 |
else: |
|
57ba118…
|
lmata
|
32 |
result.result_set = [] |
|
57ba118…
|
lmata
|
33 |
return result |
|
57ba118…
|
lmata
|
34 |
|
|
57ba118…
|
lmata
|
35 |
store.query.side_effect = query_side_effect |
|
57ba118…
|
lmata
|
36 |
store.clear = MagicMock() |
|
57ba118…
|
lmata
|
37 |
return store |
|
57ba118…
|
lmata
|
38 |
|
|
57ba118…
|
lmata
|
39 |
|
|
57ba118…
|
lmata
|
40 |
# ── export_graph ───────────────────────────────────────────────────────────── |
|
57ba118…
|
lmata
|
41 |
|
|
57ba118…
|
lmata
|
42 |
class TestExportGraph: |
|
57ba118…
|
lmata
|
43 |
def test_creates_output_file(self): |
|
57ba118…
|
lmata
|
44 |
store = _mock_store(nodes=[], edges=[]) |
|
57ba118…
|
lmata
|
45 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
46 |
output = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
47 |
export_graph(store, output) |
|
57ba118…
|
lmata
|
48 |
assert output.exists() |
|
57ba118…
|
lmata
|
49 |
|
|
57ba118…
|
lmata
|
50 |
def test_returns_counts(self): |
|
57ba118…
|
lmata
|
51 |
nodes = [["Function", {"name": "foo", "file_path": "app.py"}]] |
|
57ba118…
|
lmata
|
52 |
store = _mock_store(nodes=nodes, edges=[]) |
|
57ba118…
|
lmata
|
53 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
54 |
stats = export_graph(store, Path(tmpdir) / "graph.jsonl") |
|
57ba118…
|
lmata
|
55 |
assert stats["nodes"] == 1 |
|
57ba118…
|
lmata
|
56 |
assert stats["edges"] == 0 |
|
57ba118…
|
lmata
|
57 |
|
|
57ba118…
|
lmata
|
58 |
def test_writes_valid_jsonl(self): |
|
57ba118…
|
lmata
|
59 |
nodes = [ |
|
57ba118…
|
lmata
|
60 |
["Function", {"name": "foo", "file_path": "app.py"}], |
|
57ba118…
|
lmata
|
61 |
["Class", {"name": "Bar", "file_path": "bar.py"}], |
|
57ba118…
|
lmata
|
62 |
] |
|
57ba118…
|
lmata
|
63 |
edges = [["CALLS", "Function", "foo", "app.py", "Function", "bar", "bar.py"]] |
|
57ba118…
|
lmata
|
64 |
store = _mock_store(nodes=nodes, edges=edges) |
|
57ba118…
|
lmata
|
65 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
66 |
output = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
67 |
export_graph(store, output) |
|
57ba118…
|
lmata
|
68 |
lines = output.read_text().strip().split("\n") |
|
57ba118…
|
lmata
|
69 |
assert len(lines) == 3 # 2 nodes + 1 edge |
|
57ba118…
|
lmata
|
70 |
for line in lines: |
|
57ba118…
|
lmata
|
71 |
record = json.loads(line) |
|
57ba118…
|
lmata
|
72 |
assert record["kind"] in ("node", "edge") |
|
57ba118…
|
lmata
|
73 |
|
|
57ba118…
|
lmata
|
74 |
def test_output_is_sorted(self): |
|
57ba118…
|
lmata
|
75 |
nodes = [ |
|
57ba118…
|
lmata
|
76 |
["Function", {"name": "z_func", "file_path": "z.py"}], |
|
57ba118…
|
lmata
|
77 |
["Class", {"name": "a_class", "file_path": "a.py"}], |
|
57ba118…
|
lmata
|
78 |
] |
|
57ba118…
|
lmata
|
79 |
store = _mock_store(nodes=nodes, edges=[]) |
|
57ba118…
|
lmata
|
80 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
81 |
output = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
82 |
export_graph(store, output) |
|
57ba118…
|
lmata
|
83 |
lines = output.read_text().strip().split("\n") |
|
57ba118…
|
lmata
|
84 |
labels = [json.loads(line)["label"] for line in lines] |
|
57ba118…
|
lmata
|
85 |
# Class comes before Function alphabetically |
|
57ba118…
|
lmata
|
86 |
assert labels[0] == "Class" |
|
57ba118…
|
lmata
|
87 |
assert labels[1] == "Function" |
|
57ba118…
|
lmata
|
88 |
|
|
57ba118…
|
lmata
|
89 |
def test_creates_parent_dirs(self): |
|
57ba118…
|
lmata
|
90 |
store = _mock_store(nodes=[], edges=[]) |
|
57ba118…
|
lmata
|
91 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
92 |
output = Path(tmpdir) / "sub" / "dir" / "graph.jsonl" |
|
57ba118…
|
lmata
|
93 |
export_graph(store, output) |
|
57ba118…
|
lmata
|
94 |
assert output.exists() |
|
57ba118…
|
lmata
|
95 |
|
|
57ba118…
|
lmata
|
96 |
|
|
57ba118…
|
lmata
|
97 |
# ── import_graph ───────────────────────────────────────────────────────────── |
|
57ba118…
|
lmata
|
98 |
|
|
57ba118…
|
lmata
|
99 |
class TestImportGraph: |
|
57ba118…
|
lmata
|
100 |
def test_raises_on_missing_file(self): |
|
57ba118…
|
lmata
|
101 |
store = MagicMock() |
|
57ba118…
|
lmata
|
102 |
with pytest.raises(FileNotFoundError): |
|
57ba118…
|
lmata
|
103 |
import_graph(store, "/nonexistent/graph.jsonl") |
|
57ba118…
|
lmata
|
104 |
|
|
57ba118…
|
lmata
|
105 |
def test_clears_graph_by_default(self): |
|
57ba118…
|
lmata
|
106 |
store = MagicMock() |
|
57ba118…
|
lmata
|
107 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
108 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
109 |
f = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
110 |
f.write_text("") |
|
57ba118…
|
lmata
|
111 |
import_graph(store, f) |
|
57ba118…
|
lmata
|
112 |
store.clear.assert_called_once() |
|
57ba118…
|
lmata
|
113 |
|
|
57ba118…
|
lmata
|
114 |
def test_no_clear_flag(self): |
|
57ba118…
|
lmata
|
115 |
store = MagicMock() |
|
57ba118…
|
lmata
|
116 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
117 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
118 |
f = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
119 |
f.write_text("") |
|
57ba118…
|
lmata
|
120 |
import_graph(store, f, clear=False) |
|
57ba118…
|
lmata
|
121 |
store.clear.assert_not_called() |
|
57ba118…
|
lmata
|
122 |
|
|
57ba118…
|
lmata
|
123 |
def test_imports_nodes(self): |
|
57ba118…
|
lmata
|
124 |
store = MagicMock() |
|
57ba118…
|
lmata
|
125 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
126 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
127 |
f = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
128 |
node = {"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": "app.py"}} |
|
57ba118…
|
lmata
|
129 |
f.write_text(json.dumps(node) + "\n") |
|
57ba118…
|
lmata
|
130 |
stats = import_graph(store, f) |
|
57ba118…
|
lmata
|
131 |
assert stats["nodes"] == 1 |
|
57ba118…
|
lmata
|
132 |
|
|
57ba118…
|
lmata
|
133 |
def test_imports_edges(self): |
|
57ba118…
|
lmata
|
134 |
store = MagicMock() |
|
57ba118…
|
lmata
|
135 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
136 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
137 |
f = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
138 |
edge = { |
|
57ba118…
|
lmata
|
139 |
"kind": "edge", |
|
57ba118…
|
lmata
|
140 |
"type": "CALLS", |
|
57ba118…
|
lmata
|
141 |
"from": {"label": "Function", "name": "foo", "path": "app.py"}, |
|
57ba118…
|
lmata
|
142 |
"to": {"label": "Function", "name": "bar", "path": "bar.py"}, |
|
57ba118…
|
lmata
|
143 |
} |
|
57ba118…
|
lmata
|
144 |
f.write_text(json.dumps(edge) + "\n") |
|
57ba118…
|
lmata
|
145 |
stats = import_graph(store, f) |
|
57ba118…
|
lmata
|
146 |
assert stats["edges"] == 1 |
|
57ba118…
|
lmata
|
147 |
|
|
57ba118…
|
lmata
|
148 |
def test_returns_counts(self): |
|
57ba118…
|
lmata
|
149 |
store = MagicMock() |
|
57ba118…
|
lmata
|
150 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
151 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
152 |
f = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
153 |
lines = [ |
|
57ba118…
|
lmata
|
154 |
json.dumps({"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": "app.py"}}), |
|
57ba118…
|
lmata
|
155 |
json.dumps({"kind": "node", "label": "Class", "props": {"name": "Bar", "file_path": "bar.py"}}), |
|
57ba118…
|
lmata
|
156 |
json.dumps({"kind": "edge", "type": "CALLS", |
|
57ba118…
|
lmata
|
157 |
"from": {"label": "Function", "name": "foo", "path": ""}, |
|
57ba118…
|
lmata
|
158 |
"to": {"label": "Class", "name": "Bar", "path": ""}}), |
|
57ba118…
|
lmata
|
159 |
] |
|
57ba118…
|
lmata
|
160 |
f.write_text("\n".join(lines) + "\n") |
|
57ba118…
|
lmata
|
161 |
stats = import_graph(store, f) |
|
57ba118…
|
lmata
|
162 |
assert stats["nodes"] == 2 |
|
57ba118…
|
lmata
|
163 |
assert stats["edges"] == 1 |
|
57ba118…
|
lmata
|
164 |
|
|
57ba118…
|
lmata
|
165 |
def test_skips_blank_lines(self): |
|
57ba118…
|
lmata
|
166 |
store = MagicMock() |
|
57ba118…
|
lmata
|
167 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
168 |
with tempfile.TemporaryDirectory() as tmpdir: |
|
57ba118…
|
lmata
|
169 |
f = Path(tmpdir) / "graph.jsonl" |
|
57ba118…
|
lmata
|
170 |
node = json.dumps({"kind": "node", "label": "Function", "props": {"name": "foo", "file_path": ""}}) |
|
57ba118…
|
lmata
|
171 |
f.write_text(f"\n{node}\n\n") |
|
57ba118…
|
lmata
|
172 |
stats = import_graph(store, f) |
|
57ba118…
|
lmata
|
173 |
assert stats["nodes"] == 1 |
|
57ba118…
|
lmata
|
174 |
|
|
57ba118…
|
lmata
|
175 |
|
|
57ba118…
|
lmata
|
176 |
# ── _export_nodes / _export_edges ──────────────────────────────────────────── |
|
57ba118…
|
lmata
|
177 |
|
|
57ba118…
|
lmata
|
178 |
class TestExportHelpers: |
|
57ba118…
|
lmata
|
179 |
def test_export_nodes_handles_non_dict_props(self): |
|
57ba118…
|
lmata
|
180 |
store = MagicMock() |
|
57ba118…
|
lmata
|
181 |
store.query.return_value = MagicMock(result_set=[["Function", "not_a_dict"]]) |
|
57ba118…
|
lmata
|
182 |
nodes = _export_nodes(store) |
|
57ba118…
|
lmata
|
183 |
assert len(nodes) == 1 |
|
57ba118…
|
lmata
|
184 |
assert nodes[0]["props"] == {} |
|
57ba118…
|
lmata
|
185 |
|
|
57ba118…
|
lmata
|
186 |
def test_export_edges_returns_structured_data(self): |
|
57ba118…
|
lmata
|
187 |
store = MagicMock() |
|
57ba118…
|
lmata
|
188 |
store.query.return_value = MagicMock( |
|
57ba118…
|
lmata
|
189 |
result_set=[["CALLS", "Function", "foo", "app.py", "Function", "bar", "bar.py"]] |
|
57ba118…
|
lmata
|
190 |
) |
|
57ba118…
|
lmata
|
191 |
edges = _export_edges(store) |
|
57ba118…
|
lmata
|
192 |
assert len(edges) == 1 |
|
57ba118…
|
lmata
|
193 |
assert edges[0]["type"] == "CALLS" |
|
57ba118…
|
lmata
|
194 |
assert edges[0]["from"]["name"] == "foo" |
|
57ba118…
|
lmata
|
195 |
assert edges[0]["to"]["name"] == "bar" |
|
57ba118…
|
lmata
|
196 |
|
|
57ba118…
|
lmata
|
197 |
|
|
57ba118…
|
lmata
|
198 |
# ── _import_node / _import_edge ────────────────────────────────────────────── |
|
57ba118…
|
lmata
|
199 |
|
|
57ba118…
|
lmata
|
200 |
class TestImportHelpers: |
|
57ba118…
|
lmata
|
201 |
def test_import_node_adds_missing_file_path(self): |
|
57ba118…
|
lmata
|
202 |
store = MagicMock() |
|
57ba118…
|
lmata
|
203 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
204 |
record = {"kind": "node", "label": "Concept", "props": {"name": "JWT"}} |
|
57ba118…
|
lmata
|
205 |
_import_node(store, record) |
|
57ba118…
|
lmata
|
206 |
store.query.assert_called_once() |
|
57ba118…
|
lmata
|
207 |
cypher = store.query.call_args[0][0] |
|
57ba118…
|
lmata
|
208 |
assert "MERGE" in cypher |
|
57ba118…
|
lmata
|
209 |
|
|
57ba118…
|
lmata
|
210 |
def test_import_node_adds_missing_name(self): |
|
57ba118…
|
lmata
|
211 |
store = MagicMock() |
|
57ba118…
|
lmata
|
212 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
213 |
record = {"kind": "node", "label": "Domain", "props": {"description": "Auth domain"}} |
|
57ba118…
|
lmata
|
214 |
_import_node(store, record) |
|
57ba118…
|
lmata
|
215 |
# Should have added name="" to props |
|
57ba118…
|
lmata
|
216 |
store.query.assert_called_once() |
|
57ba118…
|
lmata
|
217 |
params = store.query.call_args[0][1] |
|
57ba118…
|
lmata
|
218 |
assert params["name"] == "" |
|
57ba118…
|
lmata
|
219 |
|
|
57ba118…
|
lmata
|
220 |
def test_import_node_uses_path_key_for_repos(self): |
|
57ba118…
|
lmata
|
221 |
store = MagicMock() |
|
57ba118…
|
lmata
|
222 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
223 |
record = {"kind": "node", "label": "Repository", "props": {"name": "myrepo", "path": "/code/myrepo"}} |
|
57ba118…
|
lmata
|
224 |
_import_node(store, record) |
|
57ba118…
|
lmata
|
225 |
cypher = store.query.call_args[0][0] |
|
57ba118…
|
lmata
|
226 |
assert "path" in cypher |
|
57ba118…
|
lmata
|
227 |
|
|
57ba118…
|
lmata
|
228 |
def test_import_edge_with_paths(self): |
|
57ba118…
|
lmata
|
229 |
store = MagicMock() |
|
57ba118…
|
lmata
|
230 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
231 |
record = { |
|
57ba118…
|
lmata
|
232 |
"kind": "edge", |
|
57ba118…
|
lmata
|
233 |
"type": "CALLS", |
|
57ba118…
|
lmata
|
234 |
"from": {"label": "Function", "name": "foo", "path": "app.py"}, |
|
57ba118…
|
lmata
|
235 |
"to": {"label": "Function", "name": "bar", "path": "bar.py"}, |
|
57ba118…
|
lmata
|
236 |
} |
|
57ba118…
|
lmata
|
237 |
_import_edge(store, record) |
|
57ba118…
|
lmata
|
238 |
store.query.assert_called_once() |
|
57ba118…
|
lmata
|
239 |
|
|
57ba118…
|
lmata
|
240 |
def test_import_edge_without_paths(self): |
|
57ba118…
|
lmata
|
241 |
store = MagicMock() |
|
57ba118…
|
lmata
|
242 |
store.query.return_value = MagicMock(result_set=[]) |
|
57ba118…
|
lmata
|
243 |
record = { |
|
57ba118…
|
lmata
|
244 |
"kind": "edge", |
|
57ba118…
|
lmata
|
245 |
"type": "RELATED_TO", |
|
57ba118…
|
lmata
|
246 |
"from": {"label": "Concept", "name": "JWT", "path": ""}, |
|
57ba118…
|
lmata
|
247 |
"to": {"label": "Concept", "name": "OAuth", "path": ""}, |
|
57ba118…
|
lmata
|
248 |
} |
|
57ba118…
|
lmata
|
249 |
_import_edge(store, record) |
|
57ba118…
|
lmata
|
250 |
store.query.assert_called_once() |
|
57ba118…
|
lmata
|
251 |
cypher = store.query.call_args[0][0] |
|
57ba118…
|
lmata
|
252 |
assert "file_path" not in cypher |