github/openapi-core
1
0
Fork 0
mirror of https://github.com/correl/openapi-core.git synced 2024-11-21 19:18:41 +00:00
Code Issues Projects Releases Packages Wiki Activity

Object validation

Browse source
This commit is contained in:
p1c2u 2018-08-21 18:33:24 +01:00 committed by Artur Maciag
parent 17855ae145
commit 0915e94530
7 changed files with 272 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
openapi_core/extensions/models/factories.py
View file

@ -1,12 +1,25 @@
"""OpenAPI X-Model extension factories module"""
from openapi_core.extensions.models.models import BaseModel
from openapi_core.extensions.models.models import Model
class ModelClassFactory(object):
base_class = Model
def create(self, name):
return type(name, (self.base_class, ), {})
class ModelFactory(object):
def create(self, properties, name=None):
model = BaseModel
if name is not None:
model = type(name, (BaseModel, ), {})
def __init__(self, model_class_factory=None):
self.model_class_factory = model_class_factory or ModelClassFactory()
return model(**properties)
def create(self, properties, name=None):
name = name or 'Model'
model_class = self._create_class(name)
return model_class(properties)
def _create_class(self, name):
return self.model_class_factory.create(name)

33
openapi_core/extensions/models/models.py
View file

@ -1,17 +1,26 @@
"""OpenAPI X-Model extension models module"""
class BaseModel(dict):
class BaseModel(object):
"""Base class for OpenAPI X-Model."""
def __getattr__(self, attr_name):
"""Only search through properties if attribute not found normally.
:type attr_name: str
"""
try:
return self[attr_name]
except KeyError:
raise AttributeError(
'type object {0!r} has no attribute {1!r}'
.format(type(self).__name__, attr_name)
)
@property
def __dict__(self):
raise NotImplementedError
class Model(BaseModel):
"""Model class for OpenAPI X-Model."""
def __init__(self, properties=None):
self.__properties = properties or {}
@property
def __dict__(self):
return self.__properties
def __getattr__(self, name):
if name not in self.__properties:
raise AttributeError
return self.__properties[name]

91
openapi_core/schema/schemas/models.py
View file

@ -170,10 +170,12 @@ class Schema(object):
def _unmarshal_collection(self, value):
return list(map(self.items.unmarshal, value))
def _unmarshal_object(self, value):
def _unmarshal_object(self, value, model_factory=None):
if not isinstance(value, (dict, )):
raise InvalidSchemaValue(
"Value of {0} not an object".format(value))
"Value of {0} not a dict".format(value))
model_factory = model_factory or ModelFactory()
if self.one_of:
properties = None
@ -197,7 +199,7 @@ class Schema(object):
else:
properties = self._unmarshal_properties(value)
return ModelFactory().create(properties, name=self.model)
return model_factory.create(properties, name=self.model)
def _unmarshal_properties(self, value, one_of_schema=None):
all_props = self.get_all_properties()
@ -234,20 +236,99 @@ class Schema(object):
continue
prop_value = prop.default
properties[prop_name] = prop.unmarshal(prop_value)
self._validate_properties(properties, one_of_schema=one_of_schema)
return properties
def get_validator_mapping(self):
mapping = self.VALIDATOR_CALLABLE_GETTER.copy()
mapping.update({
SchemaType.OBJECT: self._validate_object,
})
return defaultdict(lambda: lambda x: x, mapping)
def validate(self, value):
if value is None:
if not self.nullable:
raise InvalidSchemaValue("Null value for non-nullable schema")
return self.default
validator = self.VALIDATOR_CALLABLE_GETTER[self.type]
validator_mapping = self.get_validator_mapping()
validator_callable = validator_mapping[self.type]
if not validator(value):
if not validator_callable(value):
raise InvalidSchemaValue(
"Value of {0} not valid type of {1}".format(
value, self.type.value)
)
return value
def _validate_object(self, value):
if not hasattr(value, '__dict__'):
raise InvalidSchemaValue(
"Value of {0} not an object".format(value))
properties = value.__dict__
if self.one_of:
valid_one_of_schema = None
for one_of_schema in self.one_of:
try:
self._validate_properties(properties, one_of_schema)
except OpenAPISchemaError:
pass
else:
if valid_one_of_schema is not None:
raise MultipleOneOfSchema(
"Exactly one schema should be valid,"
"multiple found")
valid_one_of_schema = True
if valid_one_of_schema is None:
raise NoOneOfSchema(
"Exactly one valid schema should be valid, None found.")
else:
self._validate_properties(properties)
return True
def _validate_properties(self, value, one_of_schema=None):
all_props = self.get_all_properties()
all_props_names = self.get_all_properties_names()
all_req_props_names = self.get_all_required_properties_names()
if one_of_schema is not None:
all_props.update(one_of_schema.get_all_properties())
all_props_names |= one_of_schema.\
get_all_properties_names()
all_req_props_names |= one_of_schema.\
get_all_required_properties_names()
value_props_names = value.keys()
extra_props = set(value_props_names) - set(all_props_names)
if extra_props and self.additional_properties is None:
raise UndefinedSchemaProperty(
"Undefined properties in schema: {0}".format(extra_props))
for prop_name in extra_props:
prop_value = value[prop_name]
self.additional_properties.validate(
prop_value)
for prop_name, prop in iteritems(all_props):
try:
prop_value = value[prop_name]
except KeyError:
if prop_name in all_req_props_names:
raise MissingSchemaProperty(
"Missing schema property {0}".format(prop_name))
if not prop.nullable and not prop.default:
continue
prop_value = prop.default
prop.validate(prop_value)
return True

