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_autoload_client_from_module(self):
with temppath() as module_dpath:
os.mkdir(module_dpath)
sys.path.append(module_dpath)
module_fpath = osp.join(module_dpath, 'mclient.py')
self._write_client_module(module_fpath, 'ModuleClient')
try:
with temppath() as config_path:
config = Config(config_path)
config.add_section(config.global_section)
config.set(config.global_section, 'autoload.modules', 'mclient')
config._autoload()
client = Client.from_options({'url': ''}, 'ModuleClient')
eq_(client.one, 1)
finally:
sys.path.remove(module_dpath)
def test_autoload_client_from_path(self):
with temppath() as module_path:
self._write_client_module(module_path, 'PathClient')
with temppath() as config_path:
config = Config(config_path)
config.add_section(config.global_section)
config.set(config.global_section, 'autoload.paths', module_path)
config._autoload()
client = Client.from_options({'url': ''}, 'PathClient')
eq_(client.one, 1)
not self.has_section(self.global_section) or
not self.has_option(self.global_section, 'default.alias')
):
raise HdfsError('No alias specified and no default alias found.')
alias = self.get(self.global_section, 'default.alias')
if not alias in self._clients:
for suffix in ('.alias', '_alias'):
section = '%s%s' % (alias, suffix)
if self.has_section(section):
options = dict(self.items(section))
class_name = options.pop('client', 'InsecureClient')
# Massage options.
if 'timeout' in options:
timeout = tuple(int(s) for s in options['timeout'].split(','))
options['timeout'] = timeout[0] if len(timeout) == 1 else timeout
self._clients[alias] = Client.from_options(options, class_name)
break
else:
raise HdfsError('Alias %r not found in %r.', alias, self.path)
return self._clients[alias]