|
1
|
"""Tests for navegador.graph.store.GraphStore.""" |
|
2
|
|
|
3
|
import tempfile |
|
4
|
from pathlib import Path |
|
5
|
from unittest.mock import MagicMock, patch |
|
6
|
|
|
7
|
import pytest |
|
8
|
|
|
9
|
from navegador.graph.store import GraphStore |
|
10
|
|
|
11
|
|
|
12
|
def _mock_client(): |
|
13
|
"""Create a mock FalkorDB client.""" |
|
14
|
client = MagicMock() |
|
15
|
graph = MagicMock() |
|
16
|
graph.query.return_value = MagicMock(result_set=None) |
|
17
|
client.select_graph.return_value = graph |
|
18
|
return client, graph |
|
19
|
|
|
20
|
|
|
21
|
# ── Constructor ─────────────────────────────────────────────────────────────── |
|
22
|
|
|
23
|
class TestGraphStoreInit: |
|
24
|
def test_calls_select_graph(self): |
|
25
|
client, graph = _mock_client() |
|
26
|
GraphStore(client) |
|
27
|
client.select_graph.assert_called_once_with(GraphStore.GRAPH_NAME) |
|
28
|
|
|
29
|
def test_stores_graph(self): |
|
30
|
client, graph = _mock_client() |
|
31
|
store = GraphStore(client) |
|
32
|
assert store._graph is graph |
|
33
|
|
|
34
|
def test_graph_name_constant(self): |
|
35
|
assert GraphStore.GRAPH_NAME == "navegador" |
|
36
|
|
|
37
|
|
|
38
|
# ── sqlite() classmethod ────────────────────────────────────────────────────── |
|
39
|
|
|
40
|
class TestSqliteConstructor: |
|
41
|
def test_creates_db_directory(self): |
|
42
|
with tempfile.TemporaryDirectory() as tmpdir: |
|
43
|
db_path = Path(tmpdir) / "sub" / "graph.db" |
|
44
|
mock_client = MagicMock() |
|
45
|
mock_graph = MagicMock() |
|
46
|
mock_graph.query.return_value = MagicMock(result_set=None) |
|
47
|
mock_client.select_graph.return_value = mock_graph |
|
48
|
mock_falkordb = MagicMock(return_value=mock_client) |
|
49
|
|
|
50
|
with patch.dict("sys.modules", {"redislite": MagicMock(FalkorDB=mock_falkordb)}): |
|
51
|
store = GraphStore.sqlite(str(db_path)) |
|
52
|
assert isinstance(store, GraphStore) |
|
53
|
mock_falkordb.assert_called_once_with(str(db_path)) |
|
54
|
|
|
55
|
def test_raises_import_error_if_not_installed(self): |
|
56
|
with patch.dict("sys.modules", {"redislite": None}): |
|
57
|
with pytest.raises(ImportError, match="falkordblite"): |
|
58
|
GraphStore.sqlite("/tmp/test.db") |
|
59
|
|
|
60
|
|
|
61
|
# ── redis() classmethod ─────────────────────────────────────────────────────── |
|
62
|
|
|
63
|
class TestRedisConstructor: |
|
64
|
def test_creates_redis_store(self): |
|
65
|
mock_client = MagicMock() |
|
66
|
mock_graph = MagicMock() |
|
67
|
mock_graph.query.return_value = MagicMock(result_set=None) |
|
68
|
mock_client.select_graph.return_value = mock_graph |
|
69
|
|
|
70
|
mock_falkordb_module = MagicMock() |
|
71
|
mock_falkordb_module.FalkorDB.from_url.return_value = mock_client |
|
72
|
|
|
73
|
with patch.dict("sys.modules", {"falkordb": mock_falkordb_module}): |
|
74
|
store = GraphStore.redis("redis://localhost:6379") |
|
75
|
assert isinstance(store, GraphStore) |
|
76
|
mock_falkordb_module.FalkorDB.from_url.assert_called_once_with("redis://localhost:6379") |
|
77
|
|
|
78
|
def test_raises_import_error_if_not_installed(self): |
|
79
|
with patch.dict("sys.modules", {"falkordb": None}): |
|
80
|
with pytest.raises(ImportError, match="falkordb"): |
|
81
|
GraphStore.redis("redis://localhost:6379") |
|
82
|
|
|
83
|
|
|
84
|
# ── query() ─────────────────────────────────────────────────────────────────── |
|
85
|
|
|
86
|
class TestQuery: |
|
87
|
def test_delegates_to_graph(self): |
|
88
|
client, graph = _mock_client() |
|
89
|
graph.query.return_value = MagicMock(result_set=[["a", "b"]]) |
|
90
|
store = GraphStore(client) |
|
91
|
result = store.query("MATCH (n) RETURN n", {"x": 1}) |
|
92
|
graph.query.assert_called_once_with("MATCH (n) RETURN n", {"x": 1}) |
|
93
|
assert result.result_set == [["a", "b"]] |
|
94
|
|
|
95
|
def test_passes_empty_dict_when_no_params(self): |
|
96
|
client, graph = _mock_client() |
|
97
|
store = GraphStore(client) |
|
98
|
store.query("MATCH (n) RETURN n") |
|
99
|
graph.query.assert_called_once_with("MATCH (n) RETURN n", {}) |
|
100
|
|
|
101
|
|
|
102
|
# ── create_node() ───────────────────────────────────────────────────────────── |
|
103
|
|
|
104
|
class TestCreateNode: |
|
105
|
def test_generates_merge_cypher(self): |
|
106
|
client, graph = _mock_client() |
|
107
|
store = GraphStore(client) |
|
108
|
store.create_node("Function", {"name": "foo", "file_path": "a.py", "docstring": "doc"}) |
|
109
|
call_args = graph.query.call_args |
|
110
|
cypher = call_args[0][0] |
|
111
|
assert "MERGE" in cypher |
|
112
|
assert "Function" in cypher |
|
113
|
assert "name" in cypher |
|
114
|
assert "file_path" in cypher |
|
115
|
|
|
116
|
def test_passes_props_as_params(self): |
|
117
|
client, graph = _mock_client() |
|
118
|
store = GraphStore(client) |
|
119
|
props = {"name": "bar", "file_path": "b.py"} |
|
120
|
store.create_node("Class", props) |
|
121
|
call_params = graph.query.call_args[0][1] |
|
122
|
assert call_params["name"] == "bar" |
|
123
|
assert call_params["file_path"] == "b.py" |
|
124
|
|
|
125
|
|
|
126
|
# ── create_edge() ───────────────────────────────────────────────────────────── |
|
127
|
|
|
128
|
class TestCreateEdge: |
|
129
|
def test_generates_match_merge_cypher(self): |
|
130
|
client, graph = _mock_client() |
|
131
|
store = GraphStore(client) |
|
132
|
store.create_edge( |
|
133
|
"Function", {"name": "foo"}, |
|
134
|
"CALLS", |
|
135
|
"Function", {"name": "bar"}, |
|
136
|
) |
|
137
|
call_args = graph.query.call_args |
|
138
|
cypher = call_args[0][0] |
|
139
|
assert "MATCH" in cypher |
|
140
|
assert "MERGE" in cypher |
|
141
|
assert "CALLS" in cypher |
|
142
|
|
|
143
|
def test_passes_from_and_to_params(self): |
|
144
|
client, graph = _mock_client() |
|
145
|
store = GraphStore(client) |
|
146
|
store.create_edge( |
|
147
|
"Function", {"name": "foo"}, |
|
148
|
"CALLS", |
|
149
|
"Function", {"name": "bar"}, |
|
150
|
) |
|
151
|
params = graph.query.call_args[0][1] |
|
152
|
assert params["from_name"] == "foo" |
|
153
|
assert params["to_name"] == "bar" |
|
154
|
|
|
155
|
def test_includes_edge_props_when_provided(self): |
|
156
|
client, graph = _mock_client() |
|
157
|
store = GraphStore(client) |
|
158
|
store.create_edge( |
|
159
|
"Class", {"name": "A"}, |
|
160
|
"INHERITS", |
|
161
|
"Class", {"name": "B"}, |
|
162
|
props={"weight": 1}, |
|
163
|
) |
|
164
|
call_args = graph.query.call_args |
|
165
|
cypher = call_args[0][0] |
|
166
|
params = call_args[0][1] |
|
167
|
assert "SET" in cypher |
|
168
|
assert params["p_weight"] == 1 |
|
169
|
|
|
170
|
def test_no_set_clause_without_props(self): |
|
171
|
client, graph = _mock_client() |
|
172
|
store = GraphStore(client) |
|
173
|
store.create_edge("A", {"name": "x"}, "REL", "B", {"name": "y"}) |
|
174
|
cypher = graph.query.call_args[0][0] |
|
175
|
assert "SET" not in cypher |
|
176
|
|
|
177
|
|
|
178
|
# ── clear() ─────────────────────────────────────────────────────────────────── |
|
179
|
|
|
180
|
class TestClear: |
|
181
|
def test_executes_delete_query(self): |
|
182
|
client, graph = _mock_client() |
|
183
|
store = GraphStore(client) |
|
184
|
store.clear() |
|
185
|
cypher = graph.query.call_args[0][0] |
|
186
|
assert "DETACH DELETE" in cypher |
|
187
|
|
|
188
|
|
|
189
|
# ── node_count / edge_count ─────────────────────────────────────────────────── |
|
190
|
|
|
191
|
class TestCounts: |
|
192
|
def test_node_count_returns_value(self): |
|
193
|
client, graph = _mock_client() |
|
194
|
graph.query.return_value = MagicMock(result_set=[[42]]) |
|
195
|
store = GraphStore(client) |
|
196
|
assert store.node_count() == 42 |
|
197
|
|
|
198
|
def test_node_count_returns_zero_on_empty(self): |
|
199
|
client, graph = _mock_client() |
|
200
|
graph.query.return_value = MagicMock(result_set=[]) |
|
201
|
store = GraphStore(client) |
|
202
|
assert store.node_count() == 0 |
|
203
|
|
|
204
|
def test_edge_count_returns_value(self): |
|
205
|
client, graph = _mock_client() |
|
206
|
graph.query.return_value = MagicMock(result_set=[[7]]) |
|
207
|
store = GraphStore(client) |
|
208
|
assert store.edge_count() == 7 |
|
209
|
|
|
210
|
def test_edge_count_returns_zero_on_empty(self): |
|
211
|
client, graph = _mock_client() |
|
212
|
graph.query.return_value = MagicMock(result_set=[]) |
|
213
|
store = GraphStore(client) |
|
214
|
assert store.edge_count() == 0 |
|
215
|
|