54
tests/integration/test_petstore.py
View file

@ -2,6 +2,7 @@ import json
import pytest
from six import iteritems
from openapi_core.extensions.models.models import BaseModel
from openapi_core.schema.media_types.exceptions import (
InvalidContentType, InvalidMediaTypeValue,
)
@ -221,7 +222,8 @@ class TestPetstore(object):
response_result = response_validator.validate(request, response)
assert response_result.errors == []
assert response_result.data == data_json
assert isinstance(response_result.data, BaseModel)
assert response_result.data.data == []
def test_get_pets_ids_param(self, spec, response_validator):
host_url = 'http://petstore.swagger.io/v1'
@ -258,7 +260,8 @@ class TestPetstore(object):
response_result = response_validator.validate(request, response)
assert response_result.errors == []
assert response_result.data == data_json
assert isinstance(response_result.data, BaseModel)
assert response_result.data.data == []
def test_get_pets_tags_param(self, spec, response_validator):
host_url = 'http://petstore.swagger.io/v1'
@ -295,7 +298,8 @@ class TestPetstore(object):
response_result = response_validator.validate(request, response)
assert response_result.errors == []
assert response_result.data == data_json
assert isinstance(response_result.data, BaseModel)
assert response_result.data.data == []
def test_get_pets_parameter_deserialization_error(self, spec):
host_url = 'http://petstore.swagger.io/v1'
@ -810,10 +814,12 @@ class TestPetstore(object):
assert body is None
data_id = 1
data_name = 'test'
data_json = {
'data': {
'id': 1,
'name': 'test',
'id': data_id,
'name': data_name,
},
}
data = json.dumps(data_json)
@ -822,7 +828,10 @@ class TestPetstore(object):
response_result = response_validator.validate(request, response)
assert response_result.errors == []
assert response_result.data == data_json
assert isinstance(response_result.data, BaseModel)
assert isinstance(response_result.data.data, BaseModel)
assert response_result.data.data.id == data_id
assert response_result.data.data.name == data_name
def test_get_pet_not_found(self, spec, response_validator):
host_url = 'http://petstore.swagger.io/v1'
@ -847,10 +856,13 @@ class TestPetstore(object):
assert body is None
code = 404
message = 'Not found'
rootCause = 'Pet not found'
data_json = {
'code': 404,
'message': 'Not found',
'rootCause': 'Pet not found',
'message': message,
'rootCause': rootCause,
}
data = json.dumps(data_json)
response = MockResponse(data, status_code=404)
@ -858,7 +870,10 @@ class TestPetstore(object):
response_result = response_validator.validate(request, response)
assert response_result.errors == []
assert response_result.data == data_json
assert isinstance(response_result.data, BaseModel)
assert response_result.data.code == code
assert response_result.data.message == message
assert response_result.data.rootCause == rootCause
def test_get_pet_wildcard(self, spec, response_validator):
host_url = 'http://petstore.swagger.io/v1'
@ -993,13 +1008,18 @@ class TestPetstore(object):
body = request.get_body(spec)
assert parameters == {}
assert body == data_json
assert isinstance(body, BaseModel)
assert body.name == pet_name
code = 400
message = 'Bad request'
rootCause = 'Tag already exist'
additionalinfo = 'Tag Dog already exist'
data_json = {
'code': 400,
'message': 'Bad request',
'rootCause': 'Tag already exist',
'additionalinfo': 'Tag Dog already exist',
'code': code,
'message': message,
'rootCause': rootCause,
'additionalinfo': additionalinfo,
}
data = json.dumps(data_json)
response = MockResponse(data, status_code=404)
@ -1007,4 +1027,8 @@ class TestPetstore(object):
response_result = response_validator.validate(request, response)
assert response_result.errors == []
assert response_result.data == data_json
assert isinstance(response_result.data, BaseModel)
assert response_result.data.code == code
assert response_result.data.message == message
assert response_result.data.rootCause == rootCause
assert response_result.data.additionalinfo == additionalinfo

