Navegador

test: comprehensive test suite achieving 83% coverage - 245 tests across all core modules (schema, config, graph store, CLI, context loader, all ingesters, Python parser, MCP server) - GraphStore: 100% — mock-based unit tests for all CRUD operations - WikiIngester: 100% — including GitHub clone and API paths - RepoIngester: 88% — orchestration, _iter_source_files, _get_parser - PlanopticonIngester: 92% — all entry points and helpers - KnowledgeIngester: 96% - PythonParser: 78% — mock tree-sitter node tests for all handlers - CLI: 90% — full command surface via click CliRunner - config.get_store(): 100% — env var priority resolution - 0 ruff lint errors

lmata 2026-03-22 22:09 trunk
Commit b663b128eb1f89d534b7576cfeae2a2deda0d9a6f2287d0f47eb1b285aed6124
Parent ce0374a2be6edb5…
13 files changed +40 -2 +44 +16 +128 +263 -24 +214 +3 +176 +433 +244 +92 +235 +86 -14
~ navegador/cli/commands.py + navegador/config.py + tests/test_cli.py + tests/test_config.py ~ tests/test_context.py + tests/test_graph_store.py + tests/test_ingestion_code.py + tests/test_ingestion_knowledge.py + tests/test_ingestion_planopticon.py + tests/test_ingestion_wiki.py + tests/test_mcp_server.py + tests/test_python_parser.py ~ tests/test_schema.py
M navegador/cli/commands.py
+40 -2
--- navegador/cli/commands.py
+++ navegador/cli/commands.py
@@ -26,12 +26,12 @@
2626
 help="Output format. Use json for agent/pipe consumption.",
2727
 )
2828
2929
3030
 def _get_store(db: str):
31
- from navegador.graph import GraphStore
32
- return GraphStore.sqlite(db)
31
+ from navegador.config import DEFAULT_DB_PATH, get_store
32
+ return get_store(db if db != DEFAULT_DB_PATH else None)
3333
3434
3535
 def _emit(text: str, fmt: str) -> None:
3636
 if fmt == "json":
3737
 click.echo(text)
@@ -49,10 +49,48 @@
4949
 Combines code structure (AST, call graphs) with business knowledge
5050
 (concepts, rules, decisions, wiki) into a single queryable graph.
