mirror of
https://github.com/correl/openapi-core.git
synced 2024-12-29 11:09:25 +00:00
Merge pull request #188 from p1c2u/refactor/move-unmarshal-out-of-schema-models
Move unmarshal out of schema models
This commit is contained in:
commit
b003cccd54
15 changed files with 285 additions and 273 deletions
61
README.rst
61
README.rst
|
@ -78,15 +78,6 @@ and unmarshal request data from validation result
|
|||
# get body
|
||||
validated_body = result.body
|
||||
|
||||
or use shortcuts for simple validation
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
from openapi_core import validate_parameters, validate_body
|
||||
|
||||
validated_params = validate_parameters(spec, request)
|
||||
validated_body = validate_body(spec, request)
|
||||
|
||||
Request object should be instance of OpenAPIRequest class (See `Integrations`_).
|
||||
|
||||
Response
|
||||
|
@ -117,14 +108,6 @@ and unmarshal response data from validation result
|
|||
# get data
|
||||
validated_data = result.data
|
||||
|
||||
or use shortcuts for simple validation
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
from openapi_core import validate_data
|
||||
|
||||
validated_data = validate_data(spec, request, response)
|
||||
|
||||
Response object should be instance of OpenAPIResponse class (See `Integrations`_).
|
||||
|
||||
|
||||
|
@ -145,17 +128,6 @@ For Django 2.2 you can use DjangoOpenAPIRequest a Django request factory:
|
|||
validator = RequestValidator(spec)
|
||||
result = validator.validate(openapi_request)
|
||||
|
||||
or simply specify request factory for shortcuts
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
from openapi_core import validate_parameters, validate_body
|
||||
|
||||
validated_params = validate_parameters(
|
||||
spec, request, request_factory=DjangoOpenAPIRequest)
|
||||
validated_body = validate_body(
|
||||
spec, request, request_factory=DjangoOpenAPIRequest)
|
||||
|
||||
You can use DjangoOpenAPIResponse as a Django response factory:
|
||||
|
||||
.. code-block:: python
|
||||
|
@ -167,17 +139,6 @@ You can use DjangoOpenAPIResponse as a Django response factory:
|
|||
validator = ResponseValidator(spec)
|
||||
result = validator.validate(openapi_request, openapi_response)
|
||||
|
||||
or simply specify response factory for shortcuts
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
from openapi_core import validate_parameters, validate_body
|
||||
|
||||
validated_data = validate_data(
|
||||
spec, request, response,
|
||||
request_factory=DjangoOpenAPIRequest,
|
||||
response_factory=DjangoOpenAPIResponse)
|
||||
|
||||
Flask
|
||||
*****
|
||||
|
||||
|
@ -247,17 +208,6 @@ You can use FlaskOpenAPIRequest a Flask/Werkzeug request factory:
|
|||
validator = RequestValidator(spec)
|
||||
result = validator.validate(openapi_request)
|
||||
|
||||
or simply specify request factory for shortcuts
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
from openapi_core import validate_parameters, validate_body
|
||||
|
||||
validated_params = validate_parameters(
|
||||
spec, request, request_factory=FlaskOpenAPIRequest)
|
||||
validated_body = validate_body(
|
||||
spec, request, request_factory=FlaskOpenAPIRequest)
|
||||
|
||||
You can use FlaskOpenAPIResponse as a Flask/Werkzeug response factory:
|
||||
|
||||
.. code-block:: python
|
||||
|
@ -269,17 +219,6 @@ You can use FlaskOpenAPIResponse as a Flask/Werkzeug response factory:
|
|||
validator = ResponseValidator(spec)
|
||||
result = validator.validate(openapi_request, openapi_response)
|
||||
|
||||
or simply specify response factory for shortcuts
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
from openapi_core import validate_parameters, validate_body
|
||||
|
||||
validated_data = validate_data(
|
||||
spec, request, response,
|
||||
request_factory=FlaskOpenAPIRequest,
|
||||
response_factory=FlaskOpenAPIResponse)
|
||||
|
||||
Pyramid
|
||||
*******
|
||||
|
||||
|
|
|
@ -4,9 +4,6 @@ from collections import defaultdict
|
|||
from openapi_core.schema.media_types.exceptions import InvalidMediaTypeValue
|
||||
from openapi_core.schema.media_types.util import json_loads
|
||||
from openapi_core.casting.schemas.exceptions import CastError
|
||||
from openapi_core.unmarshalling.schemas.exceptions import (
|
||||
UnmarshalError, ValidateError,
|
||||
)
|
||||
|
||||
|
||||
MEDIA_TYPE_DESERIALIZERS = {
|
||||
|
@ -47,13 +44,3 @@ class MediaType(object):
|
|||
return self.schema.cast(deserialized)
|
||||
except CastError as exc:
|
||||
raise InvalidMediaTypeValue(exc)
|
||||
|
||||
def unmarshal(self, value, custom_formatters=None, resolver=None):
|
||||
if not self.schema:
|
||||
return value
|
||||
|
||||
try:
|
||||
return self.schema.unmarshal(
|
||||
value, resolver=resolver, custom_formatters=custom_formatters)
|
||||
except (ValidateError, UnmarshalError) as exc:
|
||||
raise InvalidMediaTypeValue(exc)
|
||||
|
|
|
@ -11,9 +11,6 @@ from openapi_core.schema.parameters.exceptions import (
|
|||
)
|
||||
from openapi_core.schema.schemas.enums import SchemaType
|
||||
from openapi_core.casting.schemas.exceptions import CastError
|
||||
from openapi_core.unmarshalling.schemas.exceptions import (
|
||||
UnmarshalError, ValidateError,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
@ -114,16 +111,3 @@ class Parameter(object):
|
|||
return self.schema.cast(deserialized)
|
||||
except CastError as exc:
|
||||
raise InvalidParameterValue(self.name, exc)
|
||||
|
||||
def unmarshal(self, value, custom_formatters=None, resolver=None):
|
||||
if not self.schema:
|
||||
return value
|
||||
|
||||
try:
|
||||
return self.schema.unmarshal(
|
||||
value,
|
||||
resolver=resolver,
|
||||
custom_formatters=custom_formatters,
|
||||
)
|
||||
except (ValidateError, UnmarshalError) as exc:
|
||||
raise InvalidParameterValue(self.name, exc)
|
||||
|
|
|
@ -2,6 +2,7 @@ from copy import deepcopy
|
|||
import warnings
|
||||
|
||||
from openapi_core.schema.schemas.enums import SchemaType, SchemaFormat
|
||||
from openapi_core.schema.schemas.models import Schema
|
||||
from openapi_core.schema_validator import OAS30Validator
|
||||
from openapi_core.schema_validator import oas30_format_checker
|
||||
from openapi_core.unmarshalling.schemas.exceptions import (
|
||||
|
@ -36,6 +37,8 @@ class SchemaUnmarshallersFactory(object):
|
|||
|
||||
def create(self, schema, type_override=None):
|
||||
"""Create unmarshaller from the schema."""
|
||||
if not isinstance(schema, Schema):
|
||||
raise TypeError("schema not type of Schema")
|
||||
if schema.deprecated:
|
||||
warnings.warn("The schema is deprecated", DeprecationWarning)
|
||||
|
||||
|
|
|
@ -1,5 +1,5 @@
|
|||
"""OpenAPI core validation request shortcuts module"""
|
||||
from functools import partial
|
||||
import warnings
|
||||
|
||||
from openapi_core.schema.media_types.exceptions import OpenAPIMediaTypeError
|
||||
from openapi_core.schema.parameters.exceptions import OpenAPIParameterError
|
||||
|
@ -28,8 +28,26 @@ def validate_request(validator, request, failsafe=None):
|
|||
return result
|
||||
|
||||
|
||||
validate_parameters = partial(validate_request, failsafe=ERRORS_BODY)
|
||||
validate_body = partial(validate_request, failsafe=ERRORS_PARAMETERS)
|
||||
def validate_parameters(validator, request):
|
||||
warnings.warn(
|
||||
"validate_parameters shortcut is deprecated, "
|
||||
"use validator.validate instead",
|
||||
DeprecationWarning,
|
||||
)
|
||||
result = validator._validate_parameters(request)
|
||||
result.raise_for_errors()
|
||||
return result
|
||||
|
||||
|
||||
def validate_body(validator, request):
|
||||
warnings.warn(
|
||||
"validate_body shortcut is deprecated, "
|
||||
"use validator.validate instead",
|
||||
DeprecationWarning,
|
||||
)
|
||||
result = validator._validate_body(request)
|
||||
result.raise_for_errors()
|
||||
return result
|
||||
|
||||
|
||||
def spec_validate_parameters(spec, request, request_factory=None):
|
||||
|
|
|
@ -12,6 +12,9 @@ from openapi_core.schema.parameters.exceptions import (
|
|||
from openapi_core.schema.paths.exceptions import InvalidPath
|
||||
from openapi_core.schema.request_bodies.exceptions import MissingRequestBody
|
||||
from openapi_core.schema.servers.exceptions import InvalidServer
|
||||
from openapi_core.unmarshalling.schemas.exceptions import (
|
||||
UnmarshalError, ValidateError,
|
||||
)
|
||||
from openapi_core.validation.request.datatypes import (
|
||||
RequestParameters, RequestValidationResult,
|
||||
)
|
||||
|
@ -26,25 +29,10 @@ class RequestValidator(object):
|
|||
|
||||
def validate(self, request):
|
||||
try:
|
||||
server = self.spec.get_server(request.full_url_pattern)
|
||||
# don't process if server errors
|
||||
except InvalidServer as exc:
|
||||
return RequestValidationResult([exc, ], None, None)
|
||||
|
||||
operation_pattern = get_operation_pattern(
|
||||
server.default_url, request.full_url_pattern
|
||||
)
|
||||
|
||||
try:
|
||||
path = self.spec[operation_pattern]
|
||||
except InvalidPath as exc:
|
||||
return RequestValidationResult([exc, ], None, None)
|
||||
|
||||
try:
|
||||
operation = self.spec.get_operation(
|
||||
operation_pattern, request.method)
|
||||
path = self._get_path(request)
|
||||
operation = self._get_operation(request)
|
||||
# don't process if operation errors
|
||||
except InvalidOperation as exc:
|
||||
except (InvalidServer, InvalidPath, InvalidOperation) as exc:
|
||||
return RequestValidationResult([exc, ], None, None)
|
||||
|
||||
params, params_errors = self._get_parameters(
|
||||
|
@ -59,6 +47,47 @@ class RequestValidator(object):
|
|||
errors = params_errors + body_errors
|
||||
return RequestValidationResult(errors, body, params)
|
||||
|
||||
def _validate_parameters(self, request):
|
||||
try:
|
||||
path = self._get_path(request)
|
||||
operation = self._get_operation(request)
|
||||
except (InvalidServer, InvalidPath, InvalidOperation) as exc:
|
||||
return RequestValidationResult([exc, ], None, None)
|
||||
|
||||
params, params_errors = self._get_parameters(
|
||||
request, chain(
|
||||
iteritems(operation.parameters),
|
||||
iteritems(path.parameters)
|
||||
)
|
||||
)
|
||||
return RequestValidationResult(params_errors, None, params)
|
||||
|
||||
def _validate_body(self, request):
|
||||
try:
|
||||
operation = self._get_operation(request)
|
||||
except (InvalidServer, InvalidOperation) as exc:
|
||||
return RequestValidationResult([exc, ], None, None)
|
||||
|
||||
body, body_errors = self._get_body(request, operation)
|
||||
return RequestValidationResult(body_errors, body, None)
|
||||
|
||||
def _get_operation_pattern(self, request):
|
||||
server = self.spec.get_server(request.full_url_pattern)
|
||||
|
||||
return get_operation_pattern(
|
||||
server.default_url, request.full_url_pattern
|
||||
)
|
||||
|
||||
def _get_path(self, request):
|
||||
operation_pattern = self._get_operation_pattern(request)
|
||||
|
||||
return self.spec[operation_pattern]
|
||||
|
||||
def _get_operation(self, request):
|
||||
operation_pattern = self._get_operation_pattern(request)
|
||||
|
||||
return self.spec.get_operation(operation_pattern, request.method)
|
||||
|
||||
def _get_parameters(self, request, params):
|
||||
errors = []
|
||||
seen = set()
|
||||
|
@ -86,11 +115,8 @@ class RequestValidator(object):
|
|||
continue
|
||||
|
||||
try:
|
||||
unmarshalled = param.unmarshal(
|
||||
casted, self.custom_formatters,
|
||||
resolver=self.spec._resolver,
|
||||
)
|
||||
except OpenAPIParameterError as exc:
|
||||
unmarshalled = self._unmarshal(param, casted)
|
||||
except (ValidateError, UnmarshalError) as exc:
|
||||
errors.append(exc)
|
||||
else:
|
||||
locations.setdefault(param.location.value, {})
|
||||
|
@ -121,11 +147,18 @@ class RequestValidator(object):
|
|||
errors.append(exc)
|
||||
else:
|
||||
try:
|
||||
body = media_type.unmarshal(
|
||||
casted, self.custom_formatters,
|
||||
resolver=self.spec._resolver,
|
||||
)
|
||||
except InvalidMediaTypeValue as exc:
|
||||
body = self._unmarshal(media_type, casted)
|
||||
except (ValidateError, UnmarshalError) as exc:
|
||||
errors.append(exc)
|
||||
|
||||
return body, errors
|
||||
|
||||
def _unmarshal(self, param_or_media_type, value):
|
||||
if not param_or_media_type.schema:
|
||||
return value
|
||||
|
||||
return param_or_media_type.schema.unmarshal(
|
||||
value,
|
||||
resolver=self.spec._resolver,
|
||||
custom_formatters=self.custom_formatters,
|
||||
)
|
||||
|
|
|
@ -1,4 +1,6 @@
|
|||
"""OpenAPI core validation response shortcuts module"""
|
||||
import warnings
|
||||
|
||||
from openapi_core.validation.response.validators import ResponseValidator
|
||||
|
||||
|
||||
|
@ -8,6 +10,17 @@ def validate_response(validator, request, response):
|
|||
return result
|
||||
|
||||
|
||||
def validate_data(validator, request, response):
|
||||
warnings.warn(
|
||||
"validate_data shortcut is deprecated, "
|
||||
"use validator.validate instead",
|
||||
DeprecationWarning,
|
||||
)
|
||||
result = validator._validate_data(request, response)
|
||||
result.raise_for_errors()
|
||||
return result
|
||||
|
||||
|
||||
def spec_validate_data(
|
||||
spec, request, response,
|
||||
request_factory=None,
|
||||
|
@ -18,6 +31,6 @@ def spec_validate_data(
|
|||
response = response_factory(response)
|
||||
|
||||
validator = ResponseValidator(spec)
|
||||
result = validate_response(validator, request, response)
|
||||
result = validate_data(validator, request, response)
|
||||
|
||||
return result.data
|
||||
|
|
|
@ -7,6 +7,9 @@ from openapi_core.schema.responses.exceptions import (
|
|||
InvalidResponse, MissingResponseContent,
|
||||
)
|
||||
from openapi_core.schema.servers.exceptions import InvalidServer
|
||||
from openapi_core.unmarshalling.schemas.exceptions import (
|
||||
UnmarshalError, ValidateError,
|
||||
)
|
||||
from openapi_core.validation.response.datatypes import ResponseValidationResult
|
||||
from openapi_core.validation.util import get_operation_pattern
|
||||
|
||||
|
@ -19,27 +22,10 @@ class ResponseValidator(object):
|
|||
|
||||
def validate(self, request, response):
|
||||
try:
|
||||
server = self.spec.get_server(request.full_url_pattern)
|
||||
# don't process if server errors
|
||||
except InvalidServer as exc:
|
||||
return ResponseValidationResult([exc, ], None, None)
|
||||
|
||||
operation_pattern = get_operation_pattern(
|
||||
server.default_url, request.full_url_pattern
|
||||
)
|
||||
|
||||
try:
|
||||
operation = self.spec.get_operation(
|
||||
operation_pattern, request.method)
|
||||
operation_response = self._get_operation_response(
|
||||
request, response)
|
||||
# don't process if operation errors
|
||||
except InvalidOperation as exc:
|
||||
return ResponseValidationResult([exc, ], None, None)
|
||||
|
||||
try:
|
||||
operation_response = operation.get_response(
|
||||
str(response.status_code))
|
||||
# don't process if operation response errors
|
||||
except InvalidResponse as exc:
|
||||
except (InvalidServer, InvalidOperation, InvalidResponse) as exc:
|
||||
return ResponseValidationResult([exc, ], None, None)
|
||||
|
||||
data, data_errors = self._get_data(response, operation_response)
|
||||
|
@ -50,6 +36,33 @@ class ResponseValidator(object):
|
|||
errors = data_errors + headers_errors
|
||||
return ResponseValidationResult(errors, data, headers)
|
||||
|
||||
def _get_operation_pattern(self, request):
|
||||
server = self.spec.get_server(request.full_url_pattern)
|
||||
|
||||
return get_operation_pattern(
|
||||
server.default_url, request.full_url_pattern
|
||||
)
|
||||
|
||||
def _get_operation(self, request):
|
||||
operation_pattern = self._get_operation_pattern(request)
|
||||
|
||||
return self.spec.get_operation(operation_pattern, request.method)
|
||||
|
||||
def _get_operation_response(self, request, response):
|
||||
operation = self._get_operation(request)
|
||||
|
||||
return operation.get_response(str(response.status_code))
|
||||
|
||||
def _validate_data(self, request, response):
|
||||
try:
|
||||
operation_response = self._get_operation_response(
|
||||
request, response)
|
||||
except (InvalidServer, InvalidOperation, InvalidResponse) as exc:
|
||||
return ResponseValidationResult([exc, ], None, None)
|
||||
|
||||
data, data_errors = self._get_data(response, operation_response)
|
||||
return ResponseValidationResult(data_errors, data, None)
|
||||
|
||||
def _get_data(self, response, operation_response):
|
||||
errors = []
|
||||
|
||||
|
@ -73,11 +86,8 @@ class ResponseValidator(object):
|
|||
errors.append(exc)
|
||||
else:
|
||||
try:
|
||||
data = media_type.unmarshal(
|
||||
casted, self.custom_formatters,
|
||||
resolver=self.spec._resolver,
|
||||
)
|
||||
except InvalidMediaTypeValue as exc:
|
||||
data = self._unmarshal(media_type, casted)
|
||||
except (ValidateError, UnmarshalError) as exc:
|
||||
errors.append(exc)
|
||||
|
||||
return data, errors
|
||||
|
@ -89,3 +99,13 @@ class ResponseValidator(object):
|
|||
headers = {}
|
||||
|
||||
return headers, errors
|
||||
|
||||
def _unmarshal(self, param_or_media_type, value):
|
||||
if not param_or_media_type.schema:
|
||||
return value
|
||||
|
||||
return param_or_media_type.schema.unmarshal(
|
||||
value,
|
||||
resolver=self.spec._resolver,
|
||||
custom_formatters=self.custom_formatters,
|
||||
)
|
||||
|
|
|
@ -6,9 +6,7 @@ from uuid import UUID
|
|||
from six import text_type
|
||||
|
||||
from openapi_core.extensions.models.models import BaseModel
|
||||
from openapi_core.schema.media_types.exceptions import (
|
||||
InvalidContentType, InvalidMediaTypeValue,
|
||||
)
|
||||
from openapi_core.schema.media_types.exceptions import InvalidContentType
|
||||
from openapi_core.schema.parameters.exceptions import (
|
||||
MissingRequiredParameter, InvalidParameterValue, EmptyParameterValue,
|
||||
)
|
||||
|
@ -175,14 +173,12 @@ class TestPetstore(object):
|
|||
|
||||
response_result = response_validator.validate(request, response)
|
||||
|
||||
original_exc = response_result.errors[0].original_exception
|
||||
schema_errors = response_result.errors[0].schema_errors
|
||||
assert response_result.errors == [
|
||||
InvalidMediaTypeValue(
|
||||
original_exception=InvalidSchemaValue(
|
||||
type=SchemaType.OBJECT,
|
||||
value=data_json,
|
||||
schema_errors=original_exc.schema_errors,
|
||||
),
|
||||
InvalidSchemaValue(
|
||||
type=SchemaType.OBJECT,
|
||||
value=data_json,
|
||||
schema_errors=schema_errors,
|
||||
),
|
||||
]
|
||||
assert response_result.data is None
|
||||
|
@ -575,7 +571,7 @@ class TestPetstore(object):
|
|||
},
|
||||
)
|
||||
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
with pytest.raises(InvalidSchemaValue):
|
||||
validate_body(spec, request)
|
||||
|
||||
def test_post_cats_only_required_body(self, spec, spec_dict):
|
||||
|
@ -915,7 +911,7 @@ class TestPetstore(object):
|
|||
|
||||
assert parameters == RequestParameters()
|
||||
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
with pytest.raises(InvalidSchemaValue):
|
||||
validate_body(spec, request)
|
||||
|
||||
def test_post_tags_empty_body(self, spec, spec_dict):
|
||||
|
@ -933,7 +929,7 @@ class TestPetstore(object):
|
|||
|
||||
assert parameters == RequestParameters()
|
||||
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
with pytest.raises(InvalidSchemaValue):
|
||||
validate_body(spec, request)
|
||||
|
||||
def test_post_tags_wrong_property_type(self, spec):
|
||||
|
@ -951,7 +947,7 @@ class TestPetstore(object):
|
|||
|
||||
assert parameters == RequestParameters()
|
||||
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
with pytest.raises(InvalidSchemaValue):
|
||||
validate_body(spec, request)
|
||||
|
||||
def test_post_tags_additional_properties(
|
||||
|
@ -1110,7 +1106,7 @@ class TestPetstore(object):
|
|||
)
|
||||
|
||||
parameters = validate_parameters(spec, request)
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
with pytest.raises(InvalidSchemaValue):
|
||||
validate_body(spec, request)
|
||||
|
||||
assert parameters == RequestParameters()
|
||||
|
|
|
@ -18,6 +18,7 @@ from openapi_core.schema.responses.exceptions import (
|
|||
from openapi_core.schema.servers.exceptions import InvalidServer
|
||||
from openapi_core.shortcuts import create_spec
|
||||
from openapi_core.testing import MockRequest, MockResponse
|
||||
from openapi_core.unmarshalling.schemas.exceptions import InvalidSchemaValue
|
||||
from openapi_core.validation.request.datatypes import RequestParameters
|
||||
from openapi_core.validation.request.validators import RequestValidator
|
||||
from openapi_core.validation.response.validators import ResponseValidator
|
||||
|
@ -465,7 +466,7 @@ class TestResponseValidator(object):
|
|||
result = validator.validate(request, response)
|
||||
|
||||
assert len(result.errors) == 1
|
||||
assert type(result.errors[0]) == InvalidMediaTypeValue
|
||||
assert type(result.errors[0]) == InvalidSchemaValue
|
||||
assert result.data is None
|
||||
assert result.headers == {}
|
||||
|
||||
|
@ -485,7 +486,7 @@ class TestResponseValidator(object):
|
|||
result = validator.validate(request, response)
|
||||
|
||||
assert len(result.errors) == 1
|
||||
assert type(result.errors[0]) == InvalidMediaTypeValue
|
||||
assert type(result.errors[0]) == InvalidSchemaValue
|
||||
assert result.data is None
|
||||
assert result.headers == {}
|
||||
|
||||
|
|
|
@ -1,9 +1,4 @@
|
|||
import pytest
|
||||
|
||||
from openapi_core.schema.media_types.exceptions import InvalidMediaTypeValue
|
||||
from openapi_core.schema.media_types.models import MediaType
|
||||
from openapi_core.schema.schemas.models import Schema
|
||||
from openapi_core.unmarshalling.schemas.formatters import Formatter
|
||||
|
||||
|
||||
class TestMediaTypeCast(object):
|
||||
|
@ -15,44 +10,3 @@ class TestMediaTypeCast(object):
|
|||
result = media_type.cast(value)
|
||||
|
||||
assert result == value
|
||||
|
||||
|
||||
class TestParameterUnmarshal(object):
|
||||
|
||||
def test_empty(self):
|
||||
media_type = MediaType('application/json')
|
||||
value = ''
|
||||
|
||||
result = media_type.unmarshal(value)
|
||||
|
||||
assert result == value
|
||||
|
||||
def test_schema_type_invalid(self):
|
||||
schema = Schema('integer', _source={'type': 'integer'})
|
||||
media_type = MediaType('application/json', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
media_type.unmarshal(value)
|
||||
|
||||
def test_schema_custom_format_invalid(self):
|
||||
|
||||
class CustomFormatter(Formatter):
|
||||
def unmarshal(self, value):
|
||||
raise ValueError
|
||||
formatter = CustomFormatter()
|
||||
custom_format = 'custom'
|
||||
custom_formatters = {
|
||||
custom_format: formatter,
|
||||
}
|
||||
schema = Schema(
|
||||
'string',
|
||||
schema_format=custom_format,
|
||||
_source={'type': 'string', 'format': 'custom'},
|
||||
)
|
||||
media_type = MediaType('application/json', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidMediaTypeValue):
|
||||
media_type.unmarshal(
|
||||
value, custom_formatters=custom_formatters)
|
||||
|
|
|
@ -1,12 +1,10 @@
|
|||
import pytest
|
||||
|
||||
from openapi_core.schema.parameters.exceptions import (
|
||||
EmptyParameterValue, InvalidParameterValue,
|
||||
EmptyParameterValue,
|
||||
)
|
||||
from openapi_core.schema.parameters.enums import ParameterStyle
|
||||
from openapi_core.schema.parameters.models import Parameter
|
||||
from openapi_core.schema.schemas.models import Schema
|
||||
from openapi_core.unmarshalling.schemas.formatters import Formatter
|
||||
|
||||
|
||||
class TestParameterInit(object):
|
||||
|
@ -65,51 +63,3 @@ class TestParameterCast(object):
|
|||
result = param.cast(value)
|
||||
|
||||
assert result == value
|
||||
|
||||
|
||||
class TestParameterUnmarshal(object):
|
||||
|
||||
def test_query_valid(self):
|
||||
param = Parameter('param', 'query')
|
||||
value = 'test'
|
||||
|
||||
result = param.unmarshal(value)
|
||||
|
||||
assert result == value
|
||||
|
||||
def test_query_allow_empty_value(self):
|
||||
param = Parameter('param', 'query', allow_empty_value=True)
|
||||
value = ''
|
||||
|
||||
result = param.unmarshal(value)
|
||||
|
||||
assert result == value
|
||||
|
||||
def test_query_schema_type_invalid(self):
|
||||
schema = Schema('integer', _source={'type': 'integer'})
|
||||
param = Parameter('param', 'query', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidParameterValue):
|
||||
param.unmarshal(value)
|
||||
|
||||
def test_query_schema_custom_format_invalid(self):
|
||||
|
||||
class CustomFormatter(Formatter):
|
||||
def unmarshal(self, value):
|
||||
raise ValueError
|
||||
formatter = CustomFormatter()
|
||||
custom_format = 'custom'
|
||||
custom_formatters = {
|
||||
custom_format: formatter,
|
||||
}
|
||||
schema = Schema(
|
||||
'string',
|
||||
schema_format=custom_format,
|
||||
_source={'type': 'string', 'format': 'custom'},
|
||||
)
|
||||
param = Parameter('param', 'query', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidParameterValue):
|
||||
param.unmarshal(value, custom_formatters=custom_formatters)
|
||||
|
|
102
tests/unit/unmarshalling/test_unmarshal.py
Normal file
102
tests/unit/unmarshalling/test_unmarshal.py
Normal file
|
@ -0,0 +1,102 @@
|
|||
import pytest
|
||||
|
||||
from openapi_core.schema.media_types.models import MediaType
|
||||
from openapi_core.schema.parameters.models import Parameter
|
||||
from openapi_core.schema.schemas.models import Schema
|
||||
from openapi_core.unmarshalling.schemas.exceptions import (
|
||||
InvalidSchemaFormatValue,
|
||||
)
|
||||
from openapi_core.unmarshalling.schemas.factories import (
|
||||
SchemaUnmarshallersFactory,
|
||||
)
|
||||
from openapi_core.unmarshalling.schemas.formatters import Formatter
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def unmarshaller_factory():
|
||||
def create_unmarshaller(param_or_media_type, custom_formatters=None):
|
||||
return SchemaUnmarshallersFactory(
|
||||
custom_formatters=custom_formatters).create(
|
||||
param_or_media_type.schema)
|
||||
return create_unmarshaller
|
||||
|
||||
|
||||
class TestParameterUnmarshal(object):
|
||||
|
||||
def test_no_schema(self, unmarshaller_factory):
|
||||
param = Parameter('param', 'query')
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
unmarshaller_factory(param).unmarshal(value)
|
||||
|
||||
def test_schema_type_invalid(self, unmarshaller_factory):
|
||||
schema = Schema('integer', _source={'type': 'integer'})
|
||||
param = Parameter('param', 'query', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidSchemaFormatValue):
|
||||
unmarshaller_factory(param).unmarshal(value)
|
||||
|
||||
def test_schema_custom_format_invalid(self, unmarshaller_factory):
|
||||
|
||||
class CustomFormatter(Formatter):
|
||||
def unmarshal(self, value):
|
||||
raise ValueError
|
||||
formatter = CustomFormatter()
|
||||
custom_format = 'custom'
|
||||
custom_formatters = {
|
||||
custom_format: formatter,
|
||||
}
|
||||
schema = Schema(
|
||||
'string',
|
||||
schema_format=custom_format,
|
||||
_source={'type': 'string', 'format': 'custom'},
|
||||
)
|
||||
param = Parameter('param', 'query', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidSchemaFormatValue):
|
||||
unmarshaller_factory(
|
||||
param, custom_formatters=custom_formatters).unmarshal(value)
|
||||
|
||||
|
||||
class TestMediaTypeUnmarshal(object):
|
||||
|
||||
def test_no_schema(self, unmarshaller_factory):
|
||||
media_type = MediaType('application/json')
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
unmarshaller_factory(media_type).unmarshal(value)
|
||||
|
||||
def test_schema_type_invalid(self, unmarshaller_factory):
|
||||
schema = Schema('integer', _source={'type': 'integer'})
|
||||
media_type = MediaType('application/json', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidSchemaFormatValue):
|
||||
unmarshaller_factory(media_type).unmarshal(value)
|
||||
|
||||
def test_schema_custom_format_invalid(self, unmarshaller_factory):
|
||||
|
||||
class CustomFormatter(Formatter):
|
||||
def unmarshal(self, value):
|
||||
raise ValueError
|
||||
formatter = CustomFormatter()
|
||||
custom_format = 'custom'
|
||||
custom_formatters = {
|
||||
custom_format: formatter,
|
||||
}
|
||||
schema = Schema(
|
||||
'string',
|
||||
schema_format=custom_format,
|
||||
_source={'type': 'string', 'format': 'custom'},
|
||||
)
|
||||
media_type = MediaType('application/json', schema=schema)
|
||||
value = 'test'
|
||||
|
||||
with pytest.raises(InvalidSchemaFormatValue):
|
||||
unmarshaller_factory(
|
||||
media_type, custom_formatters=custom_formatters).unmarshal(
|
||||
value)
|
|
@ -12,7 +12,8 @@ from openapi_core.validation.request.shortcuts import (
|
|||
class TestSpecValidateParameters(object):
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_parameters'
|
||||
)
|
||||
def test_no_request_factory(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -26,7 +27,8 @@ class TestSpecValidateParameters(object):
|
|||
mock_validate.aasert_called_once_with(request)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_parameters'
|
||||
)
|
||||
def test_no_request_factory_error(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -39,7 +41,8 @@ class TestSpecValidateParameters(object):
|
|||
mock_validate.aasert_called_once_with(request)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_parameters'
|
||||
)
|
||||
def test_request_factory(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -56,7 +59,8 @@ class TestSpecValidateParameters(object):
|
|||
)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_parameters'
|
||||
)
|
||||
def test_request_factory_error(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -75,7 +79,8 @@ class TestSpecValidateParameters(object):
|
|||
class TestSpecValidateBody(object):
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_body'
|
||||
)
|
||||
def test_no_request_factory(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -89,7 +94,8 @@ class TestSpecValidateBody(object):
|
|||
mock_validate.aasert_called_once_with(request)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_body'
|
||||
)
|
||||
def test_no_request_factory_error(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -102,7 +108,8 @@ class TestSpecValidateBody(object):
|
|||
mock_validate.aasert_called_once_with(request)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_body'
|
||||
)
|
||||
def test_request_factory(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -119,7 +126,8 @@ class TestSpecValidateBody(object):
|
|||
)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.validate'
|
||||
'openapi_core.validation.request.shortcuts.RequestValidator.'
|
||||
'_validate_body'
|
||||
)
|
||||
def test_request_factory_error(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
|
|
@ -10,7 +10,8 @@ from openapi_core.validation.response.shortcuts import spec_validate_data
|
|||
class TestSpecValidateData(object):
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.validate'
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.'
|
||||
'_validate_data'
|
||||
)
|
||||
def test_no_factories(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -25,7 +26,8 @@ class TestSpecValidateData(object):
|
|||
mock_validate.aasert_called_once_with(request, response)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.validate'
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.'
|
||||
'_validate_data'
|
||||
)
|
||||
def test_no_factories_error(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -39,7 +41,8 @@ class TestSpecValidateData(object):
|
|||
mock_validate.aasert_called_once_with(request, response)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.validate'
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.'
|
||||
'_validate_data'
|
||||
)
|
||||
def test_factories(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
@ -62,7 +65,8 @@ class TestSpecValidateData(object):
|
|||
)
|
||||
|
||||
@mock.patch(
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.validate'
|
||||
'openapi_core.validation.response.shortcuts.ResponseValidator.'
|
||||
'_validate_data'
|
||||
)
|
||||
def test_factories_error(self, mock_validate):
|
||||
spec = mock.sentinel.spec
|
||||
|
|
Loading…
Reference in a new issue