Navegador

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

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button