Merge pull request #1 from acsone/16.0-fastapi-multislash-route-lmi

[IMP] fastapi: Ensures endpoint matching is based on path parts

Author: PicchiSeba

fastapi/demo/fastapi_endpoint_demo.xml

@@ -51,7 +51,7 @@ methods. See documentation to learn more about how to create a new app.
         Like the other demo endpoint but with multi-slash
     </field>
     <field name="app">demo</field>
-    <field name="root_path">/fastapi/demo</field>
+    <field name="root_path">/fastapi/demo-multi</field>
     <field name="demo_auth_method">http_basic</field>
     <field name="user_id" ref="my_demo_app_user" />
   </record>

fastapi/models/fastapi_endpoint.py

@@ -198,6 +198,8 @@ def _endpoint_registry_route_unique_key(self, routing: Dict[str, Any]):
         return f"{self._name}:{self.id}:{path}"
 
     def _reset_app(self):
+        self._get_id_by_root_path_map.clear_cache(self)
+        self._get_id_for_path.clear_cache(self)
         self._reset_app_cache_marker.clear_cache(self)
 
     @tools.ormcache()
@@ -211,24 +213,71 @@ def _reset_app_cache_marker(self):
         """
 
     @api.model
-    @tools.ormcache("path")
-    def get_endpoint(self, path):
-        # try to match the request url with the most similar endpoint
-        endpoints_by_length = self.search([]).sorted(
-            lambda fe: len(fe.root_path), reverse=True
+    def _normalize_url_path(self, path) -> str:
+        """
+        Normalize a URL path:
+        * Remove redundant slashes,
+        * Remove trailing slash (unless it's the root),
+        * Lowercase for case-insensitive matching
+        """
+        parts = [part.lower() for part in path.strip().split("/") if part]
+        return "/" + "/".join(parts)
+
+    @api.model
+    def _is_suburl(self, path, prefix) -> bool:
+        """
+        Check if 'path' is a subpath of 'prefix' in URL logic:
+        * Must start with the prefix followed by a slash
+          This will ensure that the matching is done one the path
+          parts and ensures that e.g. /a/b is not prefix of /a/bc.
+        """
+        path = self._normalize_url_path(path)
+        prefix = self._normalize_url_path(prefix)
+
+        if path == prefix:
+            return True
+        if path.startswith(prefix + "/"):
+            return True
+        return False
+
+    @api.model
+    def _find_first_matching_url_path(self, paths, prefix) -> str | None:
+        """
+        Return the first path that is a subpath of 'prefix',
+        ordered by longest URL path first (most number of segments).
+        """
+        # Sort by number of segments (shallowest first)
+        sorted_paths = sorted(
+            paths,
+            key=lambda p: len(self._normalize_url_path(p).split("/")),
+            reverse=True,
         )
-        endpoint = False
-        while endpoints_by_length:
-            candidate_endpoint = endpoints_by_length[0]
-            if path.startswith(candidate_endpoint.root_path):
-                endpoint = candidate_endpoint
-                break
-            endpoints_by_length -= candidate_endpoint
-        return endpoint
+
+        for path in sorted_paths:
+            if self._is_suburl(prefix, path):
+                return path
+        return None
+
+    @api.model
+    @tools.ormcache()
+    def _get_id_by_root_path_map(self):
+        return {r.root_path: r.id for r in self.search([])}
+
+    @api.model
+    @tools.ormcache("path")
+    def _get_id_for_path(self, path):
+        id_by_path = self._get_id_by_root_path_map()
+        root_path = self._find_first_matching_url_path(id_by_path.keys(), path)
+        return id_by_path.get(root_path)
+
+    @api.model
+    def _get_endpoint(self, path):
+        id_ = self._get_id_for_path(path)
+        return self.browse(id_) if id_ else None
 
     @api.model
     def get_app(self, path):
-        record = self.get_endpoint(path)
+        record = self._get_endpoint(path)
         if not record:
             return None
         app = FastAPI()
@@ -254,7 +303,7 @@ def _clear_fastapi_exception_handlers(self, app: FastAPI) -> None:
     @api.model
     @tools.ormcache("path")
     def get_uid(self, path):
-        record = self.get_endpoint(path)
+        record = self._get_endpoint(path)
         if not record:
             return None
         return record.user_id.id

fastapi/tests/__init__.py

@@ -1,3 +1,2 @@
 from . import test_fastapi
 from . import test_fastapi_demo
-from . import test_fastapi_multislash_demo

fastapi/tests/test_fastapi.py

@@ -20,7 +20,11 @@ class FastAPIHttpCase(HttpCase):
     def setUpClass(cls):
         super().setUpClass()
         cls.fastapi_demo_app = cls.env.ref("fastapi.fastapi_endpoint_demo")
-        cls.fastapi_demo_app._handle_registry_sync()
+        cls.fastapi_multi_demo_app = cls.env.ref(
+            "fastapi.fastapi_endpoint_multislash_demo"
+        )
+        cls.fastapi_apps = cls.fastapi_demo_app + cls.fastapi_multi_demo_app
+        cls.fastapi_apps._handle_registry_sync()
         lang = (
             cls.env["res.lang"]
             .with_context(active_test=False)
@@ -169,3 +173,29 @@ def test_no_commit_on_exception(self) -> None:
             expected_message="test",
             expected_status_code=status.HTTP_409_CONFLICT,
         )
+
+    def test_url_matching(self):
+        # Test the URL mathing method on the endpoint
+        paths = ["/fastapi", "/fastapi_demo", "/fastapi/v1"]
+        EndPoint = self.env["fastapi.endpoint"]
+        self.assertEqual(
+            EndPoint._find_first_matching_url_path(paths, "/fastapi_demo/test"),
+            "/fastapi_demo",
+        )
+        self.assertEqual(
+            EndPoint._find_first_matching_url_path(paths, "/fastapi/test"), "/fastapi"
+        )
+        self.assertEqual(
+            EndPoint._find_first_matching_url_path(paths, "/fastapi/v2/test"),
+            "/fastapi",
+        )
+        self.assertEqual(
+            EndPoint._find_first_matching_url_path(paths, "/fastapi/v1/test"),
+            "/fastapi/v1",
+        )
+
+    def test_multi_slash(self):
+        route = "/fastapi/demo-multi/demo/"
+        response = self.url_open(route, timeout=20)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(self.fastapi_multi_demo_app.root_path, str(response.url))