Navegador

navegador / tests / test_ingestion_planopticon.py
Source Blame History 646 lines
b663b12… lmata 1 """Tests for navegador.ingestion.planopticon — PlanopticonIngester."""
b663b12… lmata 2
b663b12… lmata 3 import json
b663b12… lmata 4 import tempfile
b663b12… lmata 5 from pathlib import Path
b663b12… lmata 6 from unittest.mock import MagicMock
b663b12… lmata 7
7e708ec… lmata 8 import pytest
7e708ec… lmata 9
b663b12… lmata 10 from navegador.graph.schema import NodeLabel
b663b12… lmata 11 from navegador.ingestion.planopticon import (
b663b12… lmata 12 EDGE_MAP,
b663b12… lmata 13 NODE_TYPE_MAP,
b663b12… lmata 14 PLANNING_TYPE_MAP,
b663b12… lmata 15 PlanopticonIngester,
b663b12… lmata 16 )
b663b12… lmata 17
b663b12… lmata 18
b663b12… lmata 19 def _make_store():
b663b12… lmata 20 store = MagicMock()
b663b12… lmata 21 store.query.return_value = MagicMock(result_set=[])
b663b12… lmata 22 return store
b663b12… lmata 23
b663b12… lmata 24
b663b12… lmata 25 # ── Fixtures ──────────────────────────────────────────────────────────────────
b663b12… lmata 26
b663b12… lmata 27 KG_DATA = {
b663b12… lmata 28 "nodes": [
b663b12… lmata 29 {"id": "n1", "type": "concept", "name": "Payment Gateway",
b663b12… lmata 30 "description": "Handles payments"},
b663b12… lmata 31 {"id": "n2", "type": "person", "name": "Carol", "email": "[email protected]"},
b663b12… lmata 32 {"id": "n3", "type": "technology", "name": "PostgreSQL", "description": "DB"},
b663b12… lmata 33 {"id": "n4", "type": "decision", "name": "Use Redis"},
b663b12… lmata 34 {"id": "n5", "type": "unknown_type", "name": "Misc"},
b663b12… lmata 35 {"id": "n6", "type": "diagram", "name": "Service Map", "source": "http://img.png"},
b663b12… lmata 36 ],
b663b12… lmata 37 "relationships": [
b663b12… lmata 38 {"source": "Payment Gateway", "target": "PostgreSQL", "type": "uses"},
b663b12… lmata 39 {"source": "Carol", "target": "Payment Gateway", "type": "assigned_to"},
b663b12… lmata 40 {"source": "", "target": "nope", "type": "related_to"}, # bad rel — no source
b663b12… lmata 41 ],
b663b12… lmata 42 "sources": [
b663b12… lmata 43 {"title": "Meeting 2024", "url": "https://ex.com", "source_type": "meeting"},
b663b12… lmata 44 ],
b663b12… lmata 45 }
b663b12… lmata 46
b663b12… lmata 47 INTERCHANGE_DATA = {
b663b12… lmata 48 "project": {"name": "MyProject", "tags": ["backend", "payments"]},
b663b12… lmata 49 "entities": [
b663b12… lmata 50 {
b663b12… lmata 51 "planning_type": "decision",
b663b12… lmata 52 "name": "Adopt microservices",
b663b12… lmata 53 "description": "Split the monolith",
b663b12… lmata 54 "status": "accepted",
b663b12… lmata 55 "rationale": "Scale independently",
b663b12… lmata 56 },
b663b12… lmata 57 {
b663b12… lmata 58 "planning_type": "requirement",
b663b12… lmata 59 "name": "PCI compliance",
b663b12… lmata 60 "description": "Must comply with PCI-DSS",
b663b12… lmata 61 "priority": "high",
b663b12… lmata 62 },
b663b12… lmata 63 {
b663b12… lmata 64 "planning_type": "goal",
b663b12… lmata 65 "name": "Increase uptime",
b663b12… lmata 66 "description": "99.9% SLA",
b663b12… lmata 67 },
b663b12… lmata 68 {
b663b12… lmata 69 # no planning_type → falls through to _ingest_kg_node
b663b12… lmata 70 "type": "concept",
b663b12… lmata 71 "name": "Event Sourcing",
b663b12… lmata 72 },
b663b12… lmata 73 ],
b663b12… lmata 74 "relationships": [],
b663b12… lmata 75 "artifacts": [
b663b12… lmata 76 {"name": "Architecture Diagram", "content": "mermaid content here"},
b663b12… lmata 77 ],
b663b12… lmata 78 "sources": [],
b663b12… lmata 79 }
b663b12… lmata 80
b663b12… lmata 81 MANIFEST_DATA = {
b663b12… lmata 82 "video": {"title": "Sprint Planning", "url": "https://example.com/video/1"},
b663b12… lmata 83 "key_points": [
b663b12… lmata 84 {"point": "Use async everywhere", "topic": "Architecture", "details": "For scale"},
b663b12… lmata 85 ],
b663b12… lmata 86 "action_items": [
b663b12… lmata 87 {"action": "Refactor auth service", "assignee": "Bob", "context": "High priority"},
b663b12… lmata 88 ],
b663b12… lmata 89 "diagrams": [
b663b12… lmata 90 {
b663b12… lmata 91 "diagram_type": "sequence",
b663b12… lmata 92 "timestamp": 120,
b663b12… lmata 93 "description": "Auth flow",
b663b12… lmata 94 "mermaid": "sequenceDiagram ...",
b663b12… lmata 95 "elements": ["User", "Auth"],
b663b12… lmata 96 }
b663b12… lmata 97 ],
b663b12… lmata 98 }
b663b12… lmata 99
b663b12… lmata 100
b663b12… lmata 101 # ── Maps ──────────────────────────────────────────────────────────────────────
b663b12… lmata 102
b663b12… lmata 103 class TestMaps:
b663b12… lmata 104 def test_node_type_map_coverage(self):
b663b12… lmata 105 assert NODE_TYPE_MAP["concept"] == NodeLabel.Concept
b663b12… lmata 106 assert NODE_TYPE_MAP["technology"] == NodeLabel.Concept
b663b12… lmata 107 assert NODE_TYPE_MAP["organization"] == NodeLabel.Concept
b663b12… lmata 108 assert NODE_TYPE_MAP["person"] == NodeLabel.Person
b663b12… lmata 109 assert NODE_TYPE_MAP["diagram"] == NodeLabel.WikiPage
b663b12… lmata 110
b663b12… lmata 111 def test_planning_type_map_coverage(self):
b663b12… lmata 112 assert PLANNING_TYPE_MAP["decision"] == NodeLabel.Decision
b663b12… lmata 113 assert PLANNING_TYPE_MAP["requirement"] == NodeLabel.Rule
b663b12… lmata 114 assert PLANNING_TYPE_MAP["constraint"] == NodeLabel.Rule
b663b12… lmata 115 assert PLANNING_TYPE_MAP["risk"] == NodeLabel.Rule
b663b12… lmata 116 assert PLANNING_TYPE_MAP["goal"] == NodeLabel.Concept
b663b12… lmata 117
b663b12… lmata 118 def test_edge_map_coverage(self):
b663b12… lmata 119 from navegador.graph.schema import EdgeType
b663b12… lmata 120 assert EDGE_MAP["uses"] == EdgeType.DEPENDS_ON
b663b12… lmata 121 assert EDGE_MAP["related_to"] == EdgeType.RELATED_TO
b663b12… lmata 122 assert EDGE_MAP["assigned_to"] == EdgeType.ASSIGNED_TO
b663b12… lmata 123 assert EDGE_MAP["governs"] == EdgeType.GOVERNS
b663b12… lmata 124 assert EDGE_MAP["implements"] == EdgeType.IMPLEMENTS
b663b12… lmata 125
b663b12… lmata 126
b663b12… lmata 127 # ── ingest_kg ─────────────────────────────────────────────────────────────────
b663b12… lmata 128
b663b12… lmata 129 class TestIngestKg:
b663b12… lmata 130 def test_ingests_concept_nodes(self):
b663b12… lmata 131 store = _make_store()
b663b12… lmata 132 ingester = PlanopticonIngester(store)
b663b12… lmata 133 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 134 p = Path(tmpdir) / "kg.json"
b663b12… lmata 135 p.write_text(json.dumps(KG_DATA))
b663b12… lmata 136 stats = ingester.ingest_kg(p)
b663b12… lmata 137 assert stats["nodes"] >= 1
b663b12… lmata 138
b663b12… lmata 139 def test_ingests_person_nodes(self):
b663b12… lmata 140 store = _make_store()
b663b12… lmata 141 ingester = PlanopticonIngester(store)
b663b12… lmata 142 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 143 p = Path(tmpdir) / "kg.json"
b663b12… lmata 144 p.write_text(json.dumps(KG_DATA))
b663b12… lmata 145 ingester.ingest_kg(p)
b663b12… lmata 146 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 147 assert NodeLabel.Person in labels
b663b12… lmata 148
b663b12… lmata 149 def test_ingests_technology_as_concept(self):
b663b12… lmata 150 store = _make_store()
b663b12… lmata 151 ingester = PlanopticonIngester(store)
b663b12… lmata 152 data = {"nodes": [{"type": "technology", "name": "PostgreSQL"}],
b663b12… lmata 153 "relationships": [], "sources": []}
b663b12… lmata 154 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 155 p = Path(tmpdir) / "kg.json"
b663b12… lmata 156 p.write_text(json.dumps(data))
b663b12… lmata 157 ingester.ingest_kg(p)
b663b12… lmata 158 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 159 assert NodeLabel.Concept in labels
b663b12… lmata 160
b663b12… lmata 161 def test_ingests_diagram_as_wiki_page(self):
b663b12… lmata 162 store = _make_store()
b663b12… lmata 163 ingester = PlanopticonIngester(store)
b663b12… lmata 164 data = {"nodes": [{"type": "diagram", "name": "Arch Diagram", "source": "http://x.com"}],
b663b12… lmata 165 "relationships": [], "sources": []}
b663b12… lmata 166 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 167 p = Path(tmpdir) / "kg.json"
b663b12… lmata 168 p.write_text(json.dumps(data))
b663b12… lmata 169 ingester.ingest_kg(p)
b663b12… lmata 170 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 171 assert NodeLabel.WikiPage in labels
b663b12… lmata 172
b663b12… lmata 173 def test_skips_nodes_without_name(self):
b663b12… lmata 174 store = _make_store()
b663b12… lmata 175 ingester = PlanopticonIngester(store)
b663b12… lmata 176 data = {"nodes": [{"type": "concept", "name": ""}], "relationships": [], "sources": []}
b663b12… lmata 177 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 178 p = Path(tmpdir) / "kg.json"
b663b12… lmata 179 p.write_text(json.dumps(data))
b663b12… lmata 180 stats = ingester.ingest_kg(p)
b663b12… lmata 181 assert stats["nodes"] == 0
b663b12… lmata 182
b663b12… lmata 183 def test_ingests_sources_as_wiki_pages(self):
b663b12… lmata 184 store = _make_store()
b663b12… lmata 185 ingester = PlanopticonIngester(store)
b663b12… lmata 186 data = {
b663b12… lmata 187 "nodes": [], "relationships": [],
b663b12… lmata 188 "sources": [
b663b12… lmata 189 {"title": "Meeting 2024", "url": "https://ex.com", "source_type": "meeting"},
b663b12… lmata 190 ],
b663b12… lmata 191 }
b663b12… lmata 192 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 193 p = Path(tmpdir) / "kg.json"
b663b12… lmata 194 p.write_text(json.dumps(data))
b663b12… lmata 195 stats = ingester.ingest_kg(p)
b663b12… lmata 196 assert stats["nodes"] >= 1
b663b12… lmata 197 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 198 assert NodeLabel.WikiPage in labels
b663b12… lmata 199
b663b12… lmata 200 def test_ingests_relationships(self):
b663b12… lmata 201 store = _make_store()
b663b12… lmata 202 ingester = PlanopticonIngester(store)
b663b12… lmata 203 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 204 p = Path(tmpdir) / "kg.json"
b663b12… lmata 205 p.write_text(json.dumps(KG_DATA))
b663b12… lmata 206 stats = ingester.ingest_kg(p)
b663b12… lmata 207 assert stats["edges"] >= 1
b663b12… lmata 208 store.query.assert_called()
b663b12… lmata 209
b663b12… lmata 210 def test_skips_bad_relationships(self):
b663b12… lmata 211 store = _make_store()
b663b12… lmata 212 ingester = PlanopticonIngester(store)
b663b12… lmata 213 data = {"nodes": [], "relationships": [{"source": "", "target": "x", "type": "related_to"}],
b663b12… lmata 214 "sources": []}
b663b12… lmata 215 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 216 p = Path(tmpdir) / "kg.json"
b663b12… lmata 217 p.write_text(json.dumps(data))
b663b12… lmata 218 stats = ingester.ingest_kg(p)
b663b12… lmata 219 assert stats["edges"] == 0
b663b12… lmata 220
7e708ec… lmata 221 def test_missing_file_raises(self):
b663b12… lmata 222 store = _make_store()
b663b12… lmata 223 ingester = PlanopticonIngester(store)
7e708ec… lmata 224 with pytest.raises(FileNotFoundError):
7e708ec… lmata 225 ingester.ingest_kg("/nonexistent/kg.json")
b663b12… lmata 226
b663b12… lmata 227 def test_returns_stats_dict(self):
b663b12… lmata 228 store = _make_store()
b663b12… lmata 229 ingester = PlanopticonIngester(store)
b663b12… lmata 230 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 231 p = Path(tmpdir) / "kg.json"
b663b12… lmata 232 p.write_text(json.dumps({"nodes": [], "relationships": [], "sources": []}))
b663b12… lmata 233 stats = ingester.ingest_kg(p)
b663b12… lmata 234 assert "nodes" in stats
b663b12… lmata 235 assert "edges" in stats
b663b12… lmata 236
b663b12… lmata 237
b663b12… lmata 238 # ── ingest_interchange ────────────────────────────────────────────────────────
b663b12… lmata 239
b663b12… lmata 240 class TestIngestInterchange:
b663b12… lmata 241 def test_ingests_decision_entities(self):
b663b12… lmata 242 store = _make_store()
b663b12… lmata 243 ingester = PlanopticonIngester(store)
b663b12… lmata 244 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 245 p = Path(tmpdir) / "interchange.json"
b663b12… lmata 246 p.write_text(json.dumps(INTERCHANGE_DATA))
b663b12… lmata 247 ingester.ingest_interchange(p)
b663b12… lmata 248 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 249 assert NodeLabel.Decision in labels
b663b12… lmata 250
b663b12… lmata 251 def test_ingests_requirement_as_rule(self):
b663b12… lmata 252 store = _make_store()
b663b12… lmata 253 ingester = PlanopticonIngester(store)
b663b12… lmata 254 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 255 p = Path(tmpdir) / "interchange.json"
b663b12… lmata 256 p.write_text(json.dumps(INTERCHANGE_DATA))
b663b12… lmata 257 ingester.ingest_interchange(p)
b663b12… lmata 258 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 259 assert NodeLabel.Rule in labels
b663b12… lmata 260
b663b12… lmata 261 def test_creates_domain_nodes_from_project_tags(self):
b663b12… lmata 262 store = _make_store()
b663b12… lmata 263 ingester = PlanopticonIngester(store)
b663b12… lmata 264 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 265 p = Path(tmpdir) / "interchange.json"
b663b12… lmata 266 p.write_text(json.dumps(INTERCHANGE_DATA))
b663b12… lmata 267 ingester.ingest_interchange(p)
b663b12… lmata 268 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 269 assert NodeLabel.Domain in labels
b663b12… lmata 270
b663b12… lmata 271 def test_ingests_artifacts_as_wiki_pages(self):
b663b12… lmata 272 store = _make_store()
b663b12… lmata 273 ingester = PlanopticonIngester(store)
b663b12… lmata 274 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 275 p = Path(tmpdir) / "interchange.json"
b663b12… lmata 276 p.write_text(json.dumps(INTERCHANGE_DATA))
b663b12… lmata 277 ingester.ingest_interchange(p)
b663b12… lmata 278 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 279 assert NodeLabel.WikiPage in labels
b663b12… lmata 280
b663b12… lmata 281 def test_empty_entities_returns_empty_stats(self):
b663b12… lmata 282 store = _make_store()
b663b12… lmata 283 ingester = PlanopticonIngester(store)
b663b12… lmata 284 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 285 p = Path(tmpdir) / "interchange.json"
b663b12… lmata 286 p.write_text(json.dumps({"project": {}, "entities": [], "relationships": [],
b663b12… lmata 287 "artifacts": [], "sources": []}))
b663b12… lmata 288 stats = ingester.ingest_interchange(p)
b663b12… lmata 289 assert stats["nodes"] == 0
b663b12… lmata 290
b663b12… lmata 291 def test_returns_stats_dict(self):
b663b12… lmata 292 store = _make_store()
b663b12… lmata 293 ingester = PlanopticonIngester(store)
b663b12… lmata 294 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 295 p = Path(tmpdir) / "interchange.json"
b663b12… lmata 296 p.write_text(json.dumps(INTERCHANGE_DATA))
b663b12… lmata 297 stats = ingester.ingest_interchange(p)
b663b12… lmata 298 assert "nodes" in stats and "edges" in stats
b663b12… lmata 299
b663b12… lmata 300
b663b12… lmata 301 # ── ingest_manifest ────────────────────────────────────────────────────────────
b663b12… lmata 302
b663b12… lmata 303 class TestIngestManifest:
b663b12… lmata 304 def test_ingests_key_points_as_concepts(self):
b663b12… lmata 305 store = _make_store()
b663b12… lmata 306 ingester = PlanopticonIngester(store)
b663b12… lmata 307 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 308 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 309 p.write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 310 ingester.ingest_manifest(p)
b663b12… lmata 311 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 312 assert NodeLabel.Concept in labels
b663b12… lmata 313
b663b12… lmata 314 def test_ingests_action_items_as_rules(self):
b663b12… lmata 315 store = _make_store()
b663b12… lmata 316 ingester = PlanopticonIngester(store)
b663b12… lmata 317 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 318 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 319 p.write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 320 ingester.ingest_manifest(p)
b663b12… lmata 321 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 322 assert NodeLabel.Rule in labels
b663b12… lmata 323
b663b12… lmata 324 def test_ingests_action_item_assignee_as_person(self):
b663b12… lmata 325 store = _make_store()
b663b12… lmata 326 ingester = PlanopticonIngester(store)
b663b12… lmata 327 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 328 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 329 p.write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 330 ingester.ingest_manifest(p)
b663b12… lmata 331 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 332 assert NodeLabel.Person in labels
b663b12… lmata 333
b663b12… lmata 334 def test_ingests_diagrams_as_wiki_pages(self):
b663b12… lmata 335 store = _make_store()
b663b12… lmata 336 ingester = PlanopticonIngester(store)
b663b12… lmata 337 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 338 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 339 p.write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 340 ingester.ingest_manifest(p)
b663b12… lmata 341 labels = [c[0][0] for c in store.create_node.call_args_list]
b663b12… lmata 342 assert NodeLabel.WikiPage in labels
b663b12… lmata 343
b663b12… lmata 344 def test_diagram_elements_become_concepts(self):
b663b12… lmata 345 store = _make_store()
b663b12… lmata 346 ingester = PlanopticonIngester(store)
b663b12… lmata 347 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 348 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 349 p.write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 350 ingester.ingest_manifest(p)
b663b12… lmata 351 # "User" and "Auth" are diagram elements → Concept nodes
b663b12… lmata 352 names = [c[0][1].get("name") for c in store.create_node.call_args_list
b663b12… lmata 353 if isinstance(c[0][1], dict)]
b663b12… lmata 354 assert "User" in names or "Auth" in names
b663b12… lmata 355
b663b12… lmata 356 def test_loads_external_kg_json(self):
b663b12… lmata 357 store = _make_store()
b663b12… lmata 358 ingester = PlanopticonIngester(store)
b663b12… lmata 359 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 360 kg = {"nodes": [{"type": "concept", "name": "External Concept"}],
b663b12… lmata 361 "relationships": [], "sources": []}
b663b12… lmata 362 (Path(tmpdir) / "kg.json").write_text(json.dumps(kg))
b663b12… lmata 363 manifest = dict(MANIFEST_DATA)
b663b12… lmata 364 manifest["knowledge_graph_json"] = "kg.json"
b663b12… lmata 365 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 366 p.write_text(json.dumps(manifest))
b663b12… lmata 367 ingester.ingest_manifest(p)
b663b12… lmata 368 names = [c[0][1].get("name") for c in store.create_node.call_args_list
b663b12… lmata 369 if isinstance(c[0][1], dict)]
b663b12… lmata 370 assert "External Concept" in names
b663b12… lmata 371
b663b12… lmata 372 def test_empty_manifest_no_crash(self):
b663b12… lmata 373 store = _make_store()
b663b12… lmata 374 ingester = PlanopticonIngester(store)
b663b12… lmata 375 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 376 p = Path(tmpdir) / "manifest.json"
b663b12… lmata 377 p.write_text(json.dumps({}))
b663b12… lmata 378 stats = ingester.ingest_manifest(p)
b663b12… lmata 379 assert "nodes" in stats
b663b12… lmata 380
b663b12… lmata 381
b663b12… lmata 382 # ── ingest_batch ──────────────────────────────────────────────────────────────
b663b12… lmata 383
b663b12… lmata 384 class TestIngestBatch:
b663b12… lmata 385 def test_processes_merged_kg_if_present(self):
b663b12… lmata 386 store = _make_store()
b663b12… lmata 387 ingester = PlanopticonIngester(store)
b663b12… lmata 388 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 389 kg = {"nodes": [{"type": "concept", "name": "Merged"}],
b663b12… lmata 390 "relationships": [], "sources": []}
b663b12… lmata 391 (Path(tmpdir) / "merged.json").write_text(json.dumps(kg))
b663b12… lmata 392 batch = {"merged_knowledge_graph_json": "merged.json"}
b663b12… lmata 393 p = Path(tmpdir) / "batch.json"
b663b12… lmata 394 p.write_text(json.dumps(batch))
b663b12… lmata 395 ingester.ingest_batch(p)
b663b12… lmata 396 names = [c[0][1].get("name") for c in store.create_node.call_args_list
b663b12… lmata 397 if isinstance(c[0][1], dict)]
b663b12… lmata 398 assert "Merged" in names
b663b12… lmata 399
b663b12… lmata 400 def test_processes_completed_videos(self):
b663b12… lmata 401 store = _make_store()
b663b12… lmata 402 ingester = PlanopticonIngester(store)
b663b12… lmata 403 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 404 (Path(tmpdir) / "vid1.json").write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 405 batch = {
b663b12… lmata 406 "videos": [
b663b12… lmata 407 {"status": "completed", "manifest_path": "vid1.json"},
b663b12… lmata 408 {"status": "pending", "manifest_path": "vid1.json"}, # skipped
b663b12… lmata 409 ]
b663b12… lmata 410 }
b663b12… lmata 411 p = Path(tmpdir) / "batch.json"
b663b12… lmata 412 p.write_text(json.dumps(batch))
b663b12… lmata 413 stats = ingester.ingest_batch(p)
b663b12… lmata 414 assert "nodes" in stats
b663b12… lmata 415
7e708ec… lmata 416 def test_missing_manifest_raises(self):
b663b12… lmata 417 store = _make_store()
b663b12… lmata 418 ingester = PlanopticonIngester(store)
b663b12… lmata 419 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 420 batch = {
b663b12… lmata 421 "videos": [
b663b12… lmata 422 {"status": "completed", "manifest_path": "nonexistent.json"},
b663b12… lmata 423 ]
b663b12… lmata 424 }
b663b12… lmata 425 p = Path(tmpdir) / "batch.json"
b663b12… lmata 426 p.write_text(json.dumps(batch))
7e708ec… lmata 427 with pytest.raises(FileNotFoundError):
7e708ec… lmata 428 ingester.ingest_batch(p)
b663b12… lmata 429
b663b12… lmata 430 def test_merges_stats_across_videos(self):
b663b12… lmata 431 store = _make_store()
b663b12… lmata 432 ingester = PlanopticonIngester(store)
b663b12… lmata 433 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 434 (Path(tmpdir) / "v1.json").write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 435 (Path(tmpdir) / "v2.json").write_text(json.dumps(MANIFEST_DATA))
b663b12… lmata 436 batch = {
b663b12… lmata 437 "videos": [
b663b12… lmata 438 {"status": "completed", "manifest_path": "v1.json"},
b663b12… lmata 439 {"status": "completed", "manifest_path": "v2.json"},
b663b12… lmata 440 ]
b663b12… lmata 441 }
b663b12… lmata 442 p = Path(tmpdir) / "batch.json"
b663b12… lmata 443 p.write_text(json.dumps(batch))
b663b12… lmata 444 stats = ingester.ingest_batch(p)
b663b12… lmata 445 # Should have processed both, stats should be non-zero
b663b12… lmata 446 assert stats.get("nodes", 0) >= 0 # at least doesn't crash
b663b12… lmata 447
b663b12… lmata 448
b663b12… lmata 449 # ── _reset_stats / _merge_stats ───────────────────────────────────────────────
b663b12… lmata 450
b663b12… lmata 451 class TestInternalHelpers:
b663b12… lmata 452 def test_reset_stats(self):
b663b12… lmata 453 store = _make_store()
b663b12… lmata 454 ingester = PlanopticonIngester(store)
b663b12… lmata 455 ingester._stats = {"nodes": 5, "edges": 3}
b663b12… lmata 456 stats = ingester._reset_stats()
b663b12… lmata 457 assert stats == {"nodes": 0, "edges": 0}
b663b12… lmata 458
b663b12… lmata 459 def test_merge_stats(self):
b663b12… lmata 460 store = _make_store()
b663b12… lmata 461 ingester = PlanopticonIngester(store)
b663b12… lmata 462 ingester._stats = {"nodes": 2, "edges": 1}
b663b12… lmata 463 ingester._merge_stats({"nodes": 3, "edges": 2, "pages": 1})
b663b12… lmata 464 assert ingester._stats["nodes"] == 5
b663b12… lmata 465 assert ingester._stats["edges"] == 3
b663b12… lmata 466 assert ingester._stats["pages"] == 1
b663b12… lmata 467
7e708ec… lmata 468 def test_load_json_missing_file_raises(self):
b663b12… lmata 469 store = _make_store()
b663b12… lmata 470 ingester = PlanopticonIngester(store)
7e708ec… lmata 471 with pytest.raises(FileNotFoundError):
7e708ec… lmata 472 ingester._load_json(Path("/nonexistent/file.json"))
b663b12… lmata 473
7e708ec… lmata 474 def test_load_json_invalid_json_raises(self):
b663b12… lmata 475 store = _make_store()
b663b12… lmata 476 ingester = PlanopticonIngester(store)
b663b12… lmata 477 with tempfile.TemporaryDirectory() as tmpdir:
b663b12… lmata 478 p = Path(tmpdir) / "bad.json"
b663b12… lmata 479 p.write_text("{ not valid json }")
7e708ec… lmata 480 with pytest.raises((json.JSONDecodeError, ValueError)):
7e708ec… lmata 481 ingester._load_json(p)
2270d4f… lmata 482
2270d4f… lmata 483
2270d4f… lmata 484 # ── ingest_interchange relationship/source branches (lines 201, 209) ──────────
2270d4f… lmata 485
2270d4f… lmata 486 class TestInterchangeRelationshipsAndSources:
2270d4f… lmata 487 def test_ingests_relationships_in_interchange(self):
2270d4f… lmata 488 store = _make_store()
2270d4f… lmata 489 ingester = PlanopticonIngester(store)
2270d4f… lmata 490 data = {
2270d4f… lmata 491 "project": {"name": "Proj", "tags": []},
2270d4f… lmata 492 "entities": [],
2270d4f… lmata 493 "relationships": [
2270d4f… lmata 494 {"source": "Alice", "target": "Bob", "type": "related_to"}
2270d4f… lmata 495 ],
2270d4f… lmata 496 "artifacts": [],
2270d4f… lmata 497 "sources": [],
2270d4f… lmata 498 }
2270d4f… lmata 499 with tempfile.TemporaryDirectory() as tmpdir:
2270d4f… lmata 500 p = Path(tmpdir) / "interchange.json"
2270d4f… lmata 501 p.write_text(json.dumps(data))
2270d4f… lmata 502 stats = ingester.ingest_interchange(p)
2270d4f… lmata 503 store.query.assert_called()
2270d4f… lmata 504 assert stats["edges"] >= 1
2270d4f… lmata 505
2270d4f… lmata 506 def test_ingests_sources_in_interchange(self):
2270d4f… lmata 507 store = _make_store()
2270d4f… lmata 508 ingester = PlanopticonIngester(store)
2270d4f… lmata 509 data = {
2270d4f… lmata 510 "project": {"name": "Proj", "tags": []},
2270d4f… lmata 511 "entities": [],
2270d4f… lmata 512 "relationships": [],
2270d4f… lmata 513 "artifacts": [],
2270d4f… lmata 514 "sources": [{"title": "Design Doc", "url": "http://ex.com"}],
2270d4f… lmata 515 }
2270d4f… lmata 516 with tempfile.TemporaryDirectory() as tmpdir:
2270d4f… lmata 517 p = Path(tmpdir) / "interchange.json"
2270d4f… lmata 518 p.write_text(json.dumps(data))
2270d4f… lmata 519 ingester.ingest_interchange(p)
2270d4f… lmata 520 labels = [c[0][0] for c in store.create_node.call_args_list]
2270d4f… lmata 521 assert NodeLabel.WikiPage in labels
2270d4f… lmata 522
2270d4f… lmata 523
2270d4f… lmata 524 # ── _ingest_kg_node with domain (lines 274-275) ───────────────────────────────
2270d4f… lmata 525
2270d4f… lmata 526 class TestIngestKgNodeWithDomain:
2270d4f… lmata 527 def test_concept_with_domain_creates_domain_link(self):
2270d4f… lmata 528 store = _make_store()
2270d4f… lmata 529 ingester = PlanopticonIngester(store)
2270d4f… lmata 530 ingester._ingest_kg_node({"type": "concept", "name": "Auth", "domain": "Security"})
2270d4f… lmata 531 domain_calls = [c[0][0] for c in store.create_node.call_args_list]
2270d4f… lmata 532 assert NodeLabel.Domain in domain_calls
2270d4f… lmata 533 store.create_edge.assert_called()
2270d4f… lmata 534
2270d4f… lmata 535
2270d4f… lmata 536 # ── _ingest_planning_entity guards and domain (lines 293, 325-326) ────────────
2270d4f… lmata 537
2270d4f… lmata 538 class TestIngestPlanningEntityBranches:
2270d4f… lmata 539 def test_skips_entity_with_empty_name(self):
2270d4f… lmata 540 store = _make_store()
2270d4f… lmata 541 ingester = PlanopticonIngester(store)
2270d4f… lmata 542 ingester._ingest_planning_entity({"planning_type": "decision", "name": ""})
2270d4f… lmata 543 store.create_node.assert_not_called()
2270d4f… lmata 544
2270d4f… lmata 545 def test_entity_with_domain_creates_domain_link(self):
2270d4f… lmata 546 store = _make_store()
2270d4f… lmata 547 ingester = PlanopticonIngester(store)
2270d4f… lmata 548 ingester._ingest_planning_entity({
2270d4f… lmata 549 "planning_type": "decision",
2270d4f… lmata 550 "name": "Switch to Postgres",
2270d4f… lmata 551 "domain": "Infrastructure",
2270d4f… lmata 552 })
2270d4f… lmata 553 domain_calls = [c[0][0] for c in store.create_node.call_args_list]
2270d4f… lmata 554 assert NodeLabel.Domain in domain_calls
2270d4f… lmata 555 store.create_edge.assert_called()
2270d4f… lmata 556
2270d4f… lmata 557
2270d4f… lmata 558 # ── _ingest_kg_relationship exception handler (lines 353-354) ─────────────────
2270d4f… lmata 559
2270d4f… lmata 560 class TestIngestKgRelationshipException:
2270d4f… lmata 561 def test_exception_in_query_is_swallowed(self):
2270d4f… lmata 562 store = _make_store()
2270d4f… lmata 563 store.query.side_effect = Exception("graph error")
2270d4f… lmata 564 ingester = PlanopticonIngester(store)
2270d4f… lmata 565 # Should not raise
2270d4f… lmata 566 ingester._ingest_kg_relationship({"source": "A", "target": "B", "type": "related_to"})
2270d4f… lmata 567 assert ingester._stats.get("edges", 0) == 0
2270d4f… lmata 568
2270d4f… lmata 569
2270d4f… lmata 570 # ── _ingest_key_points empty-point skip (line 360) ───────────────────────────
2270d4f… lmata 571
2270d4f… lmata 572 class TestIngestKeyPointsEmptySkip:
2270d4f… lmata 573 def test_skips_empty_point(self):
2270d4f… lmata 574 store = _make_store()
2270d4f… lmata 575 ingester = PlanopticonIngester(store)
2270d4f… lmata 576 ingester._ingest_key_points([{"point": "", "topic": "foo"}], "source")
2270d4f… lmata 577 store.create_node.assert_not_called()
2270d4f… lmata 578
2270d4f… lmata 579
2270d4f… lmata 580 # ── _ingest_action_items empty-action skip (line 383) ────────────────────────
2270d4f… lmata 581
2270d4f… lmata 582 class TestIngestActionItemsEmptySkip:
2270d4f… lmata 583 def test_skips_empty_action(self):
2270d4f… lmata 584 store = _make_store()
2270d4f… lmata 585 ingester = PlanopticonIngester(store)
2270d4f… lmata 586 ingester._ingest_action_items([{"action": "", "assignee": "Bob"}], "source")
2270d4f… lmata 587 store.create_node.assert_not_called()
2270d4f… lmata 588
2270d4f… lmata 589
2270d4f… lmata 590 # ── diagram element empty-string skip (line 426) ─────────────────────────────
2270d4f… lmata 591
2270d4f… lmata 592 class TestDiagramElementEmptySkip:
2270d4f… lmata 593 def test_skips_empty_diagram_element(self):
2270d4f… lmata 594 store = _make_store()
2270d4f… lmata 595 ingester = PlanopticonIngester(store)
2270d4f… lmata 596 with tempfile.TemporaryDirectory() as tmpdir:
2270d4f… lmata 597 p = Path(tmpdir) / "manifest.json"
2270d4f… lmata 598 p.write_text(json.dumps({
2270d4f… lmata 599 "video": {"title": "T", "url": "http://x.com"},
2270d4f… lmata 600 "key_points": [],
2270d4f… lmata 601 "action_items": [],
2270d4f… lmata 602 "diagrams": [{
2270d4f… lmata 603 "diagram_type": "sequence",
2270d4f… lmata 604 "timestamp": 0,
2270d4f… lmata 605 "description": "D",
2270d4f… lmata 606 "mermaid": "",
2270d4f… lmata 607 "elements": ["", " "], # all empty/whitespace
2270d4f… lmata 608 }],
2270d4f… lmata 609 }))
2270d4f… lmata 610 ingester.ingest_manifest(p)
2270d4f… lmata 611 # Only WikiPage for the diagram itself; no Concept for elements
2270d4f… lmata 612 concept_calls = [c for c in store.create_node.call_args_list
2270d4f… lmata 613 if c[0][0] == NodeLabel.Concept]
2270d4f… lmata 614 assert len(concept_calls) == 0
2270d4f… lmata 615
2270d4f… lmata 616
2270d4f… lmata 617 # ── _ingest_source empty name guard (line 440) ───────────────────────────────
2270d4f… lmata 618
2270d4f… lmata 619 class TestIngestSourceEmptyName:
2270d4f… lmata 620 def test_skips_source_with_no_name(self):
2270d4f… lmata 621 store = _make_store()
2270d4f… lmata 622 ingester = PlanopticonIngester(store)
2270d4f… lmata 623 ingester._ingest_source({"title": "", "source_id": None, "url": ""})
2270d4f… lmata 624 store.create_node.assert_not_called()
2270d4f… lmata 625
2270d4f… lmata 626
2270d4f… lmata 627 # ── _ingest_artifact empty name guard (line 453) ─────────────────────────────
2270d4f… lmata 628
2270d4f… lmata 629 class TestIngestArtifactEmptyName:
2270d4f… lmata 630 def test_skips_artifact_with_no_name(self):
2270d4f… lmata 631 store = _make_store()
2270d4f… lmata 632 ingester = PlanopticonIngester(store)
2270d4f… lmata 633 ingester._ingest_artifact({"name": ""}, "project")
2270d4f… lmata 634 store.create_node.assert_not_called()
2270d4f… lmata 635
2270d4f… lmata 636
2270d4f… lmata 637 # ── _lazy_wiki_link exception handler (lines 476-477) ────────────────────────
2270d4f… lmata 638
2270d4f… lmata 639 class TestLazyWikiLinkException:
2270d4f… lmata 640 def test_exception_in_create_edge_is_swallowed(self):
2270d4f… lmata 641 from navegador.graph.schema import NodeLabel
2270d4f… lmata 642 store = _make_store()
2270d4f… lmata 643 store.create_edge.side_effect = Exception("no such node")
2270d4f… lmata 644 ingester = PlanopticonIngester(store)
2270d4f… lmata 645 # Should not raise
2270d4f… lmata 646 ingester._lazy_wiki_link("AuthService", NodeLabel.Concept, "source-123")

Keyboard Shortcuts

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