Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self._is_notebook is not None:
raise(RuntimeError('Multiple entry points found under sample: {}'.format(self._test_name)))
if ext_name == '.py':
self._is_notebook = False
if ext_name == '.ipynb':
self._is_notebook = True
if self._is_notebook is None:
raise(RuntimeError('No entry point found for sample: {}'.format(self._test_name)))
config_schema = yamale.make_schema(SCHEMA_CONFIG)
# Retrieve default config
try:
with open(DEFAULT_CONFIG, 'r') as f:
raw_args = yaml.safe_load(f)
default_config = yamale.make_data(DEFAULT_CONFIG)
yamale.validate(config_schema, default_config) # If fails, a ValueError will be raised.
except yaml.YAMLError as yamlerr:
raise RuntimeError('Illegal default config:{}'.format(yamlerr))
except OSError as ose:
raise FileExistsError('Default config not found:{}'.format(ose))
else:
self._run_pipeline = raw_args['run_pipeline']
# For presubmit check, do not do any image injection as for now.
# Notebook samples need to be papermilled first.
if self._is_notebook:
# Parse necessary params from config.yaml
nb_params = {}
try:
config_file = os.path.join(CONFIG_DIR, '%s.config.yaml' % self._test_name)
with open(config_file, 'r') as f:
raise RuntimeError('Illegal default config:{}'.format(yamlerr))
except OSError as ose:
raise FileExistsError('Default config not found:{}'.format(ose))
else:
self._run_pipeline = raw_args['run_pipeline']
# For presubmit check, do not do any image injection as for now.
# Notebook samples need to be papermilled first.
if self._is_notebook:
# Parse necessary params from config.yaml
nb_params = {}
try:
config_file = os.path.join(CONFIG_DIR, '%s.config.yaml' % self._test_name)
with open(config_file, 'r') as f:
raw_args = yaml.safe_load(f)
test_config = yamale.make_data(config_file)
yamale.validate(config_schema, test_config) # If fails, a ValueError will be raised.
except yaml.YAMLError as yamlerr:
print('No legit yaml config file found, use default args:{}'.format(yamlerr))
except OSError as ose:
print('Config file with the same name not found, use default args:{}'.format(ose))
else:
if 'notebook_params' in raw_args.keys():
nb_params.update(raw_args['notebook_params'])
if 'output' in raw_args['notebook_params'].keys(): # output is a special param that has to be specified dynamically.
nb_params['output'] = self._sample_test_output
if 'run_pipeline' in raw_args.keys():
self._run_pipeline = raw_args['run_pipeline']
pm.execute_notebook(
input_path='%s.ipynb' % self._test_name,
output_path='%s.ipynb' % self._test_name,
from glob import glob
import sys
import yamale
schema = yamale.make_schema('./docs/schema.yaml')
data = yamale.make_data('./docs/current.settings.yaml')
yamale.validate(schema, data)
templates = glob('mia/templates/*/settings.yaml')
for template in templates:
sys.stdout.write('Checking %s against schema... ' % template)
data = yamale.make_data(template)
yamale.validate(schema, data)
print("done!")
if type(yaml) != list:
yaml = [yaml]
if base_dir is not None:
schema = os.path.join(base_dir, schema)
yaml = {os.path.join(base_dir, y) for y in yaml}
# Run yaml through glob and flatten list
yaml = set(itertools.chain(*map(glob.glob, yaml)))
# Remove schema from set of data files
yaml = yaml - {schema}
yamale_schema = yamale.make_schema(schema, validators=validators)
yamale_data = itertools.chain(*map(yamale.make_data, yaml))
return yamale.validate(yamale_schema, yamale_data) is not None
def parse_yaml_file(file_path: _FilePath) -> Config:
_data = make_data(file_path, parser='ruamel')
_schema = make_schema(SCHEMA_FILE, parser='ruamel')
yamale_validate(_schema, _data, strict=True)
config: Config = _data[0][0]
config.setdefault('logging', {})
config['logging'].setdefault('level', 'INFO')
for config_crawler in config['crawlers']:
config_crawler.setdefault("weight", 1.0)
config_crawler.setdefault('restart_at_front_when_exhausted', False)
config_crawler.setdefault('config', {})
return config
def _validate(schema_path, data_path, parser, strict):
schema = schemas.get(schema_path)
try:
if not schema:
schema = yamale.make_schema(schema_path, parser)
schemas[schema_path] = schema
data = yamale.make_data(data_path, parser)
yamale.validate(schema, data, strict)
except Exception as e:
error = '\nError!\n'
error += 'Schema: %s\n' % schema_path
error += 'Data file: %s\n' % data_path
error += traceback.format_exc()
print(error)
raise ValueError('Validation failed!')