jython-db4o / Lib / site-packages / pip-1.2.1-py2.7.egg / pip / vcs /

"""Handles all VCS (version control) support"""

import os
import shutil

from pip.backwardcompat import urlparse, urllib
from pip.log import logger
from pip.util import (display_path, backup_dir, find_command,
                      ask, rmtree, ask_path_exists)

__all__ = ['vcs', 'get_src_requirement']

class VcsSupport(object):
    _registry = {}
    schemes = ['ssh', 'git', 'hg', 'bzr', 'sftp', 'svn']

    def __init__(self):
        # Register more schemes with urlparse for various version control systems
        # Python >= 2.7.4, 3.3 doesn't have uses_fragment
        if getattr(urlparse, 'uses_fragment', None):
        super(VcsSupport, self).__init__()

    def __iter__(self):
        return self._registry.__iter__()

    def backends(self):
        return list(self._registry.values())

    def dirnames(self):
        return [backend.dirname for backend in self.backends]

    def all_schemes(self):
        schemes = []
        for backend in self.backends:
        return schemes

    def register(self, cls):
        if not hasattr(cls, 'name'):
            logger.warn('Cannot register VCS %s' % cls.__name__)
        if not in self._registry:
            self._registry[] = cls

    def unregister(self, cls=None, name=None):
        if name in self._registry:
            del self._registry[name]
        elif cls in self._registry.values():
            del self._registry[]
            logger.warn('Cannot unregister because no class or name given')

    def get_backend_name(self, location):
        Return the name of the version control backend if found at given
        location, e.g. vcs.get_backend_name('/path/to/vcs/checkout')
        for vc_type in self._registry.values():
            path = os.path.join(location, vc_type.dirname)
            if os.path.exists(path):
        return None

    def get_backend(self, name):
        name = name.lower()
        if name in self._registry:
            return self._registry[name]

    def get_backend_from_location(self, location):
        vc_type = self.get_backend_name(location)
        if vc_type:
            return self.get_backend(vc_type)
        return None

vcs = VcsSupport()

class VersionControl(object):
    name = ''
    dirname = ''

    def __init__(self, url=None, *args, **kwargs):
        self.url = url
        self._cmd = None
        super(VersionControl, self).__init__(*args, **kwargs)

    def _filter(self, line):
        return (logger.INFO, line)

    def _is_local_repository(self, repo):
           posix absolute paths start with os.path.sep,
           win32 ones ones start with drive (like c:\\folder)
        drive, tail = os.path.splitdrive(repo)
        return repo.startswith(os.path.sep) or drive

    def cmd(self):
        if self._cmd is not None:
            return self._cmd
        command = find_command('Found command %r at %r' % (, command))
        self._cmd = command
        return command

    def get_url_rev(self):
        Returns the correct repository URL and revision by parsing the given
        repository URL
        error_message= (
           "Sorry, '%s' is a malformed VCS url. "
           "Ihe format is <vcs>+<protocol>://<url>, "
           "e.g. svn+http://myrepo/svn/MyApp#egg=MyApp")
        assert '+' in self.url, error_message % self.url
        url = self.url.split('+', 1)[1]
        scheme, netloc, path, query, frag = urlparse.urlsplit(url)
        rev = None
        if '@' in path:
            path, rev = path.rsplit('@', 1)
        url = urlparse.urlunsplit((scheme, netloc, path, query, ''))
        return url, rev

    def get_info(self, location):
        Returns (url, revision), where both are strings
        assert not location.rstrip('/').endswith(self.dirname), 'Bad directory: %s' % location
        return self.get_url(location), self.get_revision(location)

    def normalize_url(self, url):
        Normalize a URL for comparison by unquoting it and removing any trailing slash.
        return urllib.unquote(url).rstrip('/')

    def compare_urls(self, url1, url2):
        Compare two repo URLs for identity, ignoring incidental differences.
        return (self.normalize_url(url1) == self.normalize_url(url2))

    def parse_vcs_bundle_file(self, content):
        Takes the contents of the bundled text file that explains how to revert
        the stripped off version control data of the given package and returns
        the URL and revision of it.
        raise NotImplementedError

    def obtain(self, dest):
        Called when installing or updating an editable package, takes the
        source path of the checkout.
        raise NotImplementedError

    def switch(self, dest, url, rev_options):
        Switch the repo at ``dest`` to point to ``URL``.
        raise NotImplemented

    def update(self, dest, rev_options):
        Update an already-existing repo to the given ``rev_options``.
        raise NotImplementedError

    def check_destination(self, dest, url, rev_options, rev_display):
        Prepare a location to receive a checkout/clone.

        Return True if the location is ready for (and requires) a
        checkout/clone, False otherwise.
        checkout = True
        prompt = False
        if os.path.exists(dest):
            checkout = False
            if os.path.exists(os.path.join(dest, self.dirname)):
                existing_url = self.get_url(dest)
                if self.compare_urls(existing_url, url):
          '%s in %s exists, and has correct URL (%s)' %
                                (self.repo_name.title(), display_path(dest),
                    logger.notify('Updating %s %s%s' %
                                  (display_path(dest), self.repo_name,
                    self.update(dest, rev_options)
                    logger.warn('%s %s in %s exists with URL %s' %
                                (, self.repo_name,
                                 display_path(dest), existing_url))
                    prompt = ('(s)witch, (i)gnore, (w)ipe, (b)ackup ',
                              ('s', 'i', 'w', 'b'))
                logger.warn('Directory %s already exists, '
                            'and is not a %s %s.' %
                            (dest,, self.repo_name))
                prompt = ('(i)gnore, (w)ipe, (b)ackup ', ('i', 'w', 'b'))
        if prompt:
            logger.warn('The plan is to install the %s repository %s' %
                        (, url))
            response = ask_path_exists('What to do?  %s' % prompt[0],

            if response == 's':
                logger.notify('Switching %s %s to %s%s' %
                              (self.repo_name, display_path(dest), url,
                self.switch(dest, url, rev_options)
            elif response == 'i':
                # do nothing
            elif response == 'w':
                logger.warn('Deleting %s' % display_path(dest))
                checkout = True
            elif response == 'b':
                dest_dir = backup_dir(dest)
                logger.warn('Backing up %s to %s'
                            % (display_path(dest), dest_dir))
                shutil.move(dest, dest_dir)
                checkout = True
        return checkout

    def unpack(self, location):
        if os.path.exists(location):

    def get_src_requirement(self, dist, location, find_tags=False):
        raise NotImplementedError

def get_src_requirement(dist, location, find_tags):
    version_control = vcs.get_backend_from_location(location)
    if version_control:
        return version_control().get_src_requirement(dist, location, find_tags)
    logger.warn('cannot determine version of editable source in %s (is not SVN checkout, Git clone, Mercurial clone or Bazaar branch)' % location)
    return dist.as_requirement()