FlaskOpenAPIView handler change

This commit is contained in:
p1c2u 2020-01-26 23:29:19 +00:00
parent 0f7fa5287e
commit b5aabf213c
2 changed files with 120 additions and 23 deletions

View file

@ -2,6 +2,7 @@
from flask.views import MethodView from flask.views import MethodView
from openapi_core.contrib.flask.decorators import FlaskOpenAPIViewDecorator from openapi_core.contrib.flask.decorators import FlaskOpenAPIViewDecorator
from openapi_core.contrib.flask.handlers import FlaskOpenAPIErrorsHandler
from openapi_core.validation.request.validators import RequestValidator from openapi_core.validation.request.validators import RequestValidator
from openapi_core.validation.response.validators import ResponseValidator from openapi_core.validation.response.validators import ResponseValidator
@ -9,34 +10,18 @@ from openapi_core.validation.response.validators import ResponseValidator
class FlaskOpenAPIView(MethodView): class FlaskOpenAPIView(MethodView):
"""Brings OpenAPI specification validation and unmarshalling for views.""" """Brings OpenAPI specification validation and unmarshalling for views."""
def __init__(self, request_validator, response_validator): openapi_errors_handler = FlaskOpenAPIErrorsHandler
super(MethodView, self).__init__()
self.request_validator = request_validator def __init__(self, spec):
self.response_validator = response_validator super(FlaskOpenAPIView, self).__init__()
self.request_validator = RequestValidator(spec)
self.response_validator = ResponseValidator(spec)
def dispatch_request(self, *args, **kwargs): def dispatch_request(self, *args, **kwargs):
decorator = FlaskOpenAPIViewDecorator( decorator = FlaskOpenAPIViewDecorator(
request_validator=self.request_validator, request_validator=self.request_validator,
response_validator=self.response_validator, response_validator=self.response_validator,
openapi_errors_handler=self.handle_openapi_errors, openapi_errors_handler=self.openapi_errors_handler,
) )
return decorator(super(FlaskOpenAPIView, self).dispatch_request)( return decorator(super(FlaskOpenAPIView, self).dispatch_request)(
*args, **kwargs) *args, **kwargs)
def handle_openapi_errors(self, errors):
"""Handles OpenAPI request/response errors.
Should return response object::
class MyView(FlaskOpenAPIView):
def handle_openapi_errors(self, errors):
return jsonify({'errors': errors})
"""
raise NotImplementedError
@classmethod
def from_spec(cls, spec):
request_validator = RequestValidator(spec)
response_validator = ResponseValidator(spec)
return cls(request_validator, response_validator)

View file

@ -0,0 +1,112 @@
from flask import Flask, make_response, jsonify
import pytest
from openapi_core.contrib.flask.views import FlaskOpenAPIView
from openapi_core.shortcuts import create_spec
class TestFlaskOpenAPIView(object):
view_response = None
@pytest.fixture
def spec(self, factory):
specfile = 'contrib/flask/data/v3.0/flask_factory.yaml'
return create_spec(factory.spec_from_file(specfile))
@pytest.fixture
def app(self):
app = Flask("__main__")
app.config['DEBUG'] = True
app.config['TESTING'] = True
return app
@pytest.yield_fixture
def client(self, app):
with app.test_client() as client:
with app.app_context():
yield client
@pytest.fixture
def view_func(self, spec):
outer = self
class MyView(FlaskOpenAPIView):
def get(self, id):
return outer.view_response
return MyView.as_view('browse_details', spec)
@pytest.fixture(autouse=True)
def view(self, app, view_func):
app.add_url_rule("/browse/<id>/", view_func=view_func)
def test_invalid_content_type(self, client):
self.view_response = make_response('success', 200)
result = client.get('/browse/12/')
assert result.json == {
'errors': [
{
'class': (
"<class 'openapi_core.schema.media_types.exceptions."
"InvalidContentType'>"
),
'status': 415,
'title': (
'Content for following mimetype not found: text/html'
)
}
]
}
def test_server_error(self, client):
result = client.get('/browse/12/', base_url='https://localhost')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.schema.servers.exceptions."
"InvalidServer'>"
),
'status': 500,
'title': (
'Invalid request server '
'https://localhost/browse/{id}/'
),
}
]
}
assert result.json == expected_data
def test_endpoint_error(self, client):
result = client.get('/browse/invalidparameter/')
expected_data = {
'errors': [
{
'class': (
"<class 'openapi_core.schema.parameters."
"exceptions.InvalidParameterValue'>"
),
'status': 400,
'title': (
'Invalid parameter value for `id`: '
'Failed to cast value invalidparameter to type '
'SchemaType.INTEGER'
)
}
]
}
assert result.json == expected_data
def test_valid(self, client):
self.view_response = jsonify(data='data')
result = client.get('/browse/12/')
assert result.status_code == 200
assert result.json == {
'data': 'data',
}