From 548e042a05c5e1359abe1a5c6642e6fa0c249ccc Mon Sep 17 00:00:00 2001 From: touale <136764239@qq.com> Date: Tue, 6 Jan 2026 16:32:39 +0800 Subject: [PATCH 1/3] feat: add duplicate route detection in APIIngress --- src/framex/driver/ingress.py | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/src/framex/driver/ingress.py b/src/framex/driver/ingress.py index 1739184..b1aef55 100644 --- a/src/framex/driver/ingress.py +++ b/src/framex/driver/ingress.py @@ -1,3 +1,4 @@ +import re from collections.abc import Callable from enum import Enum from typing import Any @@ -123,7 +124,7 @@ def _verify_api_key(request: Request, api_key: str | None = Depends(api_key_head dependencies.append(Depends(_verify_api_key)) - app.add_api_route( + self.add_api_route( path, route_handler, methods=methods, @@ -146,3 +147,24 @@ async def inner(self) -> str: # pragma: no cover def __repr__(self): return BACKEND_NAME + + def add_api_route( + self, + path: str, + endpoint: Callable[..., Any], + *, + methods: list[str] | None = None, + **kwargs: Any, + ) -> None: + method_set: set[str] = {m.upper() for m in (methods or [])} + norm_path = re.sub(r"\{[^}]+\}", "{}", path) + + for route in app.routes: + if ( + isinstance(route, APIRoute) + and re.sub(r"\{[^}]+\}", "{}", route.path) == norm_path + and route.methods & method_set + ): + raise RuntimeError(f"Duplicate API route: {sorted(method_set)} {norm_path}") + + app.add_api_route(path, endpoint, methods=list(method_set), **kwargs) From 73ee04de5ca71f77770944275b9a1f27dce0d059 Mon Sep 17 00:00:00 2001 From: touale <136764239@qq.com> Date: Tue, 6 Jan 2026 16:41:26 +0800 Subject: [PATCH 2/3] test: add comprehensive APIIngress route validation tests --- tests/driver/test_ingress.py | 111 +++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) create mode 100644 tests/driver/test_ingress.py diff --git a/tests/driver/test_ingress.py b/tests/driver/test_ingress.py new file mode 100644 index 0000000..da61453 --- /dev/null +++ b/tests/driver/test_ingress.py @@ -0,0 +1,111 @@ +from unittest.mock import Mock, patch + +import pytest +from fastapi.routing import APIRoute +from starlette.routing import Route + +from framex.driver.ingress import APIIngress + +# ---------- helpers ---------- + + +def make_route(path: str, methods: set[str]) -> APIRoute: + route = Mock(spec=APIRoute) + route.path = path + route.methods = methods + return route + + +# ---------- fixtures ---------- + + +@pytest.fixture +def mock_app(): + with patch("framex.driver.ingress.app") as app: + app.routes = [] + app.add_api_route = Mock() + yield app + + +@pytest.fixture +def ingress(): + return APIIngress.__new__(APIIngress) + + +# ---------- tests ---------- + + +def test_add_first_route_success(ingress, mock_app): + endpoint = Mock() + + ingress.add_api_route("/users", endpoint, methods=["GET"]) + + mock_app.add_api_route.assert_called_once_with("/users", endpoint, methods=["GET"]) + + +@pytest.mark.parametrize( + ("existing_path", "new_path"), + [ + ("/users/{id}", "/users/{id}"), + ("/users/{id}", "/users/{user_id}"), + ("/users/{uid}/posts/{pid}", "/users/{id}/posts/{post_id}"), + ], +) +def test_duplicate_path_same_method_raises(ingress, mock_app, existing_path, new_path): + mock_app.routes = [make_route(existing_path, {"GET"})] + + with pytest.raises(RuntimeError, match=r"Duplicate API route"): + ingress.add_api_route(new_path, Mock(), methods=["GET"]) + + +def test_same_path_different_method_allowed(ingress, mock_app): + mock_app.routes = [make_route("/users/{id}", {"GET"})] + + ingress.add_api_route("/users/{id}", Mock(), methods=["POST"]) + + mock_app.add_api_route.assert_called_once() + + +def test_overlapping_methods_raises(ingress, mock_app): + mock_app.routes = [make_route("/users", {"GET", "POST"})] + + with pytest.raises(RuntimeError): + ingress.add_api_route("/users", Mock(), methods=["POST", "PUT"]) + + +def test_case_insensitive_methods(ingress, mock_app): + mock_app.routes = [make_route("/users", {"GET"})] + + with pytest.raises(RuntimeError): + ingress.add_api_route("/users", Mock(), methods=["get"]) + + +def test_none_methods_becomes_empty_list(ingress, mock_app): + ingress.add_api_route("/users", Mock(), methods=None) + + _, kwargs = mock_app.add_api_route.call_args + assert kwargs["methods"] == [] + + +def test_non_api_route_is_ignored(ingress, mock_app): + non_api_route = Mock(spec=Route) + non_api_route.path = "/users/{id}" + mock_app.routes = [non_api_route] + + ingress.add_api_route("/users/{id}", Mock(), methods=["GET"]) + + mock_app.add_api_route.assert_called_once() + + +def test_kwargs_are_passed_through(ingress, mock_app): + ingress.add_api_route( + "/users", + Mock(), + methods=["GET"], + tags=["users"], + response_class=Mock(), + ) + + _, kwargs = mock_app.add_api_route.call_args + assert kwargs["tags"] == ["users"] + assert "response_class" in kwargs From 4875ab2cf02b767a315943c340cf7aeb26fd9df5 Mon Sep 17 00:00:00 2001 From: touale <136764239@qq.com> Date: Tue, 6 Jan 2026 16:44:45 +0800 Subject: [PATCH 3/3] fix: default methods to GET when methods is None --- src/framex/driver/ingress.py | 2 +- tests/driver/test_ingress.py | 7 ------- 2 files changed, 1 insertion(+), 8 deletions(-) diff --git a/src/framex/driver/ingress.py b/src/framex/driver/ingress.py index b1aef55..19a859d 100644 --- a/src/framex/driver/ingress.py +++ b/src/framex/driver/ingress.py @@ -156,7 +156,7 @@ def add_api_route( methods: list[str] | None = None, **kwargs: Any, ) -> None: - method_set: set[str] = {m.upper() for m in (methods or [])} + method_set: set[str] = {m.upper() for m in methods} if methods else {"GET"} norm_path = re.sub(r"\{[^}]+\}", "{}", path) for route in app.routes: diff --git a/tests/driver/test_ingress.py b/tests/driver/test_ingress.py index da61453..fd1a5aa 100644 --- a/tests/driver/test_ingress.py +++ b/tests/driver/test_ingress.py @@ -80,13 +80,6 @@ def test_case_insensitive_methods(ingress, mock_app): ingress.add_api_route("/users", Mock(), methods=["get"]) -def test_none_methods_becomes_empty_list(ingress, mock_app): - ingress.add_api_route("/users", Mock(), methods=None) - - _, kwargs = mock_app.add_api_route.call_args - assert kwargs["methods"] == [] - - def test_non_api_route_is_ignored(ingress, mock_app): non_api_route = Mock(spec=Route) non_api_route.path = "/users/{id}"