Navegador

feat: FastAPI framework enrichment — routes, dependencies, Pydantic models, background tasks Closes #28

lmata 2026-03-23 05:21 trunk
Commit 8554cc4a2568bd22edd5758bb314887a57b1865c7a949fdfa2afecb09c419b91
Parent 396dc34f3bbf11f…
2 files changed +98 +471
+ navegador/enrichment/fastapi.py + tests/test_enrichment_fastapi.py
A navegador/enrichment/fastapi.py
+98
--- a/navegador/enrichment/fastapi.py
+++ b/navegador/enrichment/fastapi.py
@@ -0,0 +1,98 @@
1
+, "FastAPI", "APIRouter]:
2
+ return ["fastapi"]
3
+
4
+ # ── Enrichment ────────────────────────────────────────────────────────────
5
+
6
+ def enrich(self) -> EnrichmentResult:
7
+ result = EnrichmentResult()
8
+
9
+ routes = self._enrich_routes()
10
+ result.promoted += routes
11
+ result.patterns_found["routes"] = routes
12
+
13
+ dependencies = self._enrich_dependencies()
14
+ result.promoted += dependencies
15
+ result.patterns_found["dependencies"] = dependencies
16
+
17
+ pydantic_models = self._enrich_pydantic_models()
18
+ result.promoted += pydantic_models
19
+ result.patterns_found["pydantic_models"] = pydantic_models
20
+
21
+ background_tasks = self._enrich_background_tasks()
22
+ result.promoted += background_tasks
23
+ result.patterns_found["background_tasks"] = background_tasks
24
+
25
+ return result
26
+
27
+ # ── Pattern helpers ───────────────────────────────────────────────────────
28
+
29
+ def _enrich_routes(self) -> int:
30
+ """
31
+ Find Function/Method nodes linked to Decorator nodes whose names match
32
+ @app.<method> or @router.<method> patterns (e.g. app.get, router.post).
33
+
34
+ Falls back to querying the ``signature`` and ``docstring`` properties for
35
+ route decorator patterns when no Decorator nodes are present in the graph.
36
+ """
37
+ promoted = 0
38
+
39
+ # Strategy 1: Decorator nodes connected rows = result.result_ia DECORATES edges
40
+ for http_method in _HTTP_METHODS:
41
+ result = self.store.query(
42
+ "MATCH (d:Decorator)-[:DECORATES]->(n) "
43
+ "WHERE d.name CONTAINS $pattern "
44
+ "RETURN n.name, n.file_path",
45
+ {"pattern": f".{http_method}"},
46
+ )
47
+ rows = result.result_set or []
48
+ for row in rows:
49
+ name, file_path = row[0], row[1]
50
+ if name and file_path:
51
+ self._promote_node(name, file_path, "Route", {"http_method": http_method})
52
+ promoted += 1
53
+
54
+ # Strategy 2: signature / docstring heuristics (no Decorator nodes)
55
+ for http_method in _HTTP_METHODS:
56
+ for prop in ("signature", "docstring"):
57
+ rows = result.result_ via DECORATES edges
58
+ s:
59
+ name, file_path = row[0], row[1]
60
+ if name and file_path:
61
+ self._promote_node(name, file_path, "Dependency")
62
+ promoted += 1
63
+
64
+ return promoted
65
+
66
+ def _enrich_pydantic_models(self) -> int:
67
+ """
68
+ Find Class nodes that inherit from ``BaseModel`` via INHERITS edges,
69
+ or whose name appears as a base in the raw INHERITS graph.
70
+ """
71
+ promoted = 0
72
+
73
+ # Primary: INHERITS edges pointing to a node named "BaseModel"
74
+ result = self.store.query(
75
+ "MATCH (n:Class)-[:INHERITS]->(base) "
76
+ "WHERE base.name = $base_name "
77
+ "RETURN n.name, n.file_path",
78
+ {"base_name": "BaseModel"},
79
+ )
80
+ rows = result.result_set or []
81
+ for row in rows:
82
+ name, file_path = row[0], row[1]
83
+ if name and file_path:
84
+ self._promote_node(name, file_path, "PydanticModel")
85
+ promoted += 1
86
+
87
+ # Fallback: docstring or source mentions BaseModel
88
+ result = self.store.query(
89
+ "MATCH (n:Class) "
90
+ "WHERE n.docstring IS NOT NULL AND n.docstring CONTAINS $pattern "
91
+ "RETURN n.name, n.file_path",
92
+ {"pattern": "BaseModel"},
93
+ )
94
+ rows = result.result_set or []
95
+ for row in rows:
96
+ name, file_path = row[0], row[1]
97
+ if name and file_path:
98
+ self._
--- a/navegador/enrichment/fastapi.py
+++ b/navegador/enrichment/fastapi.py
@@ -0,0 +1,98 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/navegador/enrichment/fastapi.py
+++ b/navegador/enrichment/fastapi.py
@@ -0,0 +1,98 @@
1 , "FastAPI", "APIRouter]:
2 return ["fastapi"]
3
4 # ── Enrichment ────────────────────────────────────────────────────────────
5
6 def enrich(self) -> EnrichmentResult:
7 result = EnrichmentResult()
8
9 routes = self._enrich_routes()
10 result.promoted += routes
11 result.patterns_found["routes"] = routes
12
13 dependencies = self._enrich_dependencies()
14 result.promoted += dependencies
15 result.patterns_found["dependencies"] = dependencies
16
17 pydantic_models = self._enrich_pydantic_models()
18 result.promoted += pydantic_models
19 result.patterns_found["pydantic_models"] = pydantic_models
20
21 background_tasks = self._enrich_background_tasks()
22 result.promoted += background_tasks
23 result.patterns_found["background_tasks"] = background_tasks
24
25 return result
26
27 # ── Pattern helpers ───────────────────────────────────────────────────────
28
29 def _enrich_routes(self) -> int:
30 """
31 Find Function/Method nodes linked to Decorator nodes whose names match
32 @app.<method> or @router.<method> patterns (e.g. app.get, router.post).
33
34 Falls back to querying the ``signature`` and ``docstring`` properties for
35 route decorator patterns when no Decorator nodes are present in the graph.
36 """
37 promoted = 0
38
39 # Strategy 1: Decorator nodes connected rows = result.result_ia DECORATES edges
40 for http_method in _HTTP_METHODS:
41 result = self.store.query(
42 "MATCH (d:Decorator)-[:DECORATES]->(n) "
43 "WHERE d.name CONTAINS $pattern "
44 "RETURN n.name, n.file_path",
45 {"pattern": f".{http_method}"},
46 )
47 rows = result.result_set or []
48 for row in rows:
49 name, file_path = row[0], row[1]
50 if name and file_path:
51 self._promote_node(name, file_path, "Route", {"http_method": http_method})
52 promoted += 1
53
54 # Strategy 2: signature / docstring heuristics (no Decorator nodes)
55 for http_method in _HTTP_METHODS:
56 for prop in ("signature", "docstring"):
57 rows = result.result_ via DECORATES edges
58 s:
59 name, file_path = row[0], row[1]
60 if name and file_path:
61 self._promote_node(name, file_path, "Dependency")
62 promoted += 1
63
64 return promoted
65
66 def _enrich_pydantic_models(self) -> int:
67 """
68 Find Class nodes that inherit from ``BaseModel`` via INHERITS edges,
69 or whose name appears as a base in the raw INHERITS graph.
70 """
71 promoted = 0
72
73 # Primary: INHERITS edges pointing to a node named "BaseModel"
74 result = self.store.query(
75 "MATCH (n:Class)-[:INHERITS]->(base) "
76 "WHERE base.name = $base_name "
77 "RETURN n.name, n.file_path",
78 {"base_name": "BaseModel"},
79 )
80 rows = result.result_set or []
81 for row in rows:
82 name, file_path = row[0], row[1]
83 if name and file_path:
84 self._promote_node(name, file_path, "PydanticModel")
85 promoted += 1
86
87 # Fallback: docstring or source mentions BaseModel
88 result = self.store.query(
89 "MATCH (n:Class) "
90 "WHERE n.docstring IS NOT NULL AND n.docstring CONTAINS $pattern "
91 "RETURN n.name, n.file_path",
92 {"pattern": "BaseModel"},
93 )
94 rows = result.result_set or []
95 for row in rows:
96 name, file_path = row[0], row[1]
97 if name and file_path:
98 self._
A tests/test_enrichment_fastapi.py
+471
--- a/tests/test_enrichment_fastapi.py
+++ b/tests/test_enrichment_fastapi.py
@@ -0,0 +1,471 @@
1
+"""
2
+Tests for navegador.enrichment.fastapi — FastAPIEnricher.
3
+
4
+All tests use a mock GraphStore so no real database is required.
5
+"""
6
+
7
+from unittest.mock import MagicMock, call, patch
8
+
9
+import pytest
10
+
11
+from navegador.enrichment.base import EnrichmentResult, FrameworkEnricher
12
+from navegador.enrichment.fastapi import FastAPIEnricher
13
+from navegador.graph.store import GraphStore
14
+
15
+
16
+# ── Helpers ────────────────────────────────────────────────────────────────────
17
+
18
+
19
+def _mock_store(result_set=None):
20
+ """Return a GraphStore backed by a mock graph that always returns result_set."""
21
+ client = MagicMock()
22
+ graph = MagicMock()
23
+ graph.query.return_value = MagicMock(result_set=result_set or [])
24
+ client.select_graph.return_value = graph
25
+ return GraphStore(client)
26
+
27
+
28
+def _mock_store_with_responses(responses):
29
+ """
30
+ Return a GraphStore whose graph.query() returns successive mock results.
31
+
32
+ ``responses`` is a list of result_set values in call order. Once
33
+ exhausted, subsequent calls return an empty result_set.
34
+ """
35
+ client = MagicMock()
36
+ graph = MagicMock()
37
+ iter_resp = iter(responses)
38
+
39
+ def _side_effect(cypher, params=None):
40
+ rs = next(iter_resp, [])
41
+ return MagicMock(result_set=rs)
42
+
43
+ graph.query.side_effect = _side_effect
44
+ client.select_graph.return_value = graph
45
+ return GraphStore(client)
46
+
47
+
48
+# ── Identity ───────────────────────────────────────────────────────────────────
49
+
50
+
51
+class TestFastAPIEnricherIdentity:
52
+ def test_framework_name(self):
53
+ store = _mock_store()
54
+ enricher = FastAPIEnricher(store)
55
+ assert enricher.framework_name == "fastapi"
56
+
57
+ def test_detection_patterns_includes_fastapi_lowercase(self):
58
+ store = _mock_store()
59
+ assert "fastapi" in FastAPIEnricher(store).detection_patterns
60
+
61
+ def test_detection_patterns_includes_fastapi_class(self):
62
+ stoassert "FastAPI" in FastAPIEnricher(store).detection_patterns
63
+
64
+ dencludes_apirouter(self):
65
+ stoassert "APIRouter" in FastAPIEnricher(store).detection_patterns
66
+
67
+ def test_ graph.query.return_value = MagicMock(result_set=result_set or [])
68
+ client.select_graph.return_value = graph
69
+ return GraphStore(client)
70
+
71
+
72
+def _mock_store_with_responses(responses):
73
+ """
74
+ Return a GraphStore whose graph.query() returns successive mock results.
75
+
76
+ ``responses`` is a list of result_set values in call order. Once
77
+ exhausted, subsequent calls return an empty result_set.
78
+ """
79
+ client = MagicMock()
80
+ graph = MagicMock()
81
+ iter_resp = iter(responses)
82
+
83
+ def _side_effect(cypher, params=None):
84
+ rs = next(iter_resp, [])
85
+ return MagicMock(result_set=rs)
86
+
87
+ graph.query.side_effect = _side_effect
88
+ client.select_graph.return_value = graph
89
+ return GraphStore(client)
90
+
91
+
92
+# ── Identity ──────────────────────────────────────────────────────three�─────────────
93
+
94
+
95
+class TestFastAPIEnricherIdentity:
96
+ def
97
+ stor# Three detection, file_path="app/main.py"):
98
+ """
99
+ Build a FastAPIEnricher whose store returns one matching row for the
100
+ *first* Decorator-based route query (app.get), then empty for the rest.
101
+ """
102
+ client = MagicMock()
103
+ graph = MagicMock()
104
+ call_count = [0]
105
+
106
+ def _side_effect(cypher, params=None):
107
+ call_count[0] += 1
108
+ # Return a hit on the very first call only
109
+ if call_count[0] == 1:
110
+ return MagicMock(result_set=[[name, file_path]])
111
+ return MagicMock(result_set=[])
112
+
113
+ graph.query.side_effect = _side_effect
114
+ client.select_graph.return_value = graph
115
+ store = GraphStore(client)
116
+ return FastAPIEnricher(store)
117
+
118
+ def test_route_promoted_increments_count(self):
119
+ enricher = self._enricher_with_route_hit()
120
+ result = enricher.enrich()
121
+ assert result.patterns_found["routes"] >= 1
122
+
123
+ def test_route_calls_promote_node_with_route_semantic_type(self):
124
+ enricher = self._enricher_with_route_hit("list_users", "app/users.py")
125
+ with patch.object(enricher, "_promote_node") as mock_promote:
126
+ # Re-wire store so only the first query returns a hit
127
+ call_count = [0]
128
+
129
+ def _side_effect(cypher, params=None):
130
+ call_count[0] += 1
131
+ if call_count[0] == 1:
132
+ return MagicMock(result_set=[["list_users", "app/users.py"]])
133
+ return MagicMock(result_set=[])
134
+
135
+ enricher.store._graph.query.side_effect = _side_effect
136
+ enricher.enrich()
137
+
138
+ calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
139
+ assert len(calls) >= 1
140
+ assert calls[0][0][0] == "list_users"
141
+ assert calls[0][0][1] == "app/users.py"
142
+
143
+ def test_route_http_method_stored_as_prop(self):
144
+ """The http_method kwarg should be passed to _promote_node for route nodes."""
145
+ store = _mock_store()
146
+ enricher = FastAPIEnricher(store)
147
+
148
+ call_count = [0]
149
+
150
+ def _side_effect(cypher, params=None):
151
+ call_count[0] += 1
152
+ if call_count[0] == 1:
153
+ return MagicMock(result_set=[["create_item", "app/main.py"]])
154
+ return MagicMock(result_set=[])
155
+
156
+ store._graph.query.side_effect = _side_effect
157
+
158
+ with patch.object(enricher, "_promote_node") as mock_promote:
159
+ enricher.enrich()
160
+
161
+ route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
162
+ assert route_calls, "Expected at least one _promote_node call with 'Route'"
163
+ _, kwargs = route_calls[0]
164
+ props = kwargs.get("props") or (route_calls[0][0][3] if len(route_calls[0][0]) > 3 else None)
165
+ assert props is not None
166
+ assert "http_method" in props
167
+
168
+
169
+# ── Dependencies ──────────────────────────────────────────────────────────────
170
+
171
+
172
+class TestFastAPIEnricherDependencies:
173
+ def test_dependency_promoted_for_depends_in_signature(self):
174
+ store = _mock_store()
175
+ enricher = FastAPIEnricher(store)
176
+
177
+ def _side_effect(cypher, params=None):
178
+ params = params or {}
179
+ if params.get("pattern") == "Depends(":
180
+ return MagicMock(result_set=[["get_db", "app/deps.py"]])
181
+ return MagicMock(result_set=[])
182
+
183
+ store._graph.query.side_effect = _side_effect
184
+
185
+ with patch.object(enricher, "_promote_node") as mock_promote:
186
+ enricher.enrich()
187
+
188
+ dep_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Dependency"]
189
+ assert len(dep_calls) >= 1
190
+
191
+ def test_dependency_semantic_type_is_dependency(self):
192
+ store = _mock_store()
193
+ enricher = FastAPIEnricher(store)
194
+
195
+ def _side_effect(cypher, params=None):
196
+ params = params or {}
197
+ if "Depends" in params.get("pattern", ""):
198
+ return MagicMock(result_set=[["auth_user", "app/auth.py"]])
199
+ return MagicMock(result_set=[])
200
+
201
+ store._graph.query.side_effect = _side_effect
202
+
203
+ with patch.object(enricher, "_promote_node") as mock_promote:
204
+ enricher.enrich()
205
+
206
+ dep_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Dependency"]
207
+ assert dep_calls
208
+
209
+
210
+# ── Pydantic Models ───────────────────────────────────────────────────────────
211
+
212
+
213
+class TestFastAPIEnricherPydanticModels:
214
+ def test_pydantic_model_promoted_via_inherits_edge(self):
215
+ store = _mock_store()
216
+ enricher = FastAPIEnricher(store)
217
+
218
+ def _side_effect(cypher, params=None):
219
+ params = params or {}
220
+ if params.get("base_name") == "BaseModel":
221
+ return MagicMock(result_set=[["UserSchema", "app/schemas.py"]])
222
+ return MagicMock(result_set=[])
223
+
224
+ store._graph.query.side_effect = _side_effect
225
+
226
+ with patch.object(enricher, "_promote_node") as mock_promote:
227
+ enricher.enrich()
228
+
229
+ pm_calls = [c for c in mock_promote.call_args_list if c[0][2] == "PydanticModel"]
230
+ assert len(pm_calls) >= 1
231
+ assert pm_calls[0][0][0] == "UserSchema"
232
+ assert pm_calls[0][0][1] == "app/schemas.py"
233
+
234
+ def test_pydantic_model_semantic_type_string(self):
235
+ store = _mock_store()
236
+ enricher = FastAPIEnricher(store)
237
+
238
+ def _side_effect(cypher, params=None):
239
+ params = params or {}
240
+ if params.get("base_name") == "BaseModel":
241
+ return MagicMock(result_set=[["ItemModel", "app/models.py"]])
242
+ return MagicMock(result_set=[])
243
+
244
+ store._graph.query.side_effect = _side_effect
245
+
246
+ with patch.object(enricher, "_promote_node") as mock_promote:
247
+ enricher.enrich()
248
+
249
+ types = {c[0][2] for c in mock_promote.call_args_list}
250
+ assert "PydanticModel" in types
251
+
252
+ def test_pydantic_model_fallback_via_docstring(self):
253
+ """If no INHERITS edge, docstring containing 'BaseModel' is the fallback."""
254
+ store = _mock_store()
255
+ enricher = FastAPIEnricher(store)
256
+
257
+ call_count = [0]
258
+
259
+ def _side_effect(cypher, params=None):
260
+ call_count[0] += 1
261
+ params = params or {}
262
+ # Fail INHERITS query, succeed on docstring fallback
263
+ if params.get("base_name") == "BaseModel":
264
+ return MagicMock(result_set=[])
265
+ if "BaseModel" in params.get("pattern", "") and "docstring" in cypher:
266
+ return MagicMock(result_set=[["CreateUser", "app/schemas.py"]])
267
+ return MagicMock(result_set=[])
268
+
269
+ store._graph.query.side_effect = _side_effect
270
+
271
+ with patch.object(enricher, "_promote_node") as mock_promote:
272
+ enricher.enrich()
273
+
274
+ pm_calls = [c for c in mock_promote.call_args_list if c[0][2] == "PydanticModel"]
275
+ assert len(pm_calls) >= 1
276
+
277
+
278
+# ── Background Tasks ──────────────────────────────────────────────────────────
279
+
280
+
281
+class TestFastAPIEnricherBackgroundTasks:
282
+ def test_background_task_promoted_via_on_event_decorator(self):
283
+ store = _mock_store()
284
+ enricher = FastAPIEnricher(store)
285
+
286
+ def _side_effect(cypher, params=None):
287
+ params = params or {}
288
+ if "on_event" in params.get("pattern", ""):
289
+ return MagicMock(result_set=[["startup_handler", "app/events.py"]])
290
+ return MagicMock(result_set=[])
291
+
292
+ store._graph.query.side_effect = _side_effect
293
+
294
+ with patch.object(enricher, "_promote_node") as mock_promote:
295
+ enricher.enrich()
296
+
297
+ bt_calls = [c for c in mock_promote.call_args_list if c[0][2] == "BackgroundTask"]
298
+ assert len(bt_calls) >= 1
299
+
300
+ def test_background_task_promoted_via_background_tasks_in_signature(self):
301
+ store = _mock_store()
302
+ enricher = FastAPIEnricher(store)
303
+
304
+ def _side_effect(cypher, params=None):
305
+ params = params or {}
306
+ if "BackgroundTasks" in params.get("pattern", ""):
307
+ return MagicMock(result_set=[["send_email", "app/tasks.py"]])
308
+ return MagicMock(result_set=[])
309
+
310
+ store._graph.query.side_effect = _side_effect
311
+
312
+ with patch.object(enricher, "_promote_node") as mock_promote:
313
+ enricher.enrich()
314
+
315
+ bt_calls = [c for c in mock_promote.call_args_list if c[0][2] == "BackgroundTask"]
316
+ assert len(bt_calls) >= 1
317
+
318
+ def test_background_task_semantic_type_string(self):
319
+ store = _mock_store()
320
+ enricher = FastAPIEnricher(store)
321
+
322
+ def _side_effect(cypher, params=None):
323
+ params = params or {}
324
+ if "on_event" in params.get("pattern", ""):
325
+ return MagicMock(result_set=[["shutdown_handler", "app/main.py"]])
326
+ return MagicMock(result_set=[])
327
+
328
+ store._graph.query.side_effect = _side_effect
329
+
330
+ with patch.object(enricher, "_promote_node") as mock_promote:
331
+ enricher.enrich()
332
+
333
+ types = {c[0][2] for c in mock_promote.call_args_list}
334
+ assert "BackgroundTask" in types
335
+
336
+
337
+# ── _promote_node integration ─────────────────────────────────────────────────
338
+
339
+
340
+class TestFastAPIEnricherPromoteNodeIntegration:
341
+ def test_promote_node_called_for_each_matched_row(self):
342
+ """Two rows returned → two _promote_node calls for that pattern."""
343
+ store = _mock_store()
344
+ enricher = FastAPIEnricher(store)
345
+
346
+ call_count = [0]
347
+
348
+ def _side_effect(cypher, params=None):
349
+ call_count[0] += 1
350
+ if call_count[0] == 1:
351
+ return MagicMock(result_set=[
352
+ ["route_a", "app/a.py"],
353
+ ["route_b", "app/b.py"],
354
+ ])
355
+ return MagicMock(result_set=[])
356
+
357
+ store._graph.query.side_effect = _side_effect
358
+
359
+ with patch.object(enricher, "_promote_node") as mock_promote:
360
+ enricher.enrich()
361
+
362
+ route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
363
+ assert len(route_calls) == 2
364
+
365
+ def test_rows_with_none_name_are_skipped(self):
366
+ """Rows where name is None must not call _promote_node."""
367
+ store = _mock_store()
368
+ enricher = FastAPIEnricher(store)
369
+
370
+ call_count = [0]
371
+
372
+ def _side_effect(cypher, params=None):
373
+ call_count[0] += 1
374
+ if call_count[0] == 1:
375
+ return MagicMock(result_set=[[None, "app/main.py"]])
376
+ return MagicMock(result_set=[])
377
+
378
+ store._graph.query.side_effect = _side_effect
379
+
380
+ with patch.object(enricher, "_promote_node") as mock_promote:
381
+ enricher.enrich()
382
+
383
+ route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
384
+ assert len(route_calls) == 0
385
+
386
+ def test_rows_with_none_file_path_are_skipped(self):
387
+ """Rows where file_path is None must not call _promote_node."""
388
+ store = _mock_store()
389
+ enricher = FastAPIEnricher(store)
390
+
391
+ call_count = [0]
392
+
393
+ def _side_effect(cypher, params=None):
394
+ call_count[0] += 1
395
+ if call_count[0] == 1:
396
+ return MagicMock(result_set=[["get_items", None]])
397
+ return MagicMock(result_set=[])
398
+
399
+ store._graph.query.side_effect = _side_effect
400
+
401
+ with patch.object(enricher, "_promote_node") as mock_promote:
402
+ enricher.enrich()
403
+
404
+ route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
405
+ assert len(route_calls) == 0
406
+
407
+
408
+# ── HTTP method coverage ──────────────────────────────────────────────────────
409
+
410
+
411
+class TestFastAPIEnricherHTTPMethods:
412
+ def test_all_http_methods_are_queried(self):
413
+ """The enricher must issue queries for all standard HTTP verbs."""
414
+ store = _mock_store(result_set=[])
415
+ FastAPIEnricher(store).enrich()
416
+
417
+ all_params = [
418
+ call_args[0][1] if len(call_args[0]) > 1 else {}
419
+ for call_args in store._graph.query.call_args_list
420
+ ]
421
+ patterns_used = {p.get("pattern", "") for p in all_params}
422
+
423
+ for method in ("get", "post", "put", "delete", "patch"):
424
+ assert any(method in pat for pat in patterns_used), (
425
+ f"HTTP method '{method}' not found in any query pattern"
426
+ )
427
+store._graph.query.call_count == 3tore(client)
428
+
429
+
430
+# ─�richer(store).detection_patterns
431
+
432
+ def test_detection_patterns_is_list_of_strings(self):
433
+ store = _mock_store()
434
+ patterns = FastAPIEnricher(store).detection_patterns
435
+ assert isinstance(patterns, list)
436
+ assert all(isinstance(p, str) for p in patterns)
437
+
438
+ def test_detection_patterns_is_nonempty(self):
439
+ store = _mock_store()
440
+ assert len(FastAPIEnricher(store).detection_patterns) >= 1
441
+
442
+ def test_is_subclass_of_framework_enricher(self):
443
+ store = _mock_store()
444
+ assert isinstance(FastAPIEnricher(store), FrameworkEnricher)
445
+
446
+
447
+# ── detect() ──────────────────────────────────────────────────────────────────
448
+
449
+
450
+class TestFastAPIEnricherDetect:
451
+ def test_detect_returns_true_when_fastapi_import_found(self):
452
+ store = _mock_store(result_set=[[1]])
453
+ assert FastAPIEnricher(store).detect() is True
454
+
455
+ def test_detect_returns_false_when_no_match(self):
456
+ store = _mock_store(result_set=[[0]])
457
+ assert FastAPIEnricher(store).detect() is False
458
+
459
+ def test_detect_returns_false_on_empty_result_set(self):
460
+ store = _mock_store(result_set=[])
461
+ assert FastAPIEnricher(store).detect() is False
462
+
463
+ def test_detect_queries_all_patterns_when_no_match(self):
464
+ store = _mock_store(result_set=[[0]])
465
+ enricher = FastAPIEnricher(store)
466
+ enricher.detect()
467
+ # All detection_patterns queried when no match is found
468
+ expected = len(enricher.detection_patterns) + len(enricher.detection_files)
469
+ assert store._graph.query.call_count == expected
470
+
471
+ def test_detect_short_circui
--- a/tests/test_enrichment_fastapi.py
+++ b/tests/test_enrichment_fastapi.py
@@ -0,0 +1,471 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_enrichment_fastapi.py
+++ b/tests/test_enrichment_fastapi.py
@@ -0,0 +1,471 @@
1 """
2 Tests for navegador.enrichment.fastapi — FastAPIEnricher.
3
4 All tests use a mock GraphStore so no real database is required.
5 """
6
7 from unittest.mock import MagicMock, call, patch
8
9 import pytest
10
11 from navegador.enrichment.base import EnrichmentResult, FrameworkEnricher
12 from navegador.enrichment.fastapi import FastAPIEnricher
13 from navegador.graph.store import GraphStore
14
15
16 # ── Helpers ────────────────────────────────────────────────────────────────────
17
18
19 def _mock_store(result_set=None):
20 """Return a GraphStore backed by a mock graph that always returns result_set."""
21 client = MagicMock()
22 graph = MagicMock()
23 graph.query.return_value = MagicMock(result_set=result_set or [])
24 client.select_graph.return_value = graph
25 return GraphStore(client)
26
27
28 def _mock_store_with_responses(responses):
29 """
30 Return a GraphStore whose graph.query() returns successive mock results.
31
32 ``responses`` is a list of result_set values in call order. Once
33 exhausted, subsequent calls return an empty result_set.
34 """
35 client = MagicMock()
36 graph = MagicMock()
37 iter_resp = iter(responses)
38
39 def _side_effect(cypher, params=None):
40 rs = next(iter_resp, [])
41 return MagicMock(result_set=rs)
42
43 graph.query.side_effect = _side_effect
44 client.select_graph.return_value = graph
45 return GraphStore(client)
46
47
48 # ── Identity ───────────────────────────────────────────────────────────────────
49
50
51 class TestFastAPIEnricherIdentity:
52 def test_framework_name(self):
53 store = _mock_store()
54 enricher = FastAPIEnricher(store)
55 assert enricher.framework_name == "fastapi"
56
57 def test_detection_patterns_includes_fastapi_lowercase(self):
58 store = _mock_store()
59 assert "fastapi" in FastAPIEnricher(store).detection_patterns
60
61 def test_detection_patterns_includes_fastapi_class(self):
62 stoassert "FastAPI" in FastAPIEnricher(store).detection_patterns
63
64 dencludes_apirouter(self):
65 stoassert "APIRouter" in FastAPIEnricher(store).detection_patterns
66
67 def test_ graph.query.return_value = MagicMock(result_set=result_set or [])
68 client.select_graph.return_value = graph
69 return GraphStore(client)
70
71
72 def _mock_store_with_responses(responses):
73 """
74 Return a GraphStore whose graph.query() returns successive mock results.
75
76 ``responses`` is a list of result_set values in call order. Once
77 exhausted, subsequent calls return an empty result_set.
78 """
79 client = MagicMock()
80 graph = MagicMock()
81 iter_resp = iter(responses)
82
83 def _side_effect(cypher, params=None):
84 rs = next(iter_resp, [])
85 return MagicMock(result_set=rs)
86
87 graph.query.side_effect = _side_effect
88 client.select_graph.return_value = graph
89 return GraphStore(client)
90
91
92 # ── Identity ──────────────────────────────────────────────────────three�─────────────
93
94
95 class TestFastAPIEnricherIdentity:
96 def
97 stor# Three detection, file_path="app/main.py"):
98 """
99 Build a FastAPIEnricher whose store returns one matching row for the
100 *first* Decorator-based route query (app.get), then empty for the rest.
101 """
102 client = MagicMock()
103 graph = MagicMock()
104 call_count = [0]
105
106 def _side_effect(cypher, params=None):
107 call_count[0] += 1
108 # Return a hit on the very first call only
109 if call_count[0] == 1:
110 return MagicMock(result_set=[[name, file_path]])
111 return MagicMock(result_set=[])
112
113 graph.query.side_effect = _side_effect
114 client.select_graph.return_value = graph
115 store = GraphStore(client)
116 return FastAPIEnricher(store)
117
118 def test_route_promoted_increments_count(self):
119 enricher = self._enricher_with_route_hit()
120 result = enricher.enrich()
121 assert result.patterns_found["routes"] >= 1
122
123 def test_route_calls_promote_node_with_route_semantic_type(self):
124 enricher = self._enricher_with_route_hit("list_users", "app/users.py")
125 with patch.object(enricher, "_promote_node") as mock_promote:
126 # Re-wire store so only the first query returns a hit
127 call_count = [0]
128
129 def _side_effect(cypher, params=None):
130 call_count[0] += 1
131 if call_count[0] == 1:
132 return MagicMock(result_set=[["list_users", "app/users.py"]])
133 return MagicMock(result_set=[])
134
135 enricher.store._graph.query.side_effect = _side_effect
136 enricher.enrich()
137
138 calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
139 assert len(calls) >= 1
140 assert calls[0][0][0] == "list_users"
141 assert calls[0][0][1] == "app/users.py"
142
143 def test_route_http_method_stored_as_prop(self):
144 """The http_method kwarg should be passed to _promote_node for route nodes."""
145 store = _mock_store()
146 enricher = FastAPIEnricher(store)
147
148 call_count = [0]
149
150 def _side_effect(cypher, params=None):
151 call_count[0] += 1
152 if call_count[0] == 1:
153 return MagicMock(result_set=[["create_item", "app/main.py"]])
154 return MagicMock(result_set=[])
155
156 store._graph.query.side_effect = _side_effect
157
158 with patch.object(enricher, "_promote_node") as mock_promote:
159 enricher.enrich()
160
161 route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
162 assert route_calls, "Expected at least one _promote_node call with 'Route'"
163 _, kwargs = route_calls[0]
164 props = kwargs.get("props") or (route_calls[0][0][3] if len(route_calls[0][0]) > 3 else None)
165 assert props is not None
166 assert "http_method" in props
167
168
169 # ── Dependencies ──────────────────────────────────────────────────────────────
170
171
172 class TestFastAPIEnricherDependencies:
173 def test_dependency_promoted_for_depends_in_signature(self):
174 store = _mock_store()
175 enricher = FastAPIEnricher(store)
176
177 def _side_effect(cypher, params=None):
178 params = params or {}
179 if params.get("pattern") == "Depends(":
180 return MagicMock(result_set=[["get_db", "app/deps.py"]])
181 return MagicMock(result_set=[])
182
183 store._graph.query.side_effect = _side_effect
184
185 with patch.object(enricher, "_promote_node") as mock_promote:
186 enricher.enrich()
187
188 dep_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Dependency"]
189 assert len(dep_calls) >= 1
190
191 def test_dependency_semantic_type_is_dependency(self):
192 store = _mock_store()
193 enricher = FastAPIEnricher(store)
194
195 def _side_effect(cypher, params=None):
196 params = params or {}
197 if "Depends" in params.get("pattern", ""):
198 return MagicMock(result_set=[["auth_user", "app/auth.py"]])
199 return MagicMock(result_set=[])
200
201 store._graph.query.side_effect = _side_effect
202
203 with patch.object(enricher, "_promote_node") as mock_promote:
204 enricher.enrich()
205
206 dep_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Dependency"]
207 assert dep_calls
208
209
210 # ── Pydantic Models ───────────────────────────────────────────────────────────
211
212
213 class TestFastAPIEnricherPydanticModels:
214 def test_pydantic_model_promoted_via_inherits_edge(self):
215 store = _mock_store()
216 enricher = FastAPIEnricher(store)
217
218 def _side_effect(cypher, params=None):
219 params = params or {}
220 if params.get("base_name") == "BaseModel":
221 return MagicMock(result_set=[["UserSchema", "app/schemas.py"]])
222 return MagicMock(result_set=[])
223
224 store._graph.query.side_effect = _side_effect
225
226 with patch.object(enricher, "_promote_node") as mock_promote:
227 enricher.enrich()
228
229 pm_calls = [c for c in mock_promote.call_args_list if c[0][2] == "PydanticModel"]
230 assert len(pm_calls) >= 1
231 assert pm_calls[0][0][0] == "UserSchema"
232 assert pm_calls[0][0][1] == "app/schemas.py"
233
234 def test_pydantic_model_semantic_type_string(self):
235 store = _mock_store()
236 enricher = FastAPIEnricher(store)
237
238 def _side_effect(cypher, params=None):
239 params = params or {}
240 if params.get("base_name") == "BaseModel":
241 return MagicMock(result_set=[["ItemModel", "app/models.py"]])
242 return MagicMock(result_set=[])
243
244 store._graph.query.side_effect = _side_effect
245
246 with patch.object(enricher, "_promote_node") as mock_promote:
247 enricher.enrich()
248
249 types = {c[0][2] for c in mock_promote.call_args_list}
250 assert "PydanticModel" in types
251
252 def test_pydantic_model_fallback_via_docstring(self):
253 """If no INHERITS edge, docstring containing 'BaseModel' is the fallback."""
254 store = _mock_store()
255 enricher = FastAPIEnricher(store)
256
257 call_count = [0]
258
259 def _side_effect(cypher, params=None):
260 call_count[0] += 1
261 params = params or {}
262 # Fail INHERITS query, succeed on docstring fallback
263 if params.get("base_name") == "BaseModel":
264 return MagicMock(result_set=[])
265 if "BaseModel" in params.get("pattern", "") and "docstring" in cypher:
266 return MagicMock(result_set=[["CreateUser", "app/schemas.py"]])
267 return MagicMock(result_set=[])
268
269 store._graph.query.side_effect = _side_effect
270
271 with patch.object(enricher, "_promote_node") as mock_promote:
272 enricher.enrich()
273
274 pm_calls = [c for c in mock_promote.call_args_list if c[0][2] == "PydanticModel"]
275 assert len(pm_calls) >= 1
276
277
278 # ── Background Tasks ──────────────────────────────────────────────────────────
279
280
281 class TestFastAPIEnricherBackgroundTasks:
282 def test_background_task_promoted_via_on_event_decorator(self):
283 store = _mock_store()
284 enricher = FastAPIEnricher(store)
285
286 def _side_effect(cypher, params=None):
287 params = params or {}
288 if "on_event" in params.get("pattern", ""):
289 return MagicMock(result_set=[["startup_handler", "app/events.py"]])
290 return MagicMock(result_set=[])
291
292 store._graph.query.side_effect = _side_effect
293
294 with patch.object(enricher, "_promote_node") as mock_promote:
295 enricher.enrich()
296
297 bt_calls = [c for c in mock_promote.call_args_list if c[0][2] == "BackgroundTask"]
298 assert len(bt_calls) >= 1
299
300 def test_background_task_promoted_via_background_tasks_in_signature(self):
301 store = _mock_store()
302 enricher = FastAPIEnricher(store)
303
304 def _side_effect(cypher, params=None):
305 params = params or {}
306 if "BackgroundTasks" in params.get("pattern", ""):
307 return MagicMock(result_set=[["send_email", "app/tasks.py"]])
308 return MagicMock(result_set=[])
309
310 store._graph.query.side_effect = _side_effect
311
312 with patch.object(enricher, "_promote_node") as mock_promote:
313 enricher.enrich()
314
315 bt_calls = [c for c in mock_promote.call_args_list if c[0][2] == "BackgroundTask"]
316 assert len(bt_calls) >= 1
317
318 def test_background_task_semantic_type_string(self):
319 store = _mock_store()
320 enricher = FastAPIEnricher(store)
321
322 def _side_effect(cypher, params=None):
323 params = params or {}
324 if "on_event" in params.get("pattern", ""):
325 return MagicMock(result_set=[["shutdown_handler", "app/main.py"]])
326 return MagicMock(result_set=[])
327
328 store._graph.query.side_effect = _side_effect
329
330 with patch.object(enricher, "_promote_node") as mock_promote:
331 enricher.enrich()
332
333 types = {c[0][2] for c in mock_promote.call_args_list}
334 assert "BackgroundTask" in types
335
336
337 # ── _promote_node integration ─────────────────────────────────────────────────
338
339
340 class TestFastAPIEnricherPromoteNodeIntegration:
341 def test_promote_node_called_for_each_matched_row(self):
342 """Two rows returned → two _promote_node calls for that pattern."""
343 store = _mock_store()
344 enricher = FastAPIEnricher(store)
345
346 call_count = [0]
347
348 def _side_effect(cypher, params=None):
349 call_count[0] += 1
350 if call_count[0] == 1:
351 return MagicMock(result_set=[
352 ["route_a", "app/a.py"],
353 ["route_b", "app/b.py"],
354 ])
355 return MagicMock(result_set=[])
356
357 store._graph.query.side_effect = _side_effect
358
359 with patch.object(enricher, "_promote_node") as mock_promote:
360 enricher.enrich()
361
362 route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
363 assert len(route_calls) == 2
364
365 def test_rows_with_none_name_are_skipped(self):
366 """Rows where name is None must not call _promote_node."""
367 store = _mock_store()
368 enricher = FastAPIEnricher(store)
369
370 call_count = [0]
371
372 def _side_effect(cypher, params=None):
373 call_count[0] += 1
374 if call_count[0] == 1:
375 return MagicMock(result_set=[[None, "app/main.py"]])
376 return MagicMock(result_set=[])
377
378 store._graph.query.side_effect = _side_effect
379
380 with patch.object(enricher, "_promote_node") as mock_promote:
381 enricher.enrich()
382
383 route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
384 assert len(route_calls) == 0
385
386 def test_rows_with_none_file_path_are_skipped(self):
387 """Rows where file_path is None must not call _promote_node."""
388 store = _mock_store()
389 enricher = FastAPIEnricher(store)
390
391 call_count = [0]
392
393 def _side_effect(cypher, params=None):
394 call_count[0] += 1
395 if call_count[0] == 1:
396 return MagicMock(result_set=[["get_items", None]])
397 return MagicMock(result_set=[])
398
399 store._graph.query.side_effect = _side_effect
400
401 with patch.object(enricher, "_promote_node") as mock_promote:
402 enricher.enrich()
403
404 route_calls = [c for c in mock_promote.call_args_list if c[0][2] == "Route"]
405 assert len(route_calls) == 0
406
407
408 # ── HTTP method coverage ──────────────────────────────────────────────────────
409
410
411 class TestFastAPIEnricherHTTPMethods:
412 def test_all_http_methods_are_queried(self):
413 """The enricher must issue queries for all standard HTTP verbs."""
414 store = _mock_store(result_set=[])
415 FastAPIEnricher(store).enrich()
416
417 all_params = [
418 call_args[0][1] if len(call_args[0]) > 1 else {}
419 for call_args in store._graph.query.call_args_list
420 ]
421 patterns_used = {p.get("pattern", "") for p in all_params}
422
423 for method in ("get", "post", "put", "delete", "patch"):
424 assert any(method in pat for pat in patterns_used), (
425 f"HTTP method '{method}' not found in any query pattern"
426 )
427 store._graph.query.call_count == 3tore(client)
428
429
430 # ─�richer(store).detection_patterns
431
432 def test_detection_patterns_is_list_of_strings(self):
433 store = _mock_store()
434 patterns = FastAPIEnricher(store).detection_patterns
435 assert isinstance(patterns, list)
436 assert all(isinstance(p, str) for p in patterns)
437
438 def test_detection_patterns_is_nonempty(self):
439 store = _mock_store()
440 assert len(FastAPIEnricher(store).detection_patterns) >= 1
441
442 def test_is_subclass_of_framework_enricher(self):
443 store = _mock_store()
444 assert isinstance(FastAPIEnricher(store), FrameworkEnricher)
445
446
447 # ── detect() ──────────────────────────────────────────────────────────────────
448
449
450 class TestFastAPIEnricherDetect:
451 def test_detect_returns_true_when_fastapi_import_found(self):
452 store = _mock_store(result_set=[[1]])
453 assert FastAPIEnricher(store).detect() is True
454
455 def test_detect_returns_false_when_no_match(self):
456 store = _mock_store(result_set=[[0]])
457 assert FastAPIEnricher(store).detect() is False
458
459 def test_detect_returns_false_on_empty_result_set(self):
460 store = _mock_store(result_set=[])
461 assert FastAPIEnricher(store).detect() is False
462
463 def test_detect_queries_all_patterns_when_no_match(self):
464 store = _mock_store(result_set=[[0]])
465 enricher = FastAPIEnricher(store)
466 enricher.detect()
467 # All detection_patterns queried when no match is found
468 expected = len(enricher.detection_patterns) + len(enricher.detection_files)
469 assert store._graph.query.call_count == expected
470
471 def test_detect_short_circui

Keyboard Shortcuts

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