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_sysexit_with_message_and_custom_code(patched_logger_critical):
with pytest.raises(SystemExit) as e:
util.sysexit_with_message('foo', 2)
assert 2 == e.value.code
patched_logger_critical.assert_called_once_with('foo')
def execute(self):
"""
Execute the actions necessary to perform a `molecule idempotence` and
returns None.
:return: None
"""
self.print_info()
if not self._config.state.converged:
msg = 'Instances not converged. Please converge instances first.'
util.sysexit_with_message(msg)
output = self._config.provisioner.converge(out=None, err=None)
idempotent = self._is_idempotent(output)
if idempotent:
msg = 'Idempotence completed successfully.'
LOG.success(msg)
else:
msg = (
'Idempotence test failed because of the following tasks:\n' u'{}'
).format('\n'.join(self._non_idempotent_tasks(output)))
util.sysexit_with_message(msg)
def _default_scenario_exists(ctx, param, value): # pragma: no cover
if value == command_base.MOLECULE_DEFAULT_SCENARIO_NAME:
return value
default_scenario_directory = os.path.join(
'molecule', command_base.MOLECULE_DEFAULT_SCENARIO_NAME
)
if not os.path.exists(default_scenario_directory):
msg = (
'The default scenario not found. Please create a scenario '
"named '{}' first."
).format(command_base.MOLECULE_DEFAULT_SCENARIO_NAME)
util.sysexit_with_message(msg)
return value
def execute(self):
"""
Execute the actions necessary to perform a `molecule login` and
returns None.
:return: None
"""
c = self._config
if (not c.state.created) and c.driver.managed:
msg = 'Instances not created. Please create instances first.'
util.sysexit_with_message(msg)
hosts = [d['name'] for d in self._config.platforms.instances]
hostname = self._get_hostname(hosts)
self._get_login(hostname)
def _verify_inventory(self):
"""
Verify the inventory is valid and returns None.
:return: None
"""
if not self.inventory:
msg = "Instances missing from the 'platform' " "section of molecule.yml."
util.sysexit_with_message(msg)
def _init_new_role(command_args):
"""
>>> molecule init role --role-name foo
"""
role_name = command_args['role_name']
role_directory = os.getcwd()
LOG.info('Initializing new role {}...'.format(role_name))
if os.path.isdir(role_name):
msg = ('The directory {} exists. '
'Cannot create new role.').format(role_name)
util.sysexit_with_message(msg)
_process_templates('role', command_args, role_directory)
scenario_base_directory = os.path.join(role_directory, role_name)
templates = [
'scenario/driver/{driver_name}'.format(**command_args),
'scenario/verifier/{verifier_name}'.format(**command_args)
]
for template in templates:
_process_templates(template, command_args, scenario_base_directory)
role_directory = os.path.join(role_directory, role_name)
msg = 'Initialized role in {} successfully.'.format(role_directory)
LOG.success(msg)
def _interpolate(self, stream, env, keep_string):
env = set_env_from_file(env, self.env_file)
i = interpolation.Interpolator(interpolation.TemplateWithDefaults, env)
try:
return i.interpolate(stream, keep_string)
except interpolation.InvalidInterpolation as e:
msg = "parsing config file '{}'.\n\n" '{}\n{}'.format(
self.molecule_file, e.place, e.string
)
util.sysexit_with_message(msg)
def _get_hostname(self, hosts):
hostname = self._config.command_args.get('host')
if hostname is None:
if len(hosts) == 1:
hostname = hosts[0]
else:
msg = (
'There are {} running hosts. Please specify '
'which with --host.\n\n'
'Available hosts:\n{}'.format(len(hosts), '\n'.join(sorted(hosts)))
)
util.sysexit_with_message(msg)
match = [x for x in hosts if x.startswith(hostname)]
if len(match) == 0:
msg = (
"There are no hosts that match '{}'. You "
'can only login to valid hosts.'
).format(hostname)
util.sysexit_with_message(msg)
elif len(match) != 1:
# If there are multiple matches, but one of them is an exact string
# match, assume this is the one they're looking for and use it.
if hostname in match:
match = [hostname]
else:
msg = (
"There are {} hosts that match '{}'. You "
'can only login to one at a time.\n\n'
def _link_or_update_vars(self):
"""
Creates or updates the symlink to group_vars and returns None.
:returns: None
"""
for d, source in self.links.items():
target = os.path.join(self.inventory_directory, d)
source = os.path.join(self._config.scenario.directory, source)
if not os.path.exists(source):
msg = "The source path '{}' does not exist.".format(source)
util.sysexit_with_message(msg)
msg = "Inventory {} linked to {}".format(source, target)
LOG.info(msg)
os.symlink(source, target)
def __init__(self, old_molecule_file, driver_name):
self._old_molecule_file = old_molecule_file
if not os.path.isfile(old_molecule_file):
msg = 'Unable to find {}. Exiting.'.format(old_molecule_file)
util.sysexit_with_message(msg)
self._m = migrate.Migrate(old_molecule_file)
self._old_role_dir = os.path.join(os.path.dirname(old_molecule_file))
self._old_dot_molecule_dir = scenario.ephemeral_directory(self._old_role_dir)
self._old_test_dir = os.path.join(self._old_role_dir, 'tests')
self._old_playbook = os.path.join(self._old_role_dir, 'playbook.yml')
self._molecule_dir = config.molecule_directory(self._old_role_dir)
self._scenario_dir = os.path.join(self._molecule_dir, 'default')
self._test_dir = os.path.join(self._scenario_dir, 'tests')
self._molecule_file = config.molecule_file(self._scenario_dir)
self._role_name = os.path.basename(os.path.normpath(self._old_role_dir))