Request validator

Author: p1c2u

openapi_core/exceptions.py

@@ -13,6 +13,10 @@ class MissingParameterError(OpenAPIMappingError):
     pass
 
 
+class MissingBodyError(OpenAPIMappingError):
+    pass
+
+
 class MissingPropertyError(OpenAPIMappingError):
     pass
 

openapi_core/request_bodies.py

@@ -1,6 +1,7 @@
 """OpenAPI core requestBodies module"""
 from functools import lru_cache
 
+from openapi_core.exceptions import InvalidContentTypeError
 from openapi_core.media_types import MediaTypeGenerator
 
 
@@ -12,7 +13,11 @@ def __init__(self, content, required=False):
         self.required = required
 
     def __getitem__(self, mimetype):
-        return self.content[mimetype]
+        try:
+            return self.content[mimetype]
+        except KeyError:
+            raise InvalidContentTypeError(
+                "Invalid mime type `{0}`".format(mimetype))
 
 
 class RequestBodyFactory(object):

openapi_core/specs.py

@@ -6,7 +6,7 @@
 from openapi_spec_validator import openapi_v3_spec_validator
 
 from openapi_core.components import ComponentsFactory
-from openapi_core.exceptions import InvalidOperationError
+from openapi_core.exceptions import InvalidOperationError, InvalidServerError
 from openapi_core.infos import InfoFactory
 from openapi_core.paths import PathsGenerator
 from openapi_core.schemas import SchemaRegistry
@@ -32,6 +32,14 @@ def __getitem__(self, path_name):
     def default_url(self):
         return self.servers[0].default_url
 
+    def get_server(self, full_url_pattern):
+        for spec_server in self.servers:
+            if spec_server.default_url in full_url_pattern:
+                return spec_server
+
+        raise InvalidServerError(
+            "Invalid request server {0}".format(full_url_pattern))
+
     def get_server_url(self, index=0):
         return self.servers[index].default_url
 

openapi_core/validators.py

@@ -0,0 +1,129 @@
+"""OpenAPI core validators module"""
+from six import iteritems
+
+from openapi_core.exceptions import (
+    OpenAPIMappingError, MissingParameterError, MissingBodyError,
+)
+
+
+class RequestParameters(dict):
+
+    valid_locations = ['path', 'query', 'headers', 'cookies']
+
+    def __getitem__(self, location):
+        self.validate_location(location)
+
+        return self.setdefault(location, {})
+
+    def __setitem__(self, location, value):
+        raise NotImplementedError
+
+    @classmethod
+    def validate_location(cls, location):
+        if location not in cls.valid_locations:
+            raise OpenAPIMappingError(
+                "Unknown parameter location: {0}".format(str(location)))
+
+
+class BaseValidationResult(object):
+
+    def __init__(self, errors):
+        self.errors = errors
+
+    def validate(self):
+        for error in self.errors:
+            raise error
+
+
+class RequestValidationResult(BaseValidationResult):
+
+    def __init__(self, errors, body=None, parameters=None):
+        super(RequestValidationResult, self).__init__(errors)
+        self.body = body
+        self.parameters = parameters or RequestParameters()
+
+
+class RequestValidator(object):
+
+    SPEC_LOCATION_TO_REQUEST_LOCATION = {
+        'path': 'view_args',
+        'query': 'args',
+        'headers': 'headers',
+        'cookies': 'cookies',
+    }
+
+    def __init__(self, spec):
+        self.spec = spec
+
+    def validate(self, request):
+        errors = []
+        body = None
+        parameters = RequestParameters()
+
+        try:
+            server = self.spec.get_server(request.full_url_pattern)
+        # don't process if server errors
+        except OpenAPIMappingError as exc:
+            errors.append(exc)
+            return RequestValidationResult(errors, body, parameters)
+
+        operation_pattern = request.full_url_pattern.replace(
+            server.default_url, '')
+        method = request.method.lower()
+
+        try:
+            operation = self.spec.get_operation(operation_pattern, method)
+        # don't process if operation errors
+        except OpenAPIMappingError as exc:
+            errors.append(exc)
+            return RequestValidationResult(errors, body, parameters)
+
+        for param_name, param in iteritems(operation.parameters):
+            try:
+                raw_value = self._get_raw_value(request, param)
+            except MissingParameterError as exc:
+                if param.required:
+                    errors.append(exc)
+
+                if not param.schema or param.schema.default is None:
+                    continue
+                raw_value = param.schema.default
+
+            value = param.unmarshal(raw_value)
+
+            parameters[param.location][param_name] = value
+
+        if operation.request_body is not None:
+            try:
+                media_type = operation.request_body[request.mimetype]
+            except OpenAPIMappingError as exc:
+                errors.append(exc)
+            else:
+                try:
+                    raw_body = self._get_raw_body(request)
+                except MissingBodyError as exc:
+                    if operation.request_body.required:
+                        errors.append(exc)
+                else:
+                    body = media_type.unmarshal(raw_body)
+
+        return RequestValidationResult(errors, body, parameters)
+
+    def _get_request_location(self, spec_location):
+        return self.SPEC_LOCATION_TO_REQUEST_LOCATION[spec_location]
+
+    def _get_raw_value(self, request, param):
+        request_location = self._get_request_location(param.location)
+        request_attr = getattr(request, request_location)
+
+        try:
+            return request_attr[param.name]
+        except KeyError:
+            raise MissingParameterError(
+                "Missing required `{0}` parameter".format(param.name))
+
+    def _get_raw_body(self, request):
+        if not request.data:
+            raise MissingBodyError("Missing required request body")
+
+        return request.data

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

