diff --git a/openapi_core/wrappers/flask.py b/openapi_core/wrappers/flask.py index ab91665..14c330b 100644 --- a/openapi_core/wrappers/flask.py +++ b/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 @@ class FlaskOpenAPIRequest(BaseOpenAPIRequest): 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): diff --git a/tests/integration/data/v3.0/flask_wrapper.yaml b/tests/integration/data/v3.0/flask_wrapper.yaml new file mode 100644 index 0000000..0dcfdd3 --- /dev/null +++ b/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. diff --git a/tests/integration/test_wrappers.py b/tests/integration/test_wrappers.py index ff0c157..6c4b7d1 100644 --- a/tests/integration/test_wrappers.py +++ b/tests/integration/test_wrappers.py @@ -1,55 +1,62 @@ -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//', endpoint='kb/browse'), + Rule('/browse//', endpoint='kb/browse') + ]) + ], default_subdomain='www') + + +@pytest.fixture +def request_factory(map, environ_factory): + server_name = 'localhost' + + 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 response_factory(): + def create_response(data, status_code=200): + return Response(data, status=status_code) + return create_response class TestFlaskOpenAPIRequest(object): - server_name = 'localhost' - - @pytest.fixture - def environ_factory(self): - return create_environ - - @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//', endpoint='kb/browse'), - Rule('/browse//', 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 openapi_request(self, request): - return FlaskOpenAPIRequest(request) - def test_simple(self, request_factory, request): request = request_factory('GET', '/', subdomain='www') @@ -115,19 +122,13 @@ class TestFlaskOpenAPIRequest(object): 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 @@ class TestFlaskOpenAPIResponse(object): 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