diff --git a/features/feature-spec-discovery.md b/features/feature-spec-discovery.md index dea0da0..45966dd 100644 --- a/features/feature-spec-discovery.md +++ b/features/feature-spec-discovery.md @@ -110,6 +110,24 @@ Add shared discovery infrastructure for Issues #125 and #126: centralized parser **When** `TypeScriptTestMiner` executes **Then** it reports `MinerErrorKind.PARSE_ERROR` for parse failures and still returns items from valid files. +### Story 10: Python API route mining +**Scenario:** As a discovery pipeline, I need to extract Python API routes across major frameworks. +**Given** Python files selected from `ctx.file_index.files_by_language(SupportedLanguage.PYTHON)` +**When** `PythonRouteMiner` runs with `ctx.frameworks[SupportedLanguage.PYTHON] = ["fastapi"]` +**Then** it emits `DiscoveredItem(kind=API_ROUTE)` entries for decorated handlers with methods and paths. +**And** FastAPI `response_model` is captured in `ApiRouteMeta`. + +**Scenario:** As a miner maintainer, I need Flask metadata fidelity. +**Given** a Flask route `@bp.route("/items", methods=["GET", "POST"])` and `@app.route("/health")` +**When** `PythonRouteMiner` emits items +**Then** metadata validates against `ApiRouteMeta` with `http_method=["GET", "POST"]` for the first route +**And** missing `methods` defaults to `["GET"]`. + +**Scenario:** As a discovery pipeline, I need Django URL pattern support. +**Given** a module with `urlpatterns = [path(...), re_path(...)]` +**When** `PythonRouteMiner` runs with framework `django` +**Then** both `path()` and `re_path()` entries are emitted as API route items. + ## Acceptance Criteria - Language abstraction returns `SupportedLanguage` members for `.py`, `.ts`, `.tsx`, `.js`, `.jsx`, `.mjs` and `None` otherwise. - `LanguageRegistry().parse(path_to_py_file)` returns `(node, SupportedLanguage.PYTHON)` for valid Python input. @@ -145,3 +163,9 @@ Add shared discovery infrastructure for Issues #125 and #126: centralized parser - TypeScript/JavaScript test metadata validates against `TestFunctionMeta`, including `call_style`, `has_todo`, and describe-block `class_name`. - `.ts` test files emit `language=SupportedLanguage.TYPESCRIPT`; `.js` files emit `language=SupportedLanguage.JAVASCRIPT`. - Confidence scoring is `0.9` for known framework + `.spec.` filename and `0.7` otherwise. +- `PythonRouteMiner` reads candidate files from `ctx.file_index.files_by_language(SupportedLanguage.PYTHON)` and does not walk the filesystem directly. +- `PythonRouteMiner` uses precomputed Python frameworks from `ctx.frameworks[SupportedLanguage.PYTHON]` and does not re-parse manifests. +- FastAPI fixtures with `GET`, `POST`, and `PATCH` decorators produce three API route items with correct methods and paths. +- Route metadata validates against `ApiRouteMeta`, including `response_model` for FastAPI and list-form `http_method` for Flask `methods=[...]`. +- Django `urlpatterns` entries using both `path()` and `re_path()` produce API route items. +- Missing Flask `methods` defaults to `["GET"]`. diff --git a/src/specleft/discovery/miners/__init__.py b/src/specleft/discovery/miners/__init__.py index c059ee1..07504d1 100644 --- a/src/specleft/discovery/miners/__init__.py +++ b/src/specleft/discovery/miners/__init__.py @@ -4,6 +4,7 @@ """Discovery miner implementations.""" from specleft.discovery.miners.defaults import default_miners +from specleft.discovery.miners.python.routes import PythonRouteMiner from specleft.discovery.miners.python.tests import PythonTestMiner from specleft.discovery.miners.shared.docstrings import DocstringMiner from specleft.discovery.miners.shared.readme import ReadmeOverviewMiner @@ -11,6 +12,7 @@ __all__ = [ "DocstringMiner", + "PythonRouteMiner", "PythonTestMiner", "ReadmeOverviewMiner", "TypeScriptTestMiner", diff --git a/src/specleft/discovery/miners/defaults.py b/src/specleft/discovery/miners/defaults.py index a6c3a31..52be3d4 100644 --- a/src/specleft/discovery/miners/defaults.py +++ b/src/specleft/discovery/miners/defaults.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING +from specleft.discovery.miners.python.routes import PythonRouteMiner from specleft.discovery.miners.python.tests import PythonTestMiner from specleft.discovery.miners.shared.docstrings import DocstringMiner from specleft.discovery.miners.shared.readme import ReadmeOverviewMiner @@ -21,6 +22,7 @@ def default_miners() -> list[BaseMiner]: return [ ReadmeOverviewMiner(), PythonTestMiner(), + PythonRouteMiner(), TypeScriptTestMiner(), DocstringMiner(), ] diff --git a/src/specleft/discovery/miners/python/__init__.py b/src/specleft/discovery/miners/python/__init__.py index 1088682..ac77aba 100644 --- a/src/specleft/discovery/miners/python/__init__.py +++ b/src/specleft/discovery/miners/python/__init__.py @@ -4,6 +4,7 @@ """Python-specific discovery miners.""" from specleft.discovery.miners.python.docstrings import extract_python_items +from specleft.discovery.miners.python.routes import PythonRouteMiner from specleft.discovery.miners.python.tests import PythonTestMiner -__all__ = ["PythonTestMiner", "extract_python_items"] +__all__ = ["PythonRouteMiner", "PythonTestMiner", "extract_python_items"] diff --git a/src/specleft/discovery/miners/python/routes.py b/src/specleft/discovery/miners/python/routes.py new file mode 100644 index 0000000..94da5e8 --- /dev/null +++ b/src/specleft/discovery/miners/python/routes.py @@ -0,0 +1,471 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""Python API-route miner for FastAPI, Flask, and Django.""" + +from __future__ import annotations + +import ast +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from specleft.discovery.context import MinerContext +from specleft.discovery.miners.shared.common import ( + elapsed_ms, + field_text, + line_number, + node_text, + walk_tree, +) +from specleft.discovery.models import ( + ApiRouteMeta, + DiscoveredItem, + ItemKind, + MinerErrorKind, + MinerResult, + SupportedLanguage, +) + +_SUPPORTED_FRAMEWORKS = frozenset({"fastapi", "flask", "django"}) +_FASTAPI_METHODS = frozenset( + {"get", "post", "put", "patch", "delete", "options", "head"} +) + + +@dataclass(frozen=True) +class _RouteMatch: + framework: str + path: str + http_method: str | list[str] + handler_name: str | None = None + response_model: str | None = None + + +class PythonRouteMiner: + """Extract HTTP routes from Python files using framework-specific patterns.""" + + miner_id = uuid.UUID("007ab65e-e4e8-4b7e-9c33-163635168071") + name = "python_api_routes" + languages = frozenset({SupportedLanguage.PYTHON}) + + def mine(self, ctx: MinerContext) -> MinerResult: + started = time.perf_counter() + frameworks = _enabled_frameworks(ctx) + if not frameworks: + return MinerResult( + miner_id=self.miner_id, + miner_name=self.name, + items=[], + duration_ms=elapsed_ms(started), + ) + + items: list[DiscoveredItem] = [] + parse_failures: list[Path] = [] + + for rel_path in ctx.file_index.files_by_language(SupportedLanguage.PYTHON): + abs_path = ctx.root / rel_path + parsed = ctx.registry.parse(abs_path) + if parsed is None: + parse_failures.append(rel_path) + continue + + root_node, language = parsed + if language is not SupportedLanguage.PYTHON: + continue + + try: + source_bytes = abs_path.read_bytes() + except OSError: + parse_failures.append(rel_path) + continue + + items.extend( + _extract_route_items( + root_node=root_node, + source_bytes=source_bytes, + file_path=rel_path, + frameworks=frameworks, + ) + ) + + error_kind: MinerErrorKind | None = None + error: str | None = None + if parse_failures: + error_kind = MinerErrorKind.PARSE_ERROR + files = ", ".join(path.as_posix() for path in parse_failures) + error = f"Failed to parse Python route files: {files}" + + return MinerResult( + miner_id=self.miner_id, + miner_name=self.name, + items=items, + error=error, + error_kind=error_kind, + duration_ms=elapsed_ms(started), + ) + + +def _enabled_frameworks(ctx: MinerContext) -> tuple[str, ...]: + detected = ctx.frameworks.get(SupportedLanguage.PYTHON, []) + enabled: list[str] = [] + for framework in detected: + if framework in _SUPPORTED_FRAMEWORKS and framework not in enabled: + enabled.append(framework) + return tuple(enabled) + + +def _extract_route_items( + *, + root_node: Any, + source_bytes: bytes, + file_path: Path, + frameworks: tuple[str, ...], +) -> list[DiscoveredItem]: + items: list[DiscoveredItem] = [] + framework_set = set(frameworks) + + if {"fastapi", "flask"} & framework_set: + items.extend( + _extract_decorated_route_items( + root_node=root_node, + source_bytes=source_bytes, + file_path=file_path, + frameworks=framework_set, + ) + ) + + if "django" in framework_set: + items.extend( + _extract_django_items(source_bytes=source_bytes, file_path=file_path) + ) + + return items + + +def _extract_decorated_route_items( + *, + root_node: Any, + source_bytes: bytes, + file_path: Path, + frameworks: set[str], +) -> list[DiscoveredItem]: + items: list[DiscoveredItem] = [] + + for node in walk_tree(root_node): + if node.type != "decorated_definition": + continue + + definition = node.child_by_field_name("definition") + if definition is None or definition.type not in { + "function_definition", + "async_function_definition", + }: + continue + + handler_name = field_text(definition, "name", source_bytes) + if not handler_name: + continue + + docstring = _extract_docstring(definition, source_bytes) + for decorator in _decorator_texts(node, source_bytes): + route = _parse_decorator_route(decorator, frameworks) + if route is None: + continue + + metadata = ApiRouteMeta( + http_method=route.http_method, + path=route.path, + framework=route.framework, + handler_name=handler_name, + has_docstring=docstring is not None, + docstring=docstring, + response_model=route.response_model, + ) + method = _display_method(route.http_method) + items.append( + DiscoveredItem( + kind=ItemKind.API_ROUTE, + name=f"{method} {route.path}", + file_path=file_path, + line_number=line_number(definition), + language=SupportedLanguage.PYTHON, + raw_text=docstring, + metadata=metadata.model_dump(), + confidence=0.9, + ) + ) + + return items + + +def _extract_django_items( + *, source_bytes: bytes, file_path: Path +) -> list[DiscoveredItem]: + try: + source_text = source_bytes.decode("utf-8", errors="ignore") + module = ast.parse(source_text) + except SyntaxError: + return [] + + docstrings = _module_function_docstrings(module) + items: list[DiscoveredItem] = [] + for call in _iter_urlpattern_calls(module): + route = _parse_django_call(call) + if route is None: + continue + handler_doc = docstrings.get(route.handler_name or "") + metadata = ApiRouteMeta( + http_method=route.http_method, + path=route.path, + framework=route.framework, + handler_name=route.handler_name, + has_docstring=handler_doc is not None, + docstring=handler_doc, + ) + items.append( + DiscoveredItem( + kind=ItemKind.API_ROUTE, + name=f"GET {route.path}", + file_path=file_path, + line_number=call.lineno, + language=SupportedLanguage.PYTHON, + raw_text=handler_doc, + metadata=metadata.model_dump(), + confidence=0.9, + ) + ) + return items + + +def _module_function_docstrings(module: ast.Module) -> dict[str, str]: + docstrings: dict[str, str] = {} + for node in module.body: + if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef): + continue + docstring = ast.get_docstring(node) + if docstring: + docstrings[node.name] = docstring + return docstrings + + +def _iter_urlpattern_calls(module: ast.Module) -> list[ast.Call]: + calls: list[ast.Call] = [] + for node in module.body: + target_value: ast.expr | None = None + if isinstance(node, ast.Assign) and any( + isinstance(target, ast.Name) and target.id == "urlpatterns" + for target in node.targets + ): + target_value = node.value + if ( + isinstance(node, ast.AnnAssign) + and isinstance(node.target, ast.Name) + and node.target.id == "urlpatterns" + ): + target_value = node.value + if not isinstance(target_value, ast.List | ast.Tuple): + continue + for element in target_value.elts: + if isinstance(element, ast.Call): + calls.append(element) + return calls + + +def _parse_django_call(call: ast.Call) -> _RouteMatch | None: + func_name = _name_from_expr(call.func) + if func_name not in {"path", "re_path"}: + return None + path = _first_string_arg(call) + if not path: + return None + handler: str | None = None + if len(call.args) >= 2: + handler = _name_from_expr(call.args[1]) + return _RouteMatch( + framework="django", + path=path, + http_method="GET", + handler_name=handler, + ) + + +def _parse_decorator_route( + decorator_text: str, + frameworks: set[str], +) -> _RouteMatch | None: + call = _parse_decorator_call(decorator_text) + if call is None: + return None + + if "fastapi" in frameworks: + fastapi = _parse_fastapi_call(call) + if fastapi is not None: + return fastapi + + if "flask" in frameworks: + flask = _parse_flask_call(call) + if flask is not None: + return flask + + return None + + +def _parse_fastapi_call(call: ast.Call) -> _RouteMatch | None: + if not isinstance(call.func, ast.Attribute): + return None + owner = _name_from_expr(call.func.value) + method = call.func.attr.lower() + if owner not in {"app", "router"} or method not in _FASTAPI_METHODS: + return None + path = _first_string_arg(call) + if not path: + return None + return _RouteMatch( + framework="fastapi", + path=path, + http_method=method.upper(), + response_model=_keyword_as_source(call, "response_model"), + ) + + +def _parse_flask_call(call: ast.Call) -> _RouteMatch | None: + if not isinstance(call.func, ast.Attribute): + return None + if call.func.attr != "route": + return None + path = _first_string_arg(call) + if not path: + return None + return _RouteMatch( + framework="flask", + path=path, + http_method=_parse_flask_methods(call), + ) + + +def _parse_flask_methods(call: ast.Call) -> list[str]: + raw_value = _keyword_value(call, "methods") + if raw_value is None: + return ["GET"] + + elements: list[str] = [] + if isinstance(raw_value, ast.Constant) and isinstance(raw_value.value, str): + elements = [raw_value.value] + elif isinstance(raw_value, ast.List | ast.Tuple | ast.Set): + for element in raw_value.elts: + if isinstance(element, ast.Constant) and isinstance(element.value, str): + elements.append(element.value) + + methods: list[str] = [] + for method in elements: + normalized = method.strip().upper() + if normalized and normalized not in methods: + methods.append(normalized) + return methods or ["GET"] + + +def _parse_decorator_call(decorator_text: str) -> ast.Call | None: + expression = decorator_text.strip() + if expression.startswith("@"): + expression = expression[1:] + + try: + parsed = ast.parse(expression, mode="eval") + except SyntaxError: + return None + + if not isinstance(parsed, ast.Expression) or not isinstance(parsed.body, ast.Call): + return None + return parsed.body + + +def _first_string_arg(call: ast.Call) -> str | None: + if not call.args: + return None + first = call.args[0] + if not isinstance(first, ast.Constant) or not isinstance(first.value, str): + return None + value = first.value.strip() + return value or None + + +def _keyword_value(call: ast.Call, keyword: str) -> ast.expr | None: + for item in call.keywords: + if item.arg == keyword: + return item.value + return None + + +def _keyword_as_source(call: ast.Call, keyword: str) -> str | None: + value = _keyword_value(call, keyword) + if value is None: + return None + try: + rendered = ast.unparse(value).strip() + except Exception: + return None + return rendered or None + + +def _name_from_expr(node: ast.expr) -> str | None: + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + return node.attr + return None + + +def _decorator_texts(node: Any, source_bytes: bytes) -> list[str]: + decorators: list[str] = [] + for child in getattr(node, "named_children", ()): + if child.type != "decorator": + continue + text = node_text(child, source_bytes).strip() + if text: + decorators.append(text) + return decorators + + +def _extract_docstring(function_node: Any, source_bytes: bytes) -> str | None: + body = function_node.child_by_field_name("body") + if body is None: + return None + children = list(getattr(body, "named_children", ())) + if not children or children[0].type != "expression_statement": + return None + for child in getattr(children[0], "named_children", ()): + if child.type in {"string", "concatenated_string"}: + return _clean_python_string(node_text(child, source_bytes)) + return None + + +def _clean_python_string(value: str) -> str | None: + stripped = value.strip() + if not stripped: + return None + try: + parsed = ast.literal_eval(stripped) + except (SyntaxError, ValueError): + parsed = _strip_wrapping_quotes(stripped) + if not isinstance(parsed, str): + return None + cleaned = parsed.strip() + return cleaned or None + + +def _strip_wrapping_quotes(value: str) -> str: + for quote in ('"""', "'''", '"', "'"): + if value.startswith(quote) and value.endswith(quote) and len(value) >= 2: + return value[len(quote) : len(value) - len(quote)].strip() + return value + + +def _display_method(http_method: str | list[str]) -> str: + if isinstance(http_method, str): + return http_method + if not http_method: + return "GET" + return http_method[0] diff --git a/tests/discovery/miners/test_python_routes.py b/tests/discovery/miners/test_python_routes.py new file mode 100644 index 0000000..10d0aee --- /dev/null +++ b/tests/discovery/miners/test_python_routes.py @@ -0,0 +1,349 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 SpecLeft Contributors + +"""Tests for Python API-route discovery miner.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from specleft.discovery.config import DiscoveryConfig +from specleft.discovery.context import MinerContext +from specleft.discovery.file_index import FileIndex +from specleft.discovery.miners.python.routes import PythonRouteMiner +from specleft.discovery.models import ( + ApiRouteMeta as _ApiRouteMeta, + MinerErrorKind, + SupportedLanguage, +) + + +@dataclass +class _FakeNode: + type: str + text_value: str = "" + children: list[_FakeNode] = field(default_factory=list) + named_children: list[_FakeNode] = field(default_factory=list) + fields: dict[str, _FakeNode] = field(default_factory=dict) + start_point: tuple[int, int] = (0, 0) + end_point: tuple[int, int] = (0, 0) + + @property + def text(self) -> bytes: + return self.text_value.encode("utf-8") + + def child_by_field_name(self, name: str) -> _FakeNode | None: + return self.fields.get(name) + + +class _RegistryStub: + def __init__( + self, mapping: dict[Path, tuple[Any, SupportedLanguage] | None] + ) -> None: + self._mapping = mapping + self.calls: list[Path] = [] + + def parse(self, file_path: Path) -> tuple[Any, SupportedLanguage] | None: + self.calls.append(file_path) + return self._mapping.get(file_path) + + +def _context( + root: Path, + registry: _RegistryStub, + *, + frameworks: dict[SupportedLanguage, list[str]] | None = None, +) -> MinerContext: + return MinerContext( + root=root, + registry=registry, # type: ignore[arg-type] + file_index=FileIndex(root), + frameworks=frameworks or {}, + config=DiscoveryConfig.default(), + ) + + +def _fixture_dir() -> Path: + return Path(__file__).resolve().parents[2] / "fixtures" / "discovery" + + +def _identifier(value: str, row: int) -> _FakeNode: + return _FakeNode( + type="identifier", + text_value=value, + start_point=(row, 0), + end_point=(row, len(value)), + ) + + +def _docstring_expr(value: str, row: int) -> _FakeNode: + string_node = _FakeNode( + type="string", + text_value=value, + start_point=(row, 0), + end_point=(row, len(value)), + ) + return _FakeNode( + type="expression_statement", + children=[string_node], + named_children=[string_node], + start_point=(row, 0), + end_point=(row, len(value)), + ) + + +def _python_function(name: str, row: int, docstring: str | None = None) -> _FakeNode: + name_node = _identifier(name, row) + body_children: list[_FakeNode] = [] + if docstring is not None: + body_children.append(_docstring_expr(docstring, row + 1)) + body = _FakeNode( + type="block", + children=list(body_children), + named_children=list(body_children), + start_point=(row + 1, 0), + end_point=(row + 2, 0), + ) + return _FakeNode( + type="function_definition", + children=[name_node, body], + named_children=[name_node, body], + fields={"name": name_node, "body": body}, + start_point=(row, 0), + end_point=(row + 3, 0), + ) + + +def _decorated_function( + *, + name: str, + row: int, + decorators: list[str], + docstring: str | None = None, +) -> _FakeNode: + decorator_nodes: list[_FakeNode] = [] + for index, decorator in enumerate(decorators): + decorator_nodes.append( + _FakeNode( + type="decorator", + text_value=decorator, + start_point=(row + index, 0), + end_point=(row + index, len(decorator)), + ) + ) + + function_node = _python_function( + name=name, + row=row + len(decorator_nodes), + docstring=docstring, + ) + children = [*decorator_nodes, function_node] + return _FakeNode( + type="decorated_definition", + children=children, + named_children=children, + fields={"definition": function_node}, + start_point=(row, 0), + end_point=function_node.end_point, + ) + + +def _module(children: list[_FakeNode]) -> _FakeNode: + return _FakeNode( + type="module", + children=list(children), + named_children=list(children), + start_point=(0, 0), + end_point=(max((child.end_point[0] for child in children), default=0) + 1, 0), + ) + + +def _fastapi_route_tree() -> _FakeNode: + return _module( + [ + _decorated_function( + name="get_user", + row=0, + decorators=['@app.get("/users/{id}", response_model=UserResponse)'], + docstring='"""Retrieve a user by ID"""', + ), + _decorated_function( + name="create_user", + row=6, + decorators=['@app.post("/users")'], + ), + _decorated_function( + name="update_user", + row=12, + decorators=['@router.patch("/users/{id}")'], + ), + ] + ) + + +def _flask_route_tree() -> _FakeNode: + return _module( + [ + _decorated_function( + name="items", + row=0, + decorators=['@bp.route("/items", methods=["GET", "POST"])'], + ), + _decorated_function( + name="health", + row=6, + decorators=['@app.route("/health")'], + ), + ] + ) + + +def test_python_route_miner_extracts_fastapi_routes_with_response_model( + tmp_path: Path, +) -> None: + fixture = _fixture_dir() / "sample_api.py" + api_file = tmp_path / "src" / "sample_api.py" + api_file.parent.mkdir(parents=True) + api_file.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + + registry = _RegistryStub( + {api_file: (_fastapi_route_tree(), SupportedLanguage.PYTHON)} + ) + + result = PythonRouteMiner().mine( + _context(tmp_path, registry, frameworks={SupportedLanguage.PYTHON: ["fastapi"]}) + ) + + assert result.error is None + assert result.error_kind is None + assert registry.calls == [api_file] + assert len(result.items) == 3 + assert all(item.language == SupportedLanguage.PYTHON for item in result.items) + assert all(isinstance(item.typed_meta(), _ApiRouteMeta) for item in result.items) + + by_name = {item.name: item for item in result.items} + assert set(by_name) == {"GET /users/{id}", "POST /users", "PATCH /users/{id}"} + assert by_name["GET /users/{id}"].metadata["framework"] == "fastapi" + assert by_name["GET /users/{id}"].metadata["response_model"] == "UserResponse" + assert by_name["GET /users/{id}"].metadata["has_docstring"] is True + assert by_name["GET /users/{id}"].metadata["docstring"] == "Retrieve a user by ID" + + +def test_python_route_miner_extracts_flask_routes_and_default_methods( + tmp_path: Path, +) -> None: + fixture = _fixture_dir() / "sample_api.py" + api_file = tmp_path / "src" / "sample_api.py" + api_file.parent.mkdir(parents=True) + api_file.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + + registry = _RegistryStub( + {api_file: (_flask_route_tree(), SupportedLanguage.PYTHON)} + ) + + result = PythonRouteMiner().mine( + _context(tmp_path, registry, frameworks={SupportedLanguage.PYTHON: ["flask"]}) + ) + + assert result.error is None + assert result.error_kind is None + assert len(result.items) == 2 + + by_path = {item.metadata["path"]: item for item in result.items} + assert by_path["/items"].metadata["http_method"] == ["GET", "POST"] + assert by_path["/items"].metadata["framework"] == "flask" + assert by_path["/health"].metadata["http_method"] == ["GET"] + + +def test_python_route_miner_extracts_django_path_and_re_path(tmp_path: Path) -> None: + api_file = tmp_path / "src" / "urls.py" + api_file.parent.mkdir(parents=True) + api_file.write_text( + "\n".join( + [ + "from django.urls import path, re_path", + "", + "def user_detail(request, user_id: int):", + ' """Show one user."""', + " return None", + "", + "def legacy_view(request, slug: str):", + " return None", + "", + "urlpatterns = [", + ' path("users//", user_detail, name="user-detail"),', + ' re_path(r"^legacy/(?P[-\\\\w]+)/$", legacy_view),', + "]", + "", + ] + ), + encoding="utf-8", + ) + + registry = _RegistryStub({api_file: (_module([]), SupportedLanguage.PYTHON)}) + + result = PythonRouteMiner().mine( + _context(tmp_path, registry, frameworks={SupportedLanguage.PYTHON: ["django"]}) + ) + + assert result.error is None + assert result.error_kind is None + assert len(result.items) == 2 + assert all(item.metadata["framework"] == "django" for item in result.items) + + by_path = {item.metadata["path"]: item for item in result.items} + assert by_path["users//"].name == "GET users//" + assert by_path["users//"].metadata["handler_name"] == "user_detail" + assert by_path["users//"].metadata["has_docstring"] is True + legacy_item = next( + item for item in result.items if item.metadata["path"].startswith("^legacy/") + ) + assert legacy_item.name.startswith("GET ^legacy/") + + +def test_python_route_miner_reports_parse_failures_and_keeps_valid_items( + tmp_path: Path, +) -> None: + fixture = _fixture_dir() / "sample_api.py" + good_file = tmp_path / "src" / "good.py" + bad_file = tmp_path / "src" / "bad.py" + good_file.parent.mkdir(parents=True) + good_file.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + bad_file.write_text("def broken(\n", encoding="utf-8") + + registry = _RegistryStub( + { + good_file: (_fastapi_route_tree(), SupportedLanguage.PYTHON), + bad_file: None, + } + ) + + result = PythonRouteMiner().mine( + _context(tmp_path, registry, frameworks={SupportedLanguage.PYTHON: ["fastapi"]}) + ) + + assert result.error_kind == MinerErrorKind.PARSE_ERROR + assert result.error is not None + assert "src/bad.py" in result.error + assert len(result.items) == 3 + + +def test_python_route_miner_skips_when_no_supported_framework(tmp_path: Path) -> None: + fixture = _fixture_dir() / "sample_api.py" + api_file = tmp_path / "src" / "sample_api.py" + api_file.parent.mkdir(parents=True) + api_file.write_text(fixture.read_text(encoding="utf-8"), encoding="utf-8") + + registry = _RegistryStub( + {api_file: (_fastapi_route_tree(), SupportedLanguage.PYTHON)} + ) + result = PythonRouteMiner().mine( + _context(tmp_path, registry, frameworks={SupportedLanguage.PYTHON: ["pytest"]}) + ) + + assert result.error is None + assert result.error_kind is None + assert result.items == [] + assert registry.calls == [] diff --git a/tests/fixtures/discovery/sample_api.py b/tests/fixtures/discovery/sample_api.py index 1933559..018afa6 100644 --- a/tests/fixtures/discovery/sample_api.py +++ b/tests/fixtures/discovery/sample_api.py @@ -1,11 +1,54 @@ """Sample API module for discovery tests.""" +from django.urls import path, re_path +from fastapi import APIRouter, FastAPI +from flask import Blueprint, Flask +app = FastAPI() +router = APIRouter() +flask_app = Flask(__name__) +bp = Blueprint("items", __name__) + + +class UserResponse: ... + + +@app.get("/users/{id}", response_model=UserResponse) def get_user(user_id: int) -> dict[str, int]: - """Return a fake user payload.""" + """Retrieve a user by ID.""" return {"id": user_id} +@app.post("/users") def create_user(payload: dict[str, str]) -> dict[str, str]: return payload + + +@router.patch("/users/{id}") +def update_user(user_id: int, payload: dict[str, str]) -> dict[str, str]: + return {"id": user_id, **payload} + + +@bp.route("/items", methods=["GET", "POST"]) +def items() -> list[str]: + return ["a", "b"] + + +@flask_app.route("/health") +def health() -> tuple[str, int]: + return "ok", 200 + + +def user_detail(request: object, user_id: int) -> None: + """Show one user.""" + + +def legacy_view(request: object, slug: str) -> None: + return None + + +urlpatterns = [ + path("users//", user_detail, name="user-detail"), + re_path(r"^legacy/(?P[-\w]+)/$", legacy_view), +]