PlanOpticon

planopticon / tests / test_graph_query.py
Source Blame History 337 lines
b363c5b… noreply 1 """Tests for graph query engine."""
b363c5b… noreply 2
b363c5b… noreply 3 import json
b363c5b… noreply 4 from unittest.mock import MagicMock
b363c5b… noreply 5
b363c5b… noreply 6 import pytest
b363c5b… noreply 7
b363c5b… noreply 8 from video_processor.integrators.graph_query import GraphQueryEngine, QueryResult
0981a08… noreply 9 from video_processor.integrators.graph_store import InMemoryStore, SQLiteStore
b363c5b… noreply 10
b363c5b… noreply 11
b363c5b… noreply 12 def _make_populated_store():
b363c5b… noreply 13 """Create a store with test data."""
b363c5b… noreply 14 store = InMemoryStore()
b363c5b… noreply 15 store.merge_entity("Python", "technology", ["A programming language"])
b363c5b… noreply 16 store.merge_entity("Django", "technology", ["A web framework"])
b363c5b… noreply 17 store.merge_entity("Alice", "person", ["Software engineer"])
b363c5b… noreply 18 store.merge_entity("Bob", "person", ["Product manager"])
b363c5b… noreply 19 store.merge_entity("Acme Corp", "organization", ["A tech company"])
b363c5b… noreply 20 store.add_relationship("Alice", "Python", "uses")
b363c5b… noreply 21 store.add_relationship("Alice", "Bob", "works_with")
b363c5b… noreply 22 store.add_relationship("Django", "Python", "built_on")
b363c5b… noreply 23 store.add_relationship("Alice", "Acme Corp", "employed_by")
b363c5b… noreply 24 return store
b363c5b… noreply 25
b363c5b… noreply 26
b363c5b… noreply 27 class TestQueryResultToText:
b363c5b… noreply 28 def test_text_with_dict_data(self):
b363c5b… noreply 29 r = QueryResult(
b363c5b… noreply 30 data={"entity_count": 5, "relationship_count": 3},
b363c5b… noreply 31 query_type="filter",
b363c5b… noreply 32 explanation="Stats",
b363c5b… noreply 33 )
b363c5b… noreply 34 text = r.to_text()
b363c5b… noreply 35 assert "entity_count: 5" in text
b363c5b… noreply 36 assert "relationship_count: 3" in text
b363c5b… noreply 37
b363c5b… noreply 38 def test_text_with_list_of_entities(self):
b363c5b… noreply 39 r = QueryResult(
b363c5b… noreply 40 data=[{"name": "Python", "type": "technology", "descriptions": ["A language"]}],
b363c5b… noreply 41 query_type="filter",
b363c5b… noreply 42 )
b363c5b… noreply 43 text = r.to_text()
b363c5b… noreply 44 assert "Python" in text
b363c5b… noreply 45 assert "technology" in text
b363c5b… noreply 46
b363c5b… noreply 47 def test_text_with_empty_list(self):
b363c5b… noreply 48 r = QueryResult(data=[], query_type="filter")
b363c5b… noreply 49 assert "No results" in r.to_text()
b363c5b… noreply 50
b363c5b… noreply 51 def test_text_with_relationships(self):
b363c5b… noreply 52 r = QueryResult(
b363c5b… noreply 53 data=[{"source": "A", "target": "B", "type": "knows"}],
b363c5b… noreply 54 query_type="filter",
b363c5b… noreply 55 )
b363c5b… noreply 56 text = r.to_text()
b363c5b… noreply 57 assert "A" in text
b363c5b… noreply 58 assert "B" in text
b363c5b… noreply 59 assert "knows" in text
b363c5b… noreply 60
b363c5b… noreply 61
b363c5b… noreply 62 class TestQueryResultToJson:
b363c5b… noreply 63 def test_json_roundtrip(self):
b363c5b… noreply 64 r = QueryResult(data={"key": "val"}, query_type="filter", raw_query="test()")
b363c5b… noreply 65 parsed = json.loads(r.to_json())
b363c5b… noreply 66 assert parsed["query_type"] == "filter"
b363c5b… noreply 67 assert parsed["data"]["key"] == "val"
b363c5b… noreply 68 assert parsed["raw_query"] == "test()"
b363c5b… noreply 69
b363c5b… noreply 70
b363c5b… noreply 71 class TestQueryResultToMermaid:
b363c5b… noreply 72 def test_mermaid_with_entities_and_rels(self):
b363c5b… noreply 73 r = QueryResult(
b363c5b… noreply 74 data=[
b363c5b… noreply 75 {"name": "Alice", "type": "person"},
b363c5b… noreply 76 {"name": "Bob", "type": "person"},
b363c5b… noreply 77 {"source": "Alice", "target": "Bob", "type": "knows"},
b363c5b… noreply 78 ],
b363c5b… noreply 79 query_type="filter",
b363c5b… noreply 80 )
b363c5b… noreply 81 mermaid = r.to_mermaid()
b363c5b… noreply 82 assert "graph LR" in mermaid
b363c5b… noreply 83 assert "Alice" in mermaid
b363c5b… noreply 84 assert "Bob" in mermaid
b363c5b… noreply 85 assert "knows" in mermaid
b363c5b… noreply 86
b363c5b… noreply 87 def test_mermaid_empty(self):
b363c5b… noreply 88 r = QueryResult(data=[], query_type="filter")
b363c5b… noreply 89 mermaid = r.to_mermaid()
b363c5b… noreply 90 assert "graph LR" in mermaid
b363c5b… noreply 91
b363c5b… noreply 92
b363c5b… noreply 93 class TestDirectMode:
b363c5b… noreply 94 def test_stats(self):
b363c5b… noreply 95 store = _make_populated_store()
b363c5b… noreply 96 engine = GraphQueryEngine(store)
b363c5b… noreply 97 result = engine.stats()
b363c5b… noreply 98 assert result.data["entity_count"] == 5
b363c5b… noreply 99 assert result.data["relationship_count"] == 4
b363c5b… noreply 100 assert result.data["entity_types"]["technology"] == 2
b363c5b… noreply 101 assert result.data["entity_types"]["person"] == 2
b363c5b… noreply 102
b363c5b… noreply 103 def test_entities_no_filter(self):
b363c5b… noreply 104 store = _make_populated_store()
b363c5b… noreply 105 engine = GraphQueryEngine(store)
b363c5b… noreply 106 result = engine.entities()
b363c5b… noreply 107 assert len(result.data) == 5
b363c5b… noreply 108
b363c5b… noreply 109 def test_entities_filter_by_name(self):
b363c5b… noreply 110 store = _make_populated_store()
b363c5b… noreply 111 engine = GraphQueryEngine(store)
b363c5b… noreply 112 result = engine.entities(name="python")
b363c5b… noreply 113 assert len(result.data) == 1
b363c5b… noreply 114 assert result.data[0]["name"] == "Python"
b363c5b… noreply 115
b363c5b… noreply 116 def test_entities_filter_by_type(self):
b363c5b… noreply 117 store = _make_populated_store()
b363c5b… noreply 118 engine = GraphQueryEngine(store)
b363c5b… noreply 119 result = engine.entities(entity_type="person")
b363c5b… noreply 120 assert len(result.data) == 2
b363c5b… noreply 121 names = {e["name"] for e in result.data}
b363c5b… noreply 122 assert names == {"Alice", "Bob"}
b363c5b… noreply 123
b363c5b… noreply 124 def test_entities_filter_by_both(self):
b363c5b… noreply 125 store = _make_populated_store()
b363c5b… noreply 126 engine = GraphQueryEngine(store)
b363c5b… noreply 127 result = engine.entities(name="ali", entity_type="person")
b363c5b… noreply 128 assert len(result.data) == 1
b363c5b… noreply 129 assert result.data[0]["name"] == "Alice"
b363c5b… noreply 130
b363c5b… noreply 131 def test_entities_case_insensitive(self):
b363c5b… noreply 132 store = _make_populated_store()
b363c5b… noreply 133 engine = GraphQueryEngine(store)
b363c5b… noreply 134 result = engine.entities(name="PYTHON")
b363c5b… noreply 135 assert len(result.data) == 1
b363c5b… noreply 136
b363c5b… noreply 137 def test_relationships_no_filter(self):
b363c5b… noreply 138 store = _make_populated_store()
b363c5b… noreply 139 engine = GraphQueryEngine(store)
b363c5b… noreply 140 result = engine.relationships()
b363c5b… noreply 141 assert len(result.data) == 4
b363c5b… noreply 142
b363c5b… noreply 143 def test_relationships_filter_by_source(self):
b363c5b… noreply 144 store = _make_populated_store()
b363c5b… noreply 145 engine = GraphQueryEngine(store)
b363c5b… noreply 146 result = engine.relationships(source="alice")
b363c5b… noreply 147 assert len(result.data) == 3
b363c5b… noreply 148
b363c5b… noreply 149 def test_relationships_filter_by_type(self):
b363c5b… noreply 150 store = _make_populated_store()
b363c5b… noreply 151 engine = GraphQueryEngine(store)
b363c5b… noreply 152 result = engine.relationships(rel_type="uses")
b363c5b… noreply 153 assert len(result.data) == 1
b363c5b… noreply 154
b363c5b… noreply 155 def test_neighbors(self):
b363c5b… noreply 156 store = _make_populated_store()
b363c5b… noreply 157 engine = GraphQueryEngine(store)
b363c5b… noreply 158 result = engine.neighbors("Alice")
b363c5b… noreply 159 # Alice connects to Python, Bob, Acme Corp
b363c5b… noreply 160 entities = [item for item in result.data if "name" in item]
b363c5b… noreply 161 rels = [item for item in result.data if "source" in item and "target" in item]
b363c5b… noreply 162 assert len(entities) >= 2 # Alice + neighbors
b363c5b… noreply 163 assert len(rels) >= 1
b363c5b… noreply 164
b363c5b… noreply 165 def test_neighbors_not_found(self):
b363c5b… noreply 166 store = _make_populated_store()
b363c5b… noreply 167 engine = GraphQueryEngine(store)
b363c5b… noreply 168 result = engine.neighbors("Ghost")
b363c5b… noreply 169 assert result.data == []
b363c5b… noreply 170 assert "not found" in result.explanation
b363c5b… noreply 171
0981a08… noreply 172 def test_sql_raises_on_inmemory(self):
b363c5b… noreply 173 store = InMemoryStore()
b363c5b… noreply 174 engine = GraphQueryEngine(store)
b363c5b… noreply 175 with pytest.raises(NotImplementedError):
0981a08… noreply 176 engine.sql("SELECT * FROM entities")
b363c5b… noreply 177
b363c5b… noreply 178 def test_entities_limit(self):
b363c5b… noreply 179 store = _make_populated_store()
b363c5b… noreply 180 engine = GraphQueryEngine(store)
b363c5b… noreply 181 result = engine.entities(limit=2)
b363c5b… noreply 182 assert len(result.data) == 2
b363c5b… noreply 183
b363c5b… noreply 184
b363c5b… noreply 185 class TestFromJsonPath:
b363c5b… noreply 186 def test_load_from_json(self, tmp_path):
b363c5b… noreply 187 data = {
b363c5b… noreply 188 "nodes": [
b363c5b… noreply 189 {"name": "Python", "type": "technology", "descriptions": ["A language"]},
b363c5b… noreply 190 {"name": "Alice", "type": "person", "descriptions": ["Engineer"]},
b363c5b… noreply 191 ],
b363c5b… noreply 192 "relationships": [
b363c5b… noreply 193 {"source": "Alice", "target": "Python", "type": "uses"},
b363c5b… noreply 194 ],
b363c5b… noreply 195 }
b363c5b… noreply 196 jf = tmp_path / "kg.json"
b363c5b… noreply 197 jf.write_text(json.dumps(data))
b363c5b… noreply 198 engine = GraphQueryEngine.from_json_path(jf)
b363c5b… noreply 199 result = engine.stats()
b363c5b… noreply 200 assert result.data["entity_count"] == 2
b363c5b… noreply 201 assert result.data["relationship_count"] == 1
b363c5b… noreply 202
b363c5b… noreply 203
0981a08… noreply 204 class TestSQLiteQuery:
0981a08… noreply 205 def test_sql_passthrough(self, tmp_path):
0981a08… noreply 206 store = SQLiteStore(tmp_path / "test.db")
b363c5b… noreply 207 store.merge_entity("Python", "technology", ["A language"])
b363c5b… noreply 208 engine = GraphQueryEngine(store)
0981a08… noreply 209 result = engine.sql("SELECT name FROM entities")
b363c5b… noreply 210 assert len(result.data) >= 1
0981a08… noreply 211 assert result.query_type == "sql"
b363c5b… noreply 212 store.close()
b363c5b… noreply 213
b363c5b… noreply 214 def test_raw_query_on_store(self, tmp_path):
0981a08… noreply 215 store = SQLiteStore(tmp_path / "test.db")
b363c5b… noreply 216 store.merge_entity("Alice", "person", ["Engineer"])
0981a08… noreply 217 rows = store.raw_query("SELECT name FROM entities")
b363c5b… noreply 218 assert len(rows) >= 1
b363c5b… noreply 219 store.close()
b363c5b… noreply 220
b363c5b… noreply 221
b363c5b… noreply 222 class TestAgenticMode:
b363c5b… noreply 223 def test_ask_requires_provider(self):
b363c5b… noreply 224 store = _make_populated_store()
b363c5b… noreply 225 engine = GraphQueryEngine(store, provider_manager=None)
b363c5b… noreply 226 result = engine.ask("What technologies are used?")
b363c5b… noreply 227 assert result.query_type == "agentic"
b363c5b… noreply 228 assert "requires" in result.explanation.lower()
b363c5b… noreply 229
b363c5b… noreply 230 def test_ask_with_mock_llm(self):
b363c5b… noreply 231 store = _make_populated_store()
b363c5b… noreply 232 mock_pm = MagicMock()
b363c5b… noreply 233 # First call: plan generation — return entities action
b363c5b… noreply 234 # Second call: synthesis — return a summary
b363c5b… noreply 235 mock_pm.chat.side_effect = [
b363c5b… noreply 236 '{"action": "entities", "entity_type": "technology"}',
b363c5b… noreply 237 "The knowledge graph contains two technologies: Python and Django.",
b363c5b… noreply 238 ]
b363c5b… noreply 239 engine = GraphQueryEngine(store, provider_manager=mock_pm)
b363c5b… noreply 240 result = engine.ask("What technologies are in the graph?")
b363c5b… noreply 241 assert result.query_type == "agentic"
b363c5b… noreply 242 assert mock_pm.chat.call_count == 2
b363c5b… noreply 243 assert "Python" in result.explanation or len(result.data) >= 1
b363c5b… noreply 244
b363c5b… noreply 245 def test_ask_with_stats_action(self):
b363c5b… noreply 246 store = _make_populated_store()
b363c5b… noreply 247 mock_pm = MagicMock()
b363c5b… noreply 248 mock_pm.chat.side_effect = [
b363c5b… noreply 249 '{"action": "stats"}',
b363c5b… noreply 250 "The graph has 5 entities and 4 relationships.",
b363c5b… noreply 251 ]
b363c5b… noreply 252 engine = GraphQueryEngine(store, provider_manager=mock_pm)
b363c5b… noreply 253 result = engine.ask("How big is this graph?")
b363c5b… noreply 254 assert result.data["entity_count"] == 5
b363c5b… noreply 255
b363c5b… noreply 256 def test_ask_with_neighbors_action(self):
b363c5b… noreply 257 store = _make_populated_store()
b363c5b… noreply 258 mock_pm = MagicMock()
b363c5b… noreply 259 mock_pm.chat.side_effect = [
b363c5b… noreply 260 '{"action": "neighbors", "entity_name": "Alice"}',
b363c5b… noreply 261 "Alice is connected to Python, Bob, and Acme Corp.",
b363c5b… noreply 262 ]
b363c5b… noreply 263 engine = GraphQueryEngine(store, provider_manager=mock_pm)
b363c5b… noreply 264 result = engine.ask("What is Alice connected to?")
b363c5b… noreply 265 assert result.query_type == "agentic"
b363c5b… noreply 266 assert len(result.data) > 0
b363c5b… noreply 267
b363c5b… noreply 268 def test_ask_handles_unparseable_plan(self):
b363c5b… noreply 269 store = _make_populated_store()
b363c5b… noreply 270 mock_pm = MagicMock()
b363c5b… noreply 271 mock_pm.chat.return_value = "I don't understand"
b363c5b… noreply 272 engine = GraphQueryEngine(store, provider_manager=mock_pm)
b363c5b… noreply 273 result = engine.ask("Gibberish?")
b363c5b… noreply 274 assert result.data is None
b363c5b… noreply 275 assert "parse" in result.explanation.lower() or "could not" in result.explanation.lower()
4a3c1b4… noreply 276
4a3c1b4… noreply 277
4a3c1b4… noreply 278 class TestGraphAlgorithms:
4a3c1b4… noreply 279 def test_shortest_path(self):
4a3c1b4… noreply 280 store = InMemoryStore()
4a3c1b4… noreply 281 store.merge_entity("A", "concept", [])
4a3c1b4… noreply 282 store.merge_entity("B", "concept", [])
4a3c1b4… noreply 283 store.merge_entity("C", "concept", [])
4a3c1b4… noreply 284 store.add_relationship("A", "B", "connects")
4a3c1b4… noreply 285 store.add_relationship("B", "C", "connects")
4a3c1b4… noreply 286 engine = GraphQueryEngine(store)
4a3c1b4… noreply 287
4a3c1b4… noreply 288 result = engine.shortest_path("A", "C")
4a3c1b4… noreply 289 assert "Path found" in result.explanation
4a3c1b4… noreply 290 assert len(result.data) > 0
4a3c1b4… noreply 291
4a3c1b4… noreply 292 def test_shortest_path_same_entity(self):
4a3c1b4… noreply 293 store = InMemoryStore()
4a3c1b4… noreply 294 store.merge_entity("X", "concept", [])
4a3c1b4… noreply 295 engine = GraphQueryEngine(store)
4a3c1b4… noreply 296
4a3c1b4… noreply 297 result = engine.shortest_path("X", "X")
4a3c1b4… noreply 298 assert "same entity" in result.explanation.lower()
4a3c1b4… noreply 299
4a3c1b4… noreply 300 def test_shortest_path_not_found(self):
4a3c1b4… noreply 301 store = InMemoryStore()
4a3c1b4… noreply 302 store.merge_entity("A", "concept", [])
4a3c1b4… noreply 303 store.merge_entity("Z", "concept", [])
4a3c1b4… noreply 304 engine = GraphQueryEngine(store)
4a3c1b4… noreply 305
4a3c1b4… noreply 306 result = engine.shortest_path("A", "Z")
4a3c1b4… noreply 307 assert "No path found" in result.explanation
4a3c1b4… noreply 308
4a3c1b4… noreply 309 def test_shortest_path_entity_missing(self):
4a3c1b4… noreply 310 store = InMemoryStore()
4a3c1b4… noreply 311 engine = GraphQueryEngine(store)
4a3c1b4… noreply 312
4a3c1b4… noreply 313 result = engine.shortest_path("Missing", "Also Missing")
4a3c1b4… noreply 314 assert "not found" in result.explanation
4a3c1b4… noreply 315
4a3c1b4… noreply 316 def test_clusters(self):
4a3c1b4… noreply 317 store = InMemoryStore()
4a3c1b4… noreply 318 store.merge_entity("A", "concept", [])
4a3c1b4… noreply 319 store.merge_entity("B", "concept", [])
4a3c1b4… noreply 320 store.add_relationship("A", "B", "connected")
4a3c1b4… noreply 321
4a3c1b4… noreply 322 store.merge_entity("X", "concept", [])
4a3c1b4… noreply 323 store.merge_entity("Y", "concept", [])
4a3c1b4… noreply 324 store.add_relationship("X", "Y", "connected")
4a3c1b4… noreply 325
4a3c1b4… noreply 326 store.merge_entity("Lone", "concept", [])
4a3c1b4… noreply 327
4a3c1b4… noreply 328 engine = GraphQueryEngine(store)
4a3c1b4… noreply 329 result = engine.clusters()
4a3c1b4… noreply 330 assert "3 clusters" in result.explanation
4a3c1b4… noreply 331 assert result.data[0]["size"] == 2
4a3c1b4… noreply 332
4a3c1b4… noreply 333 def test_clusters_empty(self):
4a3c1b4… noreply 334 store = InMemoryStore()
4a3c1b4… noreply 335 engine = GraphQueryEngine(store)
4a3c1b4… noreply 336 result = engine.clusters()
4a3c1b4… noreply 337 assert result.data == []

Keyboard Shortcuts

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