PlanOpticon

planopticon / tests / test_visualization.py
Source Blame History 339 lines
0981a08… noreply 1 """Tests for video_processor.utils.visualization module."""
0981a08… noreply 2
0981a08… noreply 3 import pytest
0981a08… noreply 4
0981a08… noreply 5 nx = pytest.importorskip("networkx", reason="networkx not installed")
0981a08… noreply 6
0981a08… noreply 7 from video_processor.utils.visualization import ( # noqa: E402
0981a08… noreply 8 compute_graph_stats,
0981a08… noreply 9 filter_graph,
0981a08… noreply 10 generate_mermaid,
0981a08… noreply 11 graph_to_d3_json,
0981a08… noreply 12 graph_to_dot,
0981a08… noreply 13 graph_to_networkx,
0981a08… noreply 14 )
0981a08… noreply 15
0981a08… noreply 16
0981a08… noreply 17 @pytest.fixture
0981a08… noreply 18 def sample_kg_data():
0981a08… noreply 19 """Mock knowledge graph data matching to_dict() format."""
0981a08… noreply 20 return {
0981a08… noreply 21 "nodes": [
0981a08… noreply 22 {
0981a08… noreply 23 "id": "Alice",
0981a08… noreply 24 "name": "Alice",
0981a08… noreply 25 "type": "person",
0981a08… noreply 26 "descriptions": ["Project lead"],
0981a08… noreply 27 "occurrences": [{"source": "transcript_batch_0", "timestamp": 0.0}],
0981a08… noreply 28 },
0981a08… noreply 29 {
0981a08… noreply 30 "id": "Bob",
0981a08… noreply 31 "name": "Bob",
0981a08… noreply 32 "type": "person",
0981a08… noreply 33 "descriptions": ["Developer"],
0981a08… noreply 34 "occurrences": [],
0981a08… noreply 35 },
0981a08… noreply 36 {
0981a08… noreply 37 "id": "Python",
0981a08… noreply 38 "name": "Python",
0981a08… noreply 39 "type": "technology",
0981a08… noreply 40 "descriptions": ["Programming language"],
0981a08… noreply 41 "occurrences": [],
0981a08… noreply 42 },
0981a08… noreply 43 {
0981a08… noreply 44 "id": "Acme Corp",
0981a08… noreply 45 "name": "Acme Corp",
0981a08… noreply 46 "type": "organization",
0981a08… noreply 47 "descriptions": ["The company"],
0981a08… noreply 48 "occurrences": [],
0981a08… noreply 49 },
0981a08… noreply 50 {
0981a08… noreply 51 "id": "Microservices",
0981a08… noreply 52 "name": "Microservices",
0981a08… noreply 53 "type": "concept",
0981a08… noreply 54 "descriptions": ["Architecture pattern"],
0981a08… noreply 55 "occurrences": [],
0981a08… noreply 56 },
0981a08… noreply 57 ],
0981a08… noreply 58 "relationships": [
0981a08… noreply 59 {
0981a08… noreply 60 "source": "Alice",
0981a08… noreply 61 "target": "Python",
0981a08… noreply 62 "type": "uses",
0981a08… noreply 63 "content_source": "transcript_batch_0",
0981a08… noreply 64 "timestamp": 1.5,
0981a08… noreply 65 },
0981a08… noreply 66 {
0981a08… noreply 67 "source": "Bob",
0981a08… noreply 68 "target": "Python",
0981a08… noreply 69 "type": "uses",
0981a08… noreply 70 "content_source": "transcript_batch_0",
0981a08… noreply 71 "timestamp": 2.0,
0981a08… noreply 72 },
0981a08… noreply 73 {
0981a08… noreply 74 "source": "Alice",
0981a08… noreply 75 "target": "Bob",
0981a08… noreply 76 "type": "works_with",
0981a08… noreply 77 "content_source": "transcript_batch_0",
0981a08… noreply 78 "timestamp": 3.0,
0981a08… noreply 79 },
0981a08… noreply 80 {
0981a08… noreply 81 "source": "Alice",
0981a08… noreply 82 "target": "Acme Corp",
0981a08… noreply 83 "type": "employed_by",
0981a08… noreply 84 "content_source": "transcript_batch_1",
0981a08… noreply 85 "timestamp": 10.0,
0981a08… noreply 86 },
0981a08… noreply 87 {
0981a08… noreply 88 "source": "Acme Corp",
0981a08… noreply 89 "target": "Microservices",
0981a08… noreply 90 "type": "adopts",
0981a08… noreply 91 "content_source": "transcript_batch_1",
0981a08… noreply 92 "timestamp": 12.0,
0981a08… noreply 93 },
0981a08… noreply 94 ],
0981a08… noreply 95 }
0981a08… noreply 96
0981a08… noreply 97
0981a08… noreply 98 @pytest.fixture
0981a08… noreply 99 def sample_graph(sample_kg_data):
0981a08… noreply 100 """Pre-built NetworkX graph from sample data."""
0981a08… noreply 101 return graph_to_networkx(sample_kg_data)
0981a08… noreply 102
0981a08… noreply 103
0981a08… noreply 104 class TestGraphToNetworkx:
0981a08… noreply 105 def test_node_count(self, sample_graph):
0981a08… noreply 106 assert sample_graph.number_of_nodes() == 5
0981a08… noreply 107
0981a08… noreply 108 def test_edge_count(self, sample_graph):
0981a08… noreply 109 assert sample_graph.number_of_edges() == 5
0981a08… noreply 110
0981a08… noreply 111 def test_node_attributes(self, sample_graph):
0981a08… noreply 112 alice = sample_graph.nodes["Alice"]
0981a08… noreply 113 assert alice["type"] == "person"
0981a08… noreply 114 assert alice["descriptions"] == ["Project lead"]
0981a08… noreply 115
0981a08… noreply 116 def test_edge_attributes(self, sample_graph):
0981a08… noreply 117 edge = sample_graph.edges["Alice", "Python"]
0981a08… noreply 118 assert edge["type"] == "uses"
0981a08… noreply 119 assert edge["content_source"] == "transcript_batch_0"
0981a08… noreply 120 assert edge["timestamp"] == 1.5
0981a08… noreply 121
0981a08… noreply 122 def test_empty_data(self):
0981a08… noreply 123 G = graph_to_networkx({})
0981a08… noreply 124 assert G.number_of_nodes() == 0
0981a08… noreply 125 assert G.number_of_edges() == 0
0981a08… noreply 126
0981a08… noreply 127 def test_nodes_only(self):
0981a08… noreply 128 data = {"nodes": [{"name": "X", "type": "concept"}]}
0981a08… noreply 129 G = graph_to_networkx(data)
0981a08… noreply 130 assert G.number_of_nodes() == 1
0981a08… noreply 131 assert G.number_of_edges() == 0
0981a08… noreply 132
0981a08… noreply 133 def test_skips_empty_names(self):
0981a08… noreply 134 data = {"nodes": [{"name": "", "type": "concept"}, {"name": "A"}]}
0981a08… noreply 135 G = graph_to_networkx(data)
0981a08… noreply 136 assert G.number_of_nodes() == 1
0981a08… noreply 137
0981a08… noreply 138 def test_skips_empty_relationship_endpoints(self):
0981a08… noreply 139 data = {
0981a08… noreply 140 "nodes": [{"name": "A"}],
0981a08… noreply 141 "relationships": [{"source": "", "target": "A", "type": "x"}],
0981a08… noreply 142 }
0981a08… noreply 143 G = graph_to_networkx(data)
0981a08… noreply 144 assert G.number_of_edges() == 0
0981a08… noreply 145
0981a08… noreply 146
0981a08… noreply 147 class TestComputeGraphStats:
0981a08… noreply 148 def test_basic_counts(self, sample_graph):
0981a08… noreply 149 stats = compute_graph_stats(sample_graph)
0981a08… noreply 150 assert stats["node_count"] == 5
0981a08… noreply 151 assert stats["edge_count"] == 5
0981a08… noreply 152
0981a08… noreply 153 def test_density_range(self, sample_graph):
0981a08… noreply 154 stats = compute_graph_stats(sample_graph)
0981a08… noreply 155 assert 0.0 <= stats["density"] <= 1.0
0981a08… noreply 156
0981a08… noreply 157 def test_connected_components(self, sample_graph):
0981a08… noreply 158 stats = compute_graph_stats(sample_graph)
0981a08… noreply 159 assert stats["connected_components"] == 1
0981a08… noreply 160
0981a08… noreply 161 def test_type_breakdown(self, sample_graph):
0981a08… noreply 162 stats = compute_graph_stats(sample_graph)
0981a08… noreply 163 assert stats["type_breakdown"]["person"] == 2
0981a08… noreply 164 assert stats["type_breakdown"]["technology"] == 1
0981a08… noreply 165 assert stats["type_breakdown"]["organization"] == 1
0981a08… noreply 166 assert stats["type_breakdown"]["concept"] == 1
0981a08… noreply 167
0981a08… noreply 168 def test_top_entities(self, sample_graph):
0981a08… noreply 169 stats = compute_graph_stats(sample_graph)
0981a08… noreply 170 top = stats["top_entities"]
0981a08… noreply 171 assert len(top) <= 10
0981a08… noreply 172 # Alice has degree 4 (3 out + 0 in? No: 3 out-edges, 0 in-edges = degree 3 undirected...
0981a08… noreply 173 # Actually in DiGraph, degree = in + out. Alice: out=3 (Python, Bob, Acme), in=0 => 3
0981a08… noreply 174 # Python: in=2, out=0 => 2
0981a08… noreply 175 assert top[0]["name"] == "Alice"
0981a08… noreply 176
0981a08… noreply 177 def test_empty_graph(self):
0981a08… noreply 178 import networkx as nx
0981a08… noreply 179
0981a08… noreply 180 G = nx.DiGraph()
0981a08… noreply 181 stats = compute_graph_stats(G)
0981a08… noreply 182 assert stats["node_count"] == 0
0981a08… noreply 183 assert stats["connected_components"] == 0
0981a08… noreply 184 assert stats["top_entities"] == []
0981a08… noreply 185
0981a08… noreply 186
0981a08… noreply 187 class TestFilterGraph:
0981a08… noreply 188 def test_filter_by_type(self, sample_graph):
0981a08… noreply 189 filtered = filter_graph(sample_graph, entity_types=["person"])
0981a08… noreply 190 assert filtered.number_of_nodes() == 2
0981a08… noreply 191 for _, data in filtered.nodes(data=True):
0981a08… noreply 192 assert data["type"] == "person"
0981a08… noreply 193
0981a08… noreply 194 def test_filter_by_min_degree(self, sample_graph):
0981a08… noreply 195 # Alice has degree 3 (3 out-edges), Python has degree 2 (2 in-edges)
0981a08… noreply 196 filtered = filter_graph(sample_graph, min_degree=3)
0981a08… noreply 197 assert "Alice" in filtered.nodes
0981a08… noreply 198 assert filtered.number_of_nodes() >= 1
0981a08… noreply 199
0981a08… noreply 200 def test_filter_combined(self, sample_graph):
0981a08… noreply 201 filtered = filter_graph(sample_graph, entity_types=["person"], min_degree=1)
0981a08… noreply 202 assert all(filtered.nodes[n]["type"] == "person" for n in filtered.nodes)
0981a08… noreply 203
0981a08… noreply 204 def test_filter_no_criteria(self, sample_graph):
0981a08… noreply 205 filtered = filter_graph(sample_graph)
0981a08… noreply 206 assert filtered.number_of_nodes() == sample_graph.number_of_nodes()
0981a08… noreply 207
0981a08… noreply 208 def test_filter_nonexistent_type(self, sample_graph):
0981a08… noreply 209 filtered = filter_graph(sample_graph, entity_types=["alien"])
0981a08… noreply 210 assert filtered.number_of_nodes() == 0
0981a08… noreply 211
0981a08… noreply 212 def test_filter_preserves_edges(self, sample_graph):
0981a08… noreply 213 filtered = filter_graph(sample_graph, entity_types=["person"])
0981a08… noreply 214 # Alice -> Bob edge should be preserved
0981a08… noreply 215 assert filtered.has_edge("Alice", "Bob")
0981a08… noreply 216
0981a08… noreply 217 def test_filter_returns_copy(self, sample_graph):
0981a08… noreply 218 filtered = filter_graph(sample_graph, entity_types=["person"])
0981a08… noreply 219 # Modifying filtered should not affect original
0981a08… noreply 220 filtered.add_node("NewNode")
0981a08… noreply 221 assert "NewNode" not in sample_graph
0981a08… noreply 222
0981a08… noreply 223
0981a08… noreply 224 class TestGenerateMermaid:
0981a08… noreply 225 def test_output_starts_with_graph(self, sample_graph):
0981a08… noreply 226 mermaid = generate_mermaid(sample_graph)
0981a08… noreply 227 assert mermaid.startswith("graph LR")
0981a08… noreply 228
0981a08… noreply 229 def test_custom_layout(self, sample_graph):
0981a08… noreply 230 mermaid = generate_mermaid(sample_graph, layout="TD")
0981a08… noreply 231 assert mermaid.startswith("graph TD")
0981a08… noreply 232
0981a08… noreply 233 def test_contains_nodes(self, sample_graph):
0981a08… noreply 234 mermaid = generate_mermaid(sample_graph)
0981a08… noreply 235 assert "Alice" in mermaid
0981a08… noreply 236 assert "Python" in mermaid
0981a08… noreply 237
0981a08… noreply 238 def test_contains_edges(self, sample_graph):
0981a08… noreply 239 mermaid = generate_mermaid(sample_graph)
0981a08… noreply 240 assert "uses" in mermaid
0981a08… noreply 241
0981a08… noreply 242 def test_contains_class_defs(self, sample_graph):
0981a08… noreply 243 mermaid = generate_mermaid(sample_graph)
0981a08… noreply 244 assert "classDef person" in mermaid
0981a08… noreply 245 assert "classDef concept" in mermaid
0981a08… noreply 246
0981a08… noreply 247 def test_max_nodes_limit(self, sample_graph):
0981a08… noreply 248 mermaid = generate_mermaid(sample_graph, max_nodes=2)
0981a08… noreply 249 # Should only have top-2 nodes by degree
0981a08… noreply 250 lines = [ln for ln in mermaid.split("\n") if '["' in ln]
0981a08… noreply 251 assert len(lines) <= 2
0981a08… noreply 252
0981a08… noreply 253 def test_empty_graph(self):
0981a08… noreply 254 import networkx as nx
0981a08… noreply 255
0981a08… noreply 256 G = nx.DiGraph()
0981a08… noreply 257 mermaid = generate_mermaid(G)
0981a08… noreply 258 assert "graph LR" in mermaid
0981a08… noreply 259
0981a08… noreply 260 def test_sanitizes_special_chars(self):
0981a08… noreply 261 import networkx as nx
0981a08… noreply 262
0981a08… noreply 263 G = nx.DiGraph()
0981a08… noreply 264 G.add_node("foo bar/baz", type="concept")
0981a08… noreply 265 mermaid = generate_mermaid(G)
0981a08… noreply 266 # Node ID should be sanitized but label preserved
0981a08… noreply 267 assert "foo_bar_baz" in mermaid
0981a08… noreply 268 assert "foo bar/baz" in mermaid
0981a08… noreply 269
0981a08… noreply 270
0981a08… noreply 271 class TestGraphToD3Json:
0981a08… noreply 272 def test_structure(self, sample_graph):
0981a08… noreply 273 d3 = graph_to_d3_json(sample_graph)
0981a08… noreply 274 assert "nodes" in d3
0981a08… noreply 275 assert "links" in d3
0981a08… noreply 276
0981a08… noreply 277 def test_node_format(self, sample_graph):
0981a08… noreply 278 d3 = graph_to_d3_json(sample_graph)
0981a08… noreply 279 node_ids = {n["id"] for n in d3["nodes"]}
0981a08… noreply 280 assert "Alice" in node_ids
0981a08… noreply 281 alice = next(n for n in d3["nodes"] if n["id"] == "Alice")
0981a08… noreply 282 assert alice["group"] == "person"
0981a08… noreply 283
0981a08… noreply 284 def test_link_format(self, sample_graph):
0981a08… noreply 285 d3 = graph_to_d3_json(sample_graph)
0981a08… noreply 286 assert len(d3["links"]) == 5
0981a08… noreply 287 link = d3["links"][0]
0981a08… noreply 288 assert "source" in link
0981a08… noreply 289 assert "target" in link
0981a08… noreply 290 assert "type" in link
0981a08… noreply 291
0981a08… noreply 292 def test_empty_graph(self):
0981a08… noreply 293 import networkx as nx
0981a08… noreply 294
0981a08… noreply 295 G = nx.DiGraph()
0981a08… noreply 296 d3 = graph_to_d3_json(G)
0981a08… noreply 297 assert d3 == {"nodes": [], "links": []}
0981a08… noreply 298
0981a08… noreply 299
0981a08… noreply 300 class TestGraphToDot:
0981a08… noreply 301 def test_starts_with_digraph(self, sample_graph):
0981a08… noreply 302 dot = graph_to_dot(sample_graph)
0981a08… noreply 303 assert dot.startswith("digraph KnowledgeGraph {")
0981a08… noreply 304
0981a08… noreply 305 def test_ends_with_closing_brace(self, sample_graph):
0981a08… noreply 306 dot = graph_to_dot(sample_graph)
0981a08… noreply 307 assert dot.strip().endswith("}")
0981a08… noreply 308
0981a08… noreply 309 def test_contains_nodes(self, sample_graph):
0981a08… noreply 310 dot = graph_to_dot(sample_graph)
0981a08… noreply 311 assert '"Alice"' in dot
0981a08… noreply 312 assert '"Python"' in dot
0981a08… noreply 313
0981a08… noreply 314 def test_contains_edges(self, sample_graph):
0981a08… noreply 315 dot = graph_to_dot(sample_graph)
0981a08… noreply 316 assert '"Alice" -> "Python"' in dot
0981a08… noreply 317
0981a08… noreply 318 def test_edge_labels(self, sample_graph):
0981a08… noreply 319 dot = graph_to_dot(sample_graph)
0981a08… noreply 320 assert 'label="uses"' in dot
0981a08… noreply 321
0981a08… noreply 322 def test_node_colors(self, sample_graph):
0981a08… noreply 323 dot = graph_to_dot(sample_graph)
0981a08… noreply 324 assert 'fillcolor="#f9d5e5"' in dot # person color for Alice
0981a08… noreply 325
0981a08… noreply 326 def test_empty_graph(self):
0981a08… noreply 327 import networkx as nx
0981a08… noreply 328
0981a08… noreply 329 G = nx.DiGraph()
0981a08… noreply 330 dot = graph_to_dot(G)
0981a08… noreply 331 assert "digraph" in dot
0981a08… noreply 332
0981a08… noreply 333 def test_special_chars_escaped(self):
0981a08… noreply 334 import networkx as nx
0981a08… noreply 335
0981a08… noreply 336 G = nx.DiGraph()
0981a08… noreply 337 G.add_node('He said "hello"', type="person")
0981a08… noreply 338 dot = graph_to_dot(G)
0981a08… noreply 339 assert 'He said \\"hello\\"' in dot

Keyboard Shortcuts

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