|
2602093…
|
lmata
|
1 |
# Copyright CONFLICT LLC 2026 (weareconflict.com) |
|
2602093…
|
lmata
|
2 |
""" |
|
2602093…
|
lmata
|
3 |
Tests for navegador.explorer — ExplorerServer, API endpoints, HTML template, |
|
2602093…
|
lmata
|
4 |
and the CLI `explore` command. |
|
2602093…
|
lmata
|
5 |
""" |
|
2602093…
|
lmata
|
6 |
|
|
2602093…
|
lmata
|
7 |
from __future__ import annotations |
|
2602093…
|
lmata
|
8 |
|
|
2602093…
|
lmata
|
9 |
import json |
|
2602093…
|
lmata
|
10 |
import urllib.error |
|
2602093…
|
lmata
|
11 |
import urllib.request |
|
2602093…
|
lmata
|
12 |
from unittest.mock import MagicMock, patch |
|
2602093…
|
lmata
|
13 |
|
|
2602093…
|
lmata
|
14 |
import pytest |
|
2602093…
|
lmata
|
15 |
from click.testing import CliRunner |
|
2602093…
|
lmata
|
16 |
|
|
2602093…
|
lmata
|
17 |
from navegador.cli.commands import main |
|
2602093…
|
lmata
|
18 |
from navegador.explorer import ExplorerServer |
|
2602093…
|
lmata
|
19 |
from navegador.explorer.templates import HTML_TEMPLATE |
|
2602093…
|
lmata
|
20 |
|
|
2602093…
|
lmata
|
21 |
# ── Helpers ──────────────────────────────────────────────────────────────── |
|
2602093…
|
lmata
|
22 |
|
|
2602093…
|
lmata
|
23 |
|
|
2602093…
|
lmata
|
24 |
def _mock_store( |
|
2602093…
|
lmata
|
25 |
*, |
|
2602093…
|
lmata
|
26 |
nodes: list | None = None, |
|
2602093…
|
lmata
|
27 |
edges: list | None = None, |
|
2602093…
|
lmata
|
28 |
node_count: int = 3, |
|
2602093…
|
lmata
|
29 |
edge_count: int = 2, |
|
2602093…
|
lmata
|
30 |
): |
|
2602093…
|
lmata
|
31 |
"""Return a minimal GraphStore mock suitable for explorer tests.""" |
|
2602093…
|
lmata
|
32 |
store = MagicMock() |
|
2602093…
|
lmata
|
33 |
store.node_count.return_value = node_count |
|
2602093…
|
lmata
|
34 |
store.edge_count.return_value = edge_count |
|
2602093…
|
lmata
|
35 |
|
|
2602093…
|
lmata
|
36 |
# Each query() call returns a result-set mock. We cycle through prebuilt |
|
2602093…
|
lmata
|
37 |
# responses so different Cypher patterns get appropriate data. |
|
2602093…
|
lmata
|
38 |
_node_rows = nodes or [] |
|
2602093…
|
lmata
|
39 |
_edge_rows = edges or [] |
|
2602093…
|
lmata
|
40 |
|
|
2602093…
|
lmata
|
41 |
def _query_side_effect(cypher: str, params=None): |
|
2602093…
|
lmata
|
42 |
result = MagicMock() |
|
2602093…
|
lmata
|
43 |
cypher_lower = cypher.lower() |
|
2602093…
|
lmata
|
44 |
if "match (a)-[r]->(b)" in cypher_lower: |
|
2602093…
|
lmata
|
45 |
result.result_set = _edge_rows |
|
2602093…
|
lmata
|
46 |
elif "match (n)-[r]->(nb)" in cypher_lower or "match (nb)-[r]->(n)" in cypher_lower: |
|
2602093…
|
lmata
|
47 |
result.result_set = [] |
|
2602093…
|
lmata
|
48 |
elif "match (n) where n.name" in cypher_lower and "properties" in cypher_lower: |
|
2602093…
|
lmata
|
49 |
# node detail: single node row |
|
2602093…
|
lmata
|
50 |
result.result_set = [["Function", {"name": "foo", "file_path": "app.py"}]] |
|
2602093…
|
lmata
|
51 |
elif "match (n)" in cypher_lower and "tolow" in cypher_lower: |
|
2602093…
|
lmata
|
52 |
result.result_set = [ |
|
2602093…
|
lmata
|
53 |
["Function", "foo", "app.py", ""], |
|
2602093…
|
lmata
|
54 |
] |
|
2602093…
|
lmata
|
55 |
elif "labels(n)" in cypher_lower and "count" in cypher_lower: |
|
2602093…
|
lmata
|
56 |
result.result_set = [["Function", 2], ["Class", 1]] |
|
2602093…
|
lmata
|
57 |
elif "type(r)" in cypher_lower and "count" in cypher_lower: |
|
2602093…
|
lmata
|
58 |
result.result_set = [["CALLS", 2]] |
|
2602093…
|
lmata
|
59 |
else: |
|
2602093…
|
lmata
|
60 |
result.result_set = _node_rows |
|
2602093…
|
lmata
|
61 |
return result |
|
2602093…
|
lmata
|
62 |
|
|
2602093…
|
lmata
|
63 |
store.query.side_effect = _query_side_effect |
|
2602093…
|
lmata
|
64 |
return store |
|
2602093…
|
lmata
|
65 |
|
|
2602093…
|
lmata
|
66 |
|
|
2602093…
|
lmata
|
67 |
def _free_port() -> int: |
|
2602093…
|
lmata
|
68 |
"""Return an available TCP port on localhost.""" |
|
2602093…
|
lmata
|
69 |
import socket |
|
2602093…
|
lmata
|
70 |
with socket.socket() as s: |
|
2602093…
|
lmata
|
71 |
s.bind(("127.0.0.1", 0)) |
|
2602093…
|
lmata
|
72 |
return s.getsockname()[1] |
|
2602093…
|
lmata
|
73 |
|
|
2602093…
|
lmata
|
74 |
|
|
2602093…
|
lmata
|
75 |
def _fetch(url: str, timeout: float = 5.0) -> tuple[int, str]: |
|
2602093…
|
lmata
|
76 |
"""GET *url* and return (status_code, response_body_str).""" |
|
2602093…
|
lmata
|
77 |
with urllib.request.urlopen(url, timeout=timeout) as resp: |
|
2602093…
|
lmata
|
78 |
return resp.status, resp.read().decode() |
|
2602093…
|
lmata
|
79 |
|
|
2602093…
|
lmata
|
80 |
|
|
2602093…
|
lmata
|
81 |
def _fetch_json(url: str, timeout: float = 5.0) -> tuple[int, dict | list]: |
|
2602093…
|
lmata
|
82 |
status, body = _fetch(url, timeout) |
|
2602093…
|
lmata
|
83 |
return status, json.loads(body) |
|
2602093…
|
lmata
|
84 |
|
|
2602093…
|
lmata
|
85 |
|
|
2602093…
|
lmata
|
86 |
# ── ExplorerServer creation ──────────────────────────────────────────────── |
|
2602093…
|
lmata
|
87 |
|
|
2602093…
|
lmata
|
88 |
|
|
2602093…
|
lmata
|
89 |
class TestExplorerServerCreation: |
|
2602093…
|
lmata
|
90 |
def test_default_host_and_port(self): |
|
2602093…
|
lmata
|
91 |
store = _mock_store() |
|
2602093…
|
lmata
|
92 |
server = ExplorerServer(store) |
|
2602093…
|
lmata
|
93 |
assert server.host == "127.0.0.1" |
|
2602093…
|
lmata
|
94 |
assert server.port == 8080 |
|
2602093…
|
lmata
|
95 |
assert server.store is store |
|
2602093…
|
lmata
|
96 |
|
|
2602093…
|
lmata
|
97 |
def test_custom_host_and_port(self): |
|
2602093…
|
lmata
|
98 |
store = _mock_store() |
|
2602093…
|
lmata
|
99 |
server = ExplorerServer(store, host="0.0.0.0", port=9999) |
|
2602093…
|
lmata
|
100 |
assert server.host == "0.0.0.0" |
|
2602093…
|
lmata
|
101 |
assert server.port == 9999 |
|
2602093…
|
lmata
|
102 |
|
|
2602093…
|
lmata
|
103 |
def test_url_property(self): |
|
2602093…
|
lmata
|
104 |
server = ExplorerServer(_mock_store(), host="127.0.0.1", port=8080) |
|
2602093…
|
lmata
|
105 |
assert server.url == "http://127.0.0.1:8080" |
|
2602093…
|
lmata
|
106 |
|
|
2602093…
|
lmata
|
107 |
def test_not_running_by_default(self): |
|
2602093…
|
lmata
|
108 |
server = ExplorerServer(_mock_store(), port=_free_port()) |
|
2602093…
|
lmata
|
109 |
assert server._server is None |
|
2602093…
|
lmata
|
110 |
assert server._thread is None |
|
2602093…
|
lmata
|
111 |
|
|
2602093…
|
lmata
|
112 |
def test_double_start_raises(self): |
|
2602093…
|
lmata
|
113 |
port = _free_port() |
|
2602093…
|
lmata
|
114 |
server = ExplorerServer(_mock_store(), port=port) |
|
2602093…
|
lmata
|
115 |
server.start() |
|
2602093…
|
lmata
|
116 |
try: |
|
2602093…
|
lmata
|
117 |
with pytest.raises(RuntimeError, match="already running"): |
|
2602093…
|
lmata
|
118 |
server.start() |
|
2602093…
|
lmata
|
119 |
finally: |
|
2602093…
|
lmata
|
120 |
server.stop() |
|
2602093…
|
lmata
|
121 |
|
|
2602093…
|
lmata
|
122 |
def test_stop_when_not_started_is_noop(self): |
|
2602093…
|
lmata
|
123 |
server = ExplorerServer(_mock_store(), port=_free_port()) |
|
2602093…
|
lmata
|
124 |
server.stop() # should not raise |
|
2602093…
|
lmata
|
125 |
|
|
2602093…
|
lmata
|
126 |
def test_context_manager(self): |
|
2602093…
|
lmata
|
127 |
port = _free_port() |
|
2602093…
|
lmata
|
128 |
store = _mock_store() |
|
2602093…
|
lmata
|
129 |
with ExplorerServer(store, port=port) as srv: |
|
2602093…
|
lmata
|
130 |
assert srv._server is not None |
|
2602093…
|
lmata
|
131 |
assert srv._server is None |
|
2602093…
|
lmata
|
132 |
|
|
2602093…
|
lmata
|
133 |
|
|
2602093…
|
lmata
|
134 |
# ── Start / stop lifecycle ───────────────────────────────────────────────── |
|
2602093…
|
lmata
|
135 |
|
|
2602093…
|
lmata
|
136 |
|
|
2602093…
|
lmata
|
137 |
class TestExplorerServerLifecycle: |
|
2602093…
|
lmata
|
138 |
def test_start_makes_server_accessible(self): |
|
2602093…
|
lmata
|
139 |
port = _free_port() |
|
2602093…
|
lmata
|
140 |
server = ExplorerServer(_mock_store(), port=port) |
|
2602093…
|
lmata
|
141 |
server.start() |
|
2602093…
|
lmata
|
142 |
try: |
|
2602093…
|
lmata
|
143 |
status, _ = _fetch(f"http://127.0.0.1:{port}/") |
|
2602093…
|
lmata
|
144 |
assert status == 200 |
|
2602093…
|
lmata
|
145 |
finally: |
|
2602093…
|
lmata
|
146 |
server.stop() |
|
2602093…
|
lmata
|
147 |
|
|
2602093…
|
lmata
|
148 |
def test_stop_takes_server_offline(self): |
|
2602093…
|
lmata
|
149 |
port = _free_port() |
|
2602093…
|
lmata
|
150 |
server = ExplorerServer(_mock_store(), port=port) |
|
2602093…
|
lmata
|
151 |
server.start() |
|
2602093…
|
lmata
|
152 |
server.stop() |
|
2602093…
|
lmata
|
153 |
with pytest.raises(Exception): |
|
2602093…
|
lmata
|
154 |
_fetch(f"http://127.0.0.1:{port}/", timeout=1.0) |
|
2602093…
|
lmata
|
155 |
|
|
2602093…
|
lmata
|
156 |
def test_thread_is_daemon(self): |
|
2602093…
|
lmata
|
157 |
port = _free_port() |
|
2602093…
|
lmata
|
158 |
server = ExplorerServer(_mock_store(), port=port) |
|
2602093…
|
lmata
|
159 |
server.start() |
|
2602093…
|
lmata
|
160 |
try: |
|
2602093…
|
lmata
|
161 |
assert server._thread is not None |
|
2602093…
|
lmata
|
162 |
assert server._thread.daemon is True |
|
2602093…
|
lmata
|
163 |
finally: |
|
2602093…
|
lmata
|
164 |
server.stop() |
|
2602093…
|
lmata
|
165 |
|
|
2602093…
|
lmata
|
166 |
|
|
2602093…
|
lmata
|
167 |
# ── API endpoint: GET / ──────────────────────────────────────────────────── |
|
2602093…
|
lmata
|
168 |
|
|
2602093…
|
lmata
|
169 |
|
|
2602093…
|
lmata
|
170 |
class TestRootEndpoint: |
|
2602093…
|
lmata
|
171 |
def test_returns_html(self): |
|
2602093…
|
lmata
|
172 |
port = _free_port() |
|
2602093…
|
lmata
|
173 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
174 |
status, body = _fetch(f"http://127.0.0.1:{port}/") |
|
2602093…
|
lmata
|
175 |
assert status == 200 |
|
2602093…
|
lmata
|
176 |
assert "<!DOCTYPE html>" in body or "<!doctype html>" in body.lower() |
|
2602093…
|
lmata
|
177 |
|
|
2602093…
|
lmata
|
178 |
def test_html_contains_canvas(self): |
|
2602093…
|
lmata
|
179 |
port = _free_port() |
|
2602093…
|
lmata
|
180 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
181 |
_, body = _fetch(f"http://127.0.0.1:{port}/") |
|
2602093…
|
lmata
|
182 |
assert "graph-canvas" in body |
|
2602093…
|
lmata
|
183 |
|
|
2602093…
|
lmata
|
184 |
def test_html_contains_search_box(self): |
|
2602093…
|
lmata
|
185 |
port = _free_port() |
|
2602093…
|
lmata
|
186 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
187 |
_, body = _fetch(f"http://127.0.0.1:{port}/") |
|
2602093…
|
lmata
|
188 |
assert "search-box" in body |
|
2602093…
|
lmata
|
189 |
|
|
2602093…
|
lmata
|
190 |
def test_html_contains_api_calls(self): |
|
2602093…
|
lmata
|
191 |
port = _free_port() |
|
2602093…
|
lmata
|
192 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
193 |
_, body = _fetch(f"http://127.0.0.1:{port}/") |
|
2602093…
|
lmata
|
194 |
assert "/api/graph" in body |
|
2602093…
|
lmata
|
195 |
|
|
2602093…
|
lmata
|
196 |
|
|
2602093…
|
lmata
|
197 |
# ── API endpoint: GET /api/graph ────────────────────────────────────────── |
|
2602093…
|
lmata
|
198 |
|
|
2602093…
|
lmata
|
199 |
|
|
2602093…
|
lmata
|
200 |
class TestGraphEndpoint: |
|
2602093…
|
lmata
|
201 |
def _make_node_rows(self): |
|
2602093…
|
lmata
|
202 |
# Rows returned for the full-node Cypher query |
|
2602093…
|
lmata
|
203 |
return [ |
|
2602093…
|
lmata
|
204 |
[1, "Function", "foo", {"name": "foo", "file_path": "app.py"}], |
|
2602093…
|
lmata
|
205 |
[2, "Class", "Bar", {"name": "Bar", "file_path": "app.py"}], |
|
2602093…
|
lmata
|
206 |
] |
|
2602093…
|
lmata
|
207 |
|
|
2602093…
|
lmata
|
208 |
def test_returns_nodes_and_edges_keys(self): |
|
2602093…
|
lmata
|
209 |
port = _free_port() |
|
2602093…
|
lmata
|
210 |
store = _mock_store(nodes=self._make_node_rows(), edges=[]) |
|
2602093…
|
lmata
|
211 |
with ExplorerServer(store, port=port): |
|
2602093…
|
lmata
|
212 |
status, data = _fetch_json(f"http://127.0.0.1:{port}/api/graph") |
|
2602093…
|
lmata
|
213 |
assert status == 200 |
|
2602093…
|
lmata
|
214 |
assert "nodes" in data |
|
2602093…
|
lmata
|
215 |
assert "edges" in data |
|
2602093…
|
lmata
|
216 |
|
|
2602093…
|
lmata
|
217 |
def test_nodes_have_required_fields(self): |
|
2602093…
|
lmata
|
218 |
port = _free_port() |
|
2602093…
|
lmata
|
219 |
store = _mock_store(nodes=self._make_node_rows()) |
|
2602093…
|
lmata
|
220 |
with ExplorerServer(store, port=port): |
|
2602093…
|
lmata
|
221 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/graph") |
|
2602093…
|
lmata
|
222 |
for node in data["nodes"]: |
|
2602093…
|
lmata
|
223 |
assert "id" in node |
|
2602093…
|
lmata
|
224 |
assert "label" in node |
|
2602093…
|
lmata
|
225 |
assert "name" in node |
|
2602093…
|
lmata
|
226 |
|
|
2602093…
|
lmata
|
227 |
def test_empty_graph(self): |
|
2602093…
|
lmata
|
228 |
port = _free_port() |
|
2602093…
|
lmata
|
229 |
store = _mock_store(nodes=[], edges=[], node_count=0, edge_count=0) |
|
2602093…
|
lmata
|
230 |
with ExplorerServer(store, port=port): |
|
2602093…
|
lmata
|
231 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/graph") |
|
2602093…
|
lmata
|
232 |
assert data["nodes"] == [] |
|
2602093…
|
lmata
|
233 |
assert data["edges"] == [] |
|
2602093…
|
lmata
|
234 |
|
|
2602093…
|
lmata
|
235 |
def test_edges_have_required_fields(self): |
|
2602093…
|
lmata
|
236 |
port = _free_port() |
|
2602093…
|
lmata
|
237 |
edge_rows = [[1, 2, "CALLS"]] |
|
2602093…
|
lmata
|
238 |
store = _mock_store(nodes=self._make_node_rows(), edges=edge_rows) |
|
2602093…
|
lmata
|
239 |
with ExplorerServer(store, port=port): |
|
2602093…
|
lmata
|
240 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/graph") |
|
2602093…
|
lmata
|
241 |
for edge in data["edges"]: |
|
2602093…
|
lmata
|
242 |
assert "source" in edge |
|
2602093…
|
lmata
|
243 |
assert "target" in edge |
|
2602093…
|
lmata
|
244 |
assert "type" in edge |
|
2602093…
|
lmata
|
245 |
|
|
2602093…
|
lmata
|
246 |
|
|
2602093…
|
lmata
|
247 |
# ── API endpoint: GET /api/search ───────────────────────────────────────── |
|
2602093…
|
lmata
|
248 |
|
|
2602093…
|
lmata
|
249 |
|
|
2602093…
|
lmata
|
250 |
class TestSearchEndpoint: |
|
2602093…
|
lmata
|
251 |
def test_returns_nodes_key(self): |
|
2602093…
|
lmata
|
252 |
port = _free_port() |
|
2602093…
|
lmata
|
253 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
254 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/search?q=foo") |
|
2602093…
|
lmata
|
255 |
assert "nodes" in data |
|
2602093…
|
lmata
|
256 |
|
|
2602093…
|
lmata
|
257 |
def test_empty_query_returns_empty(self): |
|
2602093…
|
lmata
|
258 |
port = _free_port() |
|
2602093…
|
lmata
|
259 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
260 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/search?q=") |
|
2602093…
|
lmata
|
261 |
assert data["nodes"] == [] |
|
2602093…
|
lmata
|
262 |
|
|
2602093…
|
lmata
|
263 |
def test_missing_q_returns_empty(self): |
|
2602093…
|
lmata
|
264 |
port = _free_port() |
|
2602093…
|
lmata
|
265 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
266 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/search") |
|
2602093…
|
lmata
|
267 |
assert data["nodes"] == [] |
|
2602093…
|
lmata
|
268 |
|
|
2602093…
|
lmata
|
269 |
def test_result_nodes_have_name_and_label(self): |
|
2602093…
|
lmata
|
270 |
port = _free_port() |
|
2602093…
|
lmata
|
271 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
272 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/search?q=foo") |
|
2602093…
|
lmata
|
273 |
for node in data["nodes"]: |
|
2602093…
|
lmata
|
274 |
assert "name" in node |
|
2602093…
|
lmata
|
275 |
assert "label" in node |
|
2602093…
|
lmata
|
276 |
|
|
2602093…
|
lmata
|
277 |
|
|
2602093…
|
lmata
|
278 |
# ── API endpoint: GET /api/node/<name> ──────────────────────────────────── |
|
2602093…
|
lmata
|
279 |
|
|
2602093…
|
lmata
|
280 |
|
|
2602093…
|
lmata
|
281 |
class TestNodeDetailEndpoint: |
|
2602093…
|
lmata
|
282 |
def test_returns_name_label_props_neighbors(self): |
|
2602093…
|
lmata
|
283 |
port = _free_port() |
|
2602093…
|
lmata
|
284 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
285 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/node/foo") |
|
2602093…
|
lmata
|
286 |
assert "name" in data |
|
2602093…
|
lmata
|
287 |
assert "label" in data |
|
2602093…
|
lmata
|
288 |
assert "props" in data |
|
2602093…
|
lmata
|
289 |
assert "neighbors" in data |
|
2602093…
|
lmata
|
290 |
|
|
2602093…
|
lmata
|
291 |
def test_name_matches_request(self): |
|
2602093…
|
lmata
|
292 |
port = _free_port() |
|
2602093…
|
lmata
|
293 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
294 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/node/foo") |
|
2602093…
|
lmata
|
295 |
assert data["name"] == "foo" |
|
2602093…
|
lmata
|
296 |
|
|
2602093…
|
lmata
|
297 |
def test_url_encoded_name(self): |
|
2602093…
|
lmata
|
298 |
port = _free_port() |
|
2602093…
|
lmata
|
299 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
300 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/node/my%20node") |
|
2602093…
|
lmata
|
301 |
assert data["name"] == "my node" |
|
2602093…
|
lmata
|
302 |
|
|
2602093…
|
lmata
|
303 |
def test_unknown_node_returns_empty_detail(self): |
|
2602093…
|
lmata
|
304 |
port = _free_port() |
|
2602093…
|
lmata
|
305 |
store = _mock_store() |
|
2602093…
|
lmata
|
306 |
# Override query to return empty for the node-detail lookup |
|
2602093…
|
lmata
|
307 |
original_side_effect = store.query.side_effect |
|
2602093…
|
lmata
|
308 |
|
|
2602093…
|
lmata
|
309 |
def _empty_node(cypher, params=None): |
|
2602093…
|
lmata
|
310 |
if "where n.name" in cypher.lower() and "properties" in cypher.lower(): |
|
2602093…
|
lmata
|
311 |
r = MagicMock() |
|
2602093…
|
lmata
|
312 |
r.result_set = [] |
|
2602093…
|
lmata
|
313 |
return r |
|
2602093…
|
lmata
|
314 |
return original_side_effect(cypher, params) |
|
2602093…
|
lmata
|
315 |
|
|
2602093…
|
lmata
|
316 |
store.query.side_effect = _empty_node |
|
2602093…
|
lmata
|
317 |
with ExplorerServer(store, port=port): |
|
2602093…
|
lmata
|
318 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/node/nonexistent") |
|
2602093…
|
lmata
|
319 |
assert data["neighbors"] == [] |
|
2602093…
|
lmata
|
320 |
|
|
2602093…
|
lmata
|
321 |
|
|
2602093…
|
lmata
|
322 |
# ── API endpoint: GET /api/stats ────────────────────────────────────────── |
|
2602093…
|
lmata
|
323 |
|
|
2602093…
|
lmata
|
324 |
|
|
2602093…
|
lmata
|
325 |
class TestStatsEndpoint: |
|
2602093…
|
lmata
|
326 |
def test_returns_nodes_and_edges_counts(self): |
|
2602093…
|
lmata
|
327 |
port = _free_port() |
|
2602093…
|
lmata
|
328 |
with ExplorerServer(_mock_store(node_count=5, edge_count=3), port=port): |
|
2602093…
|
lmata
|
329 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/stats") |
|
2602093…
|
lmata
|
330 |
assert data["nodes"] == 5 |
|
2602093…
|
lmata
|
331 |
assert data["edges"] == 3 |
|
2602093…
|
lmata
|
332 |
|
|
2602093…
|
lmata
|
333 |
def test_returns_node_types_and_edge_types(self): |
|
2602093…
|
lmata
|
334 |
port = _free_port() |
|
2602093…
|
lmata
|
335 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
336 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/stats") |
|
2602093…
|
lmata
|
337 |
assert "node_types" in data |
|
2602093…
|
lmata
|
338 |
assert "edge_types" in data |
|
2602093…
|
lmata
|
339 |
assert isinstance(data["node_types"], dict) |
|
2602093…
|
lmata
|
340 |
assert isinstance(data["edge_types"], dict) |
|
2602093…
|
lmata
|
341 |
|
|
2602093…
|
lmata
|
342 |
def test_node_type_counts_sum(self): |
|
2602093…
|
lmata
|
343 |
port = _free_port() |
|
2602093…
|
lmata
|
344 |
with ExplorerServer(_mock_store(node_count=3), port=port): |
|
2602093…
|
lmata
|
345 |
_, data = _fetch_json(f"http://127.0.0.1:{port}/api/stats") |
|
2602093…
|
lmata
|
346 |
total = sum(data["node_types"].values()) |
|
2602093…
|
lmata
|
347 |
# The mock returns Function:2, Class:1 → total 3 |
|
2602093…
|
lmata
|
348 |
assert total == 3 |
|
2602093…
|
lmata
|
349 |
|
|
2602093…
|
lmata
|
350 |
|
|
2602093…
|
lmata
|
351 |
# ── 404 for unknown routes ───────────────────────────────────────────────── |
|
2602093…
|
lmata
|
352 |
|
|
2602093…
|
lmata
|
353 |
|
|
2602093…
|
lmata
|
354 |
class TestNotFound: |
|
2602093…
|
lmata
|
355 |
def test_unknown_path_returns_404(self): |
|
2602093…
|
lmata
|
356 |
port = _free_port() |
|
2602093…
|
lmata
|
357 |
with ExplorerServer(_mock_store(), port=port): |
|
2602093…
|
lmata
|
358 |
with pytest.raises(urllib.error.HTTPError) as exc_info: |
|
2602093…
|
lmata
|
359 |
urllib.request.urlopen(f"http://127.0.0.1:{port}/api/nonexistent") |
|
2602093…
|
lmata
|
360 |
assert exc_info.value.code == 404 |
|
2602093…
|
lmata
|
361 |
|
|
2602093…
|
lmata
|
362 |
|
|
2602093…
|
lmata
|
363 |
# ── HTML template ───────────────────────────────────────────────────────── |
|
2602093…
|
lmata
|
364 |
|
|
2602093…
|
lmata
|
365 |
|
|
2602093…
|
lmata
|
366 |
class TestHtmlTemplate: |
|
2602093…
|
lmata
|
367 |
def test_is_string(self): |
|
2602093…
|
lmata
|
368 |
assert isinstance(HTML_TEMPLATE, str) |
|
2602093…
|
lmata
|
369 |
|
|
2602093…
|
lmata
|
370 |
def test_contains_doctype(self): |
|
2602093…
|
lmata
|
371 |
assert "<!DOCTYPE html>" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
372 |
|
|
2602093…
|
lmata
|
373 |
def test_contains_canvas(self): |
|
2602093…
|
lmata
|
374 |
assert "graph-canvas" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
375 |
|
|
2602093…
|
lmata
|
376 |
def test_contains_search_box(self): |
|
2602093…
|
lmata
|
377 |
assert "search-box" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
378 |
|
|
2602093…
|
lmata
|
379 |
def test_contains_detail_panel(self): |
|
2602093…
|
lmata
|
380 |
assert "detail-panel" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
381 |
|
|
2602093…
|
lmata
|
382 |
def test_contains_api_graph_fetch(self): |
|
2602093…
|
lmata
|
383 |
assert "/api/graph" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
384 |
|
|
2602093…
|
lmata
|
385 |
def test_contains_api_search_fetch(self): |
|
2602093…
|
lmata
|
386 |
assert "/api/search" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
387 |
|
|
2602093…
|
lmata
|
388 |
def test_contains_api_node_fetch(self): |
|
2602093…
|
lmata
|
389 |
assert "/api/node/" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
390 |
|
|
2602093…
|
lmata
|
391 |
def test_contains_api_stats_fetch(self): |
|
2602093…
|
lmata
|
392 |
assert "/api/stats" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
393 |
|
|
2602093…
|
lmata
|
394 |
def test_no_external_deps(self): |
|
2602093…
|
lmata
|
395 |
"""No CDN or external URLs should appear in the template.""" |
|
2602093…
|
lmata
|
396 |
import re |
|
2602093…
|
lmata
|
397 |
# Look for any http(s):// URLs — internal /api/ paths are fine |
|
2602093…
|
lmata
|
398 |
external = re.findall(r'https?://\S+', HTML_TEMPLATE) |
|
2602093…
|
lmata
|
399 |
assert external == [], f"External URLs found: {external}" |
|
2602093…
|
lmata
|
400 |
|
|
2602093…
|
lmata
|
401 |
def test_contains_force_directed_physics(self): |
|
2602093…
|
lmata
|
402 |
lower = HTML_TEMPLATE.lower() |
|
2602093…
|
lmata
|
403 |
assert "REPEL" in HTML_TEMPLATE or "repulsion" in lower or "force" in lower |
|
2602093…
|
lmata
|
404 |
|
|
2602093…
|
lmata
|
405 |
def test_colors_injected(self): |
|
2602093…
|
lmata
|
406 |
assert "Function" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
407 |
assert "Class" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
408 |
|
|
2602093…
|
lmata
|
409 |
def test_self_contained_script_tag(self): |
|
2602093…
|
lmata
|
410 |
assert "<script>" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
411 |
|
|
2602093…
|
lmata
|
412 |
def test_self_contained_style_tag(self): |
|
2602093…
|
lmata
|
413 |
assert "<style>" in HTML_TEMPLATE |
|
2602093…
|
lmata
|
414 |
|
|
2602093…
|
lmata
|
415 |
|
|
2602093…
|
lmata
|
416 |
# ── CLI command: navegador explore ──────────────────────────────────────── |
|
2602093…
|
lmata
|
417 |
|
|
2602093…
|
lmata
|
418 |
|
|
2602093…
|
lmata
|
419 |
class TestExploreCLI: |
|
2602093…
|
lmata
|
420 |
def test_help_text(self): |
|
2602093…
|
lmata
|
421 |
runner = CliRunner() |
|
2602093…
|
lmata
|
422 |
result = runner.invoke(main, ["explore", "--help"]) |
|
2602093…
|
lmata
|
423 |
assert result.exit_code == 0 |
|
2602093…
|
lmata
|
424 |
assert "explore" in result.output.lower() or "graph" in result.output.lower() |
|
2602093…
|
lmata
|
425 |
|
|
2602093…
|
lmata
|
426 |
def test_explore_command_registered(self): |
|
2602093…
|
lmata
|
427 |
"""Verify the explore command is registered under the main group.""" |
|
2602093…
|
lmata
|
428 |
from navegador.cli.commands import main as cli_main |
|
2602093…
|
lmata
|
429 |
assert "explore" in cli_main.commands |
|
2602093…
|
lmata
|
430 |
|
|
2602093…
|
lmata
|
431 |
def test_explore_starts_and_stops(self): |
|
2602093…
|
lmata
|
432 |
"""CLI explore should start ExplorerServer and stop cleanly on KeyboardInterrupt.""" |
|
2602093…
|
lmata
|
433 |
runner = CliRunner() |
|
2602093…
|
lmata
|
434 |
port = _free_port() |
|
2602093…
|
lmata
|
435 |
|
|
2602093…
|
lmata
|
436 |
mock_srv = MagicMock() |
|
2602093…
|
lmata
|
437 |
mock_srv.url = f"http://127.0.0.1:{port}" |
|
2602093…
|
lmata
|
438 |
|
|
2602093…
|
lmata
|
439 |
call_count = [0] |
|
2602093…
|
lmata
|
440 |
|
|
2602093…
|
lmata
|
441 |
def _fake_sleep(seconds): |
|
2602093…
|
lmata
|
442 |
# Let the first call (browser delay) pass, raise on second (main loop) |
|
2602093…
|
lmata
|
443 |
call_count[0] += 1 |
|
2602093…
|
lmata
|
444 |
if call_count[0] >= 2: |
|
2602093…
|
lmata
|
445 |
raise KeyboardInterrupt |
|
2602093…
|
lmata
|
446 |
|
|
2602093…
|
lmata
|
447 |
# The explore command does local imports, so patch at the source modules. |
|
2602093…
|
lmata
|
448 |
with ( |
|
2602093…
|
lmata
|
449 |
patch("navegador.explorer.ExplorerServer", return_value=mock_srv), |
|
2602093…
|
lmata
|
450 |
patch("navegador.cli.commands._get_store", return_value=MagicMock()), |
|
2602093…
|
lmata
|
451 |
patch("time.sleep", side_effect=_fake_sleep), |
|
2602093…
|
lmata
|
452 |
patch("webbrowser.open"), |
|
2602093…
|
lmata
|
453 |
): |
|
2602093…
|
lmata
|
454 |
result = runner.invoke(main, ["explore", "--port", str(port)]) |
|
2602093…
|
lmata
|
455 |
|
|
2602093…
|
lmata
|
456 |
mock_srv.start.assert_called_once() |
|
2602093…
|
lmata
|
457 |
mock_srv.stop.assert_called_once() |
|
2602093…
|
lmata
|
458 |
assert result.exit_code == 0 |
|
2602093…
|
lmata
|
459 |
|
|
2602093…
|
lmata
|
460 |
def test_explore_no_browser_flag(self): |
|
2602093…
|
lmata
|
461 |
"""--no-browser should skip webbrowser.open.""" |
|
2602093…
|
lmata
|
462 |
runner = CliRunner() |
|
2602093…
|
lmata
|
463 |
port = _free_port() |
|
2602093…
|
lmata
|
464 |
|
|
2602093…
|
lmata
|
465 |
mock_srv = MagicMock() |
|
2602093…
|
lmata
|
466 |
mock_srv.url = f"http://127.0.0.1:{port}" |
|
2602093…
|
lmata
|
467 |
|
|
2602093…
|
lmata
|
468 |
def _fake_sleep(seconds): |
|
2602093…
|
lmata
|
469 |
raise KeyboardInterrupt |
|
2602093…
|
lmata
|
470 |
|
|
2602093…
|
lmata
|
471 |
with ( |
|
2602093…
|
lmata
|
472 |
patch("navegador.explorer.ExplorerServer", return_value=mock_srv), |
|
2602093…
|
lmata
|
473 |
patch("navegador.cli.commands._get_store", return_value=MagicMock()), |
|
2602093…
|
lmata
|
474 |
patch("time.sleep", side_effect=_fake_sleep), |
|
2602093…
|
lmata
|
475 |
patch("webbrowser.open") as mock_open, |
|
2602093…
|
lmata
|
476 |
): |
|
2602093…
|
lmata
|
477 |
result = runner.invoke(main, ["explore", "--no-browser", "--port", str(port)]) |
|
2602093…
|
lmata
|
478 |
|
|
2602093…
|
lmata
|
479 |
mock_open.assert_not_called() |
|
2602093…
|
lmata
|
480 |
assert result.exit_code == 0 |
|
2602093…
|
lmata
|
481 |
|
|
2602093…
|
lmata
|
482 |
def test_explore_custom_port(self): |
|
2602093…
|
lmata
|
483 |
"""--port option should be forwarded to ExplorerServer.""" |
|
2602093…
|
lmata
|
484 |
runner = CliRunner() |
|
2602093…
|
lmata
|
485 |
port = _free_port() |
|
2602093…
|
lmata
|
486 |
|
|
2602093…
|
lmata
|
487 |
captured = {} |
|
2602093…
|
lmata
|
488 |
|
|
2602093…
|
lmata
|
489 |
def _fake_server(store, host, port): # noqa: A002 |
|
2602093…
|
lmata
|
490 |
captured["port"] = port |
|
2602093…
|
lmata
|
491 |
srv = MagicMock() |
|
2602093…
|
lmata
|
492 |
srv.url = f"http://{host}:{port}" |
|
2602093…
|
lmata
|
493 |
return srv |
|
2602093…
|
lmata
|
494 |
|
|
2602093…
|
lmata
|
495 |
def _fake_sleep(seconds): |
|
2602093…
|
lmata
|
496 |
raise KeyboardInterrupt |
|
2602093…
|
lmata
|
497 |
|
|
2602093…
|
lmata
|
498 |
with ( |
|
2602093…
|
lmata
|
499 |
patch("navegador.explorer.ExplorerServer", side_effect=_fake_server), |
|
2602093…
|
lmata
|
500 |
patch("navegador.cli.commands._get_store", return_value=MagicMock()), |
|
2602093…
|
lmata
|
501 |
patch("time.sleep", side_effect=_fake_sleep), |
|
2602093…
|
lmata
|
502 |
patch("webbrowser.open"), |
|
2602093…
|
lmata
|
503 |
): |
|
2602093…
|
lmata
|
504 |
runner.invoke(main, ["explore", "--port", str(port)]) |
|
2602093…
|
lmata
|
505 |
|
|
2602093…
|
lmata
|
506 |
assert captured.get("port") == port |
|
2602093…
|
lmata
|
507 |
|
|
2602093…
|
lmata
|
508 |
def test_explore_custom_host(self): |
|
2602093…
|
lmata
|
509 |
"""--host option should be forwarded to ExplorerServer.""" |
|
2602093…
|
lmata
|
510 |
runner = CliRunner() |
|
2602093…
|
lmata
|
511 |
captured = {} |
|
2602093…
|
lmata
|
512 |
|
|
2602093…
|
lmata
|
513 |
def _fake_server(store, host, port): # noqa: A002 |
|
2602093…
|
lmata
|
514 |
captured["host"] = host |
|
2602093…
|
lmata
|
515 |
srv = MagicMock() |
|
2602093…
|
lmata
|
516 |
srv.url = f"http://{host}:{port}" |
|
2602093…
|
lmata
|
517 |
return srv |
|
2602093…
|
lmata
|
518 |
|
|
2602093…
|
lmata
|
519 |
def _fake_sleep(seconds): |
|
2602093…
|
lmata
|
520 |
raise KeyboardInterrupt |
|
2602093…
|
lmata
|
521 |
|
|
2602093…
|
lmata
|
522 |
with ( |
|
2602093…
|
lmata
|
523 |
patch("navegador.explorer.ExplorerServer", side_effect=_fake_server), |
|
2602093…
|
lmata
|
524 |
patch("navegador.cli.commands._get_store", return_value=MagicMock()), |
|
2602093…
|
lmata
|
525 |
patch("time.sleep", side_effect=_fake_sleep), |
|
2602093…
|
lmata
|
526 |
patch("webbrowser.open"), |
|
2602093…
|
lmata
|
527 |
): |
|
2602093…
|
lmata
|
528 |
runner.invoke(main, ["explore", "--host", "0.0.0.0"]) |
|
2602093…
|
lmata
|
529 |
|
|
2602093…
|
lmata
|
530 |
assert captured.get("host") == "0.0.0.0" |
|
2602093…
|
lmata
|
531 |
|
|
2602093…
|
lmata
|
532 |
def test_explore_output_shows_url(self): |
|
2602093…
|
lmata
|
533 |
"""explore should print the server URL to stdout.""" |
|
2602093…
|
lmata
|
534 |
runner = CliRunner() |
|
2602093…
|
lmata
|
535 |
port = _free_port() |
|
2602093…
|
lmata
|
536 |
|
|
2602093…
|
lmata
|
537 |
mock_srv = MagicMock() |
|
2602093…
|
lmata
|
538 |
mock_srv.url = f"http://127.0.0.1:{port}" |
|
2602093…
|
lmata
|
539 |
|
|
2602093…
|
lmata
|
540 |
def _fake_sleep(seconds): |
|
2602093…
|
lmata
|
541 |
raise KeyboardInterrupt |
|
2602093…
|
lmata
|
542 |
|
|
2602093…
|
lmata
|
543 |
with ( |
|
2602093…
|
lmata
|
544 |
patch("navegador.explorer.ExplorerServer", return_value=mock_srv), |
|
2602093…
|
lmata
|
545 |
patch("navegador.cli.commands._get_store", return_value=MagicMock()), |
|
2602093…
|
lmata
|
546 |
patch("time.sleep", side_effect=_fake_sleep), |
|
2602093…
|
lmata
|
547 |
patch("webbrowser.open"), |
|
2602093…
|
lmata
|
548 |
): |
|
2602093…
|
lmata
|
549 |
result = runner.invoke(main, ["explore", "--port", str(port)]) |
|
2602093…
|
lmata
|
550 |
|
|
2602093…
|
lmata
|
551 |
assert str(port) in result.output or "127.0.0.1" in result.output |