Merge pull request #449 from p1c2u/fix/wetkzeug-flask-root-path-fix

werkzeug flask root path fix

Author: p1c2u

openapi_core/contrib/flask/requests.py

@@ -21,6 +21,7 @@ def __init__(self, request: Request):
     @property
     def path_pattern(self) -> str:
         if self.request.url_rule is None:
-            return self.request.path
+            return self.path
 
-        return self.path_regex.sub(r"{\1}", self.request.url_rule.rule)
+        path = self.get_path(self.request.url_rule.rule)
+        return self.path_regex.sub(r"{\1}", path)

openapi_core/contrib/werkzeug/requests.py

@@ -31,7 +31,7 @@ def host_url(self) -> str:
 
     @property
     def path(self) -> str:
-        return self.request.path
+        return self.get_path(self.request.path)
 
     @property
     def method(self) -> str:
@@ -44,3 +44,6 @@ def body(self) -> Optional[str]:
     @property
     def mimetype(self) -> str:
         return self.request.mimetype
+
+    def get_path(self, path: str) -> str:
+        return "".join([self.request.root_path, path])

tests/integration/contrib/flask/test_flask_validator.py

@@ -0,0 +1,73 @@
+from json import dumps
+
+import pytest
+from flask import Flask
+from flask.testing import FlaskClient
+from flask.wrappers import Response
+
+from openapi_core.contrib.flask import FlaskOpenAPIRequest
+from openapi_core.validation.request import openapi_request_validator
+
+
+class TestWerkzeugOpenAPIValidation:
+    @pytest.fixture
+    def spec(self, factory):
+        specfile = "contrib/requests/data/v3.0/requests_factory.yaml"
+        return factory.spec_from_file(specfile)
+
+    @pytest.fixture
+    def app(self):
+        app = Flask("__main__", root_path="/browse")
+        app.config["DEBUG"] = True
+        app.config["TESTING"] = True
+        return app
+
+    @pytest.fixture
+    def details_view_func(self, spec):
+        def datails_browse(id):
+            from flask import request
+
+            openapi_request = FlaskOpenAPIRequest(request)
+            result = openapi_request_validator.validate(spec, openapi_request)
+            assert not result.errors
+
+            if request.args.get("q") == "string":
+                return Response(
+                    dumps({"data": "data"}),
+                    headers={"X-Rate-Limit": "12"},
+                    mimetype="application/json",
+                    status=200,
+                )
+            else:
+                return Response("Not Found", status=404)
+
+        return datails_browse
+
+    @pytest.fixture(autouse=True)
+    def view(self, app, details_view_func):
+        app.add_url_rule(
+            "/<id>/",
+            view_func=details_view_func,
+            methods=["POST"],
+        )
+
+    @pytest.fixture
+    def client(self, app):
+        return FlaskClient(app)
+
+    def test_request_validator_root_path(self, client):
+        query_string = {
+            "q": "string",
+        }
+        headers = {"content-type": "application/json"}
+        data = {"param1": 1}
+        result = client.post(
+            "/12/",
+            base_url="http://localhost/browse",
+            query_string=query_string,
+            json=data,
+            headers=headers,
+        )
+
+        assert result.status_code == 200
+        assert result.json == {"data": "data"}

tests/integration/contrib/werkzeug/test_werkzeug_validation.py

@@ -1,7 +1,6 @@
 from json import dumps
 
 import pytest
-import requests
 import responses
 from werkzeug.test import Client
 from werkzeug.wrappers import Request
@@ -40,6 +39,23 @@ def test_app(environ, start_response):
     def client(self, app):
         return Client(app)
 
+    def test_request_validator_root_path(self, client, spec):
+        query_string = {
+            "q": "string",
+        }
+        headers = {"content-type": "application/json"}
+        data = {"param1": 1}
+        response = client.post(
+            "/12/",
+            base_url="http://localhost/browse",
+            query_string=query_string,
+            json=data,
+            headers=headers,
+        )
+        openapi_request = WerkzeugOpenAPIRequest(response.request)
+        result = openapi_request_validator.validate(spec, openapi_request)
+        assert not result.errors
+
     def test_request_validator_path_pattern(self, client, spec):
         query_string = {
             "q": "string",