PlanOpticon

planopticon / tests / test_graph_store.py
Source Blame History 274 lines
0ad36b7… noreply 1 """Tests for graph storage backends."""
0ad36b7… noreply 2
0981a08… noreply 3 from video_processor.integrators.graph_store import InMemoryStore, SQLiteStore, create_store
0ad36b7… noreply 4
0ad36b7… noreply 5
0ad36b7… noreply 6 class TestInMemoryStore:
0ad36b7… noreply 7 def test_merge_entity_creates_new(self):
0ad36b7… noreply 8 store = InMemoryStore()
0ad36b7… noreply 9 store.merge_entity("Python", "technology", ["A programming language"])
0ad36b7… noreply 10 assert store.get_entity_count() == 1
0ad36b7… noreply 11 entity = store.get_entity("python")
0ad36b7… noreply 12 assert entity is not None
0ad36b7… noreply 13 assert entity["name"] == "Python"
0ad36b7… noreply 14 assert entity["type"] == "technology"
0ad36b7… noreply 15 assert "A programming language" in entity["descriptions"]
0ad36b7… noreply 16
0ad36b7… noreply 17 def test_merge_entity_case_insensitive_dedup(self):
0ad36b7… noreply 18 store = InMemoryStore()
0ad36b7… noreply 19 store.merge_entity("Python", "technology", ["Language"])
0ad36b7… noreply 20 store.merge_entity("python", "technology", ["Snake-based"])
0ad36b7… noreply 21 store.merge_entity("PYTHON", "technology", ["Popular"])
0ad36b7… noreply 22 assert store.get_entity_count() == 1
0ad36b7… noreply 23 entity = store.get_entity("Python")
0ad36b7… noreply 24 assert entity is not None
0ad36b7… noreply 25 assert "Language" in entity["descriptions"]
0ad36b7… noreply 26 assert "Snake-based" in entity["descriptions"]
0ad36b7… noreply 27 assert "Popular" in entity["descriptions"]
0ad36b7… noreply 28
0ad36b7… noreply 29 def test_add_occurrence(self):
0ad36b7… noreply 30 store = InMemoryStore()
0ad36b7… noreply 31 store.merge_entity("Alice", "person", ["Engineer"])
0ad36b7… noreply 32 store.add_occurrence("Alice", "transcript_0", timestamp=10.5, text="Alice said...")
0ad36b7… noreply 33 entity = store.get_entity("alice")
0ad36b7… noreply 34 assert len(entity["occurrences"]) == 1
0ad36b7… noreply 35 assert entity["occurrences"][0]["source"] == "transcript_0"
0ad36b7… noreply 36 assert entity["occurrences"][0]["timestamp"] == 10.5
0ad36b7… noreply 37
0ad36b7… noreply 38 def test_add_occurrence_nonexistent_entity(self):
0ad36b7… noreply 39 store = InMemoryStore()
0ad36b7… noreply 40 store.add_occurrence("Ghost", "transcript_0")
0ad36b7… noreply 41 # Should not crash, just no-op
0ad36b7… noreply 42 assert store.get_entity_count() == 0
0ad36b7… noreply 43
0ad36b7… noreply 44 def test_add_relationship(self):
0ad36b7… noreply 45 store = InMemoryStore()
0ad36b7… noreply 46 store.merge_entity("Alice", "person", [])
0ad36b7… noreply 47 store.merge_entity("Bob", "person", [])
0ad36b7… noreply 48 store.add_relationship("Alice", "Bob", "knows", content_source="t0", timestamp=5.0)
0ad36b7… noreply 49 assert store.get_relationship_count() == 1
0ad36b7… noreply 50 rels = store.get_all_relationships()
0ad36b7… noreply 51 assert rels[0]["source"] == "Alice"
0ad36b7… noreply 52 assert rels[0]["target"] == "Bob"
0ad36b7… noreply 53 assert rels[0]["type"] == "knows"
0ad36b7… noreply 54
0ad36b7… noreply 55 def test_has_entity(self):
0ad36b7… noreply 56 store = InMemoryStore()
0ad36b7… noreply 57 assert not store.has_entity("Python")
0ad36b7… noreply 58 store.merge_entity("Python", "technology", [])
0ad36b7… noreply 59 assert store.has_entity("Python")
0ad36b7… noreply 60 assert store.has_entity("python")
0ad36b7… noreply 61 assert store.has_entity("PYTHON")
0ad36b7… noreply 62
0ad36b7… noreply 63 def test_get_entity_not_found(self):
0ad36b7… noreply 64 store = InMemoryStore()
0ad36b7… noreply 65 assert store.get_entity("nonexistent") is None
0ad36b7… noreply 66
0ad36b7… noreply 67 def test_get_all_entities(self):
0ad36b7… noreply 68 store = InMemoryStore()
0ad36b7… noreply 69 store.merge_entity("Alice", "person", ["Engineer"])
0ad36b7… noreply 70 store.merge_entity("Bob", "person", ["Manager"])
0ad36b7… noreply 71 entities = store.get_all_entities()
0ad36b7… noreply 72 assert len(entities) == 2
0ad36b7… noreply 73 names = {e["name"] for e in entities}
0ad36b7… noreply 74 assert names == {"Alice", "Bob"}
0ad36b7… noreply 75
0ad36b7… noreply 76 def test_to_dict_format(self):
0ad36b7… noreply 77 store = InMemoryStore()
0ad36b7… noreply 78 store.merge_entity("Python", "technology", ["A language"])
0ad36b7… noreply 79 store.merge_entity("Django", "technology", ["A framework"])
0ad36b7… noreply 80 store.add_relationship("Django", "Python", "uses")
0ad36b7… noreply 81 store.add_occurrence("Python", "transcript_0", timestamp=1.0, text="mentioned Python")
0ad36b7… noreply 82
0ad36b7… noreply 83 data = store.to_dict()
0ad36b7… noreply 84 assert "nodes" in data
0ad36b7… noreply 85 assert "relationships" in data
0ad36b7… noreply 86 assert len(data["nodes"]) == 2
0ad36b7… noreply 87 assert len(data["relationships"]) == 1
0ad36b7… noreply 88
0ad36b7… noreply 89 # Descriptions should be lists (not sets)
0ad36b7… noreply 90 for node in data["nodes"]:
0ad36b7… noreply 91 assert isinstance(node["descriptions"], list)
0ad36b7… noreply 92 assert "id" in node
0ad36b7… noreply 93 assert "name" in node
0ad36b7… noreply 94 assert "type" in node
0ad36b7… noreply 95
0ad36b7… noreply 96 def test_to_dict_roundtrip(self):
0ad36b7… noreply 97 """Verify to_dict produces data that can reload into a new store."""
0ad36b7… noreply 98 store = InMemoryStore()
0ad36b7… noreply 99 store.merge_entity("Alice", "person", ["Engineer"])
0ad36b7… noreply 100 store.merge_entity("Bob", "person", ["Manager"])
0ad36b7… noreply 101 store.add_relationship("Alice", "Bob", "reports_to")
0ad36b7… noreply 102 store.add_occurrence("Alice", "src", timestamp=1.0, text="hello")
0ad36b7… noreply 103
0ad36b7… noreply 104 data = store.to_dict()
0ad36b7… noreply 105
0ad36b7… noreply 106 # Reload into a new store
0ad36b7… noreply 107 store2 = InMemoryStore()
0ad36b7… noreply 108 for node in data["nodes"]:
0ad36b7… noreply 109 store2.merge_entity(
0ad36b7… noreply 110 node["name"], node["type"], node["descriptions"], node.get("source")
0ad36b7… noreply 111 )
0ad36b7… noreply 112 for occ in node.get("occurrences", []):
0ad36b7… noreply 113 store2.add_occurrence(
0ad36b7… noreply 114 node["name"], occ["source"], occ.get("timestamp"), occ.get("text")
0ad36b7… noreply 115 )
0ad36b7… noreply 116 for rel in data["relationships"]:
0ad36b7… noreply 117 store2.add_relationship(
0ad36b7… noreply 118 rel["source"],
0ad36b7… noreply 119 rel["target"],
0ad36b7… noreply 120 rel["type"],
0ad36b7… noreply 121 rel.get("content_source"),
0ad36b7… noreply 122 rel.get("timestamp"),
0ad36b7… noreply 123 )
0ad36b7… noreply 124
0ad36b7… noreply 125 assert store2.get_entity_count() == 2
0ad36b7… noreply 126 assert store2.get_relationship_count() == 1
0ad36b7… noreply 127
0ad36b7… noreply 128 def test_empty_store(self):
0ad36b7… noreply 129 store = InMemoryStore()
0ad36b7… noreply 130 assert store.get_entity_count() == 0
0ad36b7… noreply 131 assert store.get_relationship_count() == 0
0ad36b7… noreply 132 assert store.get_all_entities() == []
0ad36b7… noreply 133 assert store.get_all_relationships() == []
0ad36b7… noreply 134 data = store.to_dict()
0ad36b7… noreply 135 assert data == {"nodes": [], "relationships": []}
0ad36b7… noreply 136
0ad36b7… noreply 137
0ad36b7… noreply 138 class TestCreateStore:
0ad36b7… noreply 139 def test_returns_in_memory_without_path(self):
0ad36b7… noreply 140 store = create_store()
0ad36b7… noreply 141 assert isinstance(store, InMemoryStore)
0ad36b7… noreply 142
0ad36b7… noreply 143 def test_returns_in_memory_with_none_path(self):
0ad36b7… noreply 144 store = create_store(db_path=None)
0ad36b7… noreply 145 assert isinstance(store, InMemoryStore)
0ad36b7… noreply 146
0981a08… noreply 147 def test_returns_sqlite_with_path(self, tmp_path):
0ad36b7… noreply 148 store = create_store(db_path=tmp_path / "test.db")
0981a08… noreply 149 assert isinstance(store, SQLiteStore)
0ad36b7… noreply 150 store.merge_entity("Test", "concept", ["test entity"])
0ad36b7… noreply 151 assert store.get_entity_count() == 1
0981a08… noreply 152 store.close()
0ad36b7… noreply 153
0ad36b7… noreply 154
0981a08… noreply 155 class TestSQLiteStore:
0ad36b7… noreply 156 def test_create_and_query_entity(self, tmp_path):
0981a08… noreply 157 store = SQLiteStore(tmp_path / "test.db")
0ad36b7… noreply 158 store.merge_entity("Python", "technology", ["A language"])
0ad36b7… noreply 159 assert store.get_entity_count() == 1
0ad36b7… noreply 160 entity = store.get_entity("python")
0ad36b7… noreply 161 assert entity is not None
0ad36b7… noreply 162 assert entity["name"] == "Python"
0ad36b7… noreply 163 store.close()
0ad36b7… noreply 164
0ad36b7… noreply 165 def test_case_insensitive_merge(self, tmp_path):
0981a08… noreply 166 store = SQLiteStore(tmp_path / "test.db")
0ad36b7… noreply 167 store.merge_entity("Python", "technology", ["Language"])
0ad36b7… noreply 168 store.merge_entity("python", "technology", ["Snake-based"])
0ad36b7… noreply 169 assert store.get_entity_count() == 1
0ad36b7… noreply 170 entity = store.get_entity("python")
0ad36b7… noreply 171 assert "Language" in entity["descriptions"]
0ad36b7… noreply 172 assert "Snake-based" in entity["descriptions"]
0ad36b7… noreply 173 store.close()
0ad36b7… noreply 174
0ad36b7… noreply 175 def test_relationships(self, tmp_path):
0981a08… noreply 176 store = SQLiteStore(tmp_path / "test.db")
0ad36b7… noreply 177 store.merge_entity("Alice", "person", [])
0ad36b7… noreply 178 store.merge_entity("Bob", "person", [])
0ad36b7… noreply 179 store.add_relationship("Alice", "Bob", "knows")
0ad36b7… noreply 180 assert store.get_relationship_count() == 1
0ad36b7… noreply 181 rels = store.get_all_relationships()
0ad36b7… noreply 182 assert rels[0]["source"] == "Alice"
0ad36b7… noreply 183 assert rels[0]["target"] == "Bob"
0ad36b7… noreply 184 store.close()
0ad36b7… noreply 185
0ad36b7… noreply 186 def test_occurrences(self, tmp_path):
0981a08… noreply 187 store = SQLiteStore(tmp_path / "test.db")
0ad36b7… noreply 188 store.merge_entity("Alice", "person", ["Engineer"])
0ad36b7… noreply 189 store.add_occurrence("Alice", "transcript_0", timestamp=10.5, text="Alice said...")
0ad36b7… noreply 190 entity = store.get_entity("alice")
0ad36b7… noreply 191 assert len(entity["occurrences"]) == 1
0ad36b7… noreply 192 assert entity["occurrences"][0]["source"] == "transcript_0"
0ad36b7… noreply 193 store.close()
0ad36b7… noreply 194
0981a08… noreply 195 def test_occurrence_nonexistent_entity(self, tmp_path):
0981a08… noreply 196 store = SQLiteStore(tmp_path / "test.db")
0981a08… noreply 197 store.add_occurrence("Ghost", "transcript_0")
0981a08… noreply 198 assert store.get_entity_count() == 0
0981a08… noreply 199 store.close()
0ad36b7… noreply 200
0981a08… noreply 201 def test_persistence(self, tmp_path):
0ad36b7… noreply 202 db_path = tmp_path / "persist.db"
0ad36b7… noreply 203
0981a08… noreply 204 store1 = SQLiteStore(db_path)
0ad36b7… noreply 205 store1.merge_entity("Python", "technology", ["A language"])
0ad36b7… noreply 206 store1.close()
0ad36b7… noreply 207
0981a08… noreply 208 store2 = SQLiteStore(db_path)
0ad36b7… noreply 209 assert store2.get_entity_count() == 1
0ad36b7… noreply 210 entity = store2.get_entity("python")
0ad36b7… noreply 211 assert entity["name"] == "Python"
0ad36b7… noreply 212 store2.close()
0ad36b7… noreply 213
0ad36b7… noreply 214 def test_to_dict_format(self, tmp_path):
0981a08… noreply 215 store = SQLiteStore(tmp_path / "test.db")
0ad36b7… noreply 216 store.merge_entity("Python", "technology", ["A language"])
0ad36b7… noreply 217 store.merge_entity("Django", "technology", ["A framework"])
0ad36b7… noreply 218 store.add_relationship("Django", "Python", "uses")
0ad36b7… noreply 219
0ad36b7… noreply 220 data = store.to_dict()
0ad36b7… noreply 221 assert len(data["nodes"]) == 2
0ad36b7… noreply 222 assert len(data["relationships"]) == 1
0ad36b7… noreply 223
0ad36b7… noreply 224 for node in data["nodes"]:
0ad36b7… noreply 225 assert isinstance(node["descriptions"], list)
0ad36b7… noreply 226 assert "id" in node
0ad36b7… noreply 227 assert "name" in node
0ad36b7… noreply 228
0ad36b7… noreply 229 store.close()
0ad36b7… noreply 230
0ad36b7… noreply 231 def test_has_entity(self, tmp_path):
0981a08… noreply 232 store = SQLiteStore(tmp_path / "test.db")
0ad36b7… noreply 233 assert not store.has_entity("Python")
0ad36b7… noreply 234 store.merge_entity("Python", "technology", [])
0ad36b7… noreply 235 assert store.has_entity("Python")
0ad36b7… noreply 236 assert store.has_entity("python")
0981a08… noreply 237 store.close()
0981a08… noreply 238
0981a08… noreply 239 def test_raw_query(self, tmp_path):
0981a08… noreply 240 store = SQLiteStore(tmp_path / "test.db")
0981a08… noreply 241 store.merge_entity("Alice", "person", ["Engineer"])
0981a08… noreply 242 rows = store.raw_query("SELECT name FROM entities")
0981a08… noreply 243 assert len(rows) >= 1
0981a08… noreply 244 assert rows[0][0] == "Alice"
0981a08… noreply 245 store.close()
0981a08… noreply 246
0981a08… noreply 247 def test_typed_relationship(self, tmp_path):
0981a08… noreply 248 store = SQLiteStore(tmp_path / "test.db")
0981a08… noreply 249 store.merge_entity("Django", "technology", [])
0981a08… noreply 250 store.merge_entity("Python", "technology", [])
0981a08… noreply 251 store.add_typed_relationship("Django", "Python", "DEPENDS_ON", {"version": "3.10"})
0981a08… noreply 252 rels = store.get_all_relationships()
0981a08… noreply 253 assert len(rels) == 1
0981a08… noreply 254 assert rels[0]["type"] == "DEPENDS_ON"
0981a08… noreply 255 store.close()
0981a08… noreply 256
0981a08… noreply 257 def test_set_entity_properties(self, tmp_path):
0981a08… noreply 258 store = SQLiteStore(tmp_path / "test.db")
0981a08… noreply 259 store.merge_entity("Python", "technology", [])
0981a08… noreply 260 assert store.set_entity_properties("Python", {"version": "3.12", "stable": True})
0981a08… noreply 261 assert not store.set_entity_properties("Ghost", {"key": "val"})
0981a08… noreply 262 store.close()
0981a08… noreply 263
0981a08… noreply 264 def test_has_relationship(self, tmp_path):
0981a08… noreply 265 store = SQLiteStore(tmp_path / "test.db")
0981a08… noreply 266 store.merge_entity("Alice", "person", [])
0981a08… noreply 267 store.merge_entity("Bob", "person", [])
0981a08… noreply 268 store.add_relationship("Alice", "Bob", "knows")
0981a08… noreply 269 assert store.has_relationship("Alice", "Bob")
0981a08… noreply 270 assert store.has_relationship("alice", "bob")
0981a08… noreply 271 assert store.has_relationship("Alice", "Bob", "knows")
0981a08… noreply 272 assert not store.has_relationship("Alice", "Bob", "hates")
0981a08… noreply 273 assert not store.has_relationship("Bob", "Alice")
0ad36b7… noreply 274 store.close()

Keyboard Shortcuts

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