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_check_deviated_value_incorrect_average(log):
source = [
['row', 'name'],
[2, 'Alex'],
]
with pytest.raises(exceptions.GoodtablesException):
report = validate(source, checks=[
{'deviated-value': {'column': 3, 'average': 'incorrect-average'}},
])
raise exceptions.GoodtablesException(message % name)
check['type'] = error['type']
check['context'] = error['context']
elif not error and type != 'custom':
message = 'Check "%s" is not a part of the spec should have type "custom"'
raise exceptions.GoodtablesException(message % name)
# Validate position
if position:
try:
position = position.split(':', 1)
assert position[0] in ['before', 'after']
assert self.__checks.get(position[1])
except (TypeError, AssertionError):
message = 'Check "%s" has been registered at invalid position "%s"'
raise exceptions.GoodtablesException(message % (name, position))
# Insert into checks
checks = OrderedDict()
self.__checks.pop(name, None)
for item_name, item_check in self.__checks.items():
if position == 'before:%s' % item_name:
checks[name] = check
checks[item_name] = item_check
if position == 'after:%s' % item_name:
checks[name] = check
if not position:
checks[name] = check
self.__checks = checks
def nested(source, presets):
warnings = []
tables = []
# Add warnings, tables
source = deepcopy(source)
for item in source:
preset = item.pop('preset', 'table')
checks = item.pop('checks', [])
if preset == 'nested':
message = 'Preset "nested" supports only one level depth'
raise exceptions.GoodtablesException(message)
try:
preset_func = presets[preset]['func']
except KeyError:
message = 'Preset "%s" is not registered' % preset
raise exceptions.GoodtablesException(message)
item_warnings, item_tables = preset_func(**item)
if checks:
for item_table in item_tables:
item_table['checks'] = checks
warnings.extend(item_warnings)
tables.extend(item_tables)
return warnings, tables
def __get_preset(self, preset):
try:
return self.__presets[preset]
except KeyError:
message = 'Preset "%s" is not registered' % preset
raise exceptions.GoodtablesException(message)
item_name = list(item.keys())[0]
item_config = list(item.values())[0]
if item_name not in self.__checks:
message = 'Check "%s" is not registered'
raise exceptions.GoodtablesException(message % item_name)
if item_name == name:
compiled_check = deepcopy(check)
if isinstance(check['func'], type):
check_options = deepcopy(options)
check_options.update(item_config)
try:
compiled_check['func'] = check['func'](**check_options)
except Exception as e:
message = 'Check "%s" options "%s" error'
message = message % (check['name'], check_options)
six.raise_from(exceptions.GoodtablesException(message), e)
compiled_checks.append(compiled_check)
return compiled_checks
result_include = set()
result_exclude = set()
for name, enabled in include.items():
if enabled:
result_include.add(name)
else:
result_include.add('structure')
result_include.add('schema')
result_exclude.add(name)
include = list(result_include)
exclude = list(result_exclude)
# Validate checks
if not isinstance(include, (list, tuple)):
message = 'Checks parameter "%s" is not valid' % include
raise exceptions.GoodtablesException(message)
# Expand checks
for group in ['structure', 'schema']:
for index, item in enumerate(list(include)):
if item == group:
del include[index]
for code, error in spec['errors'].items():
# It's temporal skip
# https://github.com/frictionlessdata/goodtables-py/issues/174
if code == 'schema-error':
continue
if error['type'] == group:
include.append(code)
# Compile checks
compiled_checks = []
warnings = []
tables = []
# Add warnings, tables
source = deepcopy(source)
for item in source:
preset = item.pop('preset', 'table')
checks = item.pop('checks', [])
if preset == 'nested':
message = 'Preset "nested" supports only one level depth'
raise exceptions.GoodtablesException(message)
try:
preset_func = presets[preset]['func']
except KeyError:
message = 'Preset "%s" is not registered' % preset
raise exceptions.GoodtablesException(message)
item_warnings, item_tables = preset_func(**item)
if checks:
for item_table in item_tables:
item_table['checks'] = checks
warnings.extend(item_warnings)
tables.extend(item_tables)
return warnings, tables