""" FIXME(bja, 2017-11) External and SourceTree have a circular dependancy! """ import errno import logging import os from .externals_description import ExternalsDescription from .externals_description import read_externals_description_file from .externals_description import create_externals_description from .repository_factory import create_repository from .repository_git import GitRepository from .externals_status import ExternalStatus from .utils import fatal_error, printlog from .global_constants import EMPTY_STR, LOCAL_PATH_INDICATOR from .global_constants import VERBOSITY_VERBOSE class _External(object): """ _External represents an external object inside a SourceTree """ # pylint: disable=R0902 def __init__(self, root_dir, name, ext_description, svn_ignore_ancestry): """Parse an external description file into a dictionary of externals. Input: root_dir : string - the root directory path where 'local_path' is relative to. name : string - name of the ext_description object. may or may not correspond to something in the path. ext_description : dict - source ExternalsDescription object svn_ignore_ancestry : bool - use --ignore-externals with svn switch """ self._name = name self._repo = None self._externals = EMPTY_STR self._externals_sourcetree = None self._stat = ExternalStatus() self._sparse = None # Parse the sub-elements # _path : local path relative to the containing source tree self._local_path = ext_description[ExternalsDescription.PATH] # _repo_dir : full repository directory repo_dir = os.path.join(root_dir, self._local_path) self._repo_dir_path = os.path.abspath(repo_dir) # _base_dir : base directory *containing* the repository self._base_dir_path = os.path.dirname(self._repo_dir_path) # repo_dir_name : base_dir_path + repo_dir_name = rep_dir_path self._repo_dir_name = os.path.basename(self._repo_dir_path) assert(os.path.join(self._base_dir_path, self._repo_dir_name) == self._repo_dir_path) self._required = ext_description[ExternalsDescription.REQUIRED] self._externals = ext_description[ExternalsDescription.EXTERNALS] # Treat a .gitmodules file as a backup externals config if not self._externals: if GitRepository.has_submodules(self._repo_dir_path): self._externals = ExternalsDescription.GIT_SUBMODULES_FILENAME repo = create_repository( name, ext_description[ExternalsDescription.REPO], svn_ignore_ancestry=svn_ignore_ancestry) if repo: self._repo = repo if self._externals and (self._externals.lower() != 'none'): self._create_externals_sourcetree() def get_name(self): """ Return the external object's name """ return self._name def get_local_path(self): """ Return the external object's path """ return self._local_path def status(self): """ If the repo destination directory exists, ensure it is correct (from correct URL, correct branch or tag), and possibly update the external. If the repo destination directory does not exist, checkout the correce branch or tag. If load_all is True, also load all of the the externals sub-externals. """ self._stat.path = self.get_local_path() if not self._required: self._stat.source_type = ExternalStatus.OPTIONAL elif self._local_path == LOCAL_PATH_INDICATOR: # LOCAL_PATH_INDICATOR, '.' paths, are standalone # component directories that are not managed by # checkout_externals. self._stat.source_type = ExternalStatus.STANDALONE else: # managed by checkout_externals self._stat.source_type = ExternalStatus.MANAGED ext_stats = {} if not os.path.exists(self._repo_dir_path): self._stat.sync_state = ExternalStatus.EMPTY msg = ('status check: repository directory for "{0}" does not ' 'exist.'.format(self._name)) logging.info(msg) self._stat.current_version = 'not checked out' # NOTE(bja, 2018-01) directory doesn't exist, so we cannot # use repo to determine the expected version. We just take # a best-guess based on the assumption that only tag or # branch should be set, but not both. if not self._repo: self._stat.expected_version = 'unknown' else: self._stat.expected_version = self._repo.tag() + self._repo.branch() else: if self._repo: self._repo.status(self._stat, self._repo_dir_path) if self._externals and self._externals_sourcetree: # we expect externals and they exist cwd = os.getcwd() # SourceTree expects to be called from the correct # root directory. os.chdir(self._repo_dir_path) ext_stats = self._externals_sourcetree.status(self._local_path) os.chdir(cwd) all_stats = {} # don't add the root component because we don't manage it # and can't provide useful info about it. if self._local_path != LOCAL_PATH_INDICATOR: # store the stats under tha local_path, not comp name so # it will be sorted correctly all_stats[self._stat.path] = self._stat if ext_stats: all_stats.update(ext_stats) return all_stats def checkout(self, verbosity, load_all): """ If the repo destination directory exists, ensure it is correct (from correct URL, correct branch or tag), and possibly update the external. If the repo destination directory does not exist, checkout the correct branch or tag. If load_all is True, also load all of the the externals sub-externals. """ if load_all: pass # Make sure we are in correct location if not os.path.exists(self._repo_dir_path): # repository directory doesn't exist. Need to check it # out, and for that we need the base_dir_path to exist try: os.makedirs(self._base_dir_path) except OSError as error: if error.errno != errno.EEXIST: msg = 'Could not create directory "{0}"'.format( self._base_dir_path) fatal_error(msg) if self._stat.source_type != ExternalStatus.STANDALONE: if verbosity >= VERBOSITY_VERBOSE: # NOTE(bja, 2018-01) probably do not want to pass # verbosity in this case, because if (verbosity == # VERBOSITY_DUMP), then the previous status output would # also be dumped, adding noise to the output. self._stat.log_status_message(VERBOSITY_VERBOSE) if self._repo: if self._stat.sync_state == ExternalStatus.STATUS_OK: # If we're already in sync, avoid showing verbose output # from the checkout command, unless the verbosity level # is 2 or more. checkout_verbosity = verbosity - 1 else: checkout_verbosity = verbosity self._repo.checkout(self._base_dir_path, self._repo_dir_name, checkout_verbosity, self.clone_recursive()) def checkout_externals(self, verbosity, load_all): """Checkout the sub-externals for this object """ if self.load_externals(): if self._externals_sourcetree: # NOTE(bja, 2018-02): the subtree externals objects # were created during initial status check. Updating # the external may have changed which sub-externals # are needed. We need to delete those objects and # re-read the potentially modified externals # description file. self._externals_sourcetree = None self._create_externals_sourcetree() self._externals_sourcetree.checkout(verbosity, load_all) def load_externals(self): 'Return True iff an externals file should be loaded' load_ex = False if os.path.exists(self._repo_dir_path): if self._externals: if self._externals.lower() != 'none': load_ex = os.path.exists(os.path.join(self._repo_dir_path, self._externals)) return load_ex def clone_recursive(self): 'Return True iff any .gitmodules files should be processed' # Try recursive unless there is an externals entry recursive = not self._externals return recursive def _create_externals_sourcetree(self): """ """ if not os.path.exists(self._repo_dir_path): # NOTE(bja, 2017-10) repository has not been checked out # yet, can't process the externals file. Assume we are # checking status before code is checkoud out and this # will be handled correctly later. return cwd = os.getcwd() os.chdir(self._repo_dir_path) if self._externals.lower() == 'none': msg = ('Internal: Attempt to create source tree for ' 'externals = none in {}'.format(self._repo_dir_path)) fatal_error(msg) if not os.path.exists(self._externals): if GitRepository.has_submodules(): self._externals = ExternalsDescription.GIT_SUBMODULES_FILENAME if not os.path.exists(self._externals): # NOTE(bja, 2017-10) this check is redundent with the one # in read_externals_description_file! msg = ('External externals description file "{0}" ' 'does not exist! In directory: {1}'.format( self._externals, self._repo_dir_path)) fatal_error(msg) externals_root = self._repo_dir_path model_data = read_externals_description_file(externals_root, self._externals) externals = create_externals_description(model_data, parent_repo=self._repo) self._externals_sourcetree = SourceTree(externals_root, externals) os.chdir(cwd) class SourceTree(object): """ SourceTree represents a group of managed externals """ def __init__(self, root_dir, model, svn_ignore_ancestry=False): """ Build a SourceTree object from a model description """ self._root_dir = os.path.abspath(root_dir) self._all_components = {} self._required_compnames = [] for comp in model: src = _External(self._root_dir, comp, model[comp], svn_ignore_ancestry) self._all_components[comp] = src if model[comp][ExternalsDescription.REQUIRED]: self._required_compnames.append(comp) def status(self, relative_path_base=LOCAL_PATH_INDICATOR): """Report the status components FIXME(bja, 2017-10) what do we do about situations where the user checked out the optional components, but didn't add optional for running status? What do we do where the user didn't add optional to the checkout but did add it to the status. -- For now, we run status on all components, and try to do the right thing based on the results.... """ load_comps = self._all_components.keys() summary = {} for comp in load_comps: printlog('{0}, '.format(comp), end='') stat = self._all_components[comp].status() stat_final = {} for name in stat.keys(): # check if we need to append the relative_path_base to # the path so it will be sorted in the correct order. if stat[name].path.startswith(relative_path_base): # use as is, without any changes to path stat_final[name] = stat[name] else: # append relative_path_base to path and store under key = updated path modified_path = os.path.join(relative_path_base, stat[name].path) stat_final[modified_path] = stat[name] stat_final[modified_path].path = modified_path summary.update(stat_final) return summary def checkout(self, verbosity, load_all, load_comp=None): """ Checkout or update indicated components into the the configured subdirs. If load_all is True, recursively checkout all externals. If load_all is False, load_comp is an optional set of components to load. If load_all is True and load_comp is None, only load the required externals. """ if verbosity >= VERBOSITY_VERBOSE: printlog('Checking out externals: ') else: printlog('Checking out externals: ', end='') if load_all: tmp_comps = self._all_components.keys() elif load_comp is not None: tmp_comps = [load_comp] else: tmp_comps = self._required_compnames # Sort by path so that if paths are nested the # parent repo is checked out first. load_comps = sorted(tmp_comps, key=lambda comp: self._all_components[comp].get_local_path()) # checkout the primary externals for comp in load_comps: if verbosity < VERBOSITY_VERBOSE: printlog('{0}, '.format(comp), end='') else: # verbose output handled by the _External object, just # output a newline printlog(EMPTY_STR) self._all_components[comp].checkout(verbosity, load_all) # now give each external an opportunitity to checkout it's externals. self._all_components[comp].checkout_externals(verbosity, load_all) printlog('')