@@ -73,6 +73,7 @@ paths:
       tags:
         - pets
       requestBody:
+        required: true
         content:
           application/json:
             schema:

tests/integration/test_validators.py

@@ -0,0 +1,177 @@
+import json
+import pytest
+
+from openapi_core.exceptions import (
+    InvalidServerError, InvalidOperationError, MissingParameterError,
+    MissingBodyError, InvalidContentTypeError,
+)
+from openapi_core.shortcuts import create_spec
+from openapi_core.validators import RequestValidator
+from openapi_core.wrappers import BaseOpenAPIRequest
+
+
+class RequestMock(BaseOpenAPIRequest):
+
+    def __init__(
+            self, host_url, method, path, path_pattern=None, args=None,
+            view_args=None, headers=None, cookies=None, data=None,
+            mimetype='application/json'):
+        self.host_url = host_url
+        self.path = path
+        self.path_pattern = path_pattern or path
+        self.method = method
+
+        self.args = args or {}
+        self.view_args = view_args or {}
+        self.headers = headers or {}
+        self.cookies = cookies or {}
+        self.data = data or ''
+
+        self.mimetype = mimetype
+
+
+class TestRequestValidator(object):
+
+    host_url = 'http://petstore.swagger.io'
+
+    @pytest.fixture
+    def spec_dict(self, factory):
+        return factory.spec_from_file("data/v3.0/petstore.yaml")
+
+    @pytest.fixture
+    def spec(self, spec_dict):
+        return create_spec(spec_dict)
+
+    @pytest.fixture
+    def validator(self, spec):
+        return RequestValidator(spec)
+
+    def test_request_server_error(self, validator):
+        request = RequestMock('http://petstore.invalid.net/v1', 'get', '/')
+
+        result = validator.validate(request)
+
+        assert len(result.errors) == 1
+        assert type(result.errors[0]) == InvalidServerError
+        assert result.body is None
+        assert result.parameters == {}
+
+    def test_invalid_operation(self, validator):
+        request = RequestMock(self.host_url, 'get', '/v1')
+
+        result = validator.validate(request)
+
+        assert len(result.errors) == 1
+        assert type(result.errors[0]) == InvalidOperationError
+        assert result.body is None
+        assert result.parameters == {}
+
+    def test_missing_parameter(self, validator):
+        request = RequestMock(self.host_url, 'get', '/v1/pets')
+
+        result = validator.validate(request)
+
+        assert type(result.errors[0]) == MissingParameterError
+        assert result.body is None
+        assert result.parameters == {
+            'query': {
+                'page': 1,
+                'search': '',
+            },
+        }
+
+    def test_get_pets(self, validator):
+        request = RequestMock(
+            self.host_url, 'get', '/v1/pets',
+            path_pattern='/v1/pets', args={'limit': '10'},
+        )
+
+        result = validator.validate(request)
+
+        assert result.errors == []
+        assert result.body is None
+        assert result.parameters == {
+            'query': {
+                'limit': 10,
+                'page': 1,
+                'search': '',
+            },
+        }
+
+    def test_missing_body(self, validator):
+        request = RequestMock(
+            self.host_url, 'post', '/v1/pets',
+            path_pattern='/v1/pets',
+        )
+
+        result = validator.validate(request)
+
+        assert len(result.errors) == 1
+        assert type(result.errors[0]) == MissingBodyError
+        assert result.body is None
+        assert result.parameters == {}
+
+    def test_invalid_content_type(self, validator):
+        request = RequestMock(
+            self.host_url, 'post', '/v1/pets',
+            path_pattern='/v1/pets', mimetype='text/csv',
+        )
+
+        result = validator.validate(request)
+
+        assert len(result.errors) == 1
+        assert type(result.errors[0]) == InvalidContentTypeError
+        assert result.body is None
+        assert result.parameters == {}
+
+    def test_post_pets(self, validator, spec_dict):
+        pet_name = 'Cat'
+        pet_tag = 'cats'
+        pet_street = 'Piekna'
+        pet_city = 'Warsaw'
+        data_json = {
+            'name': pet_name,
+            'tag': pet_tag,
+            'position': '2',
+            'address': {
+                'street': pet_street,
+                'city': pet_city,
+            }
+        }
+        data = json.dumps(data_json)
+        request = RequestMock(
+            self.host_url, 'post', '/v1/pets',
+            path_pattern='/v1/pets', data=data,
+        )
+
+        result = validator.validate(request)
+
+        assert result.errors == []
+        assert result.parameters == {}
+
+        schemas = spec_dict['components']['schemas']
+        pet_model = schemas['PetCreate']['x-model']
+        address_model = schemas['Address']['x-model']
+        assert result.body.__class__.__name__ == pet_model
+        assert result.body.name == pet_name
+        assert result.body.tag == pet_tag
+        assert result.body.position == 2
+        assert result.body.address.__class__.__name__ == address_model
+        assert result.body.address.street == pet_street
+        assert result.body.address.city == pet_city
+
+    def test_get_pet(self, validator):
+        request = RequestMock(
+            self.host_url, 'get', '/v1/pets/1',
+            path_pattern='/v1/pets/{petId}', view_args={'petId': '1'},
+        )
+
+        result = validator.validate(request)
+
+        assert result.errors == []
+        assert result.body is None
+        assert result.parameters == {
+            'path': {
+                'petId': 1,
+            },
+        }