Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
('requires_schema', {
'schema': '{"requires": {"properties": {"other": {"type": "number"}}}}',
'expected': {
'requires': {
'properties': {
'other': {
'type': 'number'
},
},
},
},
}),
('requires_wrong_value', {
'schema': '{"requires": 5}',
'access': 'requires',
'raises': SchemaError(
'requires value 5 is neither a string nor an object'),
}),
('minimum_default', {
'schema': '{}',
'expected': {
'minimum': None
},
}),
('minimum_integer', {
'schema': '{"minimum": 5}',
'expected': {
'minimum': 5
},
}),
('minimum_float', {
'schema': '{"minimum": 3.5}',
('additionalProperties_false', {
'schema': '{"additionalProperties": false}',
'expected': {
"additionalProperties": False,
},
}),
('additionalProperties_object', {
'schema': '{"additionalProperties": {"type": "number"}}',
'expected': {
"additionalProperties": {"type": "number"},
},
}),
('additionalProperties_wrong_type', {
'schema': '{"additionalProperties": 5}',
'access': 'additionalProperties',
'raises': SchemaError(
'additionalProperties value 5 is neither false nor an'
' object'),
}),
('requires_default', {
'schema': '{}',
'expected': {
'requires': {},
},
}),
('requires_property_name', {
'schema': '{"requires": "other"}',
'expected': {
'requires': "other",
},
}),
('requires_schema', {
def test_save_validation(fixture, resource, return_payload):
"""Tests saving an ``Element`` including validation."""
resource.url_attribute = 'resource_uri'
httpretty.register_uri(httpretty.GET, resource.get_url(), body=fixture('movie.json'),
content_type='application/json')
movie = resource.collection.get(id=1)
movie.runtime = 'NAN'
with pytest.raises(schema.SchemaError) as excinfo:
movie.save()
assert str(excinfo.value) == 'Invalid runtime'
resource.schema = None # Deactivates validation
httpretty.register_uri(httpretty.PUT, movie.get_url(), body=return_payload,
content_type='application/json')
movie.save().runtime = 'NAN'
def main():
"""
The main function for Skelebot CLI where the config is loaded,
arguments are parsed, and commands are executed
"""
try:
env = get_env()
config = yaml.loadConfig(env)
parser = skeleParser.SkeleParser(config, env)
executor.execute(config, parser)
except SchemaError as error:
print(SCHEMA_ERROR.format(error))
sys.exit(1)
except RuntimeError as error:
print(ERROR.format(error))
sys.exit(1)
def validate_template(name: str, pattern: str, value: str) -> None:
if pattern not in value:
message = "%s should contain the string %s" % (name, pattern)
raise schema.SchemaError(message)
def _decode(data):
try:
decoded = decode_function(data)
return ContentAwareImage(blob=decoded)
except Exception as e:
if app.config['IN_DEBUG']() or isinstance(e, SchemaError):
raise e
else:
# raise generic error that is appropriate to return to a user
raise SchemaError('Could not decode image.')
def put(self):
"""The API method to PUT request for flask.
"""
move_id = request.args.get('id')
admin_id = request.args.get('admin_id', None)
if not move_id:
return {'status': 'ERROR', 'message': '\'id\' parameter is missing'}
try:
form = request.form.to_dict(flat=True)
data = MOVE_SCHEMA.validate(form)
except SchemaError as e:
return {'status': 'ERROR', 'message': e.code}
result = move_arm(admin_id, move_id, data)
return {'status': 'OK' if result else 'ERROR'}
"""
Makes sure that the lane definition dict has the correct schema
:param pipeline_dict: The lane definition dict from which the lane definition will be created
:return: (dict) The validated dict
"""
extract_format = {'class': str, Optional('kwargs'): Or({str: object}, None)}
transform_load_format = {'class': str, Optional('kwargs'): Or({str: object}, None)}
shared_format = {'resource_name': str, 'class': str, Optional('kwargs'): Or({str: object}, None)}
schema = Schema({'processors': {'extract': Or([extract_format], extract_format),
Optional('transform'): Or([transform_load_format], transform_load_format),
'load': Or([transform_load_format], transform_load_format)},
Optional('shared'): Or([shared_format], shared_format)})
try:
return schema.validate(pipeline_dict)
except SchemaError as e:
raise errors.PipelineSchemaError(**e.__dict__)
def validate_dictionaries(dicts):
schema = Schema({'name': str,
'headers': Or(str, [str]),
'units': Or(str, [str]),
Optional('reader'): str,
Optional('starts'): Or(str, [str]),
Optional('ends'): Or(str, [str])
})
for d in dicts:
try:
schema.validate(d)
except SchemaError:
raise ValueError(d)