6
tests/integration/test_validators.py
View file

@ -4,6 +4,7 @@ import pytest
from openapi_core.schema.media_types.exceptions import (
InvalidContentType, InvalidMediaTypeValue,
)
from openapi_core.extensions.models.models import BaseModel
from openapi_core.schema.operations.exceptions import InvalidOperation
from openapi_core.schema.parameters.exceptions import MissingRequiredParameter
from openapi_core.schema.request_bodies.exceptions import MissingRequestBody
@ -327,5 +328,8 @@ class TestResponseValidator(object):
result = validator.validate(request, response)
assert result.errors == []
assert result.data == response_json
assert isinstance(result.data, BaseModel)
assert len(result.data.data) == 1
assert result.data.data[0].id == 1
assert result.data.data[0].name == 'Sparky'
assert result.headers == {}

44
tests/unit/extensions/test_models.py Normal file
View file

@ -0,0 +1,44 @@
import pytest
from openapi_core.extensions.models.models import BaseModel, Model
class TestBaseModelDict(object):
def test_not_implemented(self):
model = BaseModel()
with pytest.raises(NotImplementedError):
model.__dict__
class TestModelDict(object):
def test_dict_empty(self):
model = Model()
result = model.__dict__
assert result == {}
def test_dict(self):
properties = {
'prop1': 'value1',
'prop2': 'value2',
}
model = Model(properties)
result = model.__dict__
assert result == properties
def test_attribute(self):
prop_value = 'value1'
properties = {
'prop1': prop_value,
}
model = Model(properties)
result = model.prop1
assert result == prop_value

59
tests/unit/schema/test_schemas.py
View file

@ -3,7 +3,10 @@ import datetime
import mock
import pytest
from openapi_core.schema.schemas.exceptions import InvalidSchemaValue
from openapi_core.extensions.models.models import Model
from openapi_core.schema.schemas.exceptions import (
InvalidSchemaValue, MultipleOneOfSchema, NoOneOfSchema,
)
from openapi_core.schema.schemas.models import Schema
@ -254,3 +257,57 @@ class TestSchemaValidate(object):
with pytest.raises(InvalidSchemaValue):
schema.validate(value)
@pytest.mark.parametrize('value', ['true', False, 1, 3.14, [1, 3]])
def test_object_not_an_object(self, value):
schema = Schema('object')
with pytest.raises(InvalidSchemaValue):
schema.validate(value)
@pytest.mark.parametrize('value', [Model(), ])
def test_object_multiple_one_of(self, value):
one_of = [
Schema('object'), Schema('object'),
]
schema = Schema('object', one_of=one_of)
with pytest.raises(MultipleOneOfSchema):
schema.validate(value)
@pytest.mark.parametrize('value', [Model(), ])
def test_object_defferent_type_one_of(self, value):
one_of = [
Schema('integer'), Schema('string'),
]
schema = Schema('object', one_of=one_of)
with pytest.raises(MultipleOneOfSchema):
schema.validate(value)
@pytest.mark.parametrize('value', [Model(), ])
def test_object_no_one_of(self, value):
one_of = [
Schema(
'object',
properties={'test1': Schema('string')},
required=['test1', ],
),
Schema(
'object',
properties={'test2': Schema('string')},
required=['test2', ],
),
]
schema = Schema('object', one_of=one_of)
with pytest.raises(NoOneOfSchema):
schema.validate(value)
@pytest.mark.parametrize('value', [Model(), ])
def test_object_default_property(self, value):
schema = Schema('object', default='value1')
result = schema.validate(value)
assert result == value