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