Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_empty_core_object(self, tmpdir):
"""
If createing a core object without any source or schema file an exception should be raised.
"""
with pytest.raises(CoreError) as ex:
Core()
assert "No source file/data was loaded" in str(ex.value)
# To trigger schema exception we must pass in a source file
source_f = tmpdir.join("bar.json")
source_f.write("3.14159")
with pytest.raises(CoreError) as ex:
Core(source_file=str(source_f))
assert "No schema file/data was loaded" in str(ex.value)
def test_load_unsupported_format(self, tmpdir):
"""
Try to load some fileending that is not supported.
Currently XML is not supported.
"""
source_f = tmpdir.join("foo.xml")
source_f.write("bar")
schema_f = tmpdir.join("bar.xml")
schema_f.write("bar")
with pytest.raises(CoreError) as ex:
Core(source_file=str(source_f))
assert "Unable to load source_file. Unknown file format of specified file path" in str(ex.value)
with pytest.raises(CoreError) as ex:
Core(schema_files=[str(schema_f)])
assert "Unknown file format. Supported file endings is" in str(ex.value)
res = re.match(rule.pattern, value, re.UNICODE)
except TypeError:
res = None
if res is None: # Not matching
self.errors.append(SchemaError.SchemaErrorEntry(
msg=u"Value '{value}' does not match pattern '{pattern}'. Path: '{path}'",
path=path,
value=nativestr(str(value)),
pattern=rule._pattern))
else:
log.debug("Pattern matched...")
if rule.range is not None:
if not is_scalar(value):
raise CoreError(u"value is not a valid scalar")
r = rule.range
try:
v = len(value)
value = v
except Exception:
pass
self._validate_range(
r.get("max"),
r.get("min"),
r.get("max-ex"),
r.get("min-ex"),
value,
path,
schema_data = dict(schema_data, **data)
self.schema = schema_data
# Nothing was loaded so try the source_data variable
if self.source is None:
log.debug(u"No source file loaded, trying source data variable")
self.source = source_data
if self.schema is None:
log.debug(u"No schema file loaded, trying schema data variable")
self.schema = schema_data
# Test if anything was loaded
if self.source is None:
raise CoreError(u"No source file/data was loaded")
if self.schema is None:
raise CoreError(u"No schema file/data was loaded")
# Merge any extensions defined in the schema with the provided list of extensions from the cli
for f in self.schema.get('extensions', []):
self.extensions.append(f)
if not isinstance(self.extensions, list) and all(isinstance(e, str) for e in self.extensions):
raise CoreError(u"Specified extensions must be a list of file paths")
self._load_extensions()
if self.strict_rule_validation:
log.info("Using strict rule keywords validation...")
def _validate_assert(self, rule, value, path):
if not self.allow_assertions:
raise CoreError('To allow usage of keyword "assert" you must use cli flag "--allow-assertions" or set the keyword "allow_assert" in Core class')
# Small hack to make strings work as a value.
if isinstance(value, str):
assert_value_str = '"{0}"'.format(value)
else:
assert_value_str = '{0}'.format(value)
assertion_string = "val = {0}; assert {1}".format(assert_value_str, rule.assertion)
try:
exec(assertion_string, {}, {})
except AssertionError:
self.errors.append(SchemaError.SchemaErrorEntry(
msg=u"Value: '{0}' assertion expression failed ({1})".format(value, rule.assertion),
path=path,
value=value,
))
def _validate_range(self, max_, min_, max_ex, min_ex, value, path, prefix):
"""
Validate that value is within range values.
"""
if not isinstance(value, int) and not isinstance(value, float):
raise CoreError("Value must be a integer type")
log.debug(
u"Validate range : %s : %s : %s : %s : %s : %s",
max_,
min_,
max_ex,
min_ex,
value,
path,
)
if max_ is not None and max_ < value:
self.errors.append(SchemaError.SchemaErrorEntry(
msg=u"Type '{prefix}' has size of '{value}', greater than max limit '{max_}'. Path: '{path}'",
path=path,
value=nativestr(value) if tt['str'](value) else value,
method = getattr(extension, func, None)
if method:
found_method = True
# No exception will should be caught. If one is raised it should bubble up all the way.
ret = method(value, rule, path)
if ret is not True and ret is not None:
msg = '%s. Path: {path}' % unicode(ret)
self.errors.append(SchemaError.SchemaErrorEntry(
msg=msg,
path=path,
value=None))
# If False or None or some other object that is interpreted as False
if not ret:
raise CoreError(u"Error when running extension function : {0}".format(func))
# Only run the first matched function. Sinc loading order is determined
# it should be easy to determine which file is used before others
break
if not found_method:
raise CoreError(u"Did not find method '{0}' in any loaded extension file".format(func))
if rule.length is not None:
self._validate_length(
rule.length,
value,
path,
'scalar',
)
# Validate timestamp
if rule.type == "timestamp":
self._validate_scalar_timestamp(value, path)
if rule.type == "date":
if not is_scalar(value):
raise CoreError(u'value is not a valid scalar')
date_format = rule.format
self._validate_scalar_date(value, date_format, path)
yml.constructor.add_constructor('tag:yaml.org,2002:python/none', Constructor.construct_yaml_null)
yml.constructor.add_constructor('tag:yaml.org,2002:python/str', Constructor.construct_python_str)
yml.constructor.add_constructor('tag:yaml.org,2002:python/tuple', Constructor.construct_python_tuple)
yml.constructor.add_constructor('tag:yaml.org,2002:python/unicode', Constructor.construct_python_unicode)
if data_file_obj:
try:
self.source = yml.load(data_file_obj.read())
except Exception as e:
raise CoreError("Unable to load data_file_obj input")
if schema_file_obj:
try:
self.schema = yml.load(schema_file_obj.read())
except Exception as e:
raise CoreError("Unable to load schema_file_obj")
if source_file is not None:
if not os.path.exists(source_file):
raise CoreError(u"Provided source_file do not exists on disk: {0}".format(source_file))
with open(source_file, "r", encoding=file_encoding) as stream:
if source_file.endswith(".json"):
self.source = json.load(stream)
elif source_file.endswith(".yaml") or source_file.endswith('.yml'):
self.source = yml.load(stream)
else:
raise CoreError(u"Unable to load source_file. Unknown file format of specified file path: {0}".format(source_file))
if not isinstance(schema_files, list):
raise CoreError(u"schema_files must be of list type")