Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def prepare_ssh_access(options):
print_bulleted(options, 'Prepare SSH Access to Bootstrapped VM')
generate_ssh_keypair(options)
if options.test_interactive:
cloud_config = salt.config.cloud_config('/etc/salt/cloud') # TODO Is this always the case?
# Determine which provider we are using by looking it up in the profile
vm_provider = cloud_config['profiles'][options.vm_source]['provider'].split(':')[0]
# Finally, we can get the provider password
provider_password = cloud_config['providers']['linode']['linode']['password']
# Using the password, we can construct a data structure for a salt-ssh roster
roster_data = {options.vm_name: {
'host': get_minion_ip_address(options, sync=False),
'passwd': provider_password}}
# FIXME Perms!
if not os.path.exists('/tmp/.jenkins_ssh'):
os.mkdir('/tmp/.jenkins_ssh')
with open('/tmp/.jenkins_ssh/roster', 'w') as roster_fh:
yaml.dump(roster_data, stream=roster_fh)
def setUpClass(cls):
cls.opts = salt.config.DEFAULT_MINION_OPTS
cls.name = 'my_distribution'
cls.base_ret = {'name': cls.name, 'changes': {}}
# Most attributes elided since there are so many required ones
cls.config = {'Enabled': True, 'HttpVersion': 'http2'}
cls.tags = {'test_tag1': 'value1'}
def get_config(config_for, from_scratch=False):
if from_scratch:
if config_for in ('master', 'syndic_master', 'mm_master', 'mm_sub_master'):
return salt.config.master_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for)
)
elif config_for in ('minion', 'sub_minion'):
return salt.config.minion_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for)
)
elif config_for in ('syndic',):
return salt.config.syndic_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for),
AdaptedConfigurationTestCaseMixin.get_config_file_path('minion')
)
elif config_for == 'client_config':
return salt.config.client_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path('master')
)
if config_for not in RUNTIME_VARS.RUNTIME_CONFIGS:
if config_for in ('master', 'syndic_master', 'mm_master', 'mm_sub_master'):
RUNTIME_VARS.RUNTIME_CONFIGS[config_for] = freeze(
salt.config.master_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for)
)
)
elif config_for in ('minion', 'sub_minion'):
def setUp(self):
'''
Setting up
'''
# Set mandatory CLI options
self.args = []
# Set config option names
self.logfile_config_setting_name = 'api_logfile'
# Set defaults
self.default_config = salt.config.DEFAULT_MASTER_OPTS.copy()
self.default_config.update(salt.config.DEFAULT_API_OPTS)
self.addCleanup(delattr, self, 'default_config')
# Log file
self.log_file = '/tmp/salt_api_parser_test'
self.api_logfile = '/tmp/api_logfile'
# Function to patch
self.config_func = 'salt.config.api_config'
# Mock log setup
self.setup_log()
# Assign parser
self.parser = salt.utils.parsers.SaltAPIParser
self.addCleanup(delattr, self, 'parser')
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for)
)
elif config_for in ('syndic',):
return salt.config.syndic_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for),
AdaptedConfigurationTestCaseMixin.get_config_file_path('minion')
)
elif config_for == 'client_config':
return salt.config.client_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path('master')
)
if config_for not in RUNTIME_VARS.RUNTIME_CONFIGS:
if config_for in ('master', 'syndic_master', 'mm_master', 'mm_sub_master'):
RUNTIME_VARS.RUNTIME_CONFIGS[config_for] = freeze(
salt.config.master_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for)
)
)
elif config_for in ('minion', 'sub_minion'):
RUNTIME_VARS.RUNTIME_CONFIGS[config_for] = freeze(
salt.config.minion_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for)
)
)
elif config_for in ('syndic',):
RUNTIME_VARS.RUNTIME_CONFIGS[config_for] = freeze(
salt.config.syndic_config(
AdaptedConfigurationTestCaseMixin.get_config_file_path(config_for),
AdaptedConfigurationTestCaseMixin.get_config_file_path('minion')
)
)
def setup_config(self):
return config.minion_config(self.get_config_file_path('minion'))
def get_dependencies():
'''
Warn if dependencies are not met.
'''
return config.check_driver_dependencies(
__virtualname__,
{'profitbricks': HAS_PROFITBRICKS}
)
'Salt version is lower than 0.16.0, as such, loading '
'configuration from the {0!r} environment variable will '
'fail'.format(env_var)
)
overrides = salt.config.load_config(path, env_var)
default_include = overrides.get(
'default_include', defaults['default_include']
)
include = overrides.get('include', [])
overrides.update(
salt.config.include_config(default_include, path, verbose=False)
)
overrides.update(
salt.config.include_config(include, path, verbose=True)
)
return apply_vm_profiles_config(providers, overrides, defaults)
def _opts_defaults(self, **kwargs):
'''
Set the opts dict to defaults and allow for opts to be overridden in
the kwargs
'''
# Let's start with the default salt cloud configuration
opts = salt.config.DEFAULT_CLOUD_OPTS.copy()
# Update it with the loaded configuration
opts.update(self.opts.copy())
# Reset some of the settings to sane values
opts['parallel'] = False
opts['keep_tmp'] = False
opts['deploy'] = True
opts['update_bootstrap'] = False
opts['show_deploy_args'] = False
opts['script_args'] = ''
# Update it with the passed kwargs
if 'kwargs' in kwargs:
opts.update(kwargs['kwargs'])
opts.update(kwargs)
profile = opts.get('profile', None)
# filter other profiles if one is specified
if profile: