Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:raise ValueError: error raised if file extension is not supported
"""
opener = None
mode = None
if filepath.endswith('.zip'):
logging.debug('File Type: zip')
opener, mode = zipfile.ZipFile, 'r'
elif filepath.endswith('.tar.gz') or filepath.endswith('.tgz'):
logging.debug('File Type: GZip Tar')
opener, mode = tarfile.open, 'r:gz'
elif filepath.endswith('.tar.bz2') or filepath.endswith('.tbz'):
logging.debug('File Type: Bzip Tar')
opener, mode = tarfile.open, 'r:bz2'
else:
exceptionhandler('{0}'.format('Could not extract `"{0}`" as no appropriate '
'extractor is found'.format(filepath)))
if create_dir:
to_directory = os.sep.join((to_directory,
'.'.join(filepath.split(os.sep)[-1].split('.')[:-1])))
try:
os.makedirs(to_directory)
except OSError:
if not os.path.isdir(to_directory):
raise
cwd = os.getcwd()
os.chdir(to_directory)
try:
:param packages:
:return:
"""
yum_cmd = ['sudo', 'yum', '-y', 'install']
if isinstance(packages, list):
yum_cmd.extend(packages)
else:
yum_cmd.append(packages)
rsp = subprocess.call(yum_cmd)
logging.debug(packages)
logging.debug('Return code of yum install: {0}'.format(rsp))
if rsp != 0:
exceptionhandler('Installing Salt from Yum has failed!')
def install(self, configuration):
"""
:param configuration:
:return:
"""
try:
self.config = json.loads(configuration)
except ValueError:
exceptionhandler('The configuration passed was not properly formed JSON. Execution Halted.')
self._configuration_validation()
self._prepare_for_install()
self._install_package()
self._build_salt_formula()
logging.info('Setting grain `systemprep`...')
ent_env = {'enterprise_environment': str(self.entenv)}
cmd = [self.saltcall, '--local', '--retcode-passthrough', 'grains.setval', 'systemprep',
str(json.dumps(ent_env))]
self.call_process(cmd)
if self.config['oupath']:
print('Setting grain `join-domain`...')
oupath = {'oupath': self.config['oupath']}
cmd = [self.saltcall, '--local', '--retcode-passthrough', 'grains.setval', '"join-domain"',
def cleanup(self):
"""
Removes temporary files loaded to the system.
:return: bool
"""
logging.info('+-' * 40)
logging.info('Cleanup Time...')
try:
logging.debug('{0} being cleaned up.'.format(self.workingdir))
shutil.rmtree(self.workingdir)
except Exception as exc:
# TODO: Update `except` logic
logging.fatal('Cleanup Failed!\nException: {0}'.format(exc))
exceptionhandler('Cleanup Failed.\nAborting.')
logging.info('Removed temporary data in working directory -- {0}'.format(self.workingdir))
logging.info('Exiting cleanup routine...')
logging.info('-+' * 40)
def call_process(self, cmd):
if not isinstance(cmd, list):
exceptionhandler('Command is not a list.\n{0}'.format(str(cmd)))
rsp = subprocess.call(cmd)
if rsp != 0:
exceptionhandler('Command failed.\n{0}'.format(str(cmd)))
Creates a directory in `basedir` with a prefix of `dirprefix`.
The directory will have a random 5 character string appended to `dirprefix`.
Returns the path to the working directory.
:param prefix:
:rtype : str
:param basedir: str, the directory in which to create the working directory
"""
logging.info('Creating a working directory.')
workingdir = None
original_umask = os.umask(0)
try:
workingdir = tempfile.mkdtemp(prefix=prefix, dir=basedir)
except Exception as exc:
exceptionhandler('Could not create workingdir in {0}.\n'
'Exception: {1}'.format(basedir, exc))
logging.debug('Working directory: {0}'.format(workingdir))
self.workingdir = workingdir
os.umask(original_umask)