Source code for anybox.recipe.odoo.vcs.base

import os
import shutil
import subprocess
import logging
from .. import utils

SUBPROCESS_ENV = os.environ.copy()

logger = logging.getLogger(__name__)

[docs]class UpdateError(subprocess.CalledProcessError): """Specific class for errors occurring during updates of existing repos. """
[docs]class CloneError(subprocess.CalledProcessError): """Class to easily signal errors in initial cloning. """
[docs]def wrap_check_call(exc_cls, call_fn): def wrapped_check_call(*args, **kwargs): """Variant on subprocess.check_* that raises %s.""" % exc_cls try: return call_fn(*args, **kwargs) except subprocess.CalledProcessError, e: up_exc = exc_cls(e.returncode, e.cmd) output = getattr(e, 'output', None) if output is not None: up_exc.output = output raise up_exc return wrapped_check_call
update_check_call = wrap_check_call(UpdateError, subprocess.check_call) clone_check_call = wrap_check_call(CloneError, subprocess.check_call) update_check_output = wrap_check_call(UpdateError, utils.check_output) clone_check_output = wrap_check_call(CloneError, utils.check_output)
[docs]class BaseRepo(object): """The common interface that all repository classes implement. :param target_dir: the local directory which will serve as a working tree :param offline: if ``True``, the repository instance will perform no network operation, and will fail instead if a non available revision is required. :param clear_locks: Some VCS systems can leave locks after some failures and provide a separate way to break them. If ``True``, the repo will break any locks prior to operations (mostly useful for automated agents, such as CI robots) :param clear_retry: if ``True`` failed updates by calling the instance are cleared (see :meth:`clear_target`) and retried once. This is intended for brittle VCSes from CI robots. Other options depend on the concrete repository class. Repository instances are **callable**. For each of them:: repo(rev) will take all the steps necessary so that its local directory is a clone of the remote source, at the specified revision. If needed and possible The revision format depends on the concrete class, but it is passed as a :class:`str`. """ def __init__(self, target_dir, url, clear_retry=False, offline=False, clear_locks=False, **options): self.target_dir = target_dir self.url = url self.clear_retry = clear_retry self.offline = offline self.clear_locks = clear_locks # additional options that may depend on the VCS subclass self.options = options
[docs] def clear_target(self): """Entirely remove the target directory.""" shutil.rmtree(self.target_dir)
[docs] def clean(self): """Remove unwanted untracked files. This default implementation removes Python object files and (resulting) empty directories. Subclasses are supposed to implement better vcs-specific behaviours. It is important for release-related options that this cleaning does not appear as a local modification. """ utils.clean_object_files(self.target_dir)
[docs] def revert(self, revision): """Revert any local changes, including pending merges.""" raise NotImplementedError
def __call__(self, revision): """Create if needed from remote source, and put it at wanted revision. """ if self.options.get('clean'): self.clean() try: self.get_update(revision) except UpdateError: if self.offline or not self.clear_retry: raise logger.warn("Update of %s failed, removing and re-cloning " "according to the clear-retry option. ", self) self.clear_target() self.get_update(revision) return self # nicer in particular for tests
[docs] def get_update(self, revision): """Make it so that the target directory is at the prescribed revision. The target directory need not to be initialized: this method will "clone" it from the remote source (whatever that means in the considered VCS). This method can fail under various circumstances, for instance if the wanted revision does not exist locally and offline mode has been selected. :raises CloneError: if initial cloning fails :raises UpdateError: if update of existing repository fails Must be implemented in concrete subclasses """ raise NotImplementedError
def __str__(self): return "%s at %r (remote=%r)" % ( self.__class__.__name__, self.target_dir, self.url) @classmethod
[docs] def is_versioned(cls, path): """True if path exists and is versioned under this vcs. Common implementation based on vcs_control_dir class attribute. """ return os.path.exists(os.path.join(path, cls.vcs_control_dir))
[docs] def uncommitted_changes(self): """True if we have uncommitted changes. Must be implemented by concrete subclasses """ raise NotImplementedError
[docs] def parents(self, pip_compatible=False): """Return universal identifier for parent nodes, aka current revisions. There might be more than one with some VCSes (ex: pending merge in hg). :param pip_compatible: if ``True``, only `pip compatible < reference/pip_install.html#vcs-support>`_ revision specifications are returned, depending on the VCS type. """ raise NotImplementedError
[docs] def archive(self, target_path): raise NotImplementedError