Navegador

test: achieve 100% coverage — planopticon branches, MCP command, CLI edge cases

lmata 2026-03-23 00:21 trunk
Commit 2270d4f25a7cf6bd260ba6fbdc84752515750ebb1f652ace1641cbb45ee33b6b
Parent 7ae008029f20f95…
2 files changed +139 +165
~ tests/test_cli.py ~ tests/test_ingestion_planopticon.py
M tests/test_cli.py
+139
--- tests/test_cli.py
+++ tests/test_cli.py
@@ -474,5 +474,144 @@
474474
475475
 def test_planopticon_help(self):
476476
 runner = CliRunner()
477477
 result = runner.invoke(main, ["planopticon", "--help"])
478478
 assert result.exit_code == 0
479
+
480
+
481
+# ── _get_store with custom db path (lines 31-32) ──────────────────────────────
482
+
483
+class TestGetStoreCustomPath:
484
+ def test_get_store_calls_get_store_with_custom_path(self):
485
+ """_get_store body: custom path is forwarded to config.get_store."""
486
+ from navegador.cli.commands import _get_store
487
+ with patch("navegador.config.get_store", return_value=_mock_store()) as mock_gs:
488
+ _get_store("/custom/path.db")
489
+ mock_gs.assert_called_once_with("/custom/path.db")
490
+
491
+ def test_get_store_passes_none_for_default_path(self):
492
+ from navegador.cli.commands import _get_store
493
+ from navegador.config import DEFAULT_DB_PATH
494
+ with patch("navegador.config.get_store", return_value=_mock_store()) as mock_gs:
495
+ _get_store(DEFAULT_DB_PATH)
496
+ mock_gs.assert_called_once_with(None)
497
+
498
+
499
+# ── search table output with results (lines 208-216) ─────────────────────────
500
+
501
+class TestSearchTableOutput:
502
+ def test_search_renders_table_with_results(self):
503
+ runner = CliRunner()
504
+ node = _node("process_payment", "Function", "payments.py")
505
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
506
+ patch("navegador.context.ContextLoader") as MockCL:
507
+ MockCL.return_value.search.return_value = [node]
508
+ result = runner.invoke(main, ["search", "payment"])
509
+ assert result.exit_code == 0
510
+ assert "process_payment" in result.output
511
+
512
+
513
+# ── decorated table output with results (lines 237-248) ──────────────────────
514
+
515
+class TestDecoratedTableOutput:
516
+ def test_decorated_no_results_table(self):
517
+ runner = CliRunner()
518
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
519
+ patch("navegador.context.ContextLoader") as MockCL:
520
+ MockCL.return_value.decorated_by.return_value = []
521
+ result = runner.invoke(main, ["decorated", "login_required"])
522
+ assert result.exit_code == 0
523
+ assert "login_required" in result.output
524
+
525
+ def test_decorated_renders_table_with_results(self):
526
+ runner = CliRunner()
527
+ node = _node("my_view", "Function", "views.py")
528
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
529
+ patch("navegador.context.ContextLoader") as MockCL:
530
+ MockCL.return_value.decorated_by.return_value = [node]
531
+ result = runner.invoke(main, ["decorated", "login_required"])
532
+ assert result.exit_code == 0
533
+ assert "my_view" in result.output
534
+
535
+
536
+# ── wiki ingest without --api flag (line 410) ─────────────────────────────────
537
+
538
+class TestWikiIngestGithubNoApi:
539
+ def test_ingest_github_without_api_flag(self):
540
+ runner = CliRunner()
541
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
542
+ patch("navegador.ingestion.WikiIngester") as MockWI:
543
+ MockWI.return_value.ingest_github.return_value = {"pages": 5, "links": 3}
544
+ result = runner.invoke(main, ["wiki", "ingest", "--repo", "owner/repo"])
545
+ assert result.exit_code == 0
546
+ MockWI.return_value.ingest_github.assert_called_once()
547
+ assert "5" in result.output
548
+
549
+
550
+# ── planopticon dir with no recognised files (line 497) ──────────────────────
551
+
552
+class TestPlanopticonIngestNoKnownFiles:
553
+ def test_empty_directory_raises_usage_error(self):
554
+ runner = CliRunner()
555
+ with runner.isolated_filesystem():
556
+ Path("output").mkdir()
557
+ # No manifest.json, knowledge_graph.json, or interchange.json
558
+ Path("output/readme.txt").write_text("nothing")
559
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()):
560
+ result = runner.invoke(main, ["planopticon", "ingest", "output"])
561
+ assert result.exit_code != 0
562
+
563
+
564
+# ── planopticon auto-detect interchange/batch (lines 505, 507) ───────────────
565
+
566
+class TestPlanopticonAutoDetect:
567
+ def test_auto_detect_interchange(self):
568
+ runner = CliRunner()
569
+ with runner.isolated_filesystem():
570
+ Path("interchange.json").write_text("{}")
571
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
572
+ patch("navegador.ingestion.PlanopticonIngester") as MockPI:
573
+ MockPI.return_value.ingest_interchange.return_value = {"nodes": 0, "edges": 0}
574
+ result = runner.invoke(main, ["planopticon", "ingest", "interchange.json"])
575
+ assert result.exit_code == 0
576
+ MockPI.return_value.ingest_interchange.assert_called_once()
577
+
578
+ def test_auto_detect_batch(self):
579
+ runner = CliRunner()
580
+ with runner.isolated_filesystem():
581
+ Path("batch.json").write_text("{}")
582
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
583
+ patch("navegador.ingestion.PlanopticonIngester") as MockPI:
584
+ MockPI.return_value.ingest_batch.return_value = {"nodes": 0, "edges": 0}
585
+ result = runner.invoke(main, ["planopticon", "ingest", "batch.json"])
586
+ assert result.exit_code == 0
587
+ MockPI.return_value.ingest_batch.assert_called_once()
588
+
589
+
590
+# ── mcp command (lines 538-549) ───────────────────────────────────────────────
591
+
592
+class TestMcpCommand:
593
+ def test_mcp_command_runs_server(self):
594
+ from contextlib import asynccontextmanager
595
+
596
+ runner = CliRunner()
597
+
598
+ @asynccontextmanager
599
+ async def _fake_stdio():
600
+ yield (MagicMock(), MagicMock())
601
+
602
+ async def _fake_run(*args, **kwargs):
603
+ pass
604
+
605
+ mock_server = MagicMock()
606
+ mock_server.create_initialization_options.return_value = {}
607
+ mock_server.run = _fake_run
608
+
609
+ with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
610
+ patch.dict("sys.modules", {
611
+ "mcp": MagicMock(),
612
+ "mcp.server": MagicMock(),
613
+ "mcp.server.stdio": MagicMock(stdio_server=_fake_stdio),
614
+ }), \
615
+ patch("navegador.mcp.create_mcp_server", return_value=mock_server):
616
+ result = runner.invoke(main, ["mcp"])
617
+ assert result.exit_code == 0
479618
--- tests/test_cli.py
+++ tests/test_cli.py
@@ -474,5 +474,144 @@
474
475 def test_planopticon_help(self):
476 runner = CliRunner()
477 result = runner.invoke(main, ["planopticon", "--help"])
478 assert result.exit_code == 0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
479
--- tests/test_cli.py
+++ tests/test_cli.py
@@ -474,5 +474,144 @@
474
475 def test_planopticon_help(self):
476 runner = CliRunner()
477 result = runner.invoke(main, ["planopticon", "--help"])
478 assert result.exit_code == 0
479
480
481 # ── _get_store with custom db path (lines 31-32) ──────────────────────────────
482
483 class TestGetStoreCustomPath:
484 def test_get_store_calls_get_store_with_custom_path(self):
485 """_get_store body: custom path is forwarded to config.get_store."""
486 from navegador.cli.commands import _get_store
487 with patch("navegador.config.get_store", return_value=_mock_store()) as mock_gs:
488 _get_store("/custom/path.db")
489 mock_gs.assert_called_once_with("/custom/path.db")
490
491 def test_get_store_passes_none_for_default_path(self):
492 from navegador.cli.commands import _get_store
493 from navegador.config import DEFAULT_DB_PATH
494 with patch("navegador.config.get_store", return_value=_mock_store()) as mock_gs:
495 _get_store(DEFAULT_DB_PATH)
496 mock_gs.assert_called_once_with(None)
497
498
499 # ── search table output with results (lines 208-216) ─────────────────────────
500
501 class TestSearchTableOutput:
502 def test_search_renders_table_with_results(self):
503 runner = CliRunner()
504 node = _node("process_payment", "Function", "payments.py")
505 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
506 patch("navegador.context.ContextLoader") as MockCL:
507 MockCL.return_value.search.return_value = [node]
508 result = runner.invoke(main, ["search", "payment"])
509 assert result.exit_code == 0
510 assert "process_payment" in result.output
511
512
513 # ── decorated table output with results (lines 237-248) ──────────────────────
514
515 class TestDecoratedTableOutput:
516 def test_decorated_no_results_table(self):
517 runner = CliRunner()
518 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
519 patch("navegador.context.ContextLoader") as MockCL:
520 MockCL.return_value.decorated_by.return_value = []
521 result = runner.invoke(main, ["decorated", "login_required"])
522 assert result.exit_code == 0
523 assert "login_required" in result.output
524
525 def test_decorated_renders_table_with_results(self):
526 runner = CliRunner()
527 node = _node("my_view", "Function", "views.py")
528 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
529 patch("navegador.context.ContextLoader") as MockCL:
530 MockCL.return_value.decorated_by.return_value = [node]
531 result = runner.invoke(main, ["decorated", "login_required"])
532 assert result.exit_code == 0
533 assert "my_view" in result.output
534
535
536 # ── wiki ingest without --api flag (line 410) ─────────────────────────────────
537
538 class TestWikiIngestGithubNoApi:
539 def test_ingest_github_without_api_flag(self):
540 runner = CliRunner()
541 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
542 patch("navegador.ingestion.WikiIngester") as MockWI:
543 MockWI.return_value.ingest_github.return_value = {"pages": 5, "links": 3}
544 result = runner.invoke(main, ["wiki", "ingest", "--repo", "owner/repo"])
545 assert result.exit_code == 0
546 MockWI.return_value.ingest_github.assert_called_once()
547 assert "5" in result.output
548
549
550 # ── planopticon dir with no recognised files (line 497) ──────────────────────
551
552 class TestPlanopticonIngestNoKnownFiles:
553 def test_empty_directory_raises_usage_error(self):
554 runner = CliRunner()
555 with runner.isolated_filesystem():
556 Path("output").mkdir()
557 # No manifest.json, knowledge_graph.json, or interchange.json
558 Path("output/readme.txt").write_text("nothing")
559 with patch("navegador.cli.commands._get_store", return_value=_mock_store()):
560 result = runner.invoke(main, ["planopticon", "ingest", "output"])
561 assert result.exit_code != 0
562
563
564 # ── planopticon auto-detect interchange/batch (lines 505, 507) ───────────────
565
566 class TestPlanopticonAutoDetect:
567 def test_auto_detect_interchange(self):
568 runner = CliRunner()
569 with runner.isolated_filesystem():
570 Path("interchange.json").write_text("{}")
571 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
572 patch("navegador.ingestion.PlanopticonIngester") as MockPI:
573 MockPI.return_value.ingest_interchange.return_value = {"nodes": 0, "edges": 0}
574 result = runner.invoke(main, ["planopticon", "ingest", "interchange.json"])
575 assert result.exit_code == 0
576 MockPI.return_value.ingest_interchange.assert_called_once()
577
578 def test_auto_detect_batch(self):
579 runner = CliRunner()
580 with runner.isolated_filesystem():
581 Path("batch.json").write_text("{}")
582 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
583 patch("navegador.ingestion.PlanopticonIngester") as MockPI:
584 MockPI.return_value.ingest_batch.return_value = {"nodes": 0, "edges": 0}
585 result = runner.invoke(main, ["planopticon", "ingest", "batch.json"])
586 assert result.exit_code == 0
587 MockPI.return_value.ingest_batch.assert_called_once()
588
589
590 # ── mcp command (lines 538-549) ───────────────────────────────────────────────
591
592 class TestMcpCommand:
593 def test_mcp_command_runs_server(self):
594 from contextlib import asynccontextmanager
595
596 runner = CliRunner()
597
598 @asynccontextmanager
599 async def _fake_stdio():
600 yield (MagicMock(), MagicMock())
601
602 async def _fake_run(*args, **kwargs):
603 pass
604
605 mock_server = MagicMock()
606 mock_server.create_initialization_options.return_value = {}
607 mock_server.run = _fake_run
608
609 with patch("navegador.cli.commands._get_store", return_value=_mock_store()), \
610 patch.dict("sys.modules", {
611 "mcp": MagicMock(),
612 "mcp.server": MagicMock(),
613 "mcp.server.stdio": MagicMock(stdio_server=_fake_stdio),
614 }), \
615 patch("navegador.mcp.create_mcp_server", return_value=mock_server):
616 result = runner.invoke(main, ["mcp"])
617 assert result.exit_code == 0
618
M tests/test_ingestion_planopticon.py
+165
--- tests/test_ingestion_planopticon.py
+++ tests/test_ingestion_planopticon.py
@@ -475,5 +475,170 @@
475475
 with tempfile.TemporaryDirectory() as tmpdir:
