Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'password': 'abc',
}
}
}
# Test detection
assert config.detect_old(old_data)
assert not config.detect_old(new_data)
# Try partial data
old_part = {
'peeringdb': {
'url': 'https://test.peeringdb.com/api',
'timeout': 10,
}
}
assert config.detect_old(old_part)
# empty case
assert not config.detect_old({})
# Test conversion
conv_data = config.convert_old(old_data)
assert config.CLIENT_SCHEMA.validate(conv_data)
assert not config.detect_old(conv_data)
assert conv_data == new_data
conv_part = config.convert_old(old_part)
assert config.CLIENT_SCHEMA.validate(conv_part)
def test_default_config():
DEFAULT = config.default_config()
with TemporaryDirectory() as path:
cfg = config.load_config(path)
assert DEFAULT == cfg
def check_load_config(config_dir):
convert = False
loaded = config.read_config(config_dir) or {}
if config.detect_old(loaded):
print(
"Found config file with pre-0.7 schema; backing up and converting to new format"
)
convert = True
cfg = config.convert_old(loaded)
else:
cfg = config.default_config()
config.recursive_update(cfg, loaded)
if convert:
config.write_config(cfg, config_dir, backup_existing=True)
return cfg
def check_load_config(config_dir):
convert = False
loaded = config.read_config(config_dir) or {}
if config.detect_old(loaded):
print(
"Found config file with pre-0.7 schema; backing up and converting to new format"
)
convert = True
cfg = config.convert_old(loaded)
else:
cfg = config.default_config()
config.recursive_update(cfg, loaded)
if convert:
config.write_config(cfg, config_dir, backup_existing=True)
return cfg
def __init__(self, cfg=None, **kwargs):
"""
Arguments:
- cfg : dict of complete config options (see config.ClientSchema),
by default loads from DEFAULT_CONFIG_DIR
- url: URL to connect to
- user: username to connect to api with
- password: password
- timeout: timeout to fail after
"""
if cfg is None:
cfg = config.load_config()
self.config = cfg
orm_config = cfg['orm']
orm_name = orm_config['backend']
if not peeringdb.backend_initialized():
peeringdb.initialize_backend(orm_name, **orm_config)
sync_config = cfg['sync']
# override config with kwargs
munge.util.recursive_update(sync_config, kwargs)
self._fetcher = Fetcher(**sync_config)
self._updater = Updater(self._fetcher, **sync_config)
self.update_all = self._updater.update_all
self.update = self._updater.update
self.update_where = self._updater.update_where