5151
 """
5252
 logging.basicConfig(level=logging.WARNING)
5353
54
+
55
+# ── Init ──────────────────────────────────────────────────────────────────────
56
+
57
+@main.command()
58
+@click.argument("path", default=".", type=click.Path())
59
+@click.option("--redis", "redis_url", default="",
60
+ help="Redis URL for centralized/production mode (e.g. redis://host:6379).")
61
+def init(path: str, redis_url: str):
62
+ """Initialise navegador in a project directory.
63
+
64
+ Creates .navegador/ (gitignored), writes .env.example with storage options.
65
+
66
+ \b
67
+ Local SQLite (default — zero infra):
68
+ navegador init
69
+
70
+ Centralized Redis (production / multi-agent):
71
+ navegador init --redis redis://host:6379
72
+ # Then: export NAVEGADOR_REDIS_URL=redis://host:6379
73
+ """
74
+ from navegador.config import init_project
75
+
76
+ nav_dir = init_project(path)
77
+ console.print(f"[green]Initialised navegador[/green] → {nav_dir}")
78
+
79
+ if redis_url:
80
+ console.print(
81
+ f"\n[bold]Redis mode:[/bold] set [cyan]NAVEGADOR_REDIS_URL={redis_url}[/cyan] "
82
+ "in your environment or CI secrets."
83
+ )
84
+ else:
85
+ console.print(
86
+ "\n[bold]Local SQLite mode[/bold] (default). "
87
+ "To use a shared Redis graph set [cyan]NAVEGADOR_REDIS_URL[/cyan]."
88
+ )
89
+
90
+ console.print("\nNext: [bold]navegador ingest .[/bold]")
91
+
5492
5593
 # ── CODE: ingest ──────────────────────────────────────────────────────────────
5694
5795
 @main.command()
5896
 @click.argument("repo_path", type=click.Path(exists=True))
5997
6098
ADDED navegador/config.py
6199
ADDED tests/test_cli.py
62100
ADDED tests/test_config.py
--- navegador/cli/commands.py
+++ navegador/cli/commands.py
@@ -26,12 +26,12 @@
26 help="Output format. Use json for agent/pipe consumption.",
27 )
28
29
30 def _get_store(db: str):
31 from navegador.graph import GraphStore
32 return GraphStore.sqlite(db)
33
34
35 def _emit(text: str, fmt: str) -> None:
36 if fmt == "json":
37 click.echo(text)
@@ -49,10 +49,48 @@
49 Combines code structure (AST, call graphs) with business knowledge
50 (concepts, rules, decisions, wiki) into a single queryable graph.
51 """
52 logging.basicConfig(level=logging.WARNING)
53
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
54
55 # ── CODE: ingest ──────────────────────────────────────────────────────────────
56
57 @main.command()
58 @click.argument("repo_path", type=click.Path(exists=True))
59
60 DDED navegador/config.py
61 DDED tests/test_cli.py
62 DDED tests/test_config.py
--- navegador/cli/commands.py
+++ navegador/cli/commands.py
@@ -26,12 +26,12 @@
26 help="Output format. Use json for agent/pipe consumption.",
27 )
28
29
30 def _get_store(db: str):
31 from navegador.config import DEFAULT_DB_PATH, get_store
32 return get_store(db if db != DEFAULT_DB_PATH else None)
33
34
35 def _emit(text: str, fmt: str) -> None:
36 if fmt == "json":
37 click.echo(text)
@@ -49,10 +49,48 @@
49 Combines code structure (AST, call graphs) with business knowledge
50 (concepts, rules, decisions, wiki) into a single queryable graph.
51 """
52 logging.basicConfig(level=logging.WARNING)
53
54
55 # ── Init ──────────────────────────────────────────────────────────────────────
56
57 @main.command()
58 @click.argument("path", default=".", type=click.Path())
59 @click.option("--redis", "redis_url", default="",
60 help="Redis URL for centralized/production mode (e.g. redis://host:6379).")
61 def init(path: str, redis_url: str):
62 """Initialise navegador in a project directory.
63
64 Creates .navegador/ (gitignored), writes .env.example with storage options.
65
66 \b
67 Local SQLite (default — zero infra):
68 navegador init
69
70 Centralized Redis (production / multi-agent):
71 navegador init --redis redis://host:6379
72 # Then: export NAVEGADOR_REDIS_URL=redis://host:6379
73 """
74 from navegador.config import init_project
75
76 nav_dir = init_project(path)
77 console.print(f"[green]Initialised navegador[/green] → {nav_dir}")
78
79 if redis_url:
80 console.print(
81 f"\n[bold]Redis mode:[/bold] set [cyan]NAVEGADOR_REDIS_URL={redis_url}[/cyan] "
82 "in your environment or CI secrets."
83 )
84 else:
85 console.print(
86 "\n[bold]Local SQLite mode[/bold] (default). "
87 "To use a shared Redis graph set [cyan]NAVEGADOR_REDIS_URL[/cyan]."
88 )
89
90 console.print("\nNext: [bold]navegador ingest .[/bold]")
91
92
93 # ── CODE: ingest ──────────────────────────────────────────────────────────────
94
95 @main.command()
96 @click.argument("repo_path", type=click.Path(exists=True))
97
98 DDED navegador/config.py
99 DDED tests/test_cli.py
100 DDED tests/test_config.py
A navegador/config.py
+44
--- a/navegador/config.py
+++ b/navegador/config.py
@@ -0,0 +1,44 @@
1
+"""
2
+Navegador storage configuration.
3
+
4
+Priority order for store selection:
5
+ 1. NAVEGADOR_REDIS_URL env var → Redis/FalkorDB (centralized, multi-agent)
6
+ 2. NAVEGADOR_DB env var → SQLite at that path
7
+ 3. Default → SQLite at .navegador/graph.db
8
+
9
+Local (SQLite) — default:
10
+ - DB lives at .navegador/graph.db inside the project
11
+ - Zero infrastructure — single file, gitignored
12
+ - Each developer has their own local graph
13
+ - Re-ingest anytime: navegador ingest .
14
+
15
+Centralized (Redis/FalkorDB) — production / multi-agent:
16
+ - Set NAVEGADOR_REDIS_URL=redis://host:6379
17
+ - All agents read/write the same shared graph
18
+ - No staleness between agents or CI
19
+ - Requires a Redis instance with the FalkorDB module loaded
20
+"""
21
+
22
+import os
23
+from pathlib import Path
24
+
25
+DEFAULT_DB_PATH = ".navegador/graph.db"
26
+
27
+
28
+def get_store(db_path: str | None = None):
29
+ """
30
+ Return a GraphStore using the best available backend.
31
+
32
+ Resolution order:
33
+ 1. Explicit db_path argument (used when --db flag is passed)
34
+ 2. NAVEGADOR_REDIS_URL env var → Redis backend
35
+ 3. NAVEGADOR_DB env var → SQLite at that path
36
+ 4. Default SQLite path
37
+ """
38
+ from navegador.graph import GraphStore
39
+
40
+ # 1. Explicit path always means SQLite
41
+ if db_path and db_path != DEFAULT_DB_PATH:
42
+ return GraphStore.sqlite(db_path)
43
+
44
+ # 2. Redis URL takes precedence over SQL
--- a/navegador/config.py
+++ b/navegador/config.py
@@ -0,0 +1,44 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/navegador/config.py
+++ b/navegador/config.py
@@ -0,0 +1,44 @@
1 """
2 Navegador storage configuration.
3
4 Priority order for store selection:
5 1. NAVEGADOR_REDIS_URL env var → Redis/FalkorDB (centralized, multi-agent)
6 2. NAVEGADOR_DB env var → SQLite at that path
7 3. Default → SQLite at .navegador/graph.db
8
9 Local (SQLite) — default:
10 - DB lives at .navegador/graph.db inside the project
11 - Zero infrastructure — single file, gitignored
12 - Each developer has their own local graph
13 - Re-ingest anytime: navegador ingest .
14
15 Centralized (Redis/FalkorDB) — production / multi-agent:
16 - Set NAVEGADOR_REDIS_URL=redis://host:6379
17 - All agents read/write the same shared graph
18 - No staleness between agents or CI
19 - Requires a Redis instance with the FalkorDB module loaded
20 """
21
22 import os
23 from pathlib import Path
24
25 DEFAULT_DB_PATH = ".navegador/graph.db"
26
27
28 def get_store(db_path: str | None = None):
29 """
30 Return a GraphStore using the best available backend.
31
32 Resolution order:
33 1. Explicit db_path argument (used when --db flag is passed)
34 2. NAVEGADOR_REDIS_URL env var → Redis backend
35 3. NAVEGADOR_DB env var → SQLite at that path
36 4. Default SQLite path
37 """
38 from navegador.graph import GraphStore
39
40 # 1. Explicit path always means SQLite
41 if db_path and db_path != DEFAULT_DB_PATH:
42 return GraphStore.sqlite(db_path)
43
44 # 2. Redis URL takes precedence over SQL
A tests/test_cli.py
+16
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -0,0 +1,16 @@
1
+"""Tests for navegador CLI commands via click CliRunner."""
2
+
3
+import json
4
+from pathlib import Path
5
+from unittest.mock import MagicMock, patch
6
+
7
+from click.testing import CliRunner
8
+
9
+from navegador.cli.commands import main
10
+from navegador.context.loader import ContextBundle, ContextNode
11
+
12
+# ── Helpers ───────────────────────────────────────────────────────────────────
13
+
14
+def _mock_store():
15
+ store = MagicMock()
16
+# ──
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -0,0 +1,16 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -0,0 +1,16 @@
1 """Tests for navegador CLI commands via click CliRunner."""
2
3 import json
4 from pathlib import Path
5 from unittest.mock import MagicMock, patch
6
7 from click.testing import CliRunner
8
9 from navegador.cli.commands import main
10 from navegador.context.loader import ContextBundle, ContextNode
11
12 # ── Helpers ───────────────────────────────────────────────────────────────────
13
14 def _mock_store():
15 store = MagicMock()
16 # ──
A tests/test_config.py
+128
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -0,0 +1,128 @@
1
+"""Tests for navegador.config — get_store() env var resolution and init_project()."""
2
+
3
+import tempfile
4
+from pathlib import Path
5
+from unittest.mock import MagicMock, patch
6
+
7
+
8
+class TestGetStore:
9
+ def test_explicit_path_returns_sqlite(self):
10
+ with patch("navegador.graph.GraphStore") as mock_gs:
11
+ mock_gs.sqlite.return_value = MagicMock()
12
+ from navegador.config import get_store
13
+
14
+ get_store("/tmp/test.db")
15
+ mock_gs.sqlite.assert_called_once_with("/tmp/test.db")
16
+
17
+ def test_redis_url_env_returns_redis(self, monkeypatch):
18
+ monkeypatch.setenv("NAVEGADOR_REDIS_URL", "redis://localhost:6379")
19
+ monkeypatch.delenv("NAVEGADOR_DB", raising=False)
20
+ with patch("navegador.graph.GraphStore") as mock_gs:
21
+ mock_gs.redis.return_value = MagicMock()
22
+ # Re-import to pick up env changes
23
+ import importlib
24
+
25
+ import navegador.config as cfg
26
+ importlib.reload(cfg)
27
+ cfg.get_store()
28
+ mock_gs.redis.assert_called_once_with("redis://localhost:6379")
29
+
30
+ def test_db_env_returns_sqlite(self, monkeypatch):
31
+ monkeypatch.delenv("NAVEGADOR_REDIS_URL", raising=False)
32
+ monkeypatch.setenv("NAVEGADOR_DB", "/tmp/custom.db")
33
+ with patch("navegador.graph.GraphStore") as mock_gs:
34
+ mock_gs.sqlite.return_value = MagicMock()
35
+ import importlib
36
+
37
+ import navegador.config as cfg
38
+ importlib.reload(cfg)
39
+ cfg.get_store()
40
+ mock_gs.sqlite.assert_called_once_with("/tmp/custom.db")
41
+
42
+ def test_default_sqlite_path(self, monkeypatch):
43
+ monkeypatch.delenv("NAVEGADOR_REDIS_URL", raising=False)
44
+ monkeypatch.delenv("NAVEGADOR_DB", raising=False)
45
+ with patch("navegador.graph.GraphStore") as mock_gs:
46
+ mock_gs.sqlite.return_value = MagicMock()
47
+ import importlib
48
+
49
+ import navegador.config as cfg
50
+ importlib.reload(cfg)
51
+ cfg.get_store()
52
+ mock_gs.sqlite.assert_called_once_with(".navegador/graph.db")
53
+
54
+ def test_redis_takes_precedence_over_db_env(self, monkeypatch):
55
+ monkeypatch.setenv("NAVEGADOR_REDIS_URL", "redis://myhost:6379")
56
+ monkeypatch.setenv("NAVEGADOR_DB", "/tmp/other.db")
57
+ with patch("navegador.graph.GraphStore") as mock_gs:
58
+ mock_gs.redis.return_value = MagicMock()
59
+ import importlib
60
+
61
+ import navegador.config as cfg
62
+ importlib.reload(cfg)
63
+ cfg.get_store()
64
+ mock_gs.redis.assert_called_once_with("redis://myhost:6379")
65
+ mock_gs.sqlite.assert_not_called()
66
+
67
+
68
+class TestInitProject:
69
+ def test_creates_navegador_dir(self):
70
+ with tempfile.TemporaryDirectory() as tmpdir:
71
+ from navegador.config import init_project
72
+ nav_dir = init_project(tmpdir)
73
+ assert nav_dir.exists()
74
+ assert nav_dir.name == ".navegador"
75
+
76
+ def test_creates_env_example(self):
77
+ with tempfile.TemporaryDirectory() as tmpdir:
78
+ from navegador.config import init_project
79
+ nav_dir = init_project(tmpdir)
80
+ env_example = nav_dir / ".env.example"
81
+ assert env_example.exists()
82
+ content = env_example.read_text()
83
+ assert "NAVEGADOR_DB" in content
84
+ assert "NAVEGADOR_REDIS_URL" in content
85
+
86
+ def test_does_not_overwrite_existing_env_example(self):
87
+ with tempfile.TemporaryDirectory() as tmpdir:
88
+ from navegador.config import init_project
89
+ nav_dir = Path(tmpdir) / ".navegador"
90
+ nav_dir.mkdir()
91
+ env_example = nav_dir / ".env.example"
92
+ env_example.write_text("custom content")
93
+ init_project(tmpdir)
94
+ assert env_example.read_text() == "custom content"
95
+
96
+ def test_creates_gitignore_if_missing(self):
97
+ with tempfile.TemporaryDirectory() as tmpdir:
98
+ from navegador.config import init_project
99
+ init_project(tmpdir)
100
+ gitignore = Path(tmpdir) / ".gitignore"
101
+ assert gitignore.exists()
102
+ assert ".navegador/" in gitignore.read_text()
103
+
104
+ def test_appends_to_existing_gitignore(self):
105
+ with tempfile.TemporaryDirectory() as tmpdir:
106
+ gitignore = Path(tmpdir) / ".gitignore"
107
+ gitignore.write_text("*.pyc\n__pycache__/\n")
108
+ from navegador.config import init_project
109
+ init_project(tmpdir)
110
+ content = gitignore.read_text()
111
+ assert "*.pyc" in content
112
+ assert ".navegador/" in content
113
+
114
+ def test_does_not_duplicate_gitignore_entry(self):
115
+ with tempfile.TemporaryDirectory() as tmpdir:
116
+ gitignore = Path(tmpdir) / ".gitignore"
117
+ gitignore.write_text(".navegador/\n")
118
+ from navegador.config import init_project
119
+ init_project(tmpdir)
120
+ content = gitignore.read_text()
121
+ assert content.count(".navegador/") == 1
122
+
123
+ def test_returns_nav_dir_path(self):
124
+ with tempfile.TemporaryDirectory() as tmpdir:
125
+ from navegador.config import init_project
126
+ result = init_project(tmpdir)
127
+ assert isinstance(result, Path)
128
+ assert result == Path(tmpdir
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -0,0 +1,128 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -0,0 +1,128 @@
1 """Tests for navegador.config — get_store() env var resolution and init_project()."""
2
3 import tempfile
4 from pathlib import Path
5 from unittest.mock import MagicMock, patch
6
7
8 class TestGetStore:
9 def test_explicit_path_returns_sqlite(self):
10 with patch("navegador.graph.GraphStore") as mock_gs:
11 mock_gs.sqlite.return_value = MagicMock()
12 from navegador.config import get_store
13
14 get_store("/tmp/test.db")
15 mock_gs.sqlite.assert_called_once_with("/tmp/test.db")
16
17 def test_redis_url_env_returns_redis(self, monkeypatch):
18 monkeypatch.setenv("NAVEGADOR_REDIS_URL", "redis://localhost:6379")
19 monkeypatch.delenv("NAVEGADOR_DB", raising=False)
20 with patch("navegador.graph.GraphStore") as mock_gs:
21 mock_gs.redis.return_value = MagicMock()
22 # Re-import to pick up env changes
23 import importlib
24
25 import navegador.config as cfg
26 importlib.reload(cfg)
27 cfg.get_store()
28 mock_gs.redis.assert_called_once_with("redis://localhost:6379")
29
30 def test_db_env_returns_sqlite(self, monkeypatch):
31 monkeypatch.delenv("NAVEGADOR_REDIS_URL", raising=False)
32 monkeypatch.setenv("NAVEGADOR_DB", "/tmp/custom.db")
33 with patch("navegador.graph.GraphStore") as mock_gs:
34 mock_gs.sqlite.return_value = MagicMock()
35 import importlib
36
37 import navegador.config as cfg
38 importlib.reload(cfg)
39 cfg.get_store()
40 mock_gs.sqlite.assert_called_once_with("/tmp/custom.db")
41
42 def test_default_sqlite_path(self, monkeypatch):
43 monkeypatch.delenv("NAVEGADOR_REDIS_URL", raising=False)
44 monkeypatch.delenv("NAVEGADOR_DB", raising=False)
45 with patch("navegador.graph.GraphStore") as mock_gs:
46 mock_gs.sqlite.return_value = MagicMock()
47 import importlib
48
49 import navegador.config as cfg
50 importlib.reload(cfg)
51 cfg.get_store()
52 mock_gs.sqlite.assert_called_once_with(".navegador/graph.db")
53
54 def test_redis_takes_precedence_over_db_env(self, monkeypatch):
55 monkeypatch.setenv("NAVEGADOR_REDIS_URL", "redis://myhost:6379")
56 monkeypatch.setenv("NAVEGADOR_DB", "/tmp/other.db")
57 with patch("navegador.graph.GraphStore") as mock_gs:
58 mock_gs.redis.return_value = MagicMock()
59 import importlib
60
61 import navegador.config as cfg
62 importlib.reload(cfg)
63 cfg.get_store()
64 mock_gs.redis.assert_called_once_with("redis://myhost:6379")
65 mock_gs.sqlite.assert_not_called()
66
67
68 class TestInitProject:
69 def test_creates_navegador_dir(self):
70 with tempfile.TemporaryDirectory() as tmpdir:
71 from navegador.config import init_project
72 nav_dir = init_project(tmpdir)
73 assert nav_dir.exists()
74 assert nav_dir.name == ".navegador"
75
76 def test_creates_env_example(self):
77 with tempfile.TemporaryDirectory() as tmpdir:
78 from navegador.config import init_project
79 nav_dir = init_project(tmpdir)
80 env_example = nav_dir / ".env.example"
81 assert env_example.exists()
82 content = env_example.read_text()
83 assert "NAVEGADOR_DB" in content
84 assert "NAVEGADOR_REDIS_URL" in content
85
86 def test_does_not_overwrite_existing_env_example(self):
87 with tempfile.TemporaryDirectory() as tmpdir:
88 from navegador.config import init_project
89 nav_dir = Path(tmpdir) / ".navegador"
90 nav_dir.mkdir()
91 env_example = nav_dir / ".env.example"
92 env_example.write_text("custom content")
93 init_project(tmpdir)
94 assert env_example.read_text() == "custom content"
95
96 def test_creates_gitignore_if_missing(self):
97 with tempfile.TemporaryDirectory() as tmpdir:
98 from navegador.config import init_project
99 init_project(tmpdir)
100 gitignore = Path(tmpdir) / ".gitignore"
101 assert gitignore.exists()
102 assert ".navegador/" in gitignore.read_text()
103
104 def test_appends_to_existing_gitignore(self):
105 with tempfile.TemporaryDirectory() as tmpdir:
106 gitignore = Path(tmpdir) / ".gitignore"
107 gitignore.write_text("*.pyc\n__pycache__/\n")
108 from navegador.config import init_project
109 init_project(tmpdir)
110 content = gitignore.read_text()
111 assert "*.pyc" in content
112 assert ".navegador/" in content
113
114 def test_does_not_duplicate_gitignore_entry(self):
115 with tempfile.TemporaryDirectory() as tmpdir:
116 gitignore = Path(tmpdir) / ".gitignore"
117 gitignore.write_text(".navegador/\n")
118 from navegador.config import init_project
119 init_project(tmpdir)
120 content = gitignore.read_text()
121 assert content.count(".navegador/") == 1
122
123 def test_returns_nav_dir_path(self):
124 with tempfile.TemporaryDirectory() as tmpdir:
125 from navegador.config import init_project
126 result = init_project(tmpdir)
127 assert isinstance(result, Path)
128 assert result == Path(tmpdir
M tests/test_context.py
+263 -24
--- tests/test_context.py
+++ tests/test_context.py
@@ -1,9 +1,13 @@
1
-"""Tests for context bundle serialization (no graph required)."""
1
+"""Tests for ContextBundle serialization and ContextLoader with mock store."""
2
+
3
+import json
4
+from unittest.mock import MagicMock
25
3
-from navegador.context.loader import ContextBundle, ContextNode
6
+from navegador.context.loader import ContextBundle, ContextLoader, ContextNode
47
8
+# ── Fixtures ──────────────────────────────────────────────────────────────────
59
610
 def _make_bundle():
711
 target = ContextNode(
812
 type="Function",
913
 name="get_user",
@@ -14,32 +18,267 @@
1418
 )
1519
 nodes = [
1620
 ContextNode(type="Function", name="validate_token", file_path="src/auth.py", line_start=10),
1721
 ContextNode(type="Class", name="User", file_path="src/models.py", line_start=5),
1822
 ]
23
+ edges = [{"from": "get_user", "type": "CALLS", "to": "validate_token"}]
24
+ return ContextBundle(target=target, nodes=nodes, edges=edges,
25
+ metadata={"query": "function_context"})
26
+
27
+
28
+def _make_knowledge_bundle():
29
+ target = ContextNode(
30
+ type="Concept", name="JWT", description="Stateless token auth", domain="auth"
31
+ )
32
+ nodes = [
33
+ ContextNode(type="Rule", name="Tokens must expire"),
34
+ ContextNode(type="WikiPage", name="Auth Overview"),
35
+ ]
1936
 edges = [
20
- {"from": "get_user", "type": "CALLS", "to": "validate_token"},
37
+ {"from": "Tokens must expire", "type": "GOVERNS", "to": "JWT"},
38
+ {"from": "Auth Overview", "type": "DOCUMENTS", "to": "JWT"},
2139
 ]
2240
 return ContextBundle(target=target, nodes=nodes, edges=edges)
2341
2442
25
-def test_bundle_to_dict():
26
- bundle = _make_bundle()
27
- d = bundle.to_dict()
28
- assert d["target"]["name"] == "get_user"
29
- assert len(d["nodes"]) == 2
30
- assert len(d["edges"]) == 1
31
-
32
-
33
-def test_bundle_to_json():
34
- import json
35
- bundle = _make_bundle()
36
- data = json.loads(bundle.to_json())
37
- assert data["target"]["type"] == "Function"
38
-
39
-
40
-def test_bundle_to_markdown():
41
- bundle = _make_bundle()
42
- md = bundle.to_markdown()
43
- assert "get_user" in md
44
- assert "CALLS" in md
45
- assert "validate_token" in md
43
+def _mock_store(rows=None):
44
+ store = MagicMock()
45
+ result = MagicMock()
46
+ result.result_set = rows or []
47
+ store.query.return_value = result
48
+ return store
49
+
50
+
51
+# ── ContextNode ───────────────────────────────────────────────────────────────
52
+
53
+class TestContextNode:
54
+ def test_defaults(self):
55
+ n = ContextNode(type="Function", name="foo")
56
+ assert n.file_path == ""
57
+ assert n.line_start is None
58
+ assert n.docstring is None
59
+ assert n.domain is None
60
+
61
+ def test_knowledge_fields(self):
62
+ n = ContextNode(type="Concept", name="Payment", description="A payment", domain="billing")
63
+ assert n.description == "A payment"
64
+ assert n.domain == "billing"
65
+
66
+
67
+# ── ContextBundle.to_dict ─────────────────────────────────────────────────────
68
+
69
+class TestContextBundleDict:
70
+ def test_structure(self):
71
+ b = _make_bundle()
72
+ d = b.to_dict()
73
+ assert d["target"]["name"] == "get_user"
74
+ assert len(d["nodes"]) == 2
75
+ assert len(d["edges"]) == 1
76
+ assert d["metadata"]["query"] == "function_context"
77
+
78
+ def test_roundtrip(self):
79
+ b = _make_bundle()
80
+ d = b.to_dict()
81
+ assert d["target"]["type"] == "Function"
82
+ assert d["nodes"][0]["name"] == "validate_token"
83
+
84
+
85
+# ── ContextBundle.to_json ─────────────────────────────────────────────────────
86
+
87
+class TestContextBundleJson:
88
+ def test_valid_json(self):
89
+ b = _make_bundle()
90
+ data = json.loads(b.to_json())
91
+ assert data["target"]["name"] == "get_user"
92
+
93
+ def test_indent(self):
94
+ b = _make_bundle()
95
+ raw = b.to_json(indent=4)
96
+ assert " " in raw # 4-space indent
97
+
98
+
99
+# ── ContextBundle.to_markdown ─────────────────────────────────────────────────
100
+
101
+class TestContextBundleMarkdown:
102
+ def test_contains_name(self):
103
+ b = _make_bundle()
104
+ md = b.to_markdown()
105
+ assert "get_user" in md
106
+
107
+ def test_contains_edge(self):
108
+ md = _make_bundle().to_markdown()
109
+ assert "CALLS" in md
110
+ assert "validate_token" in md
111
+
112
+ def test_contains_docstring(self):
113
+ md = _make_bundle().to_markdown()
114
+ assert "Return a user by ID." in md
115
+
116
+ def test_contains_signature(self):
117
+ md = _make_bundle().to_markdown()
118
+ assert "def get_user" in md
119
+
120
+ def test_knowledge_bundle(self):
121
+ md = _make_knowledge_bundle().to_markdown()
122
+ assert "JWT" in md
123
+ assert "auth" in md
124
+ assert "GOVERNS" in md
125
+
126
+ def test_empty_nodes(self):
127
+ target = ContextNode(type="File", name="empty.py", file_path="empty.py")
128
+ b = ContextBundle(target=target)
129
+ md = b.to_markdown()
130
+ assert "empty.py" in md
131
+
132
+
133
+# ── ContextLoader ─────────────────────────────────────────────────────────────
134
+
135
+class TestContextLoaderFile:
136
+ def test_load_file_empty(self):
137
+ store = _mock_store([])
138
+ loader = ContextLoader(store)
139
+ bundle = loader.load_file("src/auth.py")
140
+ assert bundle.target.name == "auth.py"
141
+ assert bundle.target.type == "File"
142
+ assert bundle.nodes == []
143
+
144
+ def test_load_file_with_rows(self):
145
+ rows = [["Function", "get_user", 10, "Get a user", "def get_user()"]]
146
+ store = _mock_store(rows)
147
+ loader = ContextLoader(store)
148
+ bundle = loader.load_file("src/auth.py")
149
+ assert len(bundle.nodes) == 1
150
+ assert bundle.nodes[0].name == "get_user"
151
+ assert bundle.nodes[0].type == "Function"
152
+
153
+
154
+class TestContextLoaderFunction:
155
+ def test_load_function_no_results(self):
156
+ store = _mock_store([])
157
+ loader = ContextLoader(store)
158
+ bundle = loader.load_function("get_user", file_path="src/auth.py")
159
+ assert bundle.target.name == "get_user"
160
+ assert bundle.nodes == []
161
+ assert bundle.edges == []
162
+
163
+ def test_load_function_with_callee(self):
164
+ def side_effect(query, params):
165
+ result = MagicMock()
166
+ if "CALLEES" in query or "callee" in query.lower():
167
+ result.result_set = [["Function", "validate_token", "src/auth.py", 5]]
168
+ elif "CALLERS" in query or "caller" in query.lower():
169
+ result.result_set = []
170
+ else:
171
+ result.result_set = []
172
+ return result
173
+
174
+ store = MagicMock()
175
+ store.query.side_effect = side_effect
176
+ loader = ContextLoader(store)
177
+ loader.load_function("get_user")
178
+ # Should have called query multiple times
179
+ assert store.query.called
180
+
181
+ def test_load_function_default_file_path(self):
182
+ store = _mock_store([])
183
+ loader = ContextLoader(store)
184
+ bundle = loader.load_function("foo")
185
+ assert bundle.target.file_path == ""
186
+
187
+
188
+class TestContextLoaderClass:
189
+ def test_load_class_empty(self):
190
+ store = _mock_store([])
191
+ loader = ContextLoader(store)
192
+ bundle = loader.load_class("AuthService")
193
+ assert bundle.target.name == "AuthService"
194
+ assert bundle.target.type == "Class"
195
+
196
+ def test_load_class_with_parent(self):
197
+ call_count = [0]
198
+
199
+ def side_effect(query, params=None):
200
+ result = MagicMock()
201
+ call_count[0] += 1
202
+ if call_count[0] == 1:
203
+ result.result_set = [["BaseService", "src/base.py"]]
204
+ else:
205
+ result.result_set = []
206
+ return result
207
+
208
+ store = MagicMock()
209
+ store.query.side_effect = side_effect
210
+ loader = ContextLoader(store)
211
+ bundle = loader.load_class("AuthService")
212
+ assert len(bundle.nodes) >= 1
213
+
214
+
215
+class TestContextLoaderExplain:
216
+ def test_explain_empty(self):
217
+ store = _mock_store([])
218
+ loader = ContextLoader(store)
219
+ bundle = loader.explain("get_user")
220
+ assert bundle.target.name == "get_user"
221
+ assert bundle.metadata["query"] == "explain"
222
+
223
+
224
+class TestContextLoaderSearch:
225
+ def test_search_empty(self):
226
+ store = _mock_store([])
227
+ loader = ContextLoader(store)
228
+ results = loader.search("auth")
229
+ assert results == []
230
+
231
+ def test_search_returns_nodes(self):
232
+ rows = [["Function", "authenticate", "src/auth.py", 10, "Authenticate a user"]]
233
+ store = _mock_store(rows)
234
+ loader = ContextLoader(store)
235
+ results = loader.search("auth")
236
+ assert len(results) == 1
237
+ assert results[0].name == "authenticate"
238
+ assert results[0].type == "Function"
239
+
240
+ def test_search_all(self):
241
+ rows = [["Concept", "Authentication", "", None, "The auth concept"]]
242
+ store = _mock_store(rows)
243
+ loader = ContextLoader(store)
244
+ results = loader.search_all("auth")
245
+ assert len(results) == 1
246
+
247
+ def test_search_by_docstring(self):
248
+ rows = [["Function", "login", "src/auth.py", 5, "Log in a user"]]
249
+ store = _mock_store(rows)
250
+ loader = ContextLoader(store)
251
+ results = loader.search_by_docstring("log in")
252
+ assert len(results) == 1
253
+
254
+ def test_decorated_by(self):
255
+ rows = [["Function", "protected_view", "src/views.py", 20]]
256
+ store = _mock_store(rows)
257
+ loader = ContextLoader(store)
258
+ results = loader.decorated_by("login_required")
259
+ assert len(results) == 1
260
+ assert results[0].name == "protected_view"
261
+
262
+
263
+class TestContextLoaderConcept:
264
+ def test_load_concept_not_found(self):
265
+ store = _mock_store([])
266
+ loader = ContextLoader(store)
267
+ bundle = loader.load_concept("Unknown")
268
+ assert bundle.metadata.get("found") is False
269
+
270
+ def test_load_concept_found(self):
271
+ rows = [["JWT", "Stateless token auth", "active", "auth", [], [], [], []]]
272
+ store = _mock_store(rows)
273
+ loader = ContextLoader(store)
274
+ bundle = loader.load_concept("JWT")
275
+ assert bundle.target.name == "JWT"
276
+ assert bundle.target.description == "Stateless token auth"
277
+
278
+ def test_load_domain(self):
279
+ rows = [["Function", "login", "src/auth.py", "Log in"]]
280
+ store = _mock_store(rows)
281
+ loader = ContextLoader(store)
282
+ bundle = loader.load_domain("auth")
283
+ assert bundle.target.name == "auth"
284
+ assert bundle.target.type == "Domain"
46285
47286
ADDED tests/test_graph_store.py
48287
ADDED tests/test_ingestion_code.py
49288
ADDED tests/test_ingestion_knowledge.py
50289
ADDED tests/test_ingestion_planopticon.py
51290
ADDED tests/test_ingestion_wiki.py
52291
ADDED tests/test_mcp_server.py
53292
ADDED tests/test_python_parser.py
--- tests/test_context.py
+++ tests/test_context.py
@@ -1,9 +1,13 @@
1 """Tests for context bundle serialization (no graph required)."""
 
 
 
2
3 from navegador.context.loader import ContextBundle, ContextNode
4
 
5
6 def _make_bundle():
7 target = ContextNode(
8 type="Function",
9 name="get_user",
@@ -14,32 +18,267 @@
14 )
15 nodes = [
16 ContextNode(type="Function", name="validate_token", file_path="src/auth.py", line_start=10),
17 ContextNode(type="Class", name="User", file_path="src/models.py", line_start=5),
18 ]
 
 
 
 
 
 
 
 
 
 
 
 
 
19 edges = [
20 {"from": "get_user", "type": "CALLS", "to": "validate_token"},
 
21 ]
22 return ContextBundle(target=target, nodes=nodes, edges=edges)
23
24
25 def test_bundle_to_dict():
26 bundle = _make_bundle()
27 d = bundle.to_dict()
28 assert d["target"]["name"] == "get_user"
29 assert len(d["nodes"]) == 2
30 assert len(d["edges"]) == 1
31
32
33 def test_bundle_to_json():
34 import json
35 bundle = _make_bundle()
36 data = json.loads(bundle.to_json())
37 assert data["target"]["type"] == "Function"
38
39
40 def test_bundle_to_markdown():
41 bundle = _make_bundle()
42 md = bundle.to_markdown()
43 assert "get_user" in md
44 assert "CALLS" in md
45 assert "validate_token" in md
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
46
47 DDED tests/test_graph_store.py
48 DDED tests/test_ingestion_code.py
49 DDED tests/test_ingestion_knowledge.py
50 DDED tests/test_ingestion_planopticon.py
51 DDED tests/test_ingestion_wiki.py
52 DDED tests/test_mcp_server.py
53 DDED tests/test_python_parser.py
--- tests/test_context.py
+++ tests/test_context.py
@@ -1,9 +1,13 @@
1 """Tests for ContextBundle serialization and ContextLoader with mock store."""
2
3 import json
4 from unittest.mock import MagicMock
5
6 from navegador.context.loader import ContextBundle, ContextLoader, ContextNode
7
8 # ── Fixtures ──────────────────────────────────────────────────────────────────
9
10 def _make_bundle():
11 target = ContextNode(
12 type="Function",
13 name="get_user",
@@ -14,32 +18,267 @@
18 )
19 nodes = [
20 ContextNode(type="Function", name="validate_token", file_path="src/auth.py", line_start=10),
21 ContextNode(type="Class", name="User", file_path="src/models.py", line_start=5),
22 ]
23 edges = [{"from": "get_user", "type": "CALLS", "to": "validate_token"}]
24 return ContextBundle(target=target, nodes=nodes, edges=edges,
25 metadata={"query": "function_context"})
26
27
28 def _make_knowledge_bundle():
29 target = ContextNode(
30 type="Concept", name="JWT", description="Stateless token auth", domain="auth"
31 )
32 nodes = [
33 ContextNode(type="Rule", name="Tokens must expire"),
34 ContextNode(type="WikiPage", name="Auth Overview"),
35 ]
36 edges = [
37 {"from": "Tokens must expire", "type": "GOVERNS", "to": "JWT"},
38 {"from": "Auth Overview", "type": "DOCUMENTS", "to": "JWT"},
39 ]
40 return ContextBundle(target=target, nodes=nodes, edges=edges)
41
42
43 def _mock_store(rows=None):
44 store = MagicMock()
45 result = MagicMock()
46 result.result_set = rows or []
47 store.query.return_value = result
48 return store
49
50
51 # ── ContextNode ───────────────────────────────────────────────────────────────
52
53 class TestContextNode:
54 def test_defaults(self):
55 n = ContextNode(type="Function", name="foo")
56 assert n.file_path == ""
57 assert n.line_start is None
58 assert n.docstring is None
59 assert n.domain is None
60
61 def test_knowledge_fields(self):
62 n = ContextNode(type="Concept", name="Payment", description="A payment", domain="billing")
63 assert n.description == "A payment"
64 assert n.domain == "billing"
65
66
67 # ── ContextBundle.to_dict ─────────────────────────────────────────────────────
68
69 class TestContextBundleDict:
70 def test_structure(self):
71 b = _make_bundle()
72 d = b.to_dict()
73 assert d["target"]["name"] == "get_user"
74 assert len(d["nodes"]) == 2
75 assert len(d["edges"]) == 1
76 assert d["metadata"]["query"] == "function_context"
77
78 def test_roundtrip(self):
79 b = _make_bundle()
80 d = b.to_dict()
81 assert d["target"]["type"] == "Function"
82 assert d["nodes"][0]["name"] == "validate_token"
83
84
85 # ── ContextBundle.to_json ─────────────────────────────────────────────────────
86
87 class TestContextBundleJson:
88 def test_valid_json(self):
89 b = _make_bundle()
90 data = json.loads(b.to_json())
91 assert data["target"]["name"] == "get_user"
92
93 def test_indent(self):
94 b = _make_bundle()
95 raw = b.to_json(indent=4)
96 assert " " in raw # 4-space indent
97
98
99 # ── ContextBundle.to_markdown ─────────────────────────────────────────────────
100
101 class TestContextBundleMarkdown:
102 def test_contains_name(self):
103 b = _make_bundle()
104 md = b.to_markdown()
105 assert "get_user" in md
106
107 def test_contains_edge(self):
108 md = _make_bundle().to_markdown()
109 assert "CALLS" in md
110 assert "validate_token" in md
111
112 def test_contains_docstring(self):
113 md = _make_bundle().to_markdown()
114 assert "Return a user by ID." in md
115
116 def test_contains_signature(self):
117 md = _make_bundle().to_markdown()
118 assert "def get_user" in md
119
120 def test_knowledge_bundle(self):
121 md = _make_knowledge_bundle().to_markdown()
122 assert "JWT" in md
123 assert "auth" in md
124 assert "GOVERNS" in md
125
126 def test_empty_nodes(self):
127 target = ContextNode(type="File", name="empty.py", file_path="empty.py")
128 b = ContextBundle(target=target)
129 md = b.to_markdown()
130 assert "empty.py" in md
131
132
133 # ── ContextLoader ─────────────────────────────────────────────────────────────
134
135 class TestContextLoaderFile:
136 def test_load_file_empty(self):
137 store = _mock_store([])
138 loader = ContextLoader(store)
139 bundle = loader.load_file("src/auth.py")
140 assert bundle.target.name == "auth.py"
141 assert bundle.target.type == "File"
142 assert bundle.nodes == []
143
144 def test_load_file_with_rows(self):
145 rows = [["Function", "get_user", 10, "Get a user", "def get_user()"]]
146 store = _mock_store(rows)
147 loader = ContextLoader(store)
148 bundle = loader.load_file("src/auth.py")
149 assert len(bundle.nodes) == 1
150 assert bundle.nodes[0].name == "get_user"
151 assert bundle.nodes[0].type == "Function"
152
153
154 class TestContextLoaderFunction:
155 def test_load_function_no_results(self):
156 store = _mock_store([])
157 loader = ContextLoader(store)
158 bundle = loader.load_function("get_user", file_path="src/auth.py")
159 assert bundle.target.name == "get_user"
160 assert bundle.nodes == []
161 assert bundle.edges == []
162
163 def test_load_function_with_callee(self):
164 def side_effect(query, params):
165 result = MagicMock()
166 if "CALLEES" in query or "callee" in query.lower():
167 result.result_set = [["Function", "validate_token", "src/auth.py", 5]]
168 elif "CALLERS" in query or "caller" in query.lower():
169 result.result_set = []
170 else:
171 result.result_set = []
172 return result
173
174 store = MagicMock()
175 store.query.side_effect = side_effect
176 loader = ContextLoader(store)
177 loader.load_function("get_user")
178 # Should have called query multiple times
179 assert store.query.called
180
181 def test_load_function_default_file_path(self):
182 store = _mock_store([])
183 loader = ContextLoader(store)
184 bundle = loader.load_function("foo")
185 assert bundle.target.file_path == ""
186
187
188 class TestContextLoaderClass:
189 def test_load_class_empty(self):
190 store = _mock_store([])
191 loader = ContextLoader(store)
192 bundle = loader.load_class("AuthService")
193 assert bundle.target.name == "AuthService"
194 assert bundle.target.type == "Class"
195
196 def test_load_class_with_parent(self):
197 call_count = [0]
198
199 def side_effect(query, params=None):
200 result = MagicMock()
201 call_count[0] += 1
202 if call_count[0] == 1:
203 result.result_set = [["BaseService", "src/base.py"]]
204 else:
205 result.result_set = []
206 return result
207
208 store = MagicMock()
209 store.query.side_effect = side_effect
210 loader = ContextLoader(store)
211 bundle = loader.load_class("AuthService")
212 assert len(bundle.nodes) >= 1
213
214
215 class TestContextLoaderExplain:
216 def test_explain_empty(self):
217 store = _mock_store([])
218 loader = ContextLoader(store)
219 bundle = loader.explain("get_user")
220 assert bundle.target.name == "get_user"
221 assert bundle.metadata["query"] == "explain"
222
223
224 class TestContextLoaderSearch:
225 def test_search_empty(self):
226 store = _mock_store([])
227 loader = ContextLoader(store)
228 results = loader.search("auth")
229 assert results == []
230
231 def test_search_returns_nodes(self):
232 rows = [["Function", "authenticate", "src/auth.py", 10, "Authenticate a user"]]
233 store = _mock_store(rows)
234 loader = ContextLoader(store)
235 results = loader.search("auth")
236 assert len(results) == 1
237 assert results[0].name == "authenticate"
238 assert results[0].type == "Function"
239
240 def test_search_all(self):
241 rows = [["Concept", "Authentication", "", None, "The auth concept"]]
242 store = _mock_store(rows)
243 loader = ContextLoader(store)
244 results = loader.search_all("auth")
245 assert len(results) == 1
246
247 def test_search_by_docstring(self):
248 rows = [["Function", "login", "src/auth.py", 5, "Log in a user"]]
249 store = _mock_store(rows)
250 loader = ContextLoader(store)
251 results = loader.search_by_docstring("log in")
252 assert len(results) == 1
253
254 def test_decorated_by(self):
255 rows = [["Function", "protected_view", "src/views.py", 20]]
256 store = _mock_store(rows)
257 loader = ContextLoader(store)
258 results = loader.decorated_by("login_required")
259 assert len(results) == 1
260 assert results[0].name == "protected_view"
261
262
263 class TestContextLoaderConcept:
264 def test_load_concept_not_found(self):
265 store = _mock_store([])
266 loader = ContextLoader(store)
267 bundle = loader.load_concept("Unknown")
268 assert bundle.metadata.get("found") is False
269
270 def test_load_concept_found(self):
271 rows = [["JWT", "Stateless token auth", "active", "auth", [], [], [], []]]
272 store = _mock_store(rows)
273 loader = ContextLoader(store)
274 bundle = loader.load_concept("JWT")
275 assert bundle.target.name == "JWT"
276 assert bundle.target.description == "Stateless token auth"
277
278 def test_load_domain(self):
279 rows = [["Function", "login", "src/auth.py", "Log in"]]
280 store = _mock_store(rows)
281 loader = ContextLoader(store)
282 bundle = loader.load_domain("auth")
283 assert bundle.target.name == "auth"
284 assert bundle.target.type == "Domain"
285
286 DDED tests/test_graph_store.py
287 DDED tests/test_ingestion_code.py
288 DDED tests/test_ingestion_knowledge.py
289 DDED tests/test_ingestion_planopticon.py
290 DDED tests/test_ingestion_wiki.py
291 DDED tests/test_mcp_server.py
292 DDED tests/test_python_parser.py
A tests/test_graph_store.py
+214
--- a/tests/test_graph_store.py
+++ b/tests/test_graph_store.py
@@ -0,0 +1,214 @@
1
+"""Tests for navegador.graph.store.GraphStore."""
2
+
3
+import tempfile
4
+from pathlib import Path
5
+from unittest.mock import MagicMock, patch
6
+
7
+import pytest
8
+
9
+from navegador.graph.store import GraphStore
10
+
11
+
12
+def _mock_client():
13
+ """Create a mock FalkorDB client."""
14
+ client = MagicMock()
15
+ graph = MagicMock()
16
+ graph.query.return_value = MagicMock(result_set=None)
17
+ client.select_graph.return_value = graph
18
+ return client, graph
19
+
20
+
21
+# ── Constructor ───────────────────────────────────────────────────────────────
22
+
23
+class TestGraphStoreInit:
24
+ def test_calls_select_graph(self):
25
+ client, graph = _mock_client()
26
+ GraphStore(client)
27
+ client.select_graph.assert_called_once_with(GraphStore.GRAPH_NAME)
28
+
29
+ def test_stores_graph(self):
30
+ client, graph = _mock_client()
31
+ store = GraphStore(client)
32
+ assert store._graph is graph
33
+
34
+ def test_graph_name_constant(self):
35
+ assert GraphStore.GRAPH_NAME == "navegador"
36
+
37
+
38
+# ── sqlite() classmethod ──────────────────────────────────────────────────────
39
+
40
+class TestSqliteConstructor:
41
+ def test_creates_db_directory(self):
42
+ with tempfile.TemporaryDirectory() as tmpdir:
43
+ db_path = Path(tmpdir) / "sub" / "graph.db"
44
+ mock_client = MagicMock()
45
+ mock_graph = MagicMock()
46
+ mock_graph.query.return_value = MagicMock(result_set=None)
47
+ mock_client.select_graph.return_value = mock_graph
48
+ mock_falkordb = MagicMock(return_value=mock_client)
49
+
50
+ with patch.dict("sys.modules", {"redislite": MagicMock(FalkorDB=mock_falkordb)}):
51
+ store = GraphStore.sqlite(str(db_path))
52
+ assert isinstance(store, GraphStore)
53
+ mock_falkordb.assert_called_once_with(str(db_path))
54
+
55
+ def test_raises_import_error_if_not_installed(self):
56
+ with patch.dict("sys.modules", {"redislite": None}):
57
+ with pytest.raises(ImportError, match="falkordblite"):
58
+ GraphStore.sqlite("/tmp/test.db")
59
+
60
+
61
+# ── redis() classmethod ───────────────────────────────────────────────────────
62
+
63
+class TestRedisConstructor:
64
+ def test_creates_redis_store(self):
65
+ mock_client = MagicMock()
66
+ mock_graph = MagicMock()
67
+ mock_graph.query.return_value = MagicMock(result_set=None)
68
+ mock_client.select_graph.return_value = mock_graph
69
+
70
+ mock_falkordb_module = MagicMock()
71
+ mock_falkordb_module.FalkorDB.from_url.return_value = mock_client
72
+
73
+ with patch.dict("sys.modules", {"falkordb": mock_falkordb_module}):
74
+ store = GraphStore.redis("redis://localhost:6379")
75
+ assert isinstance(store, GraphStore)
76
+ mock_falkordb_module.FalkorDB.from_url.assert_called_once_with("redis://localhost:6379")
77
+
78
+ def test_raises_import_error_if_not_installed(self):
79
+ with patch.dict("sys.modules", {"falkordb": None}):
80
+ with pytest.raises(ImportError, match="falkordb"):
81
+ GraphStore.redis("redis://localhost:6379")
82
+
83
+
84
+# ── query() ───────────────────────────────────────────────────────────────────
85
+
86
+class TestQuery:
87
+ def test_delegates_to_graph(self):
88
+ client, graph = _mock_client()
89
+ graph.query.return_value = MagicMock(result_set=[["a", "b"]])
90
+ store = GraphStore(client)
91
+ result = store.query("MATCH (n) RETURN n", {"x": 1})
92
+ graph.query.assert_called_once_with("MATCH (n) RETURN n", {"x": 1})
93
+ assert result.result_set == [["a", "b"]]
94
+
95
+ def test_passes_empty_dict_when_no_params(self):
96
+ client, graph = _mock_client()
97
+ store = GraphStore(client)
98
+ store.query("MATCH (n) RETURN n")
99
+ graph.query.assert_called_once_with("MATCH (n) RETURN n", {})
100
+
101
+
102
+# ── create_node() ─────────────────────────────────────────────────────────────
103
+
104
+class TestCreateNode:
105
+ def test_generates_merge_cypher(self):
106
+ client, graph = _mock_client()
107
+ store = GraphStore(client)
108
+ store.create_node("Function", {"name": "foo", "file_path": "a.py", "docstring": "doc"})
109
+ call_args = graph.query.call_args
110
+ cypher = call_args[0][0]
111
+ assert "MERGE" in cypher
112
+ assert "Function" in cypher
113
+ assert "name" in cypher
114
+ assert "file_path" in cypher
115
+
116
+ def test_passes_props_as_params(self):
117
+ client, graph = _mock_client()
118
+ store = GraphStore(client)
119
+ props = {"name": "bar", "file_path": "b.py"}
120
+ store.create_node("Class", props)
121
+ call_params = graph.query.call_args[0][1]
122
+ assert call_params["name"] == "bar"
123
+ assert call_params["file_path"] == "b.py"
124
+
125
+
126
+# ── create_edge() ─────────────────────────────────────────────────────────────
127
+
128
+class TestCreateEdge:
129
+ def test_generates_match_merge_cypher(self):
130
+ client, graph = _mock_client()
131
+ store = GraphStore(client)
132
+ store.create_edge(
133
+ "Function", {"name": "foo"},
134
+ "CALLS",
135
+ "Function", {"name": "bar"},
136
+ )
137
+ call_args = graph.query.call_args
138
+ cypher = call_args[0][0]
139
+ assert "MATCH" in cypher
140
+ assert "MERGE" in cypher
141
+ assert "CALLS" in cypher
142
+
143
+ def test_passes_from_and_to_params(self):
144
+ client, graph = _mock_client()
145
+ store = GraphStore(client)
146
+ store.create_edge(
147
+ "Function", {"name": "foo"},
148
+ "CALLS",
149
+ "Function", {"name": "bar"},
150
+ )
151
+ params = graph.query.call_args[0][1]
152
+ assert params["from_name"] == "foo"
153
+ assert params["to_name"] == "bar"
154
+
155
+ def test_includes_edge_props_when_provided(self):
156
+ client, graph = _mock_client()
157
+ store = GraphStore(client)
158
+ store.create_edge(
159
+ "Class", {"name": "A"},
160
+ "INHERITS",
161
+ "Class", {"name": "B"},
162
+ props={"weight": 1},
163
+ )
164
+ call_args = graph.query.call_args
165
+ cypher = call_args[0][0]
166
+ params = call_args[0][1]
167
+ assert "SET" in cypher
168
+ assert params["p_weight"] == 1
169
+
170
+ def test_no_set_clause_without_props(self):
171
+ client, graph = _mock_client()
172
+ store = GraphStore(client)
173
+ store.create_edge("A", {"name": "x"}, "REL", "B", {"name": "y"})
174
+ cypher = graph.query.call_args[0][0]
175
+ assert "SET" not in cypher
176
+
177
+
178
+# ── clear() ───────────────────────────────────────────────────────────────────
179
+
180
+class TestClear:
181
+ def test_executes_delete_query(self):
182
+ client, graph = _mock_client()
183
+ store = GraphStore(client)
184
+ store.clear()
185
+ cypher = graph.query.call_args[0][0]
186
+ assert "DETACH DELETE" in cypher
187
+
188
+
189
+# ── node_count / edge_count ───────────────────────────────────────────────────
190
+
191
+class TestCounts:
192
+ def test_node_count_returns_value(self):
193
+ client, graph = _mock_client()
194
+ graph.query.return_value = MagicMock(result_set=[[42]])
195
+ store = GraphStore(client)
196
+ assert store.node_count() == 42
197
+
198
+ def test_node_count_returns_zero_on_empty(self):
199
+ client, graph = _mock_client()
200
+ graph.query.return_value = MagicMock(result_set=[])
201
+ store = GraphStore(client)
202
+ assert store.node_count() == 0
203
+
204
+ def test_edge_count_returns_value(self):
205
+ client, graph = _mock_client()
206
+ graph.query.return_value = MagicMock(result_set=[[7]])
207
+ store = GraphStore(client)
208
+ assert store.edge_count() == 7
209
+
210
+ def test_edge_count_returns_zero_on_empty(self):
211
+ client, graph = _mock_client()
212
+ graph.query.return_value = MagicMock(result_set=[])
213
+ store = GraphStore(client)
214
+ assert store.edge_count() == 0
--- a/tests/test_graph_store.py
+++ b/tests/test_graph_store.py
@@ -0,0 +1,214 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_graph_store.py
+++ b/tests/test_graph_store.py
@@ -0,0 +1,214 @@
1 """Tests for navegador.graph.store.GraphStore."""
2
3 import tempfile
4 from pathlib import Path
5 from unittest.mock import MagicMock, patch
6
7 import pytest
8
9 from navegador.graph.store import GraphStore
10
11
12 def _mock_client():
13 """Create a mock FalkorDB client."""
14 client = MagicMock()
15 graph = MagicMock()
16 graph.query.return_value = MagicMock(result_set=None)
17 client.select_graph.return_value = graph
18 return client, graph
19
20
21 # ── Constructor ───────────────────────────────────────────────────────────────
22
23 class TestGraphStoreInit:
24 def test_calls_select_graph(self):
25 client, graph = _mock_client()
26 GraphStore(client)
27 client.select_graph.assert_called_once_with(GraphStore.GRAPH_NAME)
28
29 def test_stores_graph(self):
30 client, graph = _mock_client()
31 store = GraphStore(client)
32 assert store._graph is graph
33
34 def test_graph_name_constant(self):
35 assert GraphStore.GRAPH_NAME == "navegador"
36
37
38 # ── sqlite() classmethod ──────────────────────────────────────────────────────
39
40 class TestSqliteConstructor:
41 def test_creates_db_directory(self):
42 with tempfile.TemporaryDirectory() as tmpdir:
43 db_path = Path(tmpdir) / "sub" / "graph.db"
44 mock_client = MagicMock()
45 mock_graph = MagicMock()
46 mock_graph.query.return_value = MagicMock(result_set=None)
47 mock_client.select_graph.return_value = mock_graph
48 mock_falkordb = MagicMock(return_value=mock_client)
49
50 with patch.dict("sys.modules", {"redislite": MagicMock(FalkorDB=mock_falkordb)}):
51 store = GraphStore.sqlite(str(db_path))
52 assert isinstance(store, GraphStore)
53 mock_falkordb.assert_called_once_with(str(db_path))
54
55 def test_raises_import_error_if_not_installed(self):
56 with patch.dict("sys.modules", {"redislite": None}):
57 with pytest.raises(ImportError, match="falkordblite"):
58 GraphStore.sqlite("/tmp/test.db")
59
60
61 # ── redis() classmethod ───────────────────────────────────────────────────────
62
63 class TestRedisConstructor:
64 def test_creates_redis_store(self):
65 mock_client = MagicMock()
66 mock_graph = MagicMock()
67 mock_graph.query.return_value = MagicMock(result_set=None)
68 mock_client.select_graph.return_value = mock_graph
69
70 mock_falkordb_module = MagicMock()
71 mock_falkordb_module.FalkorDB.from_url.return_value = mock_client
72
73 with patch.dict("sys.modules", {"falkordb": mock_falkordb_module}):
74 store = GraphStore.redis("redis://localhost:6379")
75 assert isinstance(store, GraphStore)
76 mock_falkordb_module.FalkorDB.from_url.assert_called_once_with("redis://localhost:6379")
77
78 def test_raises_import_error_if_not_installed(self):
79 with patch.dict("sys.modules", {"falkordb": None}):
80 with pytest.raises(ImportError, match="falkordb"):
81 GraphStore.redis("redis://localhost:6379")
82
83
84 # ── query() ───────────────────────────────────────────────────────────────────
85
86 class TestQuery:
87 def test_delegates_to_graph(self):
88 client, graph = _mock_client()
89 graph.query.return_value = MagicMock(result_set=[["a", "b"]])
90 store = GraphStore(client)
91 result = store.query("MATCH (n) RETURN n", {"x": 1})
92 graph.query.assert_called_once_with("MATCH (n) RETURN n", {"x": 1})
93 assert result.result_set == [["a", "b"]]
94
95 def test_passes_empty_dict_when_no_params(self):
96 client, graph = _mock_client()
97 store = GraphStore(client)
98 store.query("MATCH (n) RETURN n")
99 graph.query.assert_called_once_with("MATCH (n) RETURN n", {})
100
101
102 # ── create_node() ─────────────────────────────────────────────────────────────
103
104 class TestCreateNode:
105 def test_generates_merge_cypher(self):
106 client, graph = _mock_client()
107 store = GraphStore(client)
108 store.create_node("Function", {"name": "foo", "file_path": "a.py", "docstring": "doc"})
109 call_args = graph.query.call_args
110 cypher = call_args[0][0]
111 assert "MERGE" in cypher
112 assert "Function" in cypher
113 assert "name" in cypher
114 assert "file_path" in cypher
115
116 def test_passes_props_as_params(self):
117 client, graph = _mock_client()
118 store = GraphStore(client)
119 props = {"name": "bar", "file_path": "b.py"}
120 store.create_node("Class", props)
121 call_params = graph.query.call_args[0][1]
122 assert call_params["name"] == "bar"
123 assert call_params["file_path"] == "b.py"
124
125
126 # ── create_edge() ─────────────────────────────────────────────────────────────
127
128 class TestCreateEdge:
129 def test_generates_match_merge_cypher(self):
130 client, graph = _mock_client()
131 store = GraphStore(client)
132 store.create_edge(
133 "Function", {"name": "foo"},
134 "CALLS",
135 "Function", {"name": "bar"},
136 )
137 call_args = graph.query.call_args
138 cypher = call_args[0][0]
139 assert "MATCH" in cypher
140 assert "MERGE" in cypher
141 assert "CALLS" in cypher
142
143 def test_passes_from_and_to_params(self):
144 client, graph = _mock_client()
145 store = GraphStore(client)
146 store.create_edge(
147 "Function", {"name": "foo"},
148 "CALLS",
149 "Function", {"name": "bar"},
150 )
151 params = graph.query.call_args[0][1]
152 assert params["from_name"] == "foo"
153 assert params["to_name"] == "bar"
154
155 def test_includes_edge_props_when_provided(self):
156 client, graph = _mock_client()
157 store = GraphStore(client)
158 store.create_edge(
159 "Class", {"name": "A"},
160 "INHERITS",
161 "Class", {"name": "B"},
162 props={"weight": 1},
163 )
164 call_args = graph.query.call_args
165 cypher = call_args[0][0]
166 params = call_args[0][1]
167 assert "SET" in cypher
168 assert params["p_weight"] == 1
169
170 def test_no_set_clause_without_props(self):
171 client, graph = _mock_client()
172 store = GraphStore(client)
173 store.create_edge("A", {"name": "x"}, "REL", "B", {"name": "y"})
174 cypher = graph.query.call_args[0][0]
175 assert "SET" not in cypher
176
177
178 # ── clear() ───────────────────────────────────────────────────────────────────
179
180 class TestClear:
181 def test_executes_delete_query(self):
182 client, graph = _mock_client()
183 store = GraphStore(client)
184 store.clear()
185 cypher = graph.query.call_args[0][0]
186 assert "DETACH DELETE" in cypher
187
188
189 # ── node_count / edge_count ───────────────────────────────────────────────────
190
191 class TestCounts:
192 def test_node_count_returns_value(self):
193 client, graph = _mock_client()
194 graph.query.return_value = MagicMock(result_set=[[42]])
195 store = GraphStore(client)
196 assert store.node_count() == 42
197
198 def test_node_count_returns_zero_on_empty(self):
199 client, graph = _mock_client()
200 graph.query.return_value = MagicMock(result_set=[])
201 store = GraphStore(client)
202 assert store.node_count() == 0
203
204 def test_edge_count_returns_value(self):
205 client, graph = _mock_client()
206 graph.query.return_value = MagicMock(result_set=[[7]])
207 store = GraphStore(client)
208 assert store.edge_count() == 7
209
210 def test_edge_count_returns_zero_on_empty(self):
211 client, graph = _mock_client()
212 graph.query.return_value = MagicMock(result_set=[])
213 store = GraphStore(client)
214 assert store.edge_count() == 0
A tests/test_ingestion_code.py
+3
--- a/tests/test_ingestion_code.py
+++ b/tests/test_ingestion_code.py
@@ -0,0 +1,3 @@
1
+"""T(parser.PythonParser", mock_p # Just verify cachiypython")
2
+ typescripttstypescript"] = mock_tstypescriptts_parser
3
+go
--- a/tests/test_ingestion_code.py
+++ b/tests/test_ingestion_code.py
@@ -0,0 +1,3 @@
 
 
 
