PlanOpticon

planopticon / tests / test_knowledge_graph.py
Source Blame History 538 lines
0981a08… noreply 1 """Tests for the KnowledgeGraph class."""
0981a08… noreply 2
0981a08… noreply 3 import json
0981a08… noreply 4 from unittest.mock import MagicMock, patch
0981a08… noreply 5
0981a08… noreply 6 import pytest
0981a08… noreply 7
0981a08… noreply 8 from video_processor.integrators.knowledge_graph import KnowledgeGraph
0981a08… noreply 9
0981a08… noreply 10
0981a08… noreply 11 @pytest.fixture
0981a08… noreply 12 def mock_pm():
0981a08… noreply 13 """A mock ProviderManager that returns predictable JSON from chat()."""
0981a08… noreply 14 pm = MagicMock()
0981a08… noreply 15 pm.chat.return_value = json.dumps(
0981a08… noreply 16 {
0981a08… noreply 17 "entities": [
0981a08… noreply 18 {"name": "Python", "type": "technology", "description": "A programming language"},
0981a08… noreply 19 {"name": "Alice", "type": "person", "description": "Lead developer"},
0981a08… noreply 20 ],
0981a08… noreply 21 "relationships": [
0981a08… noreply 22 {"source": "Alice", "target": "Python", "type": "uses"},
0981a08… noreply 23 ],
0981a08… noreply 24 }
0981a08… noreply 25 )
0981a08… noreply 26 return pm
0981a08… noreply 27
0981a08… noreply 28
0981a08… noreply 29 @pytest.fixture
0981a08… noreply 30 def kg_no_provider():
0981a08… noreply 31 """KnowledgeGraph with no provider (in-memory store)."""
0981a08… noreply 32 return KnowledgeGraph()
0981a08… noreply 33
0981a08… noreply 34
0981a08… noreply 35 @pytest.fixture
0981a08… noreply 36 def kg_with_provider(mock_pm):
0981a08… noreply 37 """KnowledgeGraph with a mock provider (in-memory store)."""
0981a08… noreply 38 return KnowledgeGraph(provider_manager=mock_pm)
0981a08… noreply 39
0981a08… noreply 40
0981a08… noreply 41 class TestCreation:
0981a08… noreply 42 def test_create_without_db_path(self):
0981a08… noreply 43 kg = KnowledgeGraph()
0981a08… noreply 44 assert kg.pm is None
0981a08… noreply 45 assert kg._store.get_entity_count() == 0
0981a08… noreply 46 assert kg._store.get_relationship_count() == 0
0981a08… noreply 47
0981a08… noreply 48 def test_create_with_db_path(self, tmp_path):
0981a08… noreply 49 db_path = tmp_path / "test.db"
0981a08… noreply 50 kg = KnowledgeGraph(db_path=db_path)
0981a08… noreply 51 assert kg._store.get_entity_count() == 0
0981a08… noreply 52 assert db_path.exists()
0981a08… noreply 53
0981a08… noreply 54 def test_create_with_provider(self, mock_pm):
0981a08… noreply 55 kg = KnowledgeGraph(provider_manager=mock_pm)
0981a08… noreply 56 assert kg.pm is mock_pm
0981a08… noreply 57
0981a08… noreply 58
0981a08… noreply 59 class TestProcessTranscript:
0981a08… noreply 60 def test_process_transcript_extracts_entities(self, kg_with_provider, mock_pm):
0981a08… noreply 61 transcript = {
0981a08… noreply 62 "segments": [
0981a08… noreply 63 {"text": "Alice is using Python for the project", "start": 0.0, "speaker": "Alice"},
0981a08… noreply 64 {"text": "It works great for data processing", "start": 5.0},
0981a08… noreply 65 ]
0981a08… noreply 66 }
0981a08… noreply 67 kg_with_provider.process_transcript(transcript)
0981a08… noreply 68
0981a08… noreply 69 # The mock returns Python and Alice as entities
0981a08… noreply 70 nodes = kg_with_provider.nodes
0981a08… noreply 71 assert "Python" in nodes
0981a08… noreply 72 assert "Alice" in nodes
0981a08… noreply 73 assert nodes["Python"]["type"] == "technology"
0981a08… noreply 74
0981a08… noreply 75 def test_process_transcript_registers_speakers(self, kg_with_provider):
0981a08… noreply 76 transcript = {
0981a08… noreply 77 "segments": [
0981a08… noreply 78 {"text": "Hello everyone", "start": 0.0, "speaker": "Bob"},
0981a08… noreply 79 ]
0981a08… noreply 80 }
0981a08… noreply 81 kg_with_provider.process_transcript(transcript)
0981a08… noreply 82 assert kg_with_provider._store.has_entity("Bob")
0981a08… noreply 83
0981a08… noreply 84 def test_process_transcript_missing_segments(self, kg_with_provider):
0981a08… noreply 85 """Should log warning and return without error."""
0981a08… noreply 86 kg_with_provider.process_transcript({})
0981a08… noreply 87 assert kg_with_provider._store.get_entity_count() == 0
0981a08… noreply 88
0981a08… noreply 89 def test_process_transcript_empty_text_skipped(self, kg_with_provider, mock_pm):
0981a08… noreply 90 transcript = {
0981a08… noreply 91 "segments": [
0981a08… noreply 92 {"text": " ", "start": 0.0},
0981a08… noreply 93 ]
0981a08… noreply 94 }
0981a08… noreply 95 kg_with_provider.process_transcript(transcript)
0981a08… noreply 96 # chat should not be called for empty batches (speaker registration may still happen)
0981a08… noreply 97 mock_pm.chat.assert_not_called()
0981a08… noreply 98
0981a08… noreply 99 def test_process_transcript_batching(self, kg_with_provider, mock_pm):
0981a08… noreply 100 """With batch_size=2, 5 segments should produce 3 batches."""
0981a08… noreply 101 segments = [{"text": f"Segment {i}", "start": float(i)} for i in range(5)]
0981a08… noreply 102 transcript = {"segments": segments}
0981a08… noreply 103 kg_with_provider.process_transcript(transcript, batch_size=2)
0981a08… noreply 104 assert mock_pm.chat.call_count == 3
0981a08… noreply 105
0981a08… noreply 106
0981a08… noreply 107 class TestProcessDiagrams:
0981a08… noreply 108 def test_process_diagrams_with_text(self, kg_with_provider, mock_pm):
0981a08… noreply 109 diagrams = [
0981a08… noreply 110 {"text_content": "Architecture shows Python microservices", "frame_index": 0},
0981a08… noreply 111 ]
0981a08… noreply 112 kg_with_provider.process_diagrams(diagrams)
0981a08… noreply 113
0981a08… noreply 114 # Should have called chat once for the text content
0981a08… noreply 115 assert mock_pm.chat.call_count == 1
0981a08… noreply 116 # diagram_0 entity should exist
0981a08… noreply 117 assert kg_with_provider._store.has_entity("diagram_0")
0981a08… noreply 118
0981a08… noreply 119 def test_process_diagrams_without_text(self, kg_with_provider, mock_pm):
0981a08… noreply 120 diagrams = [
0981a08… noreply 121 {"text_content": "", "frame_index": 5},
0981a08… noreply 122 ]
0981a08… noreply 123 kg_with_provider.process_diagrams(diagrams)
0981a08… noreply 124 # No chat call for empty text
0981a08… noreply 125 mock_pm.chat.assert_not_called()
0981a08… noreply 126 # But diagram entity still created
0981a08… noreply 127 assert kg_with_provider._store.has_entity("diagram_0")
0981a08… noreply 128
0981a08… noreply 129 def test_process_multiple_diagrams(self, kg_with_provider, mock_pm):
0981a08… noreply 130 diagrams = [
0981a08… noreply 131 {"text_content": "Diagram A content", "frame_index": 0},
0981a08… noreply 132 {"text_content": "Diagram B content", "frame_index": 10},
0981a08… noreply 133 ]
0981a08… noreply 134 kg_with_provider.process_diagrams(diagrams)
0981a08… noreply 135 assert kg_with_provider._store.has_entity("diagram_0")
0981a08… noreply 136 assert kg_with_provider._store.has_entity("diagram_1")
2a1b11a… noreply 137
2a1b11a… noreply 138
2a1b11a… noreply 139 class TestProcessScreenshots:
2a1b11a… noreply 140 @pytest.fixture
2a1b11a… noreply 141 def mock_pm(self):
2a1b11a… noreply 142 pm = MagicMock()
2a1b11a… noreply 143 pm.chat.return_value = json.dumps(
2a1b11a… noreply 144 [
2a1b11a… noreply 145 {"name": "Python", "type": "technology", "description": "Language"},
2a1b11a… noreply 146 {"name": "Flask", "type": "technology", "description": "Framework"},
2a1b11a… noreply 147 ]
2a1b11a… noreply 148 )
2a1b11a… noreply 149 return pm
2a1b11a… noreply 150
2a1b11a… noreply 151 @pytest.fixture
2a1b11a… noreply 152 def kg_with_provider(self, mock_pm):
2a1b11a… noreply 153 return KnowledgeGraph(provider_manager=mock_pm)
2a1b11a… noreply 154
2a1b11a… noreply 155 def test_process_screenshots_with_text(self, kg_with_provider, mock_pm):
2a1b11a… noreply 156 screenshots = [
2a1b11a… noreply 157 {
2a1b11a… noreply 158 "text_content": "import flask\napp = Flask(__name__)",
2a1b11a… noreply 159 "content_type": "code",
2a1b11a… noreply 160 "entities": ["Flask", "Python"],
2a1b11a… noreply 161 "frame_index": 3,
2a1b11a… noreply 162 },
2a1b11a… noreply 163 ]
2a1b11a… noreply 164 kg_with_provider.process_screenshots(screenshots)
2a1b11a… noreply 165 # LLM extraction from text_content
2a1b11a… noreply 166 mock_pm.chat.assert_called()
2a1b11a… noreply 167 # Explicitly listed entities should be added
2a1b11a… noreply 168 assert kg_with_provider._store.has_entity("Flask")
2a1b11a… noreply 169 assert kg_with_provider._store.has_entity("Python")
2a1b11a… noreply 170
2a1b11a… noreply 171 def test_process_screenshots_without_text(self, kg_with_provider, mock_pm):
2a1b11a… noreply 172 screenshots = [
2a1b11a… noreply 173 {
2a1b11a… noreply 174 "text_content": "",
2a1b11a… noreply 175 "content_type": "other",
2a1b11a… noreply 176 "entities": ["Docker"],
2a1b11a… noreply 177 "frame_index": 5,
2a1b11a… noreply 178 },
2a1b11a… noreply 179 ]
2a1b11a… noreply 180 kg_with_provider.process_screenshots(screenshots)
2a1b11a… noreply 181 # No chat call for empty text
2a1b11a… noreply 182 mock_pm.chat.assert_not_called()
2a1b11a… noreply 183 # But explicit entities still added
2a1b11a… noreply 184 assert kg_with_provider._store.has_entity("Docker")
2a1b11a… noreply 185
2a1b11a… noreply 186 def test_process_screenshots_empty_entities(self, kg_with_provider):
2a1b11a… noreply 187 screenshots = [
2a1b11a… noreply 188 {
2a1b11a… noreply 189 "text_content": "",
2a1b11a… noreply 190 "content_type": "slide",
2a1b11a… noreply 191 "entities": [],
2a1b11a… noreply 192 "frame_index": 0,
2a1b11a… noreply 193 },
2a1b11a… noreply 194 ]
2a1b11a… noreply 195 kg_with_provider.process_screenshots(screenshots)
2a1b11a… noreply 196 # No crash, no entities added
2a1b11a… noreply 197
2a1b11a… noreply 198 def test_process_screenshots_filters_short_names(self, kg_with_provider):
2a1b11a… noreply 199 screenshots = [
2a1b11a… noreply 200 {
2a1b11a… noreply 201 "text_content": "",
2a1b11a… noreply 202 "entities": ["A", "Go", "Python"],
2a1b11a… noreply 203 "frame_index": 0,
2a1b11a… noreply 204 },
2a1b11a… noreply 205 ]
2a1b11a… noreply 206 kg_with_provider.process_screenshots(screenshots)
2a1b11a… noreply 207 # "A" is too short (< 2 chars), filtered out
2a1b11a… noreply 208 assert not kg_with_provider._store.has_entity("A")
2a1b11a… noreply 209 assert kg_with_provider._store.has_entity("Go")
2a1b11a… noreply 210 assert kg_with_provider._store.has_entity("Python")
0981a08… noreply 211
0981a08… noreply 212
0981a08… noreply 213 class TestToDictFromDict:
0981a08… noreply 214 def test_round_trip_empty(self):
0981a08… noreply 215 kg = KnowledgeGraph()
0981a08… noreply 216 data = kg.to_dict()
0981a08… noreply 217 kg2 = KnowledgeGraph.from_dict(data)
0981a08… noreply 218 assert kg2._store.get_entity_count() == 0
0981a08… noreply 219 assert kg2._store.get_relationship_count() == 0
0981a08… noreply 220
0981a08… noreply 221 def test_round_trip_with_entities(self, kg_with_provider, mock_pm):
0981a08… noreply 222 # Add some content to populate the graph
0981a08… noreply 223 kg_with_provider.add_content("Alice uses Python", "test_source")
0981a08… noreply 224 original = kg_with_provider.to_dict()
0981a08… noreply 225
0981a08… noreply 226 restored = KnowledgeGraph.from_dict(original)
0981a08… noreply 227 restored_dict = restored.to_dict()
0981a08… noreply 228
0981a08… noreply 229 assert len(restored_dict["nodes"]) == len(original["nodes"])
0981a08… noreply 230 assert len(restored_dict["relationships"]) == len(original["relationships"])
0981a08… noreply 231
0981a08… noreply 232 original_names = {n["name"] for n in original["nodes"]}
0981a08… noreply 233 restored_names = {n["name"] for n in restored_dict["nodes"]}
0981a08… noreply 234 assert original_names == restored_names
0981a08… noreply 235
0981a08… noreply 236 def test_round_trip_with_sources(self):
0981a08… noreply 237 kg = KnowledgeGraph()
0981a08… noreply 238 kg.register_source(
0981a08… noreply 239 {
0981a08… noreply 240 "source_id": "src1",
0981a08… noreply 241 "source_type": "video",
0981a08… noreply 242 "title": "Test Video",
0981a08… noreply 243 "ingested_at": "2025-01-01T00:00:00",
0981a08… noreply 244 }
0981a08… noreply 245 )
0981a08… noreply 246 data = kg.to_dict()
0981a08… noreply 247 assert "sources" in data
0981a08… noreply 248 assert data["sources"][0]["source_id"] == "src1"
0981a08… noreply 249
0981a08… noreply 250 kg2 = KnowledgeGraph.from_dict(data)
0981a08… noreply 251 sources = kg2._store.get_sources()
0981a08… noreply 252 assert len(sources) == 1
0981a08… noreply 253 assert sources[0]["source_id"] == "src1"
0981a08… noreply 254
0981a08… noreply 255 def test_from_dict_with_db_path(self, tmp_path):
0981a08… noreply 256 data = {
0981a08… noreply 257 "nodes": [
0981a08… noreply 258 {"name": "TestEntity", "type": "concept", "descriptions": ["A test"]},
0981a08… noreply 259 ],
0981a08… noreply 260 "relationships": [],
0981a08… noreply 261 }
0981a08… noreply 262 db_path = tmp_path / "restored.db"
0981a08… noreply 263 kg = KnowledgeGraph.from_dict(data, db_path=db_path)
0981a08… noreply 264 assert kg._store.has_entity("TestEntity")
0981a08… noreply 265 assert db_path.exists()
0981a08… noreply 266
0981a08… noreply 267
0981a08… noreply 268 class TestSave:
0981a08… noreply 269 def test_save_json(self, tmp_path, kg_with_provider, mock_pm):
0981a08… noreply 270 kg_with_provider.add_content("Alice uses Python", "source1")
0981a08… noreply 271 path = tmp_path / "graph.json"
0981a08… noreply 272 result = kg_with_provider.save(path)
0981a08… noreply 273
0981a08… noreply 274 assert result == path
0981a08… noreply 275 assert path.exists()
0981a08… noreply 276 data = json.loads(path.read_text())
0981a08… noreply 277 assert "nodes" in data
0981a08… noreply 278 assert "relationships" in data
0981a08… noreply 279
0981a08… noreply 280 def test_save_db(self, tmp_path, kg_with_provider, mock_pm):
0981a08… noreply 281 kg_with_provider.add_content("Alice uses Python", "source1")
0981a08… noreply 282 path = tmp_path / "graph.db"
0981a08… noreply 283 result = kg_with_provider.save(path)
0981a08… noreply 284
0981a08… noreply 285 assert result == path
0981a08… noreply 286 assert path.exists()
0981a08… noreply 287
0981a08… noreply 288 def test_save_no_suffix_defaults_to_db(self, tmp_path, kg_with_provider, mock_pm):
0981a08… noreply 289 kg_with_provider.add_content("Alice uses Python", "source1")
0981a08… noreply 290 path = tmp_path / "graph"
0981a08… noreply 291 result = kg_with_provider.save(path)
0981a08… noreply 292 assert result.suffix == ".db"
0981a08… noreply 293 assert result.exists()
0981a08… noreply 294
0981a08… noreply 295 def test_save_creates_parent_dirs(self, tmp_path, kg_with_provider, mock_pm):
0981a08… noreply 296 kg_with_provider.add_content("Alice uses Python", "source1")
0981a08… noreply 297 path = tmp_path / "nested" / "dir" / "graph.json"
0981a08… noreply 298 result = kg_with_provider.save(path)
0981a08… noreply 299 assert result.exists()
0981a08… noreply 300
0981a08… noreply 301 def test_save_unknown_suffix_falls_back_to_json(self, tmp_path):
0981a08… noreply 302 kg = KnowledgeGraph()
0981a08… noreply 303 kg._store.merge_entity("TestNode", "concept", ["test"])
0981a08… noreply 304 path = tmp_path / "graph.xyz"
0981a08… noreply 305 result = kg.save(path)
0981a08… noreply 306 assert result.exists()
0981a08… noreply 307 # Should be valid JSON
0981a08… noreply 308 data = json.loads(path.read_text())
0981a08… noreply 309 assert "nodes" in data
0981a08… noreply 310
0981a08… noreply 311
0981a08… noreply 312 class TestMerge:
0981a08… noreply 313 def test_merge_disjoint(self):
0981a08… noreply 314 kg1 = KnowledgeGraph()
0981a08… noreply 315 kg1._store.merge_entity("Alice", "person", ["Developer"])
0981a08… noreply 316
0981a08… noreply 317 kg2 = KnowledgeGraph()
0981a08… noreply 318 kg2._store.merge_entity("Bob", "person", ["Manager"])
0981a08… noreply 319
0981a08… noreply 320 kg1.merge(kg2)
0981a08… noreply 321 assert kg1._store.has_entity("Alice")
0981a08… noreply 322 assert kg1._store.has_entity("Bob")
0981a08… noreply 323 assert kg1._store.get_entity_count() == 2
0981a08… noreply 324
0981a08… noreply 325 def test_merge_overlapping_entities_descriptions_merged(self):
0981a08… noreply 326 kg1 = KnowledgeGraph()
0981a08… noreply 327 kg1._store.merge_entity("Python", "concept", ["A language"])
0981a08… noreply 328
0981a08… noreply 329 kg2 = KnowledgeGraph()
0981a08… noreply 330 kg2._store.merge_entity("Python", "technology", ["Programming language"])
0981a08… noreply 331
0981a08… noreply 332 kg1.merge(kg2)
0981a08… noreply 333 entity = kg1._store.get_entity("Python")
0981a08… noreply 334 # Descriptions from both should be present
0981a08… noreply 335 descs = entity["descriptions"]
0981a08… noreply 336 if isinstance(descs, set):
0981a08… noreply 337 descs = list(descs)
0981a08… noreply 338 assert "A language" in descs
0981a08… noreply 339 assert "Programming language" in descs
0981a08… noreply 340
0981a08… noreply 341 def test_merge_overlapping_entities_with_sqlite(self, tmp_path):
0981a08… noreply 342 """SQLiteStore does update type on merge_entity, so type resolution works there."""
0981a08… noreply 343 kg1 = KnowledgeGraph(db_path=tmp_path / "kg1.db")
0981a08… noreply 344 kg1._store.merge_entity("Python", "concept", ["A language"])
0981a08… noreply 345
0981a08… noreply 346 kg2 = KnowledgeGraph(db_path=tmp_path / "kg2.db")
0981a08… noreply 347 kg2._store.merge_entity("Python", "technology", ["Programming language"])
0981a08… noreply 348
0981a08… noreply 349 kg1.merge(kg2)
0981a08… noreply 350 entity = kg1._store.get_entity("Python")
0981a08… noreply 351 # SQLiteStore overwrites type — merge resolves to more specific
0981a08… noreply 352 # (The merge method computes the resolved type and passes it to merge_entity,
0981a08… noreply 353 # but InMemoryStore ignores type for existing entities while SQLiteStore does not)
0981a08… noreply 354 assert entity is not None
0981a08… noreply 355 assert kg1._store.get_entity_count() == 1
0981a08… noreply 356
0981a08… noreply 357 def test_merge_fuzzy_match(self):
0981a08… noreply 358 kg1 = KnowledgeGraph()
0981a08… noreply 359 kg1._store.merge_entity("JavaScript", "technology", ["A language"])
0981a08… noreply 360
0981a08… noreply 361 kg2 = KnowledgeGraph()
0981a08… noreply 362 kg2._store.merge_entity("Javascript", "technology", ["Web language"])
0981a08… noreply 363
0981a08… noreply 364 kg1.merge(kg2)
0981a08… noreply 365 # Should fuzzy-match and merge, not create two entities
0981a08… noreply 366 assert kg1._store.get_entity_count() == 1
0981a08… noreply 367 entity = kg1._store.get_entity("JavaScript")
0981a08… noreply 368 assert entity is not None
0981a08… noreply 369
0981a08… noreply 370 def test_merge_relationships(self):
0981a08… noreply 371 kg1 = KnowledgeGraph()
0981a08… noreply 372 kg1._store.merge_entity("Alice", "person", [])
0981a08… noreply 373
0981a08… noreply 374 kg2 = KnowledgeGraph()
0981a08… noreply 375 kg2._store.merge_entity("Bob", "person", [])
0981a08… noreply 376 kg2._store.add_relationship("Alice", "Bob", "collaborates_with")
0981a08… noreply 377
0981a08… noreply 378 kg1.merge(kg2)
0981a08… noreply 379 rels = kg1._store.get_all_relationships()
0981a08… noreply 380 assert len(rels) == 1
0981a08… noreply 381 assert rels[0]["type"] == "collaborates_with"
0981a08… noreply 382
0981a08… noreply 383 def test_merge_sources(self):
0981a08… noreply 384 kg1 = KnowledgeGraph()
0981a08… noreply 385 kg2 = KnowledgeGraph()
0981a08… noreply 386 kg2.register_source(
0981a08… noreply 387 {
0981a08… noreply 388 "source_id": "vid2",
0981a08… noreply 389 "source_type": "video",
0981a08… noreply 390 "title": "Video 2",
0981a08… noreply 391 "ingested_at": "2025-01-01T00:00:00",
0981a08… noreply 392 }
0981a08… noreply 393 )
0981a08… noreply 394 kg1.merge(kg2)
0981a08… noreply 395 sources = kg1._store.get_sources()
0981a08… noreply 396 assert len(sources) == 1
0981a08… noreply 397 assert sources[0]["source_id"] == "vid2"
0981a08… noreply 398
0981a08… noreply 399 def test_merge_type_specificity_with_sqlite(self, tmp_path):
0981a08… noreply 400 """Type specificity resolution works with SQLiteStore which updates type."""
0981a08… noreply 401 kg1 = KnowledgeGraph(db_path=tmp_path / "kg1.db")
0981a08… noreply 402 kg1._store.merge_entity("React", "concept", [])
0981a08… noreply 403
0981a08… noreply 404 kg2 = KnowledgeGraph(db_path=tmp_path / "kg2.db")
0981a08… noreply 405 kg2._store.merge_entity("React", "technology", [])
0981a08… noreply 406
0981a08… noreply 407 kg1.merge(kg2)
0981a08… noreply 408 entity = kg1._store.get_entity("React")
0981a08… noreply 409 assert entity is not None
0981a08… noreply 410 assert kg1._store.get_entity_count() == 1
0981a08… noreply 411
0981a08… noreply 412
0981a08… noreply 413 class TestRegisterSource:
0981a08… noreply 414 def test_register_and_retrieve(self):
0981a08… noreply 415 kg = KnowledgeGraph()
0981a08… noreply 416 source = {
0981a08… noreply 417 "source_id": "src123",
0981a08… noreply 418 "source_type": "video",
0981a08… noreply 419 "title": "Meeting Recording",
0981a08… noreply 420 "path": "/tmp/meeting.mp4",
0981a08… noreply 421 "ingested_at": "2025-06-01T10:00:00",
0981a08… noreply 422 }
0981a08… noreply 423 kg.register_source(source)
0981a08… noreply 424 sources = kg._store.get_sources()
0981a08… noreply 425 assert len(sources) == 1
0981a08… noreply 426 assert sources[0]["source_id"] == "src123"
0981a08… noreply 427 assert sources[0]["title"] == "Meeting Recording"
0981a08… noreply 428
0981a08… noreply 429 def test_register_multiple_sources(self):
0981a08… noreply 430 kg = KnowledgeGraph()
0981a08… noreply 431 for i in range(3):
0981a08… noreply 432 kg.register_source(
0981a08… noreply 433 {
0981a08… noreply 434 "source_id": f"src{i}",
0981a08… noreply 435 "source_type": "video",
0981a08… noreply 436 "title": f"Video {i}",
0981a08… noreply 437 "ingested_at": "2025-01-01",
0981a08… noreply 438 }
0981a08… noreply 439 )
0981a08… noreply 440 assert len(kg._store.get_sources()) == 3
0981a08… noreply 441
0981a08… noreply 442
0981a08… noreply 443 class TestClassifyForPlanning:
0981a08… noreply 444 @patch("video_processor.integrators.knowledge_graph.TaxonomyClassifier", create=True)
0981a08… noreply 445 def test_classify_calls_taxonomy(self, mock_cls):
0981a08… noreply 446 """classify_for_planning should delegate to TaxonomyClassifier."""
0981a08… noreply 447 mock_instance = MagicMock()
0981a08… noreply 448 mock_instance.classify_entities.return_value = {"goals": [], "risks": []}
0981a08… noreply 449
0981a08… noreply 450 with patch(
0981a08… noreply 451 "video_processor.integrators.taxonomy.TaxonomyClassifier",
0981a08… noreply 452 return_value=mock_instance,
0981a08… noreply 453 ):
0981a08… noreply 454 kg = KnowledgeGraph()
0981a08… noreply 455 kg._store.merge_entity("Ship MVP", "concept", ["Launch the product"])
0981a08… noreply 456 kg.classify_for_planning()
0981a08… noreply 457
0981a08… noreply 458 mock_instance.classify_entities.assert_called_once()
0981a08… noreply 459
0981a08… noreply 460
0981a08… noreply 461 class TestExtractEntitiesAndRelationships:
0981a08… noreply 462 def test_returns_entities_and_relationships(self, kg_with_provider):
0981a08… noreply 463 entities, rels = kg_with_provider.extract_entities_and_relationships("Alice uses Python")
0981a08… noreply 464 assert len(entities) == 2
0981a08… noreply 465 assert len(rels) == 1
0981a08… noreply 466 assert entities[0].name == "Python"
0981a08… noreply 467 assert rels[0].source == "Alice"
0981a08… noreply 468 assert rels[0].target == "Python"
0981a08… noreply 469
0981a08… noreply 470 def test_no_provider_returns_empty(self, kg_no_provider):
0981a08… noreply 471 entities, rels = kg_no_provider.extract_entities_and_relationships("Some text")
0981a08… noreply 472 assert entities == []
0981a08… noreply 473 assert rels == []
0981a08… noreply 474
0981a08… noreply 475 def test_handles_flat_list_response(self, mock_pm):
0981a08… noreply 476 """If the model returns a flat entity list, it should still parse entities."""
0981a08… noreply 477 mock_pm.chat.return_value = json.dumps(
0981a08… noreply 478 [
0981a08… noreply 479 {"name": "Docker", "type": "technology", "description": "Container platform"},
0981a08… noreply 480 ]
0981a08… noreply 481 )
0981a08… noreply 482 kg = KnowledgeGraph(provider_manager=mock_pm)
0981a08… noreply 483 entities, rels = kg.extract_entities_and_relationships("Using Docker")
0981a08… noreply 484 assert len(entities) == 1
0981a08… noreply 485 assert entities[0].name == "Docker"
0981a08… noreply 486 assert rels == []
0981a08… noreply 487
0981a08… noreply 488 def test_handles_malformed_json(self, mock_pm):
0981a08… noreply 489 mock_pm.chat.return_value = "not valid json at all"
0981a08… noreply 490 kg = KnowledgeGraph(provider_manager=mock_pm)
0981a08… noreply 491 entities, rels = kg.extract_entities_and_relationships("text")
0981a08… noreply 492 assert entities == []
0981a08… noreply 493 assert rels == []
0981a08… noreply 494
0981a08… noreply 495
0981a08… noreply 496 class TestNodeAndRelationshipProperties:
0981a08… noreply 497 def test_nodes_property(self, kg_with_provider, mock_pm):
0981a08… noreply 498 kg_with_provider.add_content("Alice uses Python", "src")
0981a08… noreply 499 nodes = kg_with_provider.nodes
0981a08… noreply 500 assert isinstance(nodes, dict)
0981a08… noreply 501 for name, node in nodes.items():
0981a08… noreply 502 assert "name" in node
0981a08… noreply 503 assert "type" in node
0981a08… noreply 504 assert "descriptions" in node
0981a08… noreply 505
0981a08… noreply 506 def test_relationships_property(self, kg_with_provider, mock_pm):
0981a08… noreply 507 kg_with_provider.add_content("Alice uses Python", "src")
0981a08… noreply 508 rels = kg_with_provider.relationships
0981a08… noreply 509 assert isinstance(rels, list)
0981a08… noreply 510 if rels:
0981a08… noreply 511 assert "source" in rels[0]
0981a08… noreply 512 assert "target" in rels[0]
0981a08… noreply 513 assert "type" in rels[0]
0981a08… noreply 514
0981a08… noreply 515
0981a08… noreply 516 class TestToData:
0981a08… noreply 517 def test_to_data_returns_pydantic_model(self, kg_with_provider, mock_pm):
0981a08… noreply 518 kg_with_provider.add_content("Alice uses Python", "src")
0981a08… noreply 519 data = kg_with_provider.to_data()
0981a08… noreply 520 from video_processor.models import KnowledgeGraphData
0981a08… noreply 521
0981a08… noreply 522 assert isinstance(data, KnowledgeGraphData)
0981a08… noreply 523 assert len(data.nodes) > 0
0981a08… noreply 524 assert all(hasattr(n, "name") for n in data.nodes)
0981a08… noreply 525
0981a08… noreply 526 def test_to_data_includes_sources(self):
0981a08… noreply 527 kg = KnowledgeGraph()
0981a08… noreply 528 kg.register_source(
0981a08… noreply 529 {
0981a08… noreply 530 "source_id": "s1",
0981a08… noreply 531 "source_type": "video",
0981a08… noreply 532 "title": "Test",
0981a08… noreply 533 "ingested_at": "2025-01-01",
0981a08… noreply 534 }
0981a08… noreply 535 )
0981a08… noreply 536 data = kg.to_data()
0981a08… noreply 537 assert len(data.sources) == 1
0981a08… noreply 538 assert data.sources[0].source_id == "s1"

Keyboard Shortcuts

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