Source code for sdss_install.install4.Install4
# License information goes here
# -*- coding: utf-8 -*-
"""Install SDSS-IV software.
"""
from __future__ import absolute_import, division, print_function, unicode_literals
from os.path import basename, dirname, join
from subprocess import PIPE, Popen
from sdss_install.application import Argument
try:
from ConfigParser import SafeConfigParser, RawConfigParser
except ImportError:
from configparser import SafeConfigParser, RawConfigParser
[docs]class Install4:
'''Class for sdss_install'ation of SVN repositories.'''
def __init__(self, logger=None, options=None):
self.set_logger(logger=logger)
self.set_options(options=options)
self.ready = False
self.url = None
self.product = None
self.package = None
self.directory = None
self.svncommand = None
self.exists = None
self.modules = None
self.build_type = None
[docs] def set_logger(self, logger=None):
'''Set the class logger'''
self.logger = logger if logger else None
if not self.logger:
print('ERROR: %r> Unable to set logger.' % self.__class__)
[docs] def set_options(self, options=None):
'''Set command line argument options'''
self.options = options if options else None
if not self.options:
if self.logger:
self.logger.error('Unable to set_options')
else:
print('ERROR: Unable to set_options')
[docs] def set_ready(self):
'''Set self.ready after sanity check self.options.'''
self.ready = self.options and self.logger
if self.ready:
self.url = (join(self.options.url, 'public')
if self.options.public
else self.options.url)
if (self.options.product == 'NO PACKAGE' or
self.options.product_version == 'NO VERSION'):
self.logger.error("You must specify a product " +
"and the version (after a space)!")
self.ready = False
elif self.options.product:
if self.options.product.endswith('/'):
self.options.product = dirname(self.options.product)
if self.options.product_version.endswith('/'):
self.options.product_version = (
dirname(self.options.product_version))
svnroots = ['repo', 'data', 'deprecated']
validproduct = [svnroot for svnroot in svnroots
if self.options.product.startswith(svnroot)]
if not validproduct:
self.options.product = join('repo', self.options.product)
else:
self.url = None
[docs] def set_product(self):
'''Set self.product dict containing product and version names etc.'''
# Determine the product and version names.
if self.ready:
self.product = {}
self.product['root'] = (dirname(self.options.product)
if self.options.longpath
else None)
self.product['name'] = basename(self.options.product)
self.product['version'] = basename(self.options.product_version)
self.product['is_master'] = self.options.product_version == 'trunk'
self.product['is_branch'] = (
self.options.product_version.startswith('branches'))
self.product['is_master_or_branch'] = (self.product['is_master'] or
self.product['is_branch'])
self.product['url'] = (self.options.product_version
if self.product['is_master_or_branch']
else join('tags',
self.options.product_version))
self.product['url'] = join(self.url,
self.options.product,
self.product['url'])
self.product['checkout_or_export'] = (
'checkout'
if self.product['is_master_or_branch']
and not self.options.public
else 'export')
[docs] def set_svncommand(self):
'''
Set the SVN command for public,
otherwise add username to SVN command.
'''
if self.ready:
self.svncommand = ['svn']
if not self.options.public and self.options.username:
self.svncommand += ['--username', self.options.username]
[docs] def set_exists(self):
'''Check for existence of the product URL.'''
if self.ready:
self.logger.info("Contacting {url} ".format(url=self.url))
command = self.svncommand + ['ls', self.product['url']]
self.logger.debug(' '.join(command))
(out, err, proc_returncode) = self.execute_command(command=command)
self.logger.debug(out)
self.exists = proc_returncode == 0
if self.exists:
self.logger.info("Found URL at %(url)s " % self.product)
else:
self.logger.error("Nonexistent URL at %(url)s" % self.product)
self.logger.error(err)
self.ready = False
[docs] def fetch(self):
'''SVN checkout or export the product version.'''
if self.ready:
command = (self.svncommand +
[self.product['checkout_or_export'],
self.product['url'],
self.directory['work']])
self.logger.info("Running %(checkout_or_export)s of %(url)s"
% self.product)
self.logger.debug(' '.join(command))
(out, err, proc_returncode) = self.execute_command(command=command)
self.logger.debug(out)
self.ready = not len(err)
if self.ready:
self.logger.info("Completed svn %(checkout_or_export)s " +
"of %(url)s" % self.product)
else:
self.logger.error("svn error during %(checkout_or_export)s " +
"of %(url)s: " % self.product + err)
[docs] def execute_command(self, command=None):
'''Execute the passed terminal command.'''
(out, err, proc_returncode) = (None, None, None)
if command:
proc = Popen(command, stdout=PIPE, stderr=PIPE)
if proc:
(out, err) = proc.communicate() if proc else (None, None)
out = out.decode("utf-8") if isinstance(out, bytes) else out
err = err.decode("utf-8") if isinstance(err, bytes) else err
proc_returncode = proc.returncode if proc else None
else:
self.ready = False
self.logger.error('Unable to execute_command. ' +
'proc: {}'.format(proc))
else:
self.ready = False
self.logger.error('Unable to execute_command. ' +
'command: {}'.format(command))
return (out, err, proc_returncode)