--- a/tests/test_ingestion_code.py
+++ b/tests/test_ingestion_code.py
@@ -0,0 +1,3 @@
1 """T(parser.PythonParser", mock_p # Just verify cachiypython")
2 typescripttstypescript"] = mock_tstypescriptts_parser
3 go
A tests/test_ingestion_knowledge.py
+176
--- a/tests/test_ingestion_knowledge.py
+++ b/tests/test_ingestion_knowledge.py
@@ -0,0 +1,176 @@
1
+"""Tests for KnowledgeIngester — manual knowledge curation."""
2
+
3
+from unittest.mock import MagicMock
4
+
5
+import pytest
6
+
7
+from navegador.graph.schema import EdgeType, NodeLabel
8
+from navegador.ingestion.knowledge import KnowledgeIngester
9
+
10
+
11
+def _mock_store():
12
+ return MagicMock()
13
+
14
+
15
+class TestKnowledgeIngesterDomains:
16
+ def test_add_domain(self):
17
+ store = _mock_store()
18
+ k = KnowledgeIngester(store)
19
+ k.add_domain("auth", description="Authentication layer")
20
+ store.create_node.assert_called_once_with(
21
+ NodeLabel.Domain, {"name": "auth", "description": "Authentication layer"}
22
+ )
23
+
24
+ def test_add_domain_no_description(self):
25
+ store = _mock_store()
26
+ k = KnowledgeIngester(store)
27
+ k.add_domain("billing")
28
+ store.create_node.assert_called_once()
29
+
30
+
31
+class TestKnowledgeIngesterConcepts:
32
+ def test_add_concept_minimal(self):
33
+ store = _mock_store()
34
+ k = KnowledgeIngester(store)
35
+ k.add_concept("JWT")
36
+ store.create_node.assert_called_once_with(
37
+ NodeLabel.Concept,
38
+ {"name": "JWT", "description": "", "domain": "", "status": "",
39
+ "rules": "", "examples": "", "wiki_refs": ""}
40
+ )
41
+
42
+ def test_add_concept_with_domain_creates_edge(self):
43
+ store = _mock_store()
44
+ k = KnowledgeIngester(store)
45
+ k.add_concept("JWT", domain="auth")
46
+ # Should create Domain node + BELONGS_TO edge
47
+ assert store.create_node.call_count == 2
48
+ store.create_edge.assert_called_once()
49
+ edge_call = store.create_edge.call_args
50
+ assert edge_call[0][2] == EdgeType.BELONGS_TO
51
+
52
+ def test_relate_concepts(self):
53
+ store = _mock_store()
54
+ k = KnowledgeIngester(store)
55
+ k.relate_concepts("JWT", "OAuth")
56
+ store.create_edge.assert_called_once_with(
57
+ NodeLabel.Concept, {"name": "JWT"},
58
+ EdgeType.RELATED_TO,
59
+ NodeLabel.Concept, {"name": "OAuth"},
60
+ )
61
+
62
+
63
+class TestKnowledgeIngesterRules:
64
+ def test_add_rule(self):
65
+ store = _mock_store()
66
+ k = KnowledgeIngester(store)
67
+ k.add_rule("Tokens must expire", severity="critical")
68
+ call_args = store.create_node.call_args[0]
69
+ assert call_args[0] == NodeLabel.Rule
70
+ assert call_args[1]["severity"] == "critical"
71
+
72
+ def test_rule_governs(self):
73
+ store = _mock_store()
74
+ k = KnowledgeIngester(store)
75
+ k.rule_governs("Tokens must expire", "JWT", NodeLabel.Concept)
76
+ store.create_edge.assert_called_once_with(
77
+ NodeLabel.Rule, {"name": "Tokens must expire"},
78
+ EdgeType.GOVERNS,
79
+ NodeLabel.Concept, {"name": "JWT"},
80
+ )
81
+
82
+
83
+class TestKnowledgeIngesterDecisions:
84
+ def test_add_decision(self):
85
+ store = _mock_store()
86
+ k = KnowledgeIngester(store)
87
+ k.add_decision("Use JWT", status="accepted", rationale="Horizontal scaling")
88
+ call_args = store.create_node.call_args[0]
89
+ assert call_args[0] == NodeLabel.Decision
90
+ assert call_args[1]["status"] == "accepted"
91
+ assert call_args[1]["rationale"] == "Horizontal scaling"
92
+
93
+
94
+class TestKnowledgeIngesterPeople:
95
+ def test_add_person(self):
96
+ store = _mock_store()
97
+ k = KnowledgeIngester(store)
98
+ k.add_person("Alice", email="[email protected]", role="lead", team="auth")
99
+ store.create_node.assert_called_once_with(
100
+ NodeLabel.Person,
101
+ {"name": "Alice", "email": "[email protected]", "role": "lead", "team": "auth"}
102
+ )
103
+
104
+ def test_assign(self):
105
+ store = _mock_store()
106
+ k = KnowledgeIngester(store)
107
+ k.assign("validate_token", NodeLabel.Function, "Alice")
108
+ store.create_edge.assert_called_once_with(
109
+ NodeLabel.Function, {"name": "validate_token"},
110
+ EdgeType.ASSIGNED_TO,
111
+ NodeLabel.Person, {"name": "Alice"},
112
+ )
113
+
114
+
115
+class TestKnowledgeIngesterWiki:
116
+ def test_wiki_page(self):
117
+ store = _mock_store()
118
+ k = KnowledgeIngester(store)
119
+ k.wiki_page("Auth Guide", url="https://example.com", source="github", content="# Auth")
120
+ call_args = store.create_node.call_args[0]
121
+ assert call_args[0] == NodeLabel.WikiPage
122
+ assert call_args[1]["name"] == "Auth Guide"
123
+
124
+ def test_wiki_documents(self):
125
+ store = _mock_store()
126
+ k = KnowledgeIngester(store)
127
+ k.wiki_documents("Auth Guide", "JWT", {"name": "JWT"}, NodeLabel.Concept)
128
+ store.create_edge.assert_called_once_with(
129
+ NodeLabel.WikiPage, {"name": "Auth Guide"},
130
+ EdgeType.DOCUMENTS,
131
+ NodeLabel.Concept, {"name": "JWT"},
132
+ )
133
+
134
+
135
+class TestKnowledgeIngesterAnnotate:
136
+ def test_annotate_with_concept(self):
137
+ store = _mock_store()
138
+ k = KnowledgeIngester(store)
139
+ k.annotate_code("validate_token", "Function", concept="JWT")
140
+ store.create_edge.assert_called_once_with(
141
+ NodeLabel.Concept, {"name": "JWT"},
142
+ EdgeType.ANNOTATES,
143
+ NodeLabel.Function, {"name": "validate_token"},
144
+ )
145
+
146
+ def test_annotate_with_rule(self):
147
+ store = _mock_store()
148
+ k = KnowledgeIngester(store)
149
+ k.annotate_code("validate_token", "Function", rule="Tokens must expire")
150
+ store.create_edge.assert_called_once_with(
151
+ NodeLabel.Rule, {"name": "Tokens must expire"},
152
+ EdgeType.ANNOTATES,
153
+ NodeLabel.Function, {"name": "validate_token"},
154
+ )
155
+
156
+ def test_annotate_with_both(self):
157
+ store = _mock_store()
158
+ k = KnowledgeIngester(store)
159
+ k.annotate_code("validate_token", "Function", concept="JWT", rule="Tokens must expire")
160
+ assert store.create_edge.call_count == 2
161
+
162
+ def test_code_implements(self):
163
+ store = _mock_store()
164
+ k = KnowledgeIngester(store)
165
+ k.code_implements("validate_token", "Function", "JWT")
166
+ store.create_edge.assert_called_once_with(
167
+ NodeLabel.Function, {"name": "validate_token"},
168
+ EdgeType.IMPLEMENTS,
169
+ NodeLabel.Concept, {"name": "JWT"},
170
+ )
171
+
172
+ def test_annotate_invalid_label_raises(self):
173
+ store = _mock_store()
174
+ k = KnowledgeIngester(store)
175
+ with pytest.raises(ValueError):
176
+ k.annotate_code("foo", "Inv
--- a/tests/test_ingestion_knowledge.py
+++ b/tests/test_ingestion_knowledge.py
@@ -0,0 +1,176 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_ingestion_knowledge.py
+++ b/tests/test_ingestion_knowledge.py
@@ -0,0 +1,176 @@
1 """Tests for KnowledgeIngester — manual knowledge curation."""
2
3 from unittest.mock import MagicMock
4
5 import pytest
6
7 from navegador.graph.schema import EdgeType, NodeLabel
8 from navegador.ingestion.knowledge import KnowledgeIngester
9
10
11 def _mock_store():
12 return MagicMock()
13
14
15 class TestKnowledgeIngesterDomains:
16 def test_add_domain(self):
17 store = _mock_store()
18 k = KnowledgeIngester(store)
19 k.add_domain("auth", description="Authentication layer")
20 store.create_node.assert_called_once_with(
21 NodeLabel.Domain, {"name": "auth", "description": "Authentication layer"}
22 )
23
24 def test_add_domain_no_description(self):
25 store = _mock_store()
26 k = KnowledgeIngester(store)
27 k.add_domain("billing")
28 store.create_node.assert_called_once()
29
30
31 class TestKnowledgeIngesterConcepts:
32 def test_add_concept_minimal(self):
33 store = _mock_store()
34 k = KnowledgeIngester(store)
35 k.add_concept("JWT")
36 store.create_node.assert_called_once_with(
37 NodeLabel.Concept,
38 {"name": "JWT", "description": "", "domain": "", "status": "",
39 "rules": "", "examples": "", "wiki_refs": ""}
40 )
41
42 def test_add_concept_with_domain_creates_edge(self):
43 store = _mock_store()
44 k = KnowledgeIngester(store)
45 k.add_concept("JWT", domain="auth")
46 # Should create Domain node + BELONGS_TO edge
47 assert store.create_node.call_count == 2
48 store.create_edge.assert_called_once()
49 edge_call = store.create_edge.call_args
50 assert edge_call[0][2] == EdgeType.BELONGS_TO
51
52 def test_relate_concepts(self):
53 store = _mock_store()
54 k = KnowledgeIngester(store)
55 k.relate_concepts("JWT", "OAuth")
56 store.create_edge.assert_called_once_with(
57 NodeLabel.Concept, {"name": "JWT"},
58 EdgeType.RELATED_TO,
59 NodeLabel.Concept, {"name": "OAuth"},
60 )
61
62
63 class TestKnowledgeIngesterRules:
64 def test_add_rule(self):
65 store = _mock_store()
66 k = KnowledgeIngester(store)
67 k.add_rule("Tokens must expire", severity="critical")
68 call_args = store.create_node.call_args[0]
69 assert call_args[0] == NodeLabel.Rule
70 assert call_args[1]["severity"] == "critical"
71
72 def test_rule_governs(self):
73 store = _mock_store()
74 k = KnowledgeIngester(store)
75 k.rule_governs("Tokens must expire", "JWT", NodeLabel.Concept)
76 store.create_edge.assert_called_once_with(
77 NodeLabel.Rule, {"name": "Tokens must expire"},
78 EdgeType.GOVERNS,
79 NodeLabel.Concept, {"name": "JWT"},
80 )
81
82
83 class TestKnowledgeIngesterDecisions:
84 def test_add_decision(self):
85 store = _mock_store()
86 k = KnowledgeIngester(store)
87 k.add_decision("Use JWT", status="accepted", rationale="Horizontal scaling")
88 call_args = store.create_node.call_args[0]
89 assert call_args[0] == NodeLabel.Decision
90 assert call_args[1]["status"] == "accepted"
91 assert call_args[1]["rationale"] == "Horizontal scaling"
92
93
94 class TestKnowledgeIngesterPeople:
95 def test_add_person(self):
96 store = _mock_store()
97 k = KnowledgeIngester(store)
98 k.add_person("Alice", email="[email protected]", role="lead", team="auth")
99 store.create_node.assert_called_once_with(
100 NodeLabel.Person,
101 {"name": "Alice", "email": "[email protected]", "role": "lead", "team": "auth"}
102 )
103
104 def test_assign(self):
105 store = _mock_store()
106 k = KnowledgeIngester(store)
107 k.assign("validate_token", NodeLabel.Function, "Alice")
108 store.create_edge.assert_called_once_with(
109 NodeLabel.Function, {"name": "validate_token"},
110 EdgeType.ASSIGNED_TO,
111 NodeLabel.Person, {"name": "Alice"},
112 )
113
114
115 class TestKnowledgeIngesterWiki:
116 def test_wiki_page(self):
117 store = _mock_store()
118 k = KnowledgeIngester(store)
119 k.wiki_page("Auth Guide", url="https://example.com", source="github", content="# Auth")
120 call_args = store.create_node.call_args[0]
121 assert call_args[0] == NodeLabel.WikiPage
122 assert call_args[1]["name"] == "Auth Guide"
123
124 def test_wiki_documents(self):
125 store = _mock_store()
126 k = KnowledgeIngester(store)
127 k.wiki_documents("Auth Guide", "JWT", {"name": "JWT"}, NodeLabel.Concept)
128 store.create_edge.assert_called_once_with(
129 NodeLabel.WikiPage, {"name": "Auth Guide"},
130 EdgeType.DOCUMENTS,
131 NodeLabel.Concept, {"name": "JWT"},
132 )
133
134
135 class TestKnowledgeIngesterAnnotate:
136 def test_annotate_with_concept(self):
137 store = _mock_store()
138 k = KnowledgeIngester(store)
139 k.annotate_code("validate_token", "Function", concept="JWT")
140 store.create_edge.assert_called_once_with(
141 NodeLabel.Concept, {"name": "JWT"},
142 EdgeType.ANNOTATES,
143 NodeLabel.Function, {"name": "validate_token"},
144 )
145
146 def test_annotate_with_rule(self):
147 store = _mock_store()
148 k = KnowledgeIngester(store)
149 k.annotate_code("validate_token", "Function", rule="Tokens must expire")
150 store.create_edge.assert_called_once_with(
151 NodeLabel.Rule, {"name": "Tokens must expire"},
152 EdgeType.ANNOTATES,
153 NodeLabel.Function, {"name": "validate_token"},
154 )
155
156 def test_annotate_with_both(self):
157 store = _mock_store()
158 k = KnowledgeIngester(store)
159 k.annotate_code("validate_token", "Function", concept="JWT", rule="Tokens must expire")
160 assert store.create_edge.call_count == 2
161
162 def test_code_implements(self):
163 store = _mock_store()
164 k = KnowledgeIngester(store)
165 k.code_implements("validate_token", "Function", "JWT")
166 store.create_edge.assert_called_once_with(
167 NodeLabel.Function, {"name": "validate_token"},
168 EdgeType.IMPLEMENTS,
169 NodeLabel.Concept, {"name": "JWT"},
170 )
171
172 def test_annotate_invalid_label_raises(self):
173 store = _mock_store()
174 k = KnowledgeIngester(store)
175 with pytest.raises(ValueError):
176 k.annotate_code("foo", "Inv
A tests/test_ingestion_planopticon.py
+433
--- a/tests/test_ingestion_planopticon.py
+++ b/tests/test_ingestion_planopticon.py
@@ -0,0 +1,433 @@
1
+ap_coverage(self):
2
+ from navNodeLabel
3
+from navegador.ingestion.planopticon import (
4
+ EDGE_MAP,
5
+ NODE_TYPE_MAP,
6
+ PLANNING_TYPE_MAP,
7
+ PlanopticonIngester,
8
+)
9
+
10
+
11
+def _make_store():
12
+ store = MagicMock()
13
+ store.query.return_value = MagicMock(result_set=[])
14
+ return store
15
+
16
+
17
+# ── Fixtures ──────────────────────────────────────────────────────────────────
18
+
19
+KG_DATA = {
20
+ "nodes": [
21
+ {"id": "n1", "type": "concept", "name": "Payment Gateway",
22
+ "description": "Handles payments"},
23
+ {"id": "n2", "type": "person", "name": "Carol", "email": "[email protected]"},
24
+ {"id": "n3", "type": "technology", "name": "PostgreSQL", "description": "DB"},
25
+ {"id": "n4", "type": "decision", "name": "Use Redis"},
26
+ {"id": "n5", "type": "unknown_type", "name": "Misc"},
27
+ {"id": "n6", "type": "diagram", "name": "Service Map", "source": "http://img.png"},
28
+ ],
29
+ "relationships": [
30
+ {"source": "Payment Gateway", "target": "PostgreSQL", "type": "uses"},
31
+ {"source": "Carol", "target": "Payment Gateway", "type": "assigned_to"},
32
+ {"source": "", "target": "nope", "type": "related_to"}, # bad rel — no source
33
+ ],
34
+ "sources": [
35
+ {"title": "Meeting 2024", "url": "https://ex.com", "source_type": "meeting"},
36
+ ],
37
+}
38
+
39
+INTERCHANGE_DATA = {
40
+ "project": {"name": "MyProject", "tags": ["backend", "payments"]},
41
+ "entities": [
42
+ {
43
+ "planning_type": "decision",
44
+ "name": "Adopt microservices",
45
+ "description": "Split the monolith",
46
+ "status": "accepted",
47
+ "rationale": "Scale independently",
48
+ },
49
+ {
50
+ "planning_type": "requirement",
51
+ "name": "PCI compliance",
52
+ "description": "Must comply with PCI-DSS",
53
+ "priority": "high",
54
+ },
55
+ {
56
+ "planning_type": "goal",
57
+ "name": "Increase uptime",
58
+ "description": "99.9% SLA",
59
+ },
60
+ {
61
+ # no planning_type → falls through to _ingest_kg_node
62
+ "type": "concept",
63
+ "name": "Event Sourcing",
64
+ },
65
+ ],
66
+ "relationships": [],
67
+ "artifacts": [
68
+ {"name": "Architecture Diagram", "content": "mermaid content here"},
69
+ ],
70
+ "sources": [],
71
+}
72
+
73
+MANIFEST_DATA = {
74
+ "video": {"title": "Sprint Planning", "url": "https://example.com/video/1"},
75
+ "key_points": [
76
+ {"point": "Use async everywhere", "topic": "Architecture", "details": "For scale"},
77
+ ],
78
+ "action_items": [
79
+ {"action": "Refactor auth service", "assignee": "Bob", "context": "High priority"},
80
+ ],
81
+ "diagrams": [
82
+ {
83
+ "diagram_type": "sequence",
84
+ "timestamp": 120,
85
+ "description": "Auth flow",
86
+ "mermaid": "sequenceDiagram ...",
87
+ "elements": ["User", "Auth"],
88
+ }
89
+ ],
90
+}
91
+
92
+
93
+# ── Maps ──────────────────────────────────────────────────────────────────────
94
+
95
+class TestMaps:
96
+ def test_node_type_map_coverage(self):
97
+ assert NODE_TYPE_MAP["concept"] == NodeLabel.Concept
98
+ assert NODE_TYPE_MAP["technology"] == NodeLabel.Concept
99
+ assert NODE_TYPE_MAP["organization"] == NodeLabel.Concept
100
+ assert NODE_TYPE_MAP["person"] == NodeLabel.Person
101
+ assert NODE_TYPE_MAP["diagram"] == NodeLabel.WikiPage
102
+
103
+ def test_planning_type_map_coverage(self):
104
+ assert PLANNING_TYPE_MAP["decision"] == NodeLabel.Decision
105
+ assert PLANNING_TYPE_MAP["requirement"] == NodeLabel.Rule
106
+ assert PLANNING_TYPE_MAP["constraint"] == NodeLabel.Rule
107
+ assert PLANNING_TYPE_MAP["risk"] == NodeLabel.Rule
108
+ assert PLANNING_TYPE_MAP["goal"] == NodeLabel.Concept
109
+
110
+ def test_edge_map_coverage(self):
111
+ from navegador.graph.schema import EdgeType
112
+ assert EDGE_MAP["uses"] == EdgeType.DEPENDS_ON
113
+ assert EDGE_MAP["related_to"] == EdgeType.RELATED_TO
114
+ assert EDGE_MAP["assigned_to"] == EdgeType.ASSIGNED_TO
115
+ assert EDGE_MAP["governs"] == EdgeType.GOVERNS
116
+ assert EDGE_MAP["implements"] == EdgeType.IMPLEMENTS
117
+
118
+
119
+# ── ingest_kg ─────────────────────────────────────────────────────────────────
120
+
121
+class TestIngestKg:
122
+ def test_ingests_concept_nodes(self):
123
+ store = _make_store()
124
+ ingester = PlanopticonIngester(store)
125
+ with tempfile.TemporaryDirectory() as tmpdir:
126
+ p = Path(tmpdir) / "kg.json"
127
+ p.write_text(json.dumps(KG_DATA))
128
+ stats = ingester.ingest_kg(p)
129
+ assert stats["nodes"] >= 1
130
+
131
+ def test_ingests_person_nodes(self):
132
+ store = _make_store()
133
+ ingester = PlanopticonIngester(store)
134
+ with tempfile.TemporaryDirectory() as tmpdir:
135
+ p = Path(tmpdir) / "kg.json"
136
+ p.write_text(json.dumps(KG_DATA))
137
+ ingester.ingest_kg(p)
138
+ labels = [c[0][0] for c in store.create_node.call_args_list]
139
+ assert NodeLabel.Person in labels
140
+
141
+ def test_ingests_technology_as_concept(self):
142
+ store = _make_store()
143
+ ingester = PlanopticonIngester(store)
144
+ data = {"nodes": [{"type": "technology", "name": "PostgreSQL"}],
145
+ "relationships": [], "sources": []}
146
+ with tempfile.TemporaryDirectory() as tmpdir:
147
+ p = Path(tmpdir) / "kg.json"
148
+ p.write_text(json.dumps(data))
149
+ ingester.ingest_kg(p)
150
+ labels = [c[0][0] for c in store.create_node.call_args_list]
151
+ assert NodeLabel.Concept in labels
152
+
153
+ def test_ingests_diagram_as_wiki_page(self):
154
+ store = _make_store()
155
+ ingester = PlanopticonIngester(store)
156
+ data = {"nodes": [{"type": "diagram", "name": "Arch Diagram", "source": "http://x.com"}],
157
+ "relationships": [], "sources": []}
158
+ with tempfile.TemporaryDirectory() as tmpdir:
159
+ p = Path(tmpdir) / "kg.json"
160
+ p.write_text(json.dumps(data))
161
+ ingester.ingest_kg(p)
162
+ labels = [c[0][0] for c in store.create_node.call_args_list]
163
+ assert NodeLabel.WikiPage in labels
164
+
165
+ def test_skips_nodes_without_name(self):
166
+ store = _make_store()
167
+ ingester = PlanopticonIngester(store)
168
+ data = {"nodes": [{"type": "concept", "name": ""}], "relationships": [], "sources": []}
169
+ with tempfile.TemporaryDirectory() as tmpdir:
170
+ p = Path(tmpdir) / "kg.json"
171
+ p.write_text(json.dumps(data))
172
+ stats = ingester.ingest_kg(p)
173
+ assert stats["nodes"] == 0
174
+
175
+ def test_ingests_sources_as_wiki_pages(self):
176
+ store = _make_store()
177
+ ingester = PlanopticonIngester(store)
178
+ data = {
179
+ "nodes": [], "relationships": [],
180
+ "sources": [
181
+ {"title": "Meeting 2024", "url": "https://ex.com", "source_type": "meeting"},
182
+ ],
183
+ }
184
+ with tempfile.TemporaryDirectory() as tmpdir:
185
+ p = Path(tmpdir) / "kg.json"
186
+ p.write_text(json.dumps(data))
187
+ stats = ingester.ingest_kg(p)
188
+ assert stats["nodes"] >= 1
189
+ labels = [c[0][0] for c in store.create_node.call_args_list]
190
+ assert NodeLabel.WikiPage in labels
191
+
192
+ def test_ingests_relationships(self):
193
+ store = _make_store()
194
+ ingester = PlanopticonIngester(store)
195
+ with tempfile.TemporaryDirectory() as tmpdir:
196
+ p = Path(tmpdir) / "kg.json"
197
+ p.write_text(json.dumps(KG_DATA))
198
+ stats = ingester.ingest_kg(p)
199
+ assert stats["edges"] >= 1
200
+ store.query.assert_called()
201
+
202
+ def test_skips_bad_relationships(self):
203
+ store = _make_store()
204
+ ingester = PlanopticonIngester(store)
205
+ data = {"nodes": [], "relationships": [{"source": "", "target": "x", "type": "related_to"}],
206
+ "sources": []}
207
+ with tempfile.TemporaryDirectory() as tmpdir:
208
+ p = Path(tmpdir) / "kg.json"
209
+ p.write_text(json.dumps(data))
210
+ stats = ingester.ingest_kg(p)
211
+ assert stats["edges"] == 0
212
+
213
+ort MagicMock
214
+
215
+import pytest
216
+
217
+from navegador.graph.schema import NodeLabel
218
+from nav"""Tests for navegador.inge"/nonexistent/kg.json")e": "Design Doc"ster = PlanopticonIngester(store)
219
+ with tempfile.TemporaryDirectory() as tmpdir:
220
+ p = Path(tmpdir) / "interchange.json"
221
+ p.write_text(json.dumps(INTERCHANGE_DATA))
222
+ ingester.ingest_ices": []}
223
+ {"nodbels = [c[0][0] for c in stor def test_ingests_requiremkg(p)
224
+ assert "nodes" in stats
225
+ assert "edges" in stats
226
+
227
+
228
+# ── ingest_interchange ────────────────────────────────────────────────────────
229
+
230
+class TestIngestInterchange:
231
+ def test_ingests_decision_entities(self):
232
+ store = _make_store()
233
+ ingester = PlanopticonIngester(store)
234
+ with tempfile.TemporaryDirectory() as tmpdir:
235
+ p = Path(tmpdir) / "interchange.json"
236
+ p.write_text(json.dumps(INTERCHANGE_DATA))
237
+ ingester.ingest_interchange(p)
238
+ labels = [c[0][0] for c in store.create_node.call_args_list]
239
+ assert NodeLabel.Decision in labels
240
+
241
+ def test_ingests_requirement_as_rule(self):
242
+ store = _make_store()
243
+ ingester = PlanopticonIngester(store)
244
+ with tempfile.TemporaryDirectory() as tmpdir:
245
+ p = Path(tmpdir) / "interchange.json"
246
+ p.write_text(json.dumps(INTERCHANGE_DATA))
247
+ ingester.ingest_interchange(p)
248
+ labels = [c[0][0] for c in store.create_node.call_args_list]
249
+ assert NodeLabel.Rule in labels
250
+
251
+ def test_creates_domain_nodes_from_project_tags(self):
252
+ store = _make_store()
253
+ ingester = PlanopticonIngester(store)
254
+ with tempfile.TemporaryDirectory() as tmpdir:
255
+ p = Path(tmpdir) / "interchange.json"
256
+ p.write_text(json.dumps(INTERCHANGE_DATA))
257
+ ingester.ingest_interchange(p)
258
+ labels = [c[0][0] for c in store.create_node.call_args_list]
259
+ assert NodeLabel.Domain in labels
260
+
261
+ def test_ingests_artifacts_as_wiki_pages(self):
262
+ store = _make_store()
263
+ ingester = PlanopticonIngester(store)
264
+ with tempfile.TemporaryDirectory() as tmpdir:
265
+ p = Path(tmpdir) / "interchange.json"
266
+ p.write_text(json.dumps(INTERCHANGE_DATA))
267
+ ingester.ingest_interchange(p)
268
+ labels = [c[0][0] for c in store.create_node.call_args_list]
269
+ assert NodeLabel.WikiPage in labels
270
+
271
+ def test_empty_entities_returns_empty_stats(self):
272
+ store = _make_store()
273
+ ingester = PlanopticonIngester(store)
274
+ with tempfile.TemporaryDirectory() as tmpdir:
275
+ p = Path(tmpdir) / "interchange.json"
276
+ p.write_text(json.dumps({"project": {}, "entities": [], "relationships": [],
277
+ "artifacts": [], "sources": []}))
278
+ stats = ingester.ingest_interchange(p)
279
+ assert stats["nodes"] == 0
280
+
281
+ def test_returns_stats_dict(self):
282
+ store = _make_store()
283
+ ingester = PlanopticonIngester(store)
284
+ with tempfile.TemporaryDirectory() as tmpdir:
285
+ p = Path(tmpdir) / "interchange.json"
286
+ p.write_text(json.dumps(INTERCHANGE_DATA))
287
+ stats = ingester.ingest_interchange(p)
288
+ assert "nodes" in stats and "edges" in stats
289
+
290
+
291
+# ── ingest_manifest ────────────────────────────────────────────────────────────
292
+
293
+class TestIngestManifest:
294
+ def test_ingests_key_points_as_concepts(self):
295
+ store = _make_store()
296
+ ingester = PlanopticonIngester(store)
297
+ with tempfile.TemporaryDirectory() as tmpdir:
298
+ p = Path(tmpdir) / "manifest.json"
299
+ p.write_text(json.dumps(MANIFEST_DATA))
300
+ ingester.ingest_manifest(p)
301
+ labels = [c[0][0] for c in store.create_node.call_args_list]
302
+ assert NodeLabel.Concept in labels
303
+
304
+ def test_ingests_action_items_as_rules(self):
305
+ store = _make_store()
306
+ ingester = PlanopticonIngester(store)
307
+ with tempfile.TemporaryDirectory() as tmpdir:
308
+ p = Path(tmpdir) / "manifest.json"
309
+ p.write_text(json.dumps(MANIFEST_DATA))
310
+ ingester.ingest_manifest(p)
311
+ labels = [c[0][0] for c in store.create_node.call_args_list]
312
+ assert NodeLabel.Rule in labels
313
+
314
+ def test_ingests_action_item_assignee_as_person(self):
315
+ store = _make_store()
316
+ ingester = PlanopticonIngester(store)
317
+ with tempfile.TemporaryDirectory() as tmpdir:
318
+ p = Path(tmpdir) / "manifest.json"
319
+ p.write_text(json.dumps(MANIFEST_DATA))
320
+ ingester.ingest_manifest(p)
321
+ labels = [c[0][0] for c in store.create_node.call_args_list]
322
+ assert NodeLabel.Person in labels
323
+
324
+ def test_ingests_diagrams_as_wiki_pages(self):
325
+ store = _make_store()
326
+ ingester = PlanopticonIngester(store)
327
+ with tempfile.TemporaryDirectory() as tmpdir:
328
+ p = Path(tmpdir) / "manifest.json"
329
+ p.write_text(json.dumps(MANIFEST_DATA))
330
+ ingester.ingest_manifest(p)
331
+ labels = [c[0][0] for c in store.create_node.call_args_list]
332
+ assert NodeLabel.WikiPage in labels
333
+
334
+ def test_diagram_elements_become_concepts(self):
335
+ store = _make_store()
336
+ ingester = PlanopticonIngester(store)
337
+ with tempfile.TemporaryDirectory() as tmpdir:
338
+ p = Path(tmpdir) / "manifest.json"
339
+ p.write_text(json.dumps(MANIFEST_DATA))
340
+ ingester.ingest_manifest(p)
341
+ # "User" and "Auth" are diagram elements → Concept nodes
342
+ names = [c[0][1].get("name") for c in store.create_node.call_args_list
343
+ if isinstance(c[0][1], dict)]
344
+ assert "User" in names or "Auth" in names
345
+
346
+ def test_loads_external_kg_json(self):
347
+ store = _make_store()
348
+ ingester = PlanopticonIngester(store)
349
+ with tempfile.TemporaryDirectory() as tmpdir:
350
+ kg = {"nodes": [{"type": "concept", "name": "External Concept"}],
351
+ "relationships": [], "sources": []}
352
+ (Path(tmpdir) / "kg.json").write_text(json.dumps(kg))
353
+ manifest = dict(MANIFEST_DATA)
354
+ manifest["knowledge_graph_json"] = "kg.json"
355
+ p = Path(tmpdir) / "manifest.json"
356
+ p.write_text(json.dumps(manifest))
357
+ ingester.ingest_manifest(p)
358
+ names = [c[0][1].get("name") for c in store.create_node.call_args_list
359
+ if isinstance(c[0][1], dict)]
360
+ assert "External Concept" in names
361
+
362
+ def test_empty_manifest_no_crash(self):
363
+ store = _make_store()
364
+ ingester = PlanopticonIngester(store)
365
+ with tempfile.TemporaryDirectory() as tmpdir:
366
+ p = Path(tmpdir) / "manifest.json"
367
+ p.write_text(json.dumps({}))
368
+ stats = ingester.ingest_manifest(p)
369
+ assert "nodes" in stats
370
+
371
+
372
+# ── ingest_batch ──────────────────────────────────────────────────────────────
373
+
374
+class TestIngestBatch:
375
+ def test_processes_merged_kg_if_present(self):
376
+ store = _make_store()
377
+ ingester = PlanopticonIngester(store)
378
+ with tempfile.TemporaryDirectory() as tmpdir:
379
+ kg = {"nodes": [{"type": "concept", "name": "Merged"}],
380
+ "relationships": [], "sources": []}
381
+ (Path(tmpdir) / "merged.json").write_text(json.dumps(kg))
382
+ batch = {"merged_knowledge_graph_json": "merged.json"}
383
+ p = Path(tmpdir) / "batch.json"
384
+ p.write_text(json.dumps(batch))
385
+ ingester.ingest_batch(p)
386
+ names = [c[0][1].get("name") for c in store.create_node.call_args_list
387
+ if isinstance(c[0][1], dict)]
388
+ assert "Merged" in names
389
+
390
+ def test_processes_completed_videos(self):
391
+ store = _make_store()
392
+ ingester = PlanopticonIngester(store)
393
+ with tempfile.TemporaryDirectory() as tmpdir:
394
+ (Path(tmpdir) / "vid1.json").write_text(json.dumps(MANIFEST_DATA))
395
+ batch = {
396
+ "videos": [
397
+ {"status": "completed", "manifest_path": "vid1.json"},
398
+ {"status": "pending", "manifest_path": "vid1.json"}, # skipped
399
+ ]
400
+ }
401
+ p = Path(tmpdir) / "batch.json"
402
+ p.write_text(json.dumps(batch))
403
+ stats = ingester.ingest_batch(p)
404
+ assert "nodes" in stats
405
+
406
+ def test_skips_missing_manifest_gracefully(self):
407
+ store = _make_store()
408
+ ingester = PlanopticonIngester(store)
409
+ with tempfile.TemporaryDirectory() as tmpdir:
410
+ pleted", "manifest_path": "nonexistent.json"},
411
+ ]
412
+ }
413
+ p = Path(tmpdir) / "batch.json"
414
+ p.write_text(json": 0,
415
+ " ingester.ingest_batch(p)
416
+
417
+ def test_merges_stats_across_videos(self):
418
+ store = _make_store()
419
+ ingester = PlanopticonIngester(store)
420
+ with tempfile.TemporaryDirectory() as tmpdir:
421
+ (Path(tmpdir) / "v1.json").write_text(json.dumps(MANIFEST_DATA))
422
+ (Path(tmpdir) / "v2.json").write_text(json.dumps(MANIFEST_DATA))
423
+ batch = {
424
+ "videos": [
425
+ {"status": "completed", "manifest_path": "v1.json"},
426
+ {"status": "completed", "manifest_path": "v2.json"},
427
+ ]
428
+ }
429
+ p = Path(tmpdir) / "batch.json"
430
+ p.write_text(json.dumps(batch))
431
+ stats = ingester.ingest_batch(p)
432
+ # Should have processed both, stats should be non-zero
433
+ assert stats.get("nodes", 0) >= 0 # at least doesn't
--- a/tests/test_ingestion_planopticon.py
+++ b/tests/test_ingestion_planopticon.py
@@ -0,0 +1,433 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_ingestion_planopticon.py
+++ b/tests/test_ingestion_planopticon.py
@@ -0,0 +1,433 @@
1 ap_coverage(self):
2 from navNodeLabel
3 from navegador.ingestion.planopticon import (
4 EDGE_MAP,
5 NODE_TYPE_MAP,
6 PLANNING_TYPE_MAP,
7 PlanopticonIngester,
8 )
9
10
11 def _make_store():
12 store = MagicMock()
13 store.query.return_value = MagicMock(result_set=[])
14 return store
15
16
17 # ── Fixtures ──────────────────────────────────────────────────────────────────
18
19 KG_DATA = {
20 "nodes": [
21 {"id": "n1", "type": "concept", "name": "Payment Gateway",
22 "description": "Handles payments"},
23 {"id": "n2", "type": "person", "name": "Carol", "email": "[email protected]"},
24 {"id": "n3", "type": "technology", "name": "PostgreSQL", "description": "DB"},
25 {"id": "n4", "type": "decision", "name": "Use Redis"},
26 {"id": "n5", "type": "unknown_type", "name": "Misc"},
27 {"id": "n6", "type": "diagram", "name": "Service Map", "source": "http://img.png"},
28 ],
29 "relationships": [
30 {"source": "Payment Gateway", "target": "PostgreSQL", "type": "uses"},
31 {"source": "Carol", "target": "Payment Gateway", "type": "assigned_to"},
32 {"source": "", "target": "nope", "type": "related_to"}, # bad rel — no source
33 ],
34 "sources": [
35 {"title": "Meeting 2024", "url": "https://ex.com", "source_type": "meeting"},
36 ],
37 }
38
39 INTERCHANGE_DATA = {
40 "project": {"name": "MyProject", "tags": ["backend", "payments"]},
41 "entities": [
42 {
43 "planning_type": "decision",
44 "name": "Adopt microservices",
45 "description": "Split the monolith",
46 "status": "accepted",
47 "rationale": "Scale independently",
48 },
49 {
50 "planning_type": "requirement",
51 "name": "PCI compliance",
52 "description": "Must comply with PCI-DSS",
53 "priority": "high",
54 },
55 {
56 "planning_type": "goal",
57 "name": "Increase uptime",
58 "description": "99.9% SLA",
59 },
60 {
61 # no planning_type → falls through to _ingest_kg_node
62 "type": "concept",
63 "name": "Event Sourcing",
64 },
65 ],
66 "relationships": [],
67 "artifacts": [
68 {"name": "Architecture Diagram", "content": "mermaid content here"},
69 ],
70 "sources": [],
71 }
72
73 MANIFEST_DATA = {
74 "video": {"title": "Sprint Planning", "url": "https://example.com/video/1"},
75 "key_points": [
76 {"point": "Use async everywhere", "topic": "Architecture", "details": "For scale"},
77 ],
78 "action_items": [
79 {"action": "Refactor auth service", "assignee": "Bob", "context": "High priority"},
80 ],
81 "diagrams": [
82 {
83 "diagram_type": "sequence",
84 "timestamp": 120,
85 "description": "Auth flow",
86 "mermaid": "sequenceDiagram ...",
87 "elements": ["User", "Auth"],
88 }
89 ],
90 }
91
92
93 # ── Maps ──────────────────────────────────────────────────────────────────────
94
95 class TestMaps:
96 def test_node_type_map_coverage(self):
97 assert NODE_TYPE_MAP["concept"] == NodeLabel.Concept
98 assert NODE_TYPE_MAP["technology"] == NodeLabel.Concept
99 assert NODE_TYPE_MAP["organization"] == NodeLabel.Concept
100 assert NODE_TYPE_MAP["person"] == NodeLabel.Person
101 assert NODE_TYPE_MAP["diagram"] == NodeLabel.WikiPage
102
103 def test_planning_type_map_coverage(self):
104 assert PLANNING_TYPE_MAP["decision"] == NodeLabel.Decision
105 assert PLANNING_TYPE_MAP["requirement"] == NodeLabel.Rule
106 assert PLANNING_TYPE_MAP["constraint"] == NodeLabel.Rule
107 assert PLANNING_TYPE_MAP["risk"] == NodeLabel.Rule
108 assert PLANNING_TYPE_MAP["goal"] == NodeLabel.Concept
109
110 def test_edge_map_coverage(self):
111 from navegador.graph.schema import EdgeType
112 assert EDGE_MAP["uses"] == EdgeType.DEPENDS_ON
113 assert EDGE_MAP["related_to"] == EdgeType.RELATED_TO
114 assert EDGE_MAP["assigned_to"] == EdgeType.ASSIGNED_TO
115 assert EDGE_MAP["governs"] == EdgeType.GOVERNS
116 assert EDGE_MAP["implements"] == EdgeType.IMPLEMENTS
117
118
119 # ── ingest_kg ─────────────────────────────────────────────────────────────────
120
121 class TestIngestKg:
122 def test_ingests_concept_nodes(self):
123 store = _make_store()
124 ingester = PlanopticonIngester(store)
125 with tempfile.TemporaryDirectory() as tmpdir:
126 p = Path(tmpdir) / "kg.json"
127 p.write_text(json.dumps(KG_DATA))
128 stats = ingester.ingest_kg(p)
129 assert stats["nodes"] >= 1
130
131 def test_ingests_person_nodes(self):
132 store = _make_store()
133 ingester = PlanopticonIngester(store)
134 with tempfile.TemporaryDirectory() as tmpdir:
135 p = Path(tmpdir) / "kg.json"
136 p.write_text(json.dumps(KG_DATA))
137 ingester.ingest_kg(p)
138 labels = [c[0][0] for c in store.create_node.call_args_list]
139 assert NodeLabel.Person in labels
140
141 def test_ingests_technology_as_concept(self):
142 store = _make_store()
143 ingester = PlanopticonIngester(store)
144 data = {"nodes": [{"type": "technology", "name": "PostgreSQL"}],
145 "relationships": [], "sources": []}
146 with tempfile.TemporaryDirectory() as tmpdir:
147 p = Path(tmpdir) / "kg.json"
148 p.write_text(json.dumps(data))
149 ingester.ingest_kg(p)
150 labels = [c[0][0] for c in store.create_node.call_args_list]
151 assert NodeLabel.Concept in labels
152
153 def test_ingests_diagram_as_wiki_page(self):
154 store = _make_store()
155 ingester = PlanopticonIngester(store)
156 data = {"nodes": [{"type": "diagram", "name": "Arch Diagram", "source": "http://x.com"}],
157 "relationships": [], "sources": []}
158 with tempfile.TemporaryDirectory() as tmpdir:
159 p = Path(tmpdir) / "kg.json"
160 p.write_text(json.dumps(data))
161 ingester.ingest_kg(p)
162 labels = [c[0][0] for c in store.create_node.call_args_list]
163 assert NodeLabel.WikiPage in labels
164
165 def test_skips_nodes_without_name(self):
166 store = _make_store()
167 ingester = PlanopticonIngester(store)
168 data = {"nodes": [{"type": "concept", "name": ""}], "relationships": [], "sources": []}
169 with tempfile.TemporaryDirectory() as tmpdir:
170 p = Path(tmpdir) / "kg.json"
171 p.write_text(json.dumps(data))
172 stats = ingester.ingest_kg(p)
173 assert stats["nodes"] == 0
174
175 def test_ingests_sources_as_wiki_pages(self):
176 store = _make_store()
177 ingester = PlanopticonIngester(store)
178 data = {
179 "nodes": [], "relationships": [],
180 "sources": [
181 {"title": "Meeting 2024", "url": "https://ex.com", "source_type": "meeting"},
182 ],
183 }
184 with tempfile.TemporaryDirectory() as tmpdir:
185 p = Path(tmpdir) / "kg.json"
186 p.write_text(json.dumps(data))
187 stats = ingester.ingest_kg(p)
188 assert stats["nodes"] >= 1
189 labels = [c[0][0] for c in store.create_node.call_args_list]
190 assert NodeLabel.WikiPage in labels
191
192 def test_ingests_relationships(self):
193 store = _make_store()
194 ingester = PlanopticonIngester(store)
195 with tempfile.TemporaryDirectory() as tmpdir:
196 p = Path(tmpdir) / "kg.json"
197 p.write_text(json.dumps(KG_DATA))
198 stats = ingester.ingest_kg(p)
199 assert stats["edges"] >= 1
200 store.query.assert_called()
201
202 def test_skips_bad_relationships(self):
203 store = _make_store()
204 ingester = PlanopticonIngester(store)
205 data = {"nodes": [], "relationships": [{"source": "", "target": "x", "type": "related_to"}],
206 "sources": []}
207 with tempfile.TemporaryDirectory() as tmpdir:
208 p = Path(tmpdir) / "kg.json"
209 p.write_text(json.dumps(data))
210 stats = ingester.ingest_kg(p)
211 assert stats["edges"] == 0
212
213 ort MagicMock
214
215 import pytest
216
217 from navegador.graph.schema import NodeLabel
218 from nav"""Tests for navegador.inge"/nonexistent/kg.json")e": "Design Doc"ster = PlanopticonIngester(store)
219 with tempfile.TemporaryDirectory() as tmpdir:
220 p = Path(tmpdir) / "interchange.json"
221 p.write_text(json.dumps(INTERCHANGE_DATA))
222 ingester.ingest_ices": []}
223 {"nodbels = [c[0][0] for c in stor def test_ingests_requiremkg(p)
224 assert "nodes" in stats
225 assert "edges" in stats
226
227
228 # ── ingest_interchange ────────────────────────────────────────────────────────
229
230 class TestIngestInterchange:
231 def test_ingests_decision_entities(self):
232 store = _make_store()
233 ingester = PlanopticonIngester(store)
234 with tempfile.TemporaryDirectory() as tmpdir:
235 p = Path(tmpdir) / "interchange.json"
236 p.write_text(json.dumps(INTERCHANGE_DATA))
237 ingester.ingest_interchange(p)
238 labels = [c[0][0] for c in store.create_node.call_args_list]
239 assert NodeLabel.Decision in labels
240
241 def test_ingests_requirement_as_rule(self):
242 store = _make_store()
243 ingester = PlanopticonIngester(store)
244 with tempfile.TemporaryDirectory() as tmpdir:
245 p = Path(tmpdir) / "interchange.json"
246 p.write_text(json.dumps(INTERCHANGE_DATA))
247 ingester.ingest_interchange(p)
248 labels = [c[0][0] for c in store.create_node.call_args_list]
249 assert NodeLabel.Rule in labels
250
251 def test_creates_domain_nodes_from_project_tags(self):
252 store = _make_store()
253 ingester = PlanopticonIngester(store)
254 with tempfile.TemporaryDirectory() as tmpdir:
255 p = Path(tmpdir) / "interchange.json"
256 p.write_text(json.dumps(INTERCHANGE_DATA))
257 ingester.ingest_interchange(p)
258 labels = [c[0][0] for c in store.create_node.call_args_list]
259 assert NodeLabel.Domain in labels
260
261 def test_ingests_artifacts_as_wiki_pages(self):
262 store = _make_store()
263 ingester = PlanopticonIngester(store)
264 with tempfile.TemporaryDirectory() as tmpdir:
265 p = Path(tmpdir) / "interchange.json"
266 p.write_text(json.dumps(INTERCHANGE_DATA))
267 ingester.ingest_interchange(p)
268 labels = [c[0][0] for c in store.create_node.call_args_list]
269 assert NodeLabel.WikiPage in labels
270
271 def test_empty_entities_returns_empty_stats(self):
272 store = _make_store()
273 ingester = PlanopticonIngester(store)
274 with tempfile.TemporaryDirectory() as tmpdir:
275 p = Path(tmpdir) / "interchange.json"
276 p.write_text(json.dumps({"project": {}, "entities": [], "relationships": [],
277 "artifacts": [], "sources": []}))
278 stats = ingester.ingest_interchange(p)
279 assert stats["nodes"] == 0
280
281 def test_returns_stats_dict(self):
282 store = _make_store()
283 ingester = PlanopticonIngester(store)
284 with tempfile.TemporaryDirectory() as tmpdir:
285 p = Path(tmpdir) / "interchange.json"
286 p.write_text(json.dumps(INTERCHANGE_DATA))
287 stats = ingester.ingest_interchange(p)
288 assert "nodes" in stats and "edges" in stats
289
290
291 # ── ingest_manifest ────────────────────────────────────────────────────────────
292
293 class TestIngestManifest:
294 def test_ingests_key_points_as_concepts(self):
295 store = _make_store()
296 ingester = PlanopticonIngester(store)
297 with tempfile.TemporaryDirectory() as tmpdir:
298 p = Path(tmpdir) / "manifest.json"
299 p.write_text(json.dumps(MANIFEST_DATA))
300 ingester.ingest_manifest(p)
301 labels = [c[0][0] for c in store.create_node.call_args_list]
302 assert NodeLabel.Concept in labels
303
304 def test_ingests_action_items_as_rules(self):
305 store = _make_store()
306 ingester = PlanopticonIngester(store)
307 with tempfile.TemporaryDirectory() as tmpdir:
308 p = Path(tmpdir) / "manifest.json"
309 p.write_text(json.dumps(MANIFEST_DATA))
310 ingester.ingest_manifest(p)
311 labels = [c[0][0] for c in store.create_node.call_args_list]
312 assert NodeLabel.Rule in labels
313
314 def test_ingests_action_item_assignee_as_person(self):
315 store = _make_store()
316 ingester = PlanopticonIngester(store)
317 with tempfile.TemporaryDirectory() as tmpdir:
318 p = Path(tmpdir) / "manifest.json"
319 p.write_text(json.dumps(MANIFEST_DATA))
320 ingester.ingest_manifest(p)
321 labels = [c[0][0] for c in store.create_node.call_args_list]
322 assert NodeLabel.Person in labels
323
324 def test_ingests_diagrams_as_wiki_pages(self):
325 store = _make_store()
326 ingester = PlanopticonIngester(store)
327 with tempfile.TemporaryDirectory() as tmpdir:
328 p = Path(tmpdir) / "manifest.json"
329 p.write_text(json.dumps(MANIFEST_DATA))
330 ingester.ingest_manifest(p)
331 labels = [c[0][0] for c in store.create_node.call_args_list]
332 assert NodeLabel.WikiPage in labels
333
334 def test_diagram_elements_become_concepts(self):
335 store = _make_store()
336 ingester = PlanopticonIngester(store)
337 with tempfile.TemporaryDirectory() as tmpdir:
338 p = Path(tmpdir) / "manifest.json"
339 p.write_text(json.dumps(MANIFEST_DATA))
340 ingester.ingest_manifest(p)
341 # "User" and "Auth" are diagram elements → Concept nodes
342 names = [c[0][1].get("name") for c in store.create_node.call_args_list
343 if isinstance(c[0][1], dict)]
344 assert "User" in names or "Auth" in names
345
346 def test_loads_external_kg_json(self):
347 store = _make_store()
348 ingester = PlanopticonIngester(store)
349 with tempfile.TemporaryDirectory() as tmpdir:
350 kg = {"nodes": [{"type": "concept", "name": "External Concept"}],
351 "relationships": [], "sources": []}
352 (Path(tmpdir) / "kg.json").write_text(json.dumps(kg))
353 manifest = dict(MANIFEST_DATA)
354 manifest["knowledge_graph_json"] = "kg.json"
355 p = Path(tmpdir) / "manifest.json"
356 p.write_text(json.dumps(manifest))
357 ingester.ingest_manifest(p)
358 names = [c[0][1].get("name") for c in store.create_node.call_args_list
359 if isinstance(c[0][1], dict)]
360 assert "External Concept" in names
361
362 def test_empty_manifest_no_crash(self):
363 store = _make_store()
364 ingester = PlanopticonIngester(store)
365 with tempfile.TemporaryDirectory() as tmpdir:
366 p = Path(tmpdir) / "manifest.json"
367 p.write_text(json.dumps({}))
368 stats = ingester.ingest_manifest(p)
369 assert "nodes" in stats
370
371
372 # ── ingest_batch ──────────────────────────────────────────────────────────────
373
374 class TestIngestBatch:
375 def test_processes_merged_kg_if_present(self):
376 store = _make_store()
377 ingester = PlanopticonIngester(store)
378 with tempfile.TemporaryDirectory() as tmpdir:
379 kg = {"nodes": [{"type": "concept", "name": "Merged"}],
380 "relationships": [], "sources": []}
381 (Path(tmpdir) / "merged.json").write_text(json.dumps(kg))
382 batch = {"merged_knowledge_graph_json": "merged.json"}
383 p = Path(tmpdir) / "batch.json"
384 p.write_text(json.dumps(batch))
385 ingester.ingest_batch(p)
386 names = [c[0][1].get("name") for c in store.create_node.call_args_list
387 if isinstance(c[0][1], dict)]
388 assert "Merged" in names
389
390 def test_processes_completed_videos(self):
391 store = _make_store()
392 ingester = PlanopticonIngester(store)
393 with tempfile.TemporaryDirectory() as tmpdir:
394 (Path(tmpdir) / "vid1.json").write_text(json.dumps(MANIFEST_DATA))
395 batch = {
396 "videos": [
397 {"status": "completed", "manifest_path": "vid1.json"},
398 {"status": "pending", "manifest_path": "vid1.json"}, # skipped
399 ]
400 }
401 p = Path(tmpdir) / "batch.json"
402 p.write_text(json.dumps(batch))
403 stats = ingester.ingest_batch(p)
404 assert "nodes" in stats
405
406 def test_skips_missing_manifest_gracefully(self):
407 store = _make_store()
408 ingester = PlanopticonIngester(store)
409 with tempfile.TemporaryDirectory() as tmpdir:
410 pleted", "manifest_path": "nonexistent.json"},
411 ]
412 }
413 p = Path(tmpdir) / "batch.json"
414 p.write_text(json": 0,
415 " ingester.ingest_batch(p)
416
417 def test_merges_stats_across_videos(self):
418 store = _make_store()
419 ingester = PlanopticonIngester(store)
420 with tempfile.TemporaryDirectory() as tmpdir:
421 (Path(tmpdir) / "v1.json").write_text(json.dumps(MANIFEST_DATA))
422 (Path(tmpdir) / "v2.json").write_text(json.dumps(MANIFEST_DATA))
423 batch = {
424 "videos": [
425 {"status": "completed", "manifest_path": "v1.json"},
426 {"status": "completed", "manifest_path": "v2.json"},
427 ]
428 }
429 p = Path(tmpdir) / "batch.json"
430 p.write_text(json.dumps(batch))
431 stats = ingester.ingest_batch(p)
432 # Should have processed both, stats should be non-zero
433 assert stats.get("nodes", 0) >= 0 # at least doesn't
A tests/test_ingestion_wiki.py
+244
--- a/tests/test_ingestion_wiki.py
+++ b/tests/test_ingestion_wiki.py
@@ -0,0 +1,244 @@
1
+"""Tests for navegador.ingestion.wiki — WikiIngester."""
2
+
3
+import tempfile
4
+from pathlib import Path
5
+from unittest.mock import MagicMock, patch
6
+
7
+import pytest
8
+
9
+from navegador.graph.schema import NodeLabel
10
+from navegador.ingestion.wiki import WikiIngester, _extract_terms
11
+
12
+# ── Unit: _extract_terms ──────────────────────────────────────────────────────
13
+
14
+class TestExtractTerms:
15
+ def test_extracts_headings(self):
16
+ md = "# Introduction\n## Getting Started\n### Deep Dive\n"
17
+ terms = _extract_terms(md)
18
+ assert "Introduction" in terms
19
+ assert "Getting Started" in terms
20
+ assert "Deep Dive" in terms
21
+
22
+ def test_extracts_bold_asterisk(self):
23
+ md = "Use **GraphStore** for all persistence."
24
+ terms = _extract_terms(md)
25
+ assert "GraphStore" in terms
26
+
27
+ def test_extracts_bold_underscore(self):
28
+ md = "The __FalkorDB__ module is required."
29
+ terms = _extract_terms(md)
30
+ assert "FalkorDB" in terms
31
+
32
+ def test_deduplicates(self):
33
+ md = "# GraphStore\nUse **GraphStore** here too."
34
+ terms = _extract_terms(md)
35
+ assert terms.count("GraphStore") == 1
36
+
37
+ def test_empty_markdown(self):
38
+ assert _extract_terms("") == []
39
+
40
+ def test_no_headings_no_bold(self):
41
+ terms = _extract_terms("plain text with no markup")
42
+ assert terms == []
43
+
44
+ def test_preserves_order(self):
45
+ md = "# Alpha\n# Beta\n**Gamma**"
46
+ terms = _extract_terms(md)
47
+ assert terms == ["Alpha", "Beta", "Gamma"]
48
+
49
+
50
+# ── Unit: ingest_local ────────────────────────────────────────────────────────
51
+
52
+class TestIngestLocal:
53
+ def _make_store(self):
54
+ store = MagicMock()
55
+ store.query.return_value = MagicMock(result_set=[])
56
+ return store
57
+
58
+ def test_ingests_markdown_files(self):
59
+ store = self._make_store()
60
+ ingester = WikiIngester(store)
61
+ with tempfile.TemporaryDirectory() as tmpdir:
62
+ (Path(tmpdir) / "home.md").write_text("# Welcome\nThis is home.")
63
+ (Path(tmpdir) / "guide.md").write_text("## Usage\nSome guide.")
64
+ stats = ingester.ingest_local(tmpdir)
65
+ assert stats["pages"] == 2
66
+
67
+ def test_skips_non_markdown(self):
68
+ store = self._make_store()
69
+ ingester = WikiIngester(store)
70
+ with tempfile.TemporaryDirectory() as tmpdir:
71
+ (Path(tmpdir) / "readme.md").write_text("# Readme")
72
+ (Path(tmpdir) / "image.png").write_bytes(b"\x89PNG")
73
+ stats = ingester.ingest_local(tmpdir)
74
+ assert stats["pages"] == 1
75
+
76
+ def test_raises_if_dir_missing(self):
77
+ store = self._make_store()
78
+ ingester = WikiIngester(store)
79
+ with pytest.raises(FileNotFoundError):
80
+ ingester.ingest_local("/nonexistent/path")
81
+
82
+ def test_creates_wiki_page_node(self):
83
+ store = self._make_store()
84
+ ingester = WikiIngester(store)
85
+ with tempfile.TemporaryDirectory() as tmpdir:
86
+ (Path(tmpdir) / "arch.md").write_text("# Architecture")
87
+ ingester.ingest_local(tmpdir)
88
+ store.create_node.assert_called_once()
89
+ call_args = store.create_node.call_args
90
+ assert call_args[0][0] == NodeLabel.WikiPage
91
+ props = call_args[0][1]
92
+ assert props["name"] == "arch"
93
+ assert props["source"] == "local"
94
+
95
+ def test_page_name_normalisation(self):
96
+ store = self._make_store()
97
+ ingester = WikiIngester(store)
98
+ with tempfile.TemporaryDirectory() as tmpdir:
99
+ (Path(tmpdir) / "getting-started.md").write_text("# Hi")
100
+ ingester.ingest_local(tmpdir)
101
+ props = store.create_node.call_args[0][1]
102
+ assert props["name"] == "getting started"
103
+
104
+ def test_creates_documents_edge_when_term_matches(self):
105
+ store = MagicMock()
106
+ store.query.return_value = MagicMock(result_set=[["Concept", "GraphStore"]])
107
+ ingester = WikiIngester(store)
108
+ with tempfile.TemporaryDirectory() as tmpdir:
109
+ (Path(tmpdir) / "page.md").write_text("# GraphStore\nSome text.")
110
+ stats = ingester.ingest_local(tmpdir)
111
+ assert stats["links"] >= 1
112
+ store.create_edge.assert_called()
113
+
114
+ def test_no_links_when_no_term_match(self):
115
+ store = self._make_store() # query returns []
116
+ ingester = WikiIngester(store)
117
+ with tempfile.TemporaryDirectory() as tmpdir:
118
+ (Path(tmpdir) / "page.md").write_text("# UnknownTerm\nText.")
119
+ stats = ingester.ingest_local(tmpdir)
120
+ assert stats["links"] == 0
121
+ store.create_edge.assert_not_called()
122
+
123
+ def test_content_capped_at_4000_chars(self):
124
+ store = self._make_store()
125
+ ingester = WikiIngester(store)
126
+ with tempfile.TemporaryDirectory() as tmpdir:
127
+ (Path(tmpdirexception).write_text("x" * 10000)
128
+ ingester.ingest_local(tmpdir)
129
+ props = store.cConceptprops = store.create_kiIngester(store)
130
+
131
+ assert "pages" in stats
132
+ assert "links" in stats
133
+ result = iself):
134
+ store = self._make_sngester(store)
135
+ with tempfile.TemporaryDirectory() as tmpdir:
136
+ subdir = Path(tmpdir) / "sub"
137
+ subdir.mkdir()
138
+ (subdir / "nested.md").write_text("# Nested")
139
+ stats = ingester.ingest_local(tmpdir)
140
+ assert stats["pages"] == 1
141
+
142
+
143
+# ── Unit: _try_link edge-type handling ────────────────────────────────────────
144
+
145
+class TestTryLink:
146
+ def test_handles_invalid_label_gracefully(self):
147
+ store = MagicMock()
148
+ store.query.return_value = MagicMock(result_set=[["InvalidLabel", "foo"]])
149
+ ingester = WikiIngester(store)
150
+ result = ingester._try_link("page", "foo")
151
+ assert result == 0
152
+
153
+ def test_creates_edge_for_valid_label(self):
154
+ store = MagicMock()
155
+ store.query.return_value = MagicMock(result_set=[["Concept", "MyService"]])
156
+ ingester = WikiIngester(store)
157
+ result = ingester._try_link("wiki page", "MyService")
158
+ assert result == 1
159
+ store.create_edge.assert_called_once()
160
+
161
+ def test_returns_zero_on_unknown_label(self):
162
+ store = MagicMock()
163
+ store.query.return_value = MagicMock(result_set=[["UnknownLabel", "node"]])
164
+ ingester = WikiIngester(store)
165
+ result = ingester._try_link("page", "node")
166
+ assert result == 0
167
+
168
+ def test_propagates_store_error(self):
169
+ store = MagicMock()
170
+ store.query.return_value = MagicMock(result_set=[["Concept", "node"]])
171
+ store.create_edge.side_effect = Exception("DB error")
172
+ ingester = WikiIngester(store)
173
+ with pytest.raises(Exception, match="DB error"):
174
+ ingester._try_link("page", "node")
175
+
176
+
177
+# ── GitHub clone (ingest_github) ──────────────────────────────────────────────
178
+
179
+class TestIngestGithub:
180
+ def _make_store(self):
181
+ store = MagicMock()
182
+ store.query.return_value = MagicMock(result_set=[])
183
+ return store
184
+
185
+ def test_clones_wiki_and_ingests_local(self):
186
+ store = self._make_store()
187
+ ingester = WikiIngester(store)
188
+
189
+ with tempfile.TemporaryDirectory() as tmpdir:
190
+ wiki_dir = Path(tmpdir)
191
+ (wiki_dir / "home.md").write_text("# Home\nWelcome.")
192
+
193
+ mock_result = MagicMock()
194
+ mock_result.returncode = 0
195
+
196
+ with patch("subprocess.run", return_value=mock_result) as mock_run, \
197
+ patch("tempfile.mkdtemp", return_value=str(tmpdir)):
198
+ stats = ingester.ingest_github("owner/repo")
199
+ mock_run.assert_called_once()
200
+ cmd = mock_run.call_args[0][0]
201
+ assert "git" in cmd
202
+ assert "clone" in cmd
203
+ assert "https://github.com/owner/repo.wiki.git" in cmd
204
+ assert stats["pages"] == 1
205
+
206
+ def test_returns_empty_on_clone_failure(self):
207
+ store = self._make_store()
208
+ ingester = WikiIngester(store)
209
+
210
+ mock_result = MagicMock()
211
+ mock_result.returncode = 1
212
+ mock_result.stderr = "fatal: repository not found"
213
+
214
+ with patch("subprocess.run", return_value=mock_result):
215
+ stats = ingester.ingest_github("owner/empty-repo")
216
+ assert stats == {"pages": 0, "links": 0}
217
+
218
+ def test_uses_token_in_url(self):
219
+ store = self._make_store()
220
+ ingester = WikiIngester(store)
221
+
222
+ mock_result = MagicMock()
223
+ mock_result.returncode = 1
224
+ mock_result.stderr = "auth error"
225
+
226
+ with patch("subprocess.run", return_value=mock_result) as mock_run:
227
+ ingester.ingest_github("owner/repo", token="mytoken")
228
+ cmd = mock_run.call_args[0][0]
229
+ assert "[email protected]" in cmd[3]
230
+
231
+ def test_uses_explicit_clone_dir(self):
232
+ store = self._make_store()
233
+ ingester = WikiIngester(store)
234
+
235
+ with tempfile.TemporaryDirectory() as tmpdir:
236
+ mock_result = MagicMock()
237
+ mock_result.returncode = 0
238
+
239
+ with patch("subprocess.run", return_value=mock_result):
240
+ ingester.ingest_github("owner/repo", clone_dir=tmpdir)
241
+ # Should not crash
242
+
243
+
244
+# ── GitHub API (ingest_github_api) ───────────────────────────�
--- a/tests/test_ingestion_wiki.py
+++ b/tests/test_ingestion_wiki.py
@@ -0,0 +1,244 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_ingestion_wiki.py
+++ b/tests/test_ingestion_wiki.py
@@ -0,0 +1,244 @@
1 """Tests for navegador.ingestion.wiki — WikiIngester."""
2
3 import tempfile
4 from pathlib import Path
5 from unittest.mock import MagicMock, patch
6
7 import pytest
8
9 from navegador.graph.schema import NodeLabel
10 from navegador.ingestion.wiki import WikiIngester, _extract_terms
11
12 # ── Unit: _extract_terms ──────────────────────────────────────────────────────
13
14 class TestExtractTerms:
15 def test_extracts_headings(self):
16 md = "# Introduction\n## Getting Started\n### Deep Dive\n"
17 terms = _extract_terms(md)
18 assert "Introduction" in terms
19 assert "Getting Started" in terms
20 assert "Deep Dive" in terms
21
22 def test_extracts_bold_asterisk(self):
23 md = "Use **GraphStore** for all persistence."
24 terms = _extract_terms(md)
25 assert "GraphStore" in terms
26
27 def test_extracts_bold_underscore(self):
28 md = "The __FalkorDB__ module is required."
29 terms = _extract_terms(md)
30 assert "FalkorDB" in terms
31
32 def test_deduplicates(self):
33 md = "# GraphStore\nUse **GraphStore** here too."
34 terms = _extract_terms(md)
35 assert terms.count("GraphStore") == 1
36
37 def test_empty_markdown(self):
38 assert _extract_terms("") == []
39
40 def test_no_headings_no_bold(self):
41 terms = _extract_terms("plain text with no markup")
42 assert terms == []
43
44 def test_preserves_order(self):
45 md = "# Alpha\n# Beta\n**Gamma**"
46 terms = _extract_terms(md)
47 assert terms == ["Alpha", "Beta", "Gamma"]
48
49
50 # ── Unit: ingest_local ────────────────────────────────────────────────────────
51
52 class TestIngestLocal:
53 def _make_store(self):
54 store = MagicMock()
55 store.query.return_value = MagicMock(result_set=[])
56 return store
57
58 def test_ingests_markdown_files(self):
59 store = self._make_store()
60 ingester = WikiIngester(store)
61 with tempfile.TemporaryDirectory() as tmpdir:
62 (Path(tmpdir) / "home.md").write_text("# Welcome\nThis is home.")
63 (Path(tmpdir) / "guide.md").write_text("## Usage\nSome guide.")
64 stats = ingester.ingest_local(tmpdir)
65 assert stats["pages"] == 2
66
67 def test_skips_non_markdown(self):
68 store = self._make_store()
69 ingester = WikiIngester(store)
70 with tempfile.TemporaryDirectory() as tmpdir:
71 (Path(tmpdir) / "readme.md").write_text("# Readme")
72 (Path(tmpdir) / "image.png").write_bytes(b"\x89PNG")
73 stats = ingester.ingest_local(tmpdir)
74 assert stats["pages"] == 1
75
76 def test_raises_if_dir_missing(self):
77 store = self._make_store()
78 ingester = WikiIngester(store)
79 with pytest.raises(FileNotFoundError):
80 ingester.ingest_local("/nonexistent/path")
81
82 def test_creates_wiki_page_node(self):
83 store = self._make_store()
84 ingester = WikiIngester(store)
85 with tempfile.TemporaryDirectory() as tmpdir:
86 (Path(tmpdir) / "arch.md").write_text("# Architecture")
87 ingester.ingest_local(tmpdir)
88 store.create_node.assert_called_once()
89 call_args = store.create_node.call_args
90 assert call_args[0][0] == NodeLabel.WikiPage
91 props = call_args[0][1]
92 assert props["name"] == "arch"
93 assert props["source"] == "local"
94
95 def test_page_name_normalisation(self):
96 store = self._make_store()
97 ingester = WikiIngester(store)
98 with tempfile.TemporaryDirectory() as tmpdir:
99 (Path(tmpdir) / "getting-started.md").write_text("# Hi")
100 ingester.ingest_local(tmpdir)
101 props = store.create_node.call_args[0][1]
102 assert props["name"] == "getting started"
103
104 def test_creates_documents_edge_when_term_matches(self):
105 store = MagicMock()
106 store.query.return_value = MagicMock(result_set=[["Concept", "GraphStore"]])
107 ingester = WikiIngester(store)
108 with tempfile.TemporaryDirectory() as tmpdir:
109 (Path(tmpdir) / "page.md").write_text("# GraphStore\nSome text.")
110 stats = ingester.ingest_local(tmpdir)
111 assert stats["links"] >= 1
112 store.create_edge.assert_called()
113
114 def test_no_links_when_no_term_match(self):
115 store = self._make_store() # query returns []
116 ingester = WikiIngester(store)
117 with tempfile.TemporaryDirectory() as tmpdir:
118 (Path(tmpdir) / "page.md").write_text("# UnknownTerm\nText.")
119 stats = ingester.ingest_local(tmpdir)
120 assert stats["links"] == 0
121 store.create_edge.assert_not_called()
122
123 def test_content_capped_at_4000_chars(self):
124 store = self._make_store()
125 ingester = WikiIngester(store)
126 with tempfile.TemporaryDirectory() as tmpdir:
127 (Path(tmpdirexception).write_text("x" * 10000)
128 ingester.ingest_local(tmpdir)
129 props = store.cConceptprops = store.create_kiIngester(store)
130
131 assert "pages" in stats
132 assert "links" in stats
133 result = iself):
134 store = self._make_sngester(store)
135 with tempfile.TemporaryDirectory() as tmpdir:
136 subdir = Path(tmpdir) / "sub"
137 subdir.mkdir()
138 (subdir / "nested.md").write_text("# Nested")
139 stats = ingester.ingest_local(tmpdir)
140 assert stats["pages"] == 1
141
142
143 # ── Unit: _try_link edge-type handling ────────────────────────────────────────
144
145 class TestTryLink:
146 def test_handles_invalid_label_gracefully(self):
147 store = MagicMock()
148 store.query.return_value = MagicMock(result_set=[["InvalidLabel", "foo"]])
149 ingester = WikiIngester(store)
150 result = ingester._try_link("page", "foo")
151 assert result == 0
152
153 def test_creates_edge_for_valid_label(self):
154 store = MagicMock()
155 store.query.return_value = MagicMock(result_set=[["Concept", "MyService"]])
156 ingester = WikiIngester(store)
157 result = ingester._try_link("wiki page", "MyService")
158 assert result == 1
159 store.create_edge.assert_called_once()
160
161 def test_returns_zero_on_unknown_label(self):
162 store = MagicMock()
163 store.query.return_value = MagicMock(result_set=[["UnknownLabel", "node"]])
164 ingester = WikiIngester(store)
165 result = ingester._try_link("page", "node")
166 assert result == 0
167
168 def test_propagates_store_error(self):
169 store = MagicMock()
170 store.query.return_value = MagicMock(result_set=[["Concept", "node"]])
171 store.create_edge.side_effect = Exception("DB error")
172 ingester = WikiIngester(store)
173 with pytest.raises(Exception, match="DB error"):
174 ingester._try_link("page", "node")
175
176
177 # ── GitHub clone (ingest_github) ──────────────────────────────────────────────
178
179 class TestIngestGithub:
180 def _make_store(self):
181 store = MagicMock()
182 store.query.return_value = MagicMock(result_set=[])
183 return store
184
185 def test_clones_wiki_and_ingests_local(self):
186 store = self._make_store()
187 ingester = WikiIngester(store)
188
189 with tempfile.TemporaryDirectory() as tmpdir:
190 wiki_dir = Path(tmpdir)
191 (wiki_dir / "home.md").write_text("# Home\nWelcome.")
192
193 mock_result = MagicMock()
194 mock_result.returncode = 0
195
196 with patch("subprocess.run", return_value=mock_result) as mock_run, \
197 patch("tempfile.mkdtemp", return_value=str(tmpdir)):
198 stats = ingester.ingest_github("owner/repo")
199 mock_run.assert_called_once()
200 cmd = mock_run.call_args[0][0]
201 assert "git" in cmd
202 assert "clone" in cmd
203 assert "https://github.com/owner/repo.wiki.git" in cmd
204 assert stats["pages"] == 1
205
206 def test_returns_empty_on_clone_failure(self):
207 store = self._make_store()
208 ingester = WikiIngester(store)
209
210 mock_result = MagicMock()
211 mock_result.returncode = 1
212 mock_result.stderr = "fatal: repository not found"
213
214 with patch("subprocess.run", return_value=mock_result):
215 stats = ingester.ingest_github("owner/empty-repo")
216 assert stats == {"pages": 0, "links": 0}
217
218 def test_uses_token_in_url(self):
219 store = self._make_store()
220 ingester = WikiIngester(store)
221
222 mock_result = MagicMock()
223 mock_result.returncode = 1
224 mock_result.stderr = "auth error"
225
226 with patch("subprocess.run", return_value=mock_result) as mock_run:
227 ingester.ingest_github("owner/repo", token="mytoken")
228 cmd = mock_run.call_args[0][0]
229 assert "[email protected]" in cmd[3]
230
231 def test_uses_explicit_clone_dir(self):
232 store = self._make_store()
233 ingester = WikiIngester(store)
234
235 with tempfile.TemporaryDirectory() as tmpdir:
236 mock_result = MagicMock()
237 mock_result.returncode = 0
238
239 with patch("subprocess.run", return_value=mock_result):
240 ingester.ingest_github("owner/repo", clone_dir=tmpdir)
241 # Should not crash
242
243
244 # ── GitHub API (ingest_github_api) ───────────────────────────�
A tests/test_mcp_server.py
+92
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -0,0 +1,92 @@
1
+"""Tests for navegador.mcp.servertool handlers."""
2
+
3
+from unittest.mock import MagicMock, patch
4
+
5
+import pytest
6
+
7
+from navegador.context.loader import C��────────
8
+
9
+def _node(name="foo", type_="Function", file_path="app.py"):
10
+ return ContextNode(name=name, type=type_, file_path=file_path)
11
+
12
+
13
+def _bundle(name="target"):
14
+ return ContextBundle(target=_node(name), nodes=[])
15
+
16
+
17
+def _mock_store():
18
+ store = MagicMock()
19
+ store.query.return_value = MagicMock(result_set=[])
20
+ store.node_count.return_value = 5
21
+ store.edge_count.return_vadef _mock_loader(store=None):
22
+ from navegador.contexloader = MagicMoloader.stor loader.load_f)
23
+ loader.load_function.return_value = _bundle()
24
+ loader.load_cl)
25
+ loader.sereturn loader
26
+
27
+
28
+# ── create_
29
+
30
+import json
31
+from unittest.mock import MagicMock, patch
32
+
33
+import pytest
34
+
35
+from navegador.context.loader import ContextBundle, ContextNode
36
+
37
+# ── Helpers"}"Tests for navegador.mcp.server — create_mcp_server and all toolr.mcp.────────�─────────────────────────────────────
38
+
39
+class TestCallToolGraphStats:
40
+ def setup_method(self):
41
+ self.fx = _ServerFixture()
42
+
43
+ @pytest.mark.asyncio
44
+ async def test_returns_node_and_edge_counts(self):
45
+ self.fx.store.node_count.return_value = 42
46
+ self.fx.store.edge_count.return_value = 17
47
+ result = await self.fx.call_tool_fn("graph_stats", {})
48
+ data = json.loads(result[0]["text"])
49
+ assert data["nodes"] == 42
50
+ assert data["edges"] == 17
51
+
52
+
53
+# ── call_tool — unknown tool ──────────────────────────────────────────────────
54
+
55
+# ── call_tool — get_rationale ────────────────────────────────�test_returns_markdUnknownContextNode
56
+
57
+# ── Helpers ────────────────────�"""Tests for navegador.mcp.server — create_mcp_server and all tool handlers."""
58
+
59
+import json
60
+from unittest.mock import MagicMock, patch
61
+
62
+import pytest
63
+
64
+from navegador.context.loader import ContextBund loader.load_function.return_value = _bundle("fn_target")
65
+ loader.load_class.return_value = _bundle("cls_target")
66
+ loader.search.return_value = []
67
+ return loader
68
+
69
+ def _build(seven_tonavegador.mcp.server import create_mcp_serverest.mark.al handlers."""
70
+
71
+import json
72
+from unittest.mock import MagicMock, mock import MagicMock, pacreate_mcp_server — happy path
73
+ def _make_serverT@GU,A:"""Build aU@E8,o:."""
74
+ mock_loader = loader or _mock_loader()K@Xh,12: = MagicMock()
75
+
76
+ # Capture the decorated functions
77
+ T@IQ,K@Ix,1K@RH,p:nonlocal list_tools_fn
78
+ list_tools_fn16@VN,1A@Tx,T:nonlocal call_tool_fn
79
+ 8@pC,D: call_tool_fn1H@VN,1k@XA,7:_moduleS@1Lk,H:mcp_server_moduleZ@ZV,T: mock_mcp_types_moduleS@1Lk,1O:mcp_types_module.Tool = MagicMock
80
+ mock_mcp_types_module.TextContent = MagicMock
81
+9@11~,W@pK,3:
82
+ 8@2Wg,~@dD,7:_modulee@eC,7:_module14@eq,5:mock_27@fz,T@tb,K:mock_loader.store)
83
+
84
+8@1qd,p:return list_tools_fn, call_tool_fn, mock_loader
85
+
86
+ I@2TW,6:serverG@2JW,x@WT,I:ambda: lambda f: fX@Xh,J:lambda: lambda f: fP@YW,7:_moduleS@1Lk,H:mcp_server_moduleh@ZV,W@pK,3:
87
+ 8@2Wg,~@dD,7:_moduleR@eC,L:MagicMock(),
88
+ 21@qb,6:resultX@iF,t:_mock_store())
89
+ assert result is mock_server
90
+M@oE,J:f_mcp_not_availableG@2JW,W@pK,4:
91
+ 8@3OP,1: n@pp,A:,
92
+ 2V@qb,10@tI,JSh7j;
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -0,0 +1,92 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -0,0 +1,92 @@
1 """Tests for navegador.mcp.servertool handlers."""
2
3 from unittest.mock import MagicMock, patch
4
5 import pytest
6
7 from navegador.context.loader import C��────────
8
9 def _node(name="foo", type_="Function", file_path="app.py"):
10 return ContextNode(name=name, type=type_, file_path=file_path)
11
12
13 def _bundle(name="target"):
14 return ContextBundle(target=_node(name), nodes=[])
15
16
17 def _mock_store():
18 store = MagicMock()
19 store.query.return_value = MagicMock(result_set=[])
20 store.node_count.return_value = 5
21 store.edge_count.return_vadef _mock_loader(store=None):
22 from navegador.contexloader = MagicMoloader.stor loader.load_f)
23 loader.load_function.return_value = _bundle()
24 loader.load_cl)
25 loader.sereturn loader
26
27
28 # ── create_
29
30 import json
31 from unittest.mock import MagicMock, patch
32
33 import pytest
34
35 from navegador.context.loader import ContextBundle, ContextNode
36
37 # ── Helpers"}"Tests for navegador.mcp.server — create_mcp_server and all toolr.mcp.────────�─────────────────────────────────────
38
39 class TestCallToolGraphStats:
40 def setup_method(self):
41 self.fx = _ServerFixture()
42
43 @pytest.mark.asyncio
44 async def test_returns_node_and_edge_counts(self):
45 self.fx.store.node_count.return_value = 42
46 self.fx.store.edge_count.return_value = 17
47 result = await self.fx.call_tool_fn("graph_stats", {})
48 data = json.loads(result[0]["text"])
49 assert data["nodes"] == 42
50 assert data["edges"] == 17
51
52
53 # ── call_tool — unknown tool ──────────────────────────────────────────────────
54
55 # ── call_tool — get_rationale ────────────────────────────────�test_returns_markdUnknownContextNode
56
57 # ── Helpers ────────────────────�"""Tests for navegador.mcp.server — create_mcp_server and all tool handlers."""
58
59 import json
60 from unittest.mock import MagicMock, patch
61
62 import pytest
63
64 from navegador.context.loader import ContextBund loader.load_function.return_value = _bundle("fn_target")
65 loader.load_class.return_value = _bundle("cls_target")
66 loader.search.return_value = []
67 return loader
68
69 def _build(seven_tonavegador.mcp.server import create_mcp_serverest.mark.al handlers."""
70
71 import json
72 from unittest.mock import MagicMock, mock import MagicMock, pacreate_mcp_server — happy path
73 def _make_serverT@GU,A:"""Build aU@E8,o:."""
74 mock_loader = loader or _mock_loader()K@Xh,12: = MagicMock()
75
76 # Capture the decorated functions
77 T@IQ,K@Ix,1K@RH,p:nonlocal list_tools_fn
78 list_tools_fn16@VN,1A@Tx,T:nonlocal call_tool_fn
79 8@pC,D: call_tool_fn1H@VN,1k@XA,7:_moduleS@1Lk,H:mcp_server_moduleZ@ZV,T: mock_mcp_types_moduleS@1Lk,1O:mcp_types_module.Tool = MagicMock
80 mock_mcp_types_module.TextContent = MagicMock
81 9@11~,W@pK,3:
82 8@2Wg,~@dD,7:_modulee@eC,7:_module14@eq,5:mock_27@fz,T@tb,K:mock_loader.store)
83
84 8@1qd,p:return list_tools_fn, call_tool_fn, mock_loader
85
86 I@2TW,6:serverG@2JW,x@WT,I:ambda: lambda f: fX@Xh,J:lambda: lambda f: fP@YW,7:_moduleS@1Lk,H:mcp_server_moduleh@ZV,W@pK,3:
87 8@2Wg,~@dD,7:_moduleR@eC,L:MagicMock(),
88 21@qb,6:resultX@iF,t:_mock_store())
89 assert result is mock_server
90 M@oE,J:f_mcp_not_availableG@2JW,W@pK,4:
91 8@3OP,1: n@pp,A:,
92 2V@qb,10@tI,JSh7j;
A tests/test_python_parser.py
+235
--- a/tests/test_python_parser.py
+++ b/tests/test_python_parser.py
@@ -0,0 +1,235 @@
1
+"""Tests for navegador.ingestion.python — PythonParser internal methods."""
2
+
3
+from unittest.mock import MagicMock, patch
4
+
5
+import pytest
6
+
7
+from navegador.graph.schema import NodeLabel
8
+
9
+# ── Mock tree-sitter node ──────────────────────────────────────────────────────
10
+
11
+class MockNode:
12
+ """Minimal mock of a tree-sitter Node."""
13
+ def __init__(self, type_: str, text: bytes = b"", children: list = None,
14
+ start_byte: int = 0, end_byte: int = 0,
15
+ start_point: tuple = (0, 0), end_point: tuple = (0, 0)):
16
+ self.type = type_
17
+ self._text = text
18
+ self.children = children or []
19
+ self.start_byte = start_byte
20
+ self.end_byte = end_byte
21
+ self.start_point = start_point
22
+ self.end_point = end_point
23
+
24
+
25
+def _text_node(text: bytes, type_: str = "identifier") -> MockNode:
26
+ return MockNode(type_, text, start_byte=0, end_byte=len(text))
27
+
28
+
29
+def _make_store():
30
+ store = MagicMock()
31
+ store.query.return_value = MagicMock(result_set=[])
32
+ return store
33
+
34
+
35
+# ── _node_text ────────────────────────────────────────────────────────────────
36
+
37
+class TestNodeText:
38
+ def test_extracts_text_from_source(self):
39
+ from navegador.ingestion.python import _node_text
40
+ source = b"hello world"
41
+ node = MockNode("identifier", start_byte=6, end_byte=11)
42
+ assert _node_text(node, source) == "world"
43
+
44
+ def test_full_source(self):
45
+ from navegador.ingestion.python import _node_text
46
+ source = b"foo_bar"
47
+ node = MockNode("identifier", start_byte=0, end_byte=7)
48
+ assert _node_text(node, source) == "foo_bar"
49
+
50
+ def test_handles_utf8(self):
51
+ from navegador.ingestion.python import _node_text
52
+ source = "héllo".encode("utf-8")
53
+ node = MockNode("identifier", start_byte=0, end_byte=len(source))
54
+ assert "llo" in _node_text(node, source)
55
+
56
+
57
+# ── _get_docstring ────────────────────────────────────────────────────────────
58
+
59
+class TestGetDocstring:
60
+ def test_returns_none_when_no_block(self):
61
+ from navegador.ingestion.python import _get_docstring
62
+ node = MockNode("function_definition", children=[
63
+ MockNode("identifier")
64
+ ])
65
+ assert _get_docstring(node, b"def foo(): pass") is None
66
+
67
+ def test_returns_none_when_no_expression_stmt(self):
68
+ from navegador.ingestion.python import _get_docstring
69
+ block = MockNode("block", children=[
70
+ MockNode("return_statement")
71
+ ])
72
+ fn = MockNode("function_definition", children=[block])
73
+ assert _get_docstring(fn, b"") is None
74
+
75
+ def test_returns_none_when_no_string_in_expr(self):
76
+ from navegador.ingestion.python import _get_docstring
77
+ expr_stmt = MockNode("expression_statement", children=[
78
+ MockNode("assignment")
79
+ ])
80
+ block = MockNode("block", children=[expr_stmt])
81
+ fn = MockNode("function_definition", children=[block])
82
+ assert _get_docstring(fn, b"") is None
83
+
84
+ def test_extracts_docstring(self):
85
+ from navegador.ingestion.python import _get_docstring
86
+ source = b'"""My docstring."""'
87
+ string_node = MockNode("string", start_byte=0, end_byte=len(source))
88
+ expr_stmt = MockNode("expression_statement", children=[string_node])
89
+ block = MockNode("block", children=[expr_stmt])
90
+ fn = MockNode("function_definition", children=[block])
91
+ result = _get_docstring(fn, source)
92
+ assert "My docstring." in result
93
+
94
+
95
+# ── _get_python_language error ─────────────────────────────────────────────────
96
+
97
+class TestGetPythonLanguage:
98
+ def test_raises_import_error_when_not_installed(self):
99
+ from navegador.ingestion.python import _get_python_language
100
+ with patch.dict("sys.modules", {"tree_sitter_python": None, "tree_sitter": None}):
101
+ with pytest.raises(ImportError, match="tree-sitter-python"):
102
+ _get_python_language()
103
+
104
+
105
+# ── PythonParser with mocked parser ──────────────────────────────────────────
106
+
107
+class TestPythonParserHandlers:
108
+ def _make_parser(self):
109
+ """Create PythonParser bypassing tree-sitter init."""
110
+ from navegador.ingestion.python import PythonParser
111
+ with patch("navegador.ingestion.python._get_parser") as mock_get:
112
+ mock_get.return_value = MagicMock()
113
+ parser = PythonParser()
114
+ return parser
115
+
116
+ def test_handle_import(self):
117
+ parser = self._make_parser()
118
+ store = _make_store()
119
+ source = b"import os.path"
120
+
121
+ dotted = _text_node(b"os.path", "dotted_name")
122
+ import_node = MockNode("import_statement", children=[dotted],
123
+ start_point=(0, 0))
124
+ stats = {"functions": 0, "classes": 0, "edges": 0}
125
+ parser._handle_import(import_node, source, "app.py", store, stats)
126
+ store.create_node.assert_called_once()
127
+ store.create_edge.assert_called_once()
128
+ assert stats["edges"] == 1
129
+
130
+ def test_handle_import_no_dotted_name(self):
131
+ parser = self._make_parser()
132
+ store = _make_store()
133
+ import_node = MockNode("import_statement", children=[
134
+ MockNode("keyword", b"import")
135
+ ], start_point=(0, 0))
136
+ stats = {"functions": 0, "classes": 0, "edges": 0}
137
+ parser._handle_import(import_node, b"import x", "app.py", store, stats)
138
+ store.create_node.assert_not_called()
139
+
140
+ def test_handle_class(self):
141
+ parser = self._make_parser()
142
+ store = _make_store()
143
+ source = b"class MyClass: pass"
144
+ name_node = _text_node(b"MyClass")
145
+ class_node = MockNode("class_definition",
146
+ children=[name_node],
147
+ start_point=(0, 0), end_point=(0, 18))
148
+ stats = {"functions": 0, "classes": 0, "edges": 0}
149
+ parser._handle_class(class_node, source, "app.py", store, stats)
150
+ assert stats["classes"] == 1
151
+ assert stats["edges"] == 1
152
+ store.create_node.assert_called()
153
+
154
+ def test_handle_class_no_identifier(self):
155
+ parser = self._make_parser()
156
+ store = _make_store()
157
+ class_node = MockNode("class_definition", children=[
158
+ MockNode("keyword", b"class")
159
+ ], start_point=(0, 0), end_point=(0, 0))
160
+ stats = {"functions": 0, "classes": 0, "edges": 0}
161
+ parser._handle_class(class_node, b"class: pass", "app.py", store, stats)
162
+ assert stats["classes"] == 0
163
+
164
+ def test_handle_class_with_inheritance(self):
165
+ parser = self._make_parser()
166
+ store = _make_store()
167
+ source = b"class Child(Parent): pass"
168
+ name_node = _text_node(b"Child")
169
+ parent_id = _text_node(b"Parent")
170
+ arg_list = MockNode("argument_list", children=[parent_id])
171
+ class_node = MockNode("class_definition",
172
+ children=[name_node, arg_list],
173
+ start_point=(0, 0), end_point=(0, 24))
174
+ stats = {"functions": 0, "classes": 0, "edges": 0}
175
+ parser._handle_class(class_node, source, "app.py", store, stats)
176
+ # Should create class node + CONTAINS edge + INHERITS edge
177
+ assert stats["edges"] == 2
178
+
179
+ def test_handle_function(self):
180
+ parser = self._make_parser()
181
+ store = _make_store()
182
+ source = b"def foo(): pass"
183
+ name_node = _text_node(b"foo")
184
+ fn_node = MockNode("function_definition", children=[name_node],
185
+ start_point=(0, 0), end_point=(0, 14))
186
+ stats = {"functions": 0, "classes": 0, "edges": 0}
187
+ parser._handle_function(fn_node, source, "app.py", store, stats, class_name=None)
188
+ assert stats["functions"] == 1
189
+ assert stats["edges"] == 1
190
+ store.create_node.assert_called_once()
191
+ label = store.create_node.call_args[0][0]
192
+ assert label == NodeLabel.Function
193
+
194
+ def test_handle_method(self):
195
+ parser = self._make_parser()
196
+ store = _make_store()
197
+ source = b"def my_method(self): pass"
198
+ name_node = _text_node(b"my_method")
199
+ fn_node = MockNode("function_definition", children=[name_node],
200
+ start_point=(0, 0), end_point=(0, 24))
201
+ stats = {"functions": 0, "classes": 0, "edges": 0}
202
+ parser._handle_function(fn_node, source, "app.py", store, stats, class_name="MyClass")
203
+ label = store.create_node.call_args[0][0]
204
+ assert label == NodeLabel.Method
205
+
206
+ def test_handle_function_no_identifier(self):
207
+ parser = self._make_parser()
208
+ store = _make_store()
209
+ fn_node = MockNode("function_definition", children=[
210
+ MockNode("keyword", b"def")
211
+ ], start_point=(0, 0), end_point=(0, 0))
212
+ stats = {"functions": 0, "classes": 0, "edges": 0}
213
+ parser._handle_function(fn_node, b"def", "app.py", store, stats, class_name=None)
214
+ assert stats["functions"] == 0
215
+
216
+ def test_extract_calls(self):
217
+ parser = self._make_parser()
218
+ store = _make_store()
219
+ source = b"def foo():\n bar()\n"
220
+
221
+ callee = _text_node(b"bar")
222
+ call_node = MockNode("call", children=[callee])
223
+ block = MockNode("block", children=[call_node])
224
+ fn_node = MockNode("function_definition", children=[block])
225
+
226
+ stats = {"functions": 0, "classes": 0, "edges": 0}
227
+ parser._extract_calls(fn_node, source, "app.py", "foo", NodeLabel.Function, store, stats)
228
+ store.create_edge.assert_called_once()
229
+ assert stats["edges"] == 1
230
+
231
+ def test_extract_calls_no_block(self):
232
+ parser = self._make_parser()
233
+ store = _make_store()
234
+ fn_node = MockNode("function_definition", children=[])
235
+
--- a/tests/test_python_parser.py
+++ b/tests/test_python_parser.py
@@ -0,0 +1,235 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_python_parser.py
+++ b/tests/test_python_parser.py
@@ -0,0 +1,235 @@
1 """Tests for navegador.ingestion.python — PythonParser internal methods."""
2
3 from unittest.mock import MagicMock, patch
4
5 import pytest
6
7 from navegador.graph.schema import NodeLabel
8
9 # ── Mock tree-sitter node ──────────────────────────────────────────────────────
10
11 class MockNode:
12 """Minimal mock of a tree-sitter Node."""
13 def __init__(self, type_: str, text: bytes = b"", children: list = None,
14 start_byte: int = 0, end_byte: int = 0,
15 start_point: tuple = (0, 0), end_point: tuple = (0, 0)):
16 self.type = type_
17 self._text = text
18 self.children = children or []
19 self.start_byte = start_byte
20 self.end_byte = end_byte
21 self.start_point = start_point
22 self.end_point = end_point
23
24
25 def _text_node(text: bytes, type_: str = "identifier") -> MockNode:
26 return MockNode(type_, text, start_byte=0, end_byte=len(text))
27
28
29 def _make_store():
30 store = MagicMock()
31 store.query.return_value = MagicMock(result_set=[])
32 return store
33
34
35 # ── _node_text ────────────────────────────────────────────────────────────────
36
37 class TestNodeText:
38 def test_extracts_text_from_source(self):
39 from navegador.ingestion.python import _node_text
40 source = b"hello world"
41 node = MockNode("identifier", start_byte=6, end_byte=11)
42 assert _node_text(node, source) == "world"
43
44 def test_full_source(self):
45 from navegador.ingestion.python import _node_text
46 source = b"foo_bar"
47 node = MockNode("identifier", start_byte=0, end_byte=7)
48 assert _node_text(node, source) == "foo_bar"
49
50 def test_handles_utf8(self):
51 from navegador.ingestion.python import _node_text
52 source = "héllo".encode("utf-8")
53 node = MockNode("identifier", start_byte=0, end_byte=len(source))
54 assert "llo" in _node_text(node, source)
55
56
57 # ── _get_docstring ────────────────────────────────────────────────────────────
58
59 class TestGetDocstring:
60 def test_returns_none_when_no_block(self):
61 from navegador.ingestion.python import _get_docstring
62 node = MockNode("function_definition", children=[
63 MockNode("identifier")
64 ])
65 assert _get_docstring(node, b"def foo(): pass") is None
66
67 def test_returns_none_when_no_expression_stmt(self):
68 from navegador.ingestion.python import _get_docstring
69 block = MockNode("block", children=[
70 MockNode("return_statement")
71 ])
72 fn = MockNode("function_definition", children=[block])
73 assert _get_docstring(fn, b"") is None
74
75 def test_returns_none_when_no_string_in_expr(self):
76 from navegador.ingestion.python import _get_docstring
77 expr_stmt = MockNode("expression_statement", children=[
78 MockNode("assignment")
79 ])
80 block = MockNode("block", children=[expr_stmt])
81 fn = MockNode("function_definition", children=[block])
82 assert _get_docstring(fn, b"") is None
83
84 def test_extracts_docstring(self):
85 from navegador.ingestion.python import _get_docstring
86 source = b'"""My docstring."""'
87 string_node = MockNode("string", start_byte=0, end_byte=len(source))
88 expr_stmt = MockNode("expression_statement", children=[string_node])
89 block = MockNode("block", children=[expr_stmt])
90 fn = MockNode("function_definition", children=[block])
91 result = _get_docstring(fn, source)
92 assert "My docstring." in result
93
94
95 # ── _get_python_language error ─────────────────────────────────────────────────
96
97 class TestGetPythonLanguage:
98 def test_raises_import_error_when_not_installed(self):
99 from navegador.ingestion.python import _get_python_language
100 with patch.dict("sys.modules", {"tree_sitter_python": None, "tree_sitter": None}):
101 with pytest.raises(ImportError, match="tree-sitter-python"):
102 _get_python_language()
103
104
105 # ── PythonParser with mocked parser ──────────────────────────────────────────
106
107 class TestPythonParserHandlers:
108 def _make_parser(self):
109 """Create PythonParser bypassing tree-sitter init."""
110 from navegador.ingestion.python import PythonParser
111 with patch("navegador.ingestion.python._get_parser") as mock_get:
112 mock_get.return_value = MagicMock()
113 parser = PythonParser()
114 return parser
115
116 def test_handle_import(self):
117 parser = self._make_parser()
118 store = _make_store()
119 source = b"import os.path"
120
121 dotted = _text_node(b"os.path", "dotted_name")
122 import_node = MockNode("import_statement", children=[dotted],
123 start_point=(0, 0))
124 stats = {"functions": 0, "classes": 0, "edges": 0}
125 parser._handle_import(import_node, source, "app.py", store, stats)
126 store.create_node.assert_called_once()
127 store.create_edge.assert_called_once()
128 assert stats["edges"] == 1
129
130 def test_handle_import_no_dotted_name(self):
131 parser = self._make_parser()
132 store = _make_store()
133 import_node = MockNode("import_statement", children=[
134 MockNode("keyword", b"import")
135 ], start_point=(0, 0))
136 stats = {"functions": 0, "classes": 0, "edges": 0}
137 parser._handle_import(import_node, b"import x", "app.py", store, stats)
138 store.create_node.assert_not_called()
139
140 def test_handle_class(self):
141 parser = self._make_parser()
142 store = _make_store()
143 source = b"class MyClass: pass"
144 name_node = _text_node(b"MyClass")
145 class_node = MockNode("class_definition",
146 children=[name_node],
147 start_point=(0, 0), end_point=(0, 18))
148 stats = {"functions": 0, "classes": 0, "edges": 0}
149 parser._handle_class(class_node, source, "app.py", store, stats)
150 assert stats["classes"] == 1
151 assert stats["edges"] == 1
152 store.create_node.assert_called()
153
154 def test_handle_class_no_identifier(self):
155 parser = self._make_parser()
156 store = _make_store()
157 class_node = MockNode("class_definition", children=[
158 MockNode("keyword", b"class")
159 ], start_point=(0, 0), end_point=(0, 0))
160 stats = {"functions": 0, "classes": 0, "edges": 0}
161 parser._handle_class(class_node, b"class: pass", "app.py", store, stats)
162 assert stats["classes"] == 0
163
164 def test_handle_class_with_inheritance(self):
165 parser = self._make_parser()
166 store = _make_store()
167 source = b"class Child(Parent): pass"
168 name_node = _text_node(b"Child")
169 parent_id = _text_node(b"Parent")
170 arg_list = MockNode("argument_list", children=[parent_id])
171 class_node = MockNode("class_definition",
172 children=[name_node, arg_list],
173 start_point=(0, 0), end_point=(0, 24))
174 stats = {"functions": 0, "classes": 0, "edges": 0}
175 parser._handle_class(class_node, source, "app.py", store, stats)
176 # Should create class node + CONTAINS edge + INHERITS edge
177 assert stats["edges"] == 2
178
179 def test_handle_function(self):
180 parser = self._make_parser()
181 store = _make_store()
182 source = b"def foo(): pass"
183 name_node = _text_node(b"foo")
184 fn_node = MockNode("function_definition", children=[name_node],
185 start_point=(0, 0), end_point=(0, 14))
186 stats = {"functions": 0, "classes": 0, "edges": 0}
187 parser._handle_function(fn_node, source, "app.py", store, stats, class_name=None)
188 assert stats["functions"] == 1
189 assert stats["edges"] == 1
190 store.create_node.assert_called_once()
191 label = store.create_node.call_args[0][0]
192 assert label == NodeLabel.Function
193
194 def test_handle_method(self):
195 parser = self._make_parser()
196 store = _make_store()
197 source = b"def my_method(self): pass"
198 name_node = _text_node(b"my_method")
199 fn_node = MockNode("function_definition", children=[name_node],
200 start_point=(0, 0), end_point=(0, 24))
201 stats = {"functions": 0, "classes": 0, "edges": 0}
202 parser._handle_function(fn_node, source, "app.py", store, stats, class_name="MyClass")
203 label = store.create_node.call_args[0][0]
204 assert label == NodeLabel.Method
205
206 def test_handle_function_no_identifier(self):
207 parser = self._make_parser()
208 store = _make_store()
209 fn_node = MockNode("function_definition", children=[
210 MockNode("keyword", b"def")
211 ], start_point=(0, 0), end_point=(0, 0))
212 stats = {"functions": 0, "classes": 0, "edges": 0}
213 parser._handle_function(fn_node, b"def", "app.py", store, stats, class_name=None)
214 assert stats["functions"] == 0
215
216 def test_extract_calls(self):
217 parser = self._make_parser()
218 store = _make_store()
219 source = b"def foo():\n bar()\n"
220
221 callee = _text_node(b"bar")
222 call_node = MockNode("call", children=[callee])
223 block = MockNode("block", children=[call_node])
224 fn_node = MockNode("function_definition", children=[block])
225
226 stats = {"functions": 0, "classes": 0, "edges": 0}
227 parser._extract_calls(fn_node, source, "app.py", "foo", NodeLabel.Function, store, stats)
228 store.create_edge.assert_called_once()
229 assert stats["edges"] == 1
230
231 def test_extract_calls_no_block(self):
232 parser = self._make_parser()
233 store = _make_store()
234 fn_node = MockNode("function_definition", children=[])
235
M tests/test_schema.py
+86 -14
--- tests/test_schema.py
+++ tests/test_schema.py
@@ -1,14 +1,86 @@
1
-from navegador.graph.schema import EdgeType, NodeLabel
2
-
3
-
4
-def test_node_labels():
5
- assert NodeLabel.File == "File"
6
- assert NodeLabel.Function == "Function"
7
- assert NodeLabel.Class == "Class"
8
-
9
-
10
-def test_edge_types():
11
- assert EdgeType.CALLS == "CALLS"
12
- assert EdgeType.IMPORTS == "IMPORTS"
13
- assert EdgeType.CONTAINS == "CONTAINS"
14
- assert EdgeType.INHERITS == "INHERITS"
1
+"""Tests for graph schema — node labels, edge types, and node properties."""
2
+
3
+from navegador.graph.schema import NODE_PROPS, EdgeType, NodeLabel
4
+
5
+
6
+class TestNodeLabel:
7
+ def test_code_labels(self):
8
+ assert NodeLabel.Repository == "Repository"
9
+ assert NodeLabel.File == "File"
10
+ assert NodeLabel.Module == "Module"
11
+ assert NodeLabel.Class == "Class"
12
+ assert NodeLabel.Function == "Function"
13
+ assert NodeLabel.Method == "Method"
14
+ assert NodeLabel.Variable == "Variable"
15
+ assert NodeLabel.Import == "Import"
16
+ assert NodeLabel.Decorator == "Decorator"
17
+
18
+ def test_knowledge_labels(self):
19
+ assert NodeLabel.Domain == "Domain"
20
+ assert NodeLabel.Concept == "Concept"
21
+ assert NodeLabel.Rule == "Rule"
22
+ assert NodeLabel.Decision == "Decision"
23
+ assert NodeLabel.WikiPage == "WikiPage"
24
+ assert NodeLabel.Person == "Person"
25
+
26
+ def test_is_str(self):
27
+ assert isinstance(NodeLabel.Function, str)
28
+
29
+ def test_total_count(self):
30
+ assert len(set(NodeLabel)) == 15 # 9 code + 6 knowledge
31
+
32
+
33
+class TestEdgeType:
34
+ def test_code_edges(self):
35
+ assert EdgeType.CONTAINS == "CONTAINS"
36
+ assert EdgeType.DEFINES == "DEFINES"
37
+ assert EdgeType.IMPORTS == "IMPORTS"
38
+ assert EdgeType.DEPENDS_ON == "DEPENDS_ON"
39
+ assert EdgeType.CALLS == "CALLS"
40
+ assert EdgeType.REFERENCES == "REFERENCES"
41
+ assert EdgeType.INHERITS == "INHERITS"
42
+ assert EdgeType.IMPLEMENTS == "IMPLEMENTS"
43
+ assert EdgeType.DECORATES == "DECORATES"
44
+
45
+ def test_knowledge_edges(self):
46
+ assert EdgeType.BELONGS_TO == "BELONGS_TO"
47
+ assert EdgeType.RELATED_TO == "RELATED_TO"
48
+ assert EdgeType.GOVERNS == "GOVERNS"
49
+ assert EdgeType.DOCUMENTS == "DOCUMENTS"
50
+ assert EdgeType.ANNOTATES == "ANNOTATES"
51
+ assert EdgeType.ASSIGNED_TO == "ASSIGNED_TO"
52
+ assert EdgeType.DECIDED_BY == "DECIDED_BY"
53
+
54
+ def test_total_count(self):
55
+ assert len(set(EdgeType)) == 16 # 9 code + 7 knowledge
56
+
57
+
58
+class TestNodeProps:
59
+ def test_all_labels_have_props(self):
60
+ for label in NodeLabel:
61
+ assert label in NODE_PROPS, f"Missing NODE_PROPS entry for {label}"
62
+
63
+ def test_function_props(self):
64
+ assert "signature" in NODE_PROPS[NodeLabel.Function]
65
+ assert "docstring" in NODE_PROPS[NodeLabel.Function]
66
+
67
+ def test_concept_props(self):
68
+ assert "description" in NODE_PROPS[NodeLabel.Concept]
69
+ assert "domain" in NODE_PROPS[NodeLabel.Concept]
70
+
71
+ def test_rule_props(self):
72
+ assert "severity" in NODE_PROPS[NodeLabel.Rule]
73
+ assert "rationale" in NODE_PROPS[NodeLabel.Rule]
74
+
75
+ def test_decision_props(self):
76
+ assert "status" in NODE_PROPS[NodeLabel.Decision]
77
+ assert "rationale" in NODE_PROPS[NodeLabel.Decision]
78
+ assert "alternatives" in NODE_PROPS[NodeLabel.Decision]
79
+
80
+ def test_person_props(self):
81
+ assert "role" in NODE_PROPS[NodeLabel.Person]
82
+ assert "email" in NODE_PROPS[NodeLabel.Person]
83
+
84
+ def test_wiki_props(self):
85
+ assert "source" in NODE_PROPS[NodeLabel.WikiPage]
86
+ assert "content" in NODE_PROPS[NodeLabel.WikiPage]
1587
--- tests/test_schema.py
+++ tests/test_schema.py
@@ -1,14 +1,86 @@
1 from navegador.graph.schema import EdgeType, NodeLabel
2
3
4 def test_node_labels():
5 assert NodeLabel.File == "File"
6 assert NodeLabel.Function == "Function"
7 assert NodeLabel.Class == "Class"
8
9
10 def test_edge_types():
11 assert EdgeType.CALLS == "CALLS"
12 assert EdgeType.IMPORTS == "IMPORTS"
13 assert EdgeType.CONTAINS == "CONTAINS"
14 assert EdgeType.INHERITS == "INHERITS"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
15
--- tests/test_schema.py
+++ tests/test_schema.py
@@ -1,14 +1,86 @@
1 """Tests for graph schema — node labels, edge types, and node properties."""
2
3 from navegador.graph.schema import NODE_PROPS, EdgeType, NodeLabel
4
5
6 class TestNodeLabel:
7 def test_code_labels(self):
8 assert NodeLabel.Repository == "Repository"
9 assert NodeLabel.File == "File"
10 assert NodeLabel.Module == "Module"
11 assert NodeLabel.Class == "Class"
12 assert NodeLabel.Function == "Function"
13 assert NodeLabel.Method == "Method"
14 assert NodeLabel.Variable == "Variable"
15 assert NodeLabel.Import == "Import"
16 assert NodeLabel.Decorator == "Decorator"
17
18 def test_knowledge_labels(self):
19 assert NodeLabel.Domain == "Domain"
20 assert NodeLabel.Concept == "Concept"
21 assert NodeLabel.Rule == "Rule"
22 assert NodeLabel.Decision == "Decision"
23 assert NodeLabel.WikiPage == "WikiPage"
24 assert NodeLabel.Person == "Person"
25
26 def test_is_str(self):
27 assert isinstance(NodeLabel.Function, str)
28
29 def test_total_count(self):
30 assert len(set(NodeLabel)) == 15 # 9 code + 6 knowledge
31
32
33 class TestEdgeType:
34 def test_code_edges(self):
35 assert EdgeType.CONTAINS == "CONTAINS"
36 assert EdgeType.DEFINES == "DEFINES"
37 assert EdgeType.IMPORTS == "IMPORTS"
38 assert EdgeType.DEPENDS_ON == "DEPENDS_ON"
39 assert EdgeType.CALLS == "CALLS"
40 assert EdgeType.REFERENCES == "REFERENCES"
41 assert EdgeType.INHERITS == "INHERITS"
42 assert EdgeType.IMPLEMENTS == "IMPLEMENTS"
43 assert EdgeType.DECORATES == "DECORATES"
44
45 def test_knowledge_edges(self):
46 assert EdgeType.BELONGS_TO == "BELONGS_TO"
47 assert EdgeType.RELATED_TO == "RELATED_TO"
48 assert EdgeType.GOVERNS == "GOVERNS"
49 assert EdgeType.DOCUMENTS == "DOCUMENTS"
50 assert EdgeType.ANNOTATES == "ANNOTATES"
51 assert EdgeType.ASSIGNED_TO == "ASSIGNED_TO"
52 assert EdgeType.DECIDED_BY == "DECIDED_BY"
53
54 def test_total_count(self):
55 assert len(set(EdgeType)) == 16 # 9 code + 7 knowledge
56
57
58 class TestNodeProps:
59 def test_all_labels_have_props(self):
60 for label in NodeLabel:
61 assert label in NODE_PROPS, f"Missing NODE_PROPS entry for {label}"
62
63 def test_function_props(self):
64 assert "signature" in NODE_PROPS[NodeLabel.Function]
65 assert "docstring" in NODE_PROPS[NodeLabel.Function]
66
67 def test_concept_props(self):
68 assert "description" in NODE_PROPS[NodeLabel.Concept]
69 assert "domain" in NODE_PROPS[NodeLabel.Concept]
70
71 def test_rule_props(self):
72 assert "severity" in NODE_PROPS[NodeLabel.Rule]
73 assert "rationale" in NODE_PROPS[NodeLabel.Rule]
74
75 def test_decision_props(self):
76 assert "status" in NODE_PROPS[NodeLabel.Decision]
77 assert "rationale" in NODE_PROPS[NodeLabel.Decision]
78 assert "alternatives" in NODE_PROPS[NodeLabel.Decision]
79
80 def test_person_props(self):
81 assert "role" in NODE_PROPS[NodeLabel.Person]
82 assert "email" in NODE_PROPS[NodeLabel.Person]
83
84 def test_wiki_props(self):
85 assert "source" in NODE_PROPS[NodeLabel.WikiPage]
86 assert "content" in NODE_PROPS[NodeLabel.WikiPage]
87

Keyboard Shortcuts

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