Merge pull request #141 from bjmc/flask_params

Modify FlaskOpenAPIRequest to accomodate path variables

Author: p1c2u

openapi_core/wrappers/flask.py

@@ -1,9 +1,16 @@
 """OpenAPI core wrappers module"""
+import re
+
 from openapi_core.wrappers.base import BaseOpenAPIRequest, BaseOpenAPIResponse
 
+# http://flask.pocoo.org/docs/1.0/quickstart/#variable-rules
+PATH_PARAMETER_PATTERN = r'<(?:(?:string|int|float|path|uuid):)?(\w+)>'
+
 
 class FlaskOpenAPIRequest(BaseOpenAPIRequest):
 
+    path_regex = re.compile(PATH_PARAMETER_PATTERN)
+
     def __init__(self, request):
         self.request = request
 
@@ -24,7 +31,7 @@ def path_pattern(self):
         if self.request.url_rule is None:
             return self.path
 
-        return self.request.url_rule.rule
+        return self.path_regex.sub(r'{\1}', self.request.url_rule.rule)
 
     @property
     def parameters(self):

tests/integration/data/v3.0/flask_wrapper.yaml

@@ -0,0 +1,19 @@
+openapi: "3.0.0"
+info:
+  title: Basic OpenAPI specification used with test_wrappers.TestFlaskOpenAPIIValidation
+  version: "0.1"
+servers:
+  - url: 'http://localhost'
+paths:
+  '/browse/{id}/':
+    parameters:
+      - name: id
+        in: path
+        required: true
+        description: the ID of the resource to retrieve
+        schema:
+          type: integer
+    get:
+      responses:
+        default:
+          description: Return the resource.

tests/integration/test_wrappers.py

@@ -1,54 +1,61 @@
-import pytest
-
 from flask.wrappers import Request, Response
 from werkzeug.datastructures import EnvironHeaders, ImmutableMultiDict
 from werkzeug.routing import Map, Rule, Subdomain
 from werkzeug.test import create_environ
 
-from openapi_core.wrappers.flask import (
-    FlaskOpenAPIRequest, FlaskOpenAPIResponse,
-)
-
+import pytest
+from openapi_core.shortcuts import create_spec
+from openapi_core.validation.response.validators import ResponseValidator
+from openapi_core.validation.request.validators import RequestValidator
+from openapi_core.wrappers.flask import (FlaskOpenAPIRequest,
+                                         FlaskOpenAPIResponse)
+
+
+@pytest.fixture
+def environ_factory():
+    return create_environ
+
+
+@pytest.fixture
+def map():
+    return Map([
+        # Static URLs
+        Rule('/', endpoint='static/index'),
+        Rule('/about', endpoint='static/about'),
+        Rule('/help', endpoint='static/help'),
+        # Knowledge Base
+        Subdomain('kb', [
+            Rule('/', endpoint='kb/index'),
+            Rule('/browse/', endpoint='kb/browse'),
+            Rule('/browse/<int:id>/', endpoint='kb/browse'),
+            Rule('/browse/<int:id>/<int:page>', endpoint='kb/browse')
+        ])
+    ], default_subdomain='www')
 
-class TestFlaskOpenAPIRequest(object):
 
+@pytest.fixture
+def request_factory(map, environ_factory):
     server_name = 'localhost'
 
-    @pytest.fixture
-    def environ_factory(self):
-        return create_environ
+    def create_request(method, path, subdomain=None, query_string=None):
+        environ = environ_factory(query_string=query_string)
+        req = Request(environ)
+        urls = map.bind_to_environ(
+            environ, server_name=server_name, subdomain=subdomain)
+        req.url_rule, req.view_args = urls.match(
+            path, method, return_rule=True)
+        return req
+    return create_request
 
-    @pytest.fixture
-    def map(self):
-        return Map([
-            # Static URLs
-            Rule('/', endpoint='static/index'),
-            Rule('/about', endpoint='static/about'),
-            Rule('/help', endpoint='static/help'),
-            # Knowledge Base
-            Subdomain('kb', [
-                Rule('/', endpoint='kb/index'),
-                Rule('/browse/', endpoint='kb/browse'),
-                Rule('/browse/<int:id>/', endpoint='kb/browse'),
-                Rule('/browse/<int:id>/<int:page>', endpoint='kb/browse')
-            ])
-        ], default_subdomain='www')
 
-    @pytest.fixture
-    def request_factory(self, map, environ_factory):
-        def create_request(method, path, subdomain=None, query_string=None):
-            environ = environ_factory(query_string=query_string)
-            req = Request(environ)
-            urls = map.bind_to_environ(
-                environ, server_name=self.server_name, subdomain=subdomain)
-            req.url_rule, req.view_args = urls.match(
-                path, method, return_rule=True)
-            return req
-        return create_request
+@pytest.fixture
+def response_factory():
+    def create_response(data, status_code=200):
+        return Response(data, status=status_code)
+    return create_response
 
-    @pytest.fixture
-    def openapi_request(self, request):
-        return FlaskOpenAPIRequest(request)
+
+class TestFlaskOpenAPIRequest(object):
 
     def test_simple(self, request_factory, request):
         request = request_factory('GET', '/', subdomain='www')
@@ -115,19 +122,13 @@ def test_url_rule(self, request_factory, request):
         assert openapi_request.host_url == request.host_url
         assert openapi_request.path == request.path
         assert openapi_request.method == request.method.lower()
-        assert openapi_request.path_pattern == request.url_rule.rule
+        assert openapi_request.path_pattern == '/browse/{id}/'
         assert openapi_request.body == request.data
         assert openapi_request.mimetype == request.mimetype
 
 
 class TestFlaskOpenAPIResponse(object):
 
-    @pytest.fixture
-    def response_factory(self):
-        def create_response(data, status_code=200):
-            return Response(data, status=status_code)
-        return create_response
-
     def test_invalid_server(self, response_factory):
         response = response_factory('Not Found', status_code=404)
 
@@ -137,3 +138,30 @@ def test_invalid_server(self, response_factory):
         assert openapi_response.data == response.data
         assert openapi_response.status_code == response._status_code
         assert openapi_response.mimetype == response.mimetype
+
+
+class TestFlaskOpenAPIValidation(object):
+
+    @pytest.fixture
+    def flask_spec(self, factory):
+        specfile = 'data/v3.0/flask_wrapper.yaml'
+        return create_spec(factory.spec_from_file(specfile))
+
+    def test_response_validator_path_pattern(self,
+                                             flask_spec,
+                                             request_factory,
+                                             response_factory):
+        validator = ResponseValidator(flask_spec)
+        request = request_factory('GET', '/browse/12/', subdomain='kb')
+        openapi_request = FlaskOpenAPIRequest(request)
+        response = response_factory('Some item', status_code=200)
+        openapi_response = FlaskOpenAPIResponse(response)
+        result = validator.validate(openapi_request, openapi_response)
+        assert not result.errors
+
+    def test_request_validator_path_pattern(self, flask_spec, request_factory):
+        validator = RequestValidator(flask_spec)
+        request = request_factory('GET', '/browse/12/', subdomain='kb')
+        openapi_request = FlaskOpenAPIRequest(request)
+        result = validator.validate(openapi_request)
+        assert not result.errors