476476
 p = Path(tmpdir) / "bad.json"
477477
 p.write_text("{ not valid json }")
478478
 result = ingester._load_json(p)
479479
 assert result == {}
480
+
481
+
482
+# ── ingest_interchange relationship/source branches (lines 201, 209) ──────────
483
+
484
+class TestInterchangeRelationshipsAndSources:
485
+ def test_ingests_relationships_in_interchange(self):
486
+ store = _make_store()
487
+ ingester = PlanopticonIngester(store)
488
+ data = {
489
+ "project": {"name": "Proj", "tags": []},
490
+ "entities": [],
491
+ "relationships": [
492
+ {"source": "Alice", "target": "Bob", "type": "related_to"}
493
+ ],
494
+ "artifacts": [],
495
+ "sources": [],
496
+ }
497
+ with tempfile.TemporaryDirectory() as tmpdir:
498
+ p = Path(tmpdir) / "interchange.json"
499
+ p.write_text(json.dumps(data))
500
+ stats = ingester.ingest_interchange(p)
501
+ store.query.assert_called()
502
+ assert stats["edges"] >= 1
503
+
504
+ def test_ingests_sources_in_interchange(self):
505
+ store = _make_store()
506
+ ingester = PlanopticonIngester(store)
507
+ data = {
508
+ "project": {"name": "Proj", "tags": []},
509
+ "entities": [],
510
+ "relationships": [],
511
+ "artifacts": [],
512
+ "sources": [{"title": "Design Doc", "url": "http://ex.com"}],
513
+ }
514
+ with tempfile.TemporaryDirectory() as tmpdir:
515
+ p = Path(tmpdir) / "interchange.json"
516
+ p.write_text(json.dumps(data))
517
+ ingester.ingest_interchange(p)
518
+ labels = [c[0][0] for c in store.create_node.call_args_list]
519
+ assert NodeLabel.WikiPage in labels
520
+
521
+
522
+# ── _ingest_kg_node with domain (lines 274-275) ───────────────────────────────
523
+
524
+class TestIngestKgNodeWithDomain:
525
+ def test_concept_with_domain_creates_domain_link(self):
526
+ store = _make_store()
527
+ ingester = PlanopticonIngester(store)
528
+ ingester._ingest_kg_node({"type": "concept", "name": "Auth", "domain": "Security"})
529
+ domain_calls = [c[0][0] for c in store.create_node.call_args_list]
530
+ assert NodeLabel.Domain in domain_calls
531
+ store.create_edge.assert_called()
532
+
533
+
534
+# ── _ingest_planning_entity guards and domain (lines 293, 325-326) ────────────
535
+
536
+class TestIngestPlanningEntityBranches:
537
+ def test_skips_entity_with_empty_name(self):
538
+ store = _make_store()
539
+ ingester = PlanopticonIngester(store)
540
+ ingester._ingest_planning_entity({"planning_type": "decision", "name": ""})
541
+ store.create_node.assert_not_called()
542
+
543
+ def test_entity_with_domain_creates_domain_link(self):
544
+ store = _make_store()
545
+ ingester = PlanopticonIngester(store)
546
+ ingester._ingest_planning_entity({
547
+ "planning_type": "decision",
548
+ "name": "Switch to Postgres",
549
+ "domain": "Infrastructure",
550
+ })
551
+ domain_calls = [c[0][0] for c in store.create_node.call_args_list]
552
+ assert NodeLabel.Domain in domain_calls
553
+ store.create_edge.assert_called()
554
+
555
+
556
+# ── _ingest_kg_relationship exception handler (lines 353-354) ─────────────────
557
+
558
+class TestIngestKgRelationshipException:
559
+ def test_exception_in_query_is_swallowed(self):
560
+ store = _make_store()
561
+ store.query.side_effect = Exception("graph error")
562
+ ingester = PlanopticonIngester(store)
563
+ # Should not raise
564
+ ingester._ingest_kg_relationship({"source": "A", "target": "B", "type": "related_to"})
565
+ assert ingester._stats.get("edges", 0) == 0
566
+
567
+
568
+# ── _ingest_key_points empty-point skip (line 360) ───────────────────────────
569
+
570
+class TestIngestKeyPointsEmptySkip:
571
+ def test_skips_empty_point(self):
572
+ store = _make_store()
573
+ ingester = PlanopticonIngester(store)
574
+ ingester._ingest_key_points([{"point": "", "topic": "foo"}], "source")
575
+ store.create_node.assert_not_called()
576
+
577
+
578
+# ── _ingest_action_items empty-action skip (line 383) ────────────────────────
579
+
580
+class TestIngestActionItemsEmptySkip:
581
+ def test_skips_empty_action(self):
582
+ store = _make_store()
583
+ ingester = PlanopticonIngester(store)
584
+ ingester._ingest_action_items([{"action": "", "assignee": "Bob"}], "source")
585
+ store.create_node.assert_not_called()
586
+
587
+
588
+# ── diagram element empty-string skip (line 426) ─────────────────────────────
589
+
590
+class TestDiagramElementEmptySkip:
591
+ def test_skips_empty_diagram_element(self):
592
+ store = _make_store()
593
+ ingester = PlanopticonIngester(store)
594
+ with tempfile.TemporaryDirectory() as tmpdir:
595
+ p = Path(tmpdir) / "manifest.json"
596
+ p.write_text(json.dumps({
597
+ "video": {"title": "T", "url": "http://x.com"},
598
+ "key_points": [],
599
+ "action_items": [],
600
+ "diagrams": [{
601
+ "diagram_type": "sequence",
602
+ "timestamp": 0,
603
+ "description": "D",
604
+ "mermaid": "",
605
+ "elements": ["", " "], # all empty/whitespace
606
+ }],
607
+ }))
608
+ ingester.ingest_manifest(p)
609
+ # Only WikiPage for the diagram itself; no Concept for elements
610
+ concept_calls = [c for c in store.create_node.call_args_list
611
+ if c[0][0] == NodeLabel.Concept]
612
+ assert len(concept_calls) == 0
613
+
614
+
615
+# ── _ingest_source empty name guard (line 440) ───────────────────────────────
616
+
617
+class TestIngestSourceEmptyName:
618
+ def test_skips_source_with_no_name(self):
619
+ store = _make_store()
620
+ ingester = PlanopticonIngester(store)
621
+ ingester._ingest_source({"title": "", "source_id": None, "url": ""})
622
+ store.create_node.assert_not_called()
623
+
624
+
625
+# ── _ingest_artifact empty name guard (line 453) ─────────────────────────────
626
+
627
+class TestIngestArtifactEmptyName:
628
+ def test_skips_artifact_with_no_name(self):
629
+ store = _make_store()
630
+ ingester = PlanopticonIngester(store)
631
+ ingester._ingest_artifact({"name": ""}, "project")
632
+ store.create_node.assert_not_called()
633
+
634
+
635
+# ── _lazy_wiki_link exception handler (lines 476-477) ────────────────────────
636
+
637
+class TestLazyWikiLinkException:
638
+ def test_exception_in_create_edge_is_swallowed(self):
639
+ from navegador.graph.schema import NodeLabel
640
+ store = _make_store()
641
+ store.create_edge.side_effect = Exception("no such node")
642
+ ingester = PlanopticonIngester(store)
643
+ # Should not raise
644
+ ingester._lazy_wiki_link("AuthService", NodeLabel.Concept, "source-123")
480645
--- tests/test_ingestion_planopticon.py
+++ tests/test_ingestion_planopticon.py
@@ -475,5 +475,170 @@
475 with tempfile.TemporaryDirectory() as tmpdir:
476 p = Path(tmpdir) / "bad.json"
477 p.write_text("{ not valid json }")
478 result = ingester._load_json(p)
479 assert result == {}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
480
--- tests/test_ingestion_planopticon.py
+++ tests/test_ingestion_planopticon.py
@@ -475,5 +475,170 @@
475 with tempfile.TemporaryDirectory() as tmpdir:
476 p = Path(tmpdir) / "bad.json"
477 p.write_text("{ not valid json }")
478 result = ingester._load_json(p)
479 assert result == {}
480
481
482 # ── ingest_interchange relationship/source branches (lines 201, 209) ──────────
483
484 class TestInterchangeRelationshipsAndSources:
485 def test_ingests_relationships_in_interchange(self):
486 store = _make_store()
487 ingester = PlanopticonIngester(store)
488 data = {
489 "project": {"name": "Proj", "tags": []},
490 "entities": [],
491 "relationships": [
492 {"source": "Alice", "target": "Bob", "type": "related_to"}
493 ],
494 "artifacts": [],
495 "sources": [],
496 }
497 with tempfile.TemporaryDirectory() as tmpdir:
498 p = Path(tmpdir) / "interchange.json"
499 p.write_text(json.dumps(data))
500 stats = ingester.ingest_interchange(p)
501 store.query.assert_called()
502 assert stats["edges"] >= 1
503
504 def test_ingests_sources_in_interchange(self):
505 store = _make_store()
506 ingester = PlanopticonIngester(store)
507 data = {
508 "project": {"name": "Proj", "tags": []},
509 "entities": [],
510 "relationships": [],
511 "artifacts": [],
512 "sources": [{"title": "Design Doc", "url": "http://ex.com"}],
513 }
514 with tempfile.TemporaryDirectory() as tmpdir:
515 p = Path(tmpdir) / "interchange.json"
516 p.write_text(json.dumps(data))
517 ingester.ingest_interchange(p)
518 labels = [c[0][0] for c in store.create_node.call_args_list]
519 assert NodeLabel.WikiPage in labels
520
521
522 # ── _ingest_kg_node with domain (lines 274-275) ───────────────────────────────
523
524 class TestIngestKgNodeWithDomain:
525 def test_concept_with_domain_creates_domain_link(self):
526 store = _make_store()
527 ingester = PlanopticonIngester(store)
528 ingester._ingest_kg_node({"type": "concept", "name": "Auth", "domain": "Security"})
529 domain_calls = [c[0][0] for c in store.create_node.call_args_list]
530 assert NodeLabel.Domain in domain_calls
531 store.create_edge.assert_called()
532
533
534 # ── _ingest_planning_entity guards and domain (lines 293, 325-326) ────────────
535
536 class TestIngestPlanningEntityBranches:
537 def test_skips_entity_with_empty_name(self):
538 store = _make_store()
539 ingester = PlanopticonIngester(store)
540 ingester._ingest_planning_entity({"planning_type": "decision", "name": ""})
541 store.create_node.assert_not_called()
542
543 def test_entity_with_domain_creates_domain_link(self):
544 store = _make_store()
545 ingester = PlanopticonIngester(store)
546 ingester._ingest_planning_entity({
547 "planning_type": "decision",
548 "name": "Switch to Postgres",
549 "domain": "Infrastructure",
550 })
551 domain_calls = [c[0][0] for c in store.create_node.call_args_list]
552 assert NodeLabel.Domain in domain_calls
553 store.create_edge.assert_called()
554
555
556 # ── _ingest_kg_relationship exception handler (lines 353-354) ─────────────────
557
558 class TestIngestKgRelationshipException:
559 def test_exception_in_query_is_swallowed(self):
560 store = _make_store()
561 store.query.side_effect = Exception("graph error")
562 ingester = PlanopticonIngester(store)
563 # Should not raise
564 ingester._ingest_kg_relationship({"source": "A", "target": "B", "type": "related_to"})
565 assert ingester._stats.get("edges", 0) == 0
566
567
568 # ── _ingest_key_points empty-point skip (line 360) ───────────────────────────
569
570 class TestIngestKeyPointsEmptySkip:
571 def test_skips_empty_point(self):
572 store = _make_store()
573 ingester = PlanopticonIngester(store)
574 ingester._ingest_key_points([{"point": "", "topic": "foo"}], "source")
575 store.create_node.assert_not_called()
576
577
578 # ── _ingest_action_items empty-action skip (line 383) ────────────────────────
579
580 class TestIngestActionItemsEmptySkip:
581 def test_skips_empty_action(self):
582 store = _make_store()
583 ingester = PlanopticonIngester(store)
584 ingester._ingest_action_items([{"action": "", "assignee": "Bob"}], "source")
585 store.create_node.assert_not_called()
586
587
588 # ── diagram element empty-string skip (line 426) ─────────────────────────────
589
590 class TestDiagramElementEmptySkip:
591 def test_skips_empty_diagram_element(self):
592 store = _make_store()
593 ingester = PlanopticonIngester(store)
594 with tempfile.TemporaryDirectory() as tmpdir:
595 p = Path(tmpdir) / "manifest.json"
596 p.write_text(json.dumps({
597 "video": {"title": "T", "url": "http://x.com"},
598 "key_points": [],
599 "action_items": [],
600 "diagrams": [{
601 "diagram_type": "sequence",
602 "timestamp": 0,
603 "description": "D",
604 "mermaid": "",
605 "elements": ["", " "], # all empty/whitespace
606 }],
607 }))
608 ingester.ingest_manifest(p)
609 # Only WikiPage for the diagram itself; no Concept for elements
610 concept_calls = [c for c in store.create_node.call_args_list
611 if c[0][0] == NodeLabel.Concept]
612 assert len(concept_calls) == 0
613
614
615 # ── _ingest_source empty name guard (line 440) ───────────────────────────────
616
617 class TestIngestSourceEmptyName:
618 def test_skips_source_with_no_name(self):
619 store = _make_store()
620 ingester = PlanopticonIngester(store)
621 ingester._ingest_source({"title": "", "source_id": None, "url": ""})
622 store.create_node.assert_not_called()
623
624
625 # ── _ingest_artifact empty name guard (line 453) ─────────────────────────────
626
627 class TestIngestArtifactEmptyName:
628 def test_skips_artifact_with_no_name(self):
629 store = _make_store()
630 ingester = PlanopticonIngester(store)
631 ingester._ingest_artifact({"name": ""}, "project")
632 store.create_node.assert_not_called()
633
634
635 # ── _lazy_wiki_link exception handler (lines 476-477) ────────────────────────
636
637 class TestLazyWikiLinkException:
638 def test_exception_in_create_edge_is_swallowed(self):
639 from navegador.graph.schema import NodeLabel
640 store = _make_store()
641 store.create_edge.side_effect = Exception("no such node")
642 ingester = PlanopticonIngester(store)
643 # Should not raise
644 ingester._lazy_wiki_link("AuthService", NodeLabel.Concept, "source-123")
645

Keyboard Shortcuts

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