Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.computer_name = watchmaker.utils.config_none_deprecate(
self.computer_name, self.log)
self.salt_debug_log = watchmaker.utils.config_none_deprecate(
self.salt_debug_log, self.log)
self.salt_content = watchmaker.utils.config_none_deprecate(
self.salt_content, self.log)
self.salt_content_path = watchmaker.utils.config_none_deprecate(
self.salt_content_path, self.log)
self.ou_path = watchmaker.utils.config_none_deprecate(
self.ou_path, self.log)
self.admin_groups = watchmaker.utils.config_none_deprecate(
self.admin_groups, self.log)
self.admin_users = watchmaker.utils.config_none_deprecate(
self.admin_users, self.log)
self.salt_states = watchmaker.utils.config_none_deprecate(
self.salt_states, self.log)
# Init attributes used by SaltBase, overridden by inheriting classes
self.salt_working_dir = None
self.salt_working_dir_prefix = None
self.salt_log_dir = None
self.salt_conf_path = None
self.salt_conf = None
self.salt_call = None
self.salt_base_env = None
self.salt_formula_root = None
self.salt_file_roots = None
self.salt_state_args = None
self.salt_debug_logfile = None
# Pop arguments used by SaltBase
self.user_formulas = kwargs.pop('user_formulas', None) or {}
self.computer_name = kwargs.pop('computer_name', None) or ''
self.ent_env = kwargs.pop('environment', None) or ''
self.valid_envs = kwargs.pop('valid_environments', []) or []
self.salt_debug_log = kwargs.pop('salt_debug_log', None) or ''
self.salt_content = kwargs.pop('salt_content', None) or ''
self.salt_content_path = kwargs.pop('salt_content_path', None) or ''
self.ou_path = kwargs.pop('ou_path', None) or ''
self.admin_groups = kwargs.pop('admin_groups', None) or ''
self.admin_users = kwargs.pop('admin_users', None) or ''
self.salt_states = kwargs.pop('salt_states', None) or ''
self.exclude_states = kwargs.pop('exclude_states', None) or ''
self.computer_name = watchmaker.utils.config_none_deprecate(
self.computer_name, self.log)
self.salt_debug_log = watchmaker.utils.config_none_deprecate(
self.salt_debug_log, self.log)
self.salt_content = watchmaker.utils.config_none_deprecate(
self.salt_content, self.log)
self.salt_content_path = watchmaker.utils.config_none_deprecate(
self.salt_content_path, self.log)
self.ou_path = watchmaker.utils.config_none_deprecate(
self.ou_path, self.log)
self.admin_groups = watchmaker.utils.config_none_deprecate(
self.admin_groups, self.log)
self.admin_users = watchmaker.utils.config_none_deprecate(
self.admin_users, self.log)
self.salt_states = watchmaker.utils.config_none_deprecate(
self.salt_states, self.log)
def __init__(self, *args, **kwargs):
# Pop arguments used by SaltWindows
self.installer_url = kwargs.pop('installer_url', None) or ''
self.ash_role = kwargs.pop('ash_role', None) or ''
# Init inherited classes
super(SaltWindows, self).__init__(*args, **kwargs)
self.ash_role = watchmaker.utils.config_none_deprecate(
self.ash_role, self.log)
# Extra variable needed for SaltWindows.
sys_drive = os.environ['systemdrive']
# Set up variables for paths to Salt directories and applications.
self.salt_root = os.sep.join((sys_drive, 'Salt'))
self.salt_call = os.sep.join((self.salt_root, 'salt-call.bat'))
self.salt_wam_root = os.sep.join((
self.system_params['prepdir'],
'Salt'))
self.salt_conf_path = os.sep.join((self.salt_wam_root, 'conf'))
self.salt_srv = os.sep.join((self.salt_wam_root, 'srv'))
self.salt_win_repo = os.sep.join((self.salt_srv, 'winrepo'))
self.salt_log_dir = self.system_params['logdir']
def retrieve_file(self, url, filename):
"""
Retrieve a file from a provided URL.
Supports all :obj:`urllib.request` handlers, as well as S3 buckets.
Args:
url: (:obj:`str`)
URL to a file.
filename: (:obj:`str`)
Path where the file will be saved.
"""
# Convert a local path to a URI
url = watchmaker.utils.uri_from_filepath(url)
self.log.debug('Downloading: %s', url)
self.log.debug('Destination: %s', filename)
try:
self.log.debug('Establishing connection to the host, %s', url)
response = watchmaker.utils.urlopen_retry(url)
self.log.debug('Opening the file handle, %s', filename)
with open(filename, 'wb') as outfile:
self.log.debug('Saving file to local filesystem...')
shutil.copyfileobj(response, outfile)
except (ValueError, urllib.error.URLError):
self.log.critical(
'Failed to retrieve the file. url = %s. filename = %s',
url, filename
)
raise
Args:
url: (:obj:`str`)
URL to a file.
filename: (:obj:`str`)
Path where the file will be saved.
"""
# Convert a local path to a URI
url = watchmaker.utils.uri_from_filepath(url)
self.log.debug('Downloading: %s', url)
self.log.debug('Destination: %s', filename)
try:
self.log.debug('Establishing connection to the host, %s', url)
response = watchmaker.utils.urlopen_retry(url)
self.log.debug('Opening the file handle, %s', filename)
with open(filename, 'wb') as outfile:
self.log.debug('Saving file to local filesystem...')
shutil.copyfileobj(response, outfile)
except (ValueError, urllib.error.URLError):
self.log.critical(
'Failed to retrieve the file. url = %s. filename = %s',
url, filename
)
raise
self.log.info(
'Retrieved the file successfully. url=%s. filename=%s',
url, filename
)