mirror of https://github.com/pypa/pip
Merge pull request #6545 from cjerdonek/vcs-imports
Remove from pip/_internal/__init__.py the vcs module imports
This commit is contained in:
Chris Jerdonek
GitHub
commit
ce46f8524e
|
|
@ -42,7 +42,6 @@ from pip._internal.cli.main_parser import parse_command
|
|||
from pip._internal.commands import commands_dict
|
||||
from pip._internal.exceptions import PipError
|
||||
from pip._internal.utils import deprecation
|
||||
from pip._internal.vcs import git, mercurial, subversion, bazaar # noqa
|
||||
from pip._vendor.urllib3.exceptions import InsecureRequestWarning
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
|
|||
|
|
@ -37,8 +37,8 @@ from pip._internal.utils.glibc import libc_ver
|
|||
from pip._internal.utils.misc import (
|
||||
ARCHIVE_EXTENSIONS, ask, ask_input, ask_password, ask_path_exists,
|
||||
backup_dir, consume, display_path, format_size, get_installed_version,
|
||||
remove_auth_from_url, rmtree, split_auth_netloc_from_url, splitext,
|
||||
unpack_file,
|
||||
path_to_url, remove_auth_from_url, rmtree, split_auth_netloc_from_url,
|
||||
splitext, unpack_file,
|
||||
)
|
||||
from pip._internal.utils.temp_dir import TempDirectory
|
||||
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
|
||||
|
|
@ -52,7 +52,7 @@ if MYPY_CHECK_RUNNING:
|
|||
from optparse import Values
|
||||
from pip._internal.models.link import Link
|
||||
from pip._internal.utils.hashes import Hashes
|
||||
from pip._internal.vcs import AuthInfo, VersionControl
|
||||
from pip._internal.vcs.versioncontrol import AuthInfo, VersionControl
|
||||
|
||||
try:
|
||||
import ssl # noqa
|
||||
|
|
@ -693,17 +693,6 @@ def url_to_path(url):
|
|||
return path
|
||||
|
||||
|
||||
def path_to_url(path):
|
||||
# type: (Union[str, Text]) -> str
|
||||
"""
|
||||
Convert a path to a file: URL. The path will be made absolute and have
|
||||
quoted path parts.
|
||||
"""
|
||||
path = os.path.normpath(os.path.abspath(path))
|
||||
url = urllib_parse.urljoin('file:', urllib_request.pathname2url(path))
|
||||
return url
|
||||
|
||||
|
||||
def is_archive_file(name):
|
||||
# type: (str) -> bool
|
||||
"""Return True if `name` is a considered as an archive file."""
|
||||
|
|
|
|||
|
|
@ -82,8 +82,8 @@ def _match_vcs_scheme(url):
|
|||
|
||||
Returns the matched VCS scheme, or None if there's no match.
|
||||
"""
|
||||
from pip._internal.vcs import VcsSupport
|
||||
for scheme in VcsSupport.schemes:
|
||||
from pip._internal.vcs import vcs
|
||||
for scheme in vcs.schemes:
|
||||
if url.lower().startswith(scheme) and url[len(scheme)] in '+:':
|
||||
return scheme
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ from pip._vendor.retrying import retry # type: ignore
|
|||
from pip._vendor.six import PY2
|
||||
from pip._vendor.six.moves import input, shlex_quote
|
||||
from pip._vendor.six.moves.urllib import parse as urllib_parse
|
||||
from pip._vendor.six.moves.urllib import request as urllib_request
|
||||
from pip._vendor.six.moves.urllib.parse import unquote as urllib_unquote
|
||||
|
||||
from pip._internal.exceptions import CommandError, InstallationError
|
||||
|
|
@ -941,6 +942,17 @@ def enum(*sequential, **named):
|
|||
return type('Enum', (), enums)
|
||||
|
||||
|
||||
def path_to_url(path):
|
||||
# type: (Union[str, Text]) -> str
|
||||
"""
|
||||
Convert a path to a file: URL. The path will be made absolute and have
|
||||
quoted path parts.
|
||||
"""
|
||||
path = os.path.normpath(os.path.abspath(path))
|
||||
url = urllib_parse.urljoin('file:', urllib_request.pathname2url(path))
|
||||
return url
|
||||
|
||||
|
||||
def split_auth_from_netloc(netloc):
|
||||
"""
|
||||
Parse out and remove the auth information from a netloc.
|
||||
|
|
|
|||
|
|
@ -1,600 +1,12 @@
|
|||
"""Handles all VCS (version control) support"""
|
||||
from __future__ import absolute_import
|
||||
|
||||
import errno
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
from pip._vendor import pkg_resources
|
||||
from pip._vendor.six.moves.urllib import parse as urllib_parse
|
||||
|
||||
from pip._internal.exceptions import BadCommand
|
||||
from pip._internal.utils.misc import (
|
||||
ask_path_exists, backup_dir, call_subprocess, display_path, rmtree,
|
||||
# Expose a limited set of classes and functions so callers outside of
|
||||
# the vcs package don't need to import deeper than `pip._internal.vcs`.
|
||||
# (The test directory and imports protected by MYPY_CHECK_RUNNING may
|
||||
# still need to import from a vcs sub-package.)
|
||||
from pip._internal.vcs.versioncontrol import ( # noqa: F401
|
||||
RemoteNotFoundError, make_vcs_requirement_url, vcs,
|
||||
)
|
||||
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
|
||||
|
||||
if MYPY_CHECK_RUNNING:
|
||||
from typing import (
|
||||
Any, Dict, Iterable, List, Mapping, Optional, Text, Tuple, Type
|
||||
)
|
||||
from pip._internal.utils.ui import SpinnerInterface
|
||||
|
||||
AuthInfo = Tuple[Optional[str], Optional[str]]
|
||||
|
||||
__all__ = ['vcs']
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def make_vcs_requirement_url(repo_url, rev, project_name, subdir=None):
|
||||
"""
|
||||
Return the URL for a VCS requirement.
|
||||
|
||||
Args:
|
||||
repo_url: the remote VCS url, with any needed VCS prefix (e.g. "git+").
|
||||
project_name: the (unescaped) project name.
|
||||
"""
|
||||
egg_project_name = pkg_resources.to_filename(project_name)
|
||||
req = '{}@{}#egg={}'.format(repo_url, rev, egg_project_name)
|
||||
if subdir:
|
||||
req += '&subdirectory={}'.format(subdir)
|
||||
|
||||
return req
|
||||
|
||||
|
||||
class RemoteNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class RevOptions(object):
|
||||
|
||||
"""
|
||||
Encapsulates a VCS-specific revision to install, along with any VCS
|
||||
install options.
|
||||
|
||||
Instances of this class should be treated as if immutable.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
vc_class, # type: Type[VersionControl]
|
||||
rev=None, # type: Optional[str]
|
||||
extra_args=None, # type: Optional[List[str]]
|
||||
):
|
||||
# type: (...) -> None
|
||||
"""
|
||||
Args:
|
||||
vc_class: a VersionControl subclass.
|
||||
rev: the name of the revision to install.
|
||||
extra_args: a list of extra options.
|
||||
"""
|
||||
if extra_args is None:
|
||||
extra_args = []
|
||||
|
||||
self.extra_args = extra_args
|
||||
self.rev = rev
|
||||
self.vc_class = vc_class
|
||||
|
||||
def __repr__(self):
|
||||
return '<RevOptions {}: rev={!r}>'.format(self.vc_class.name, self.rev)
|
||||
|
||||
@property
|
||||
def arg_rev(self):
|
||||
# type: () -> Optional[str]
|
||||
if self.rev is None:
|
||||
return self.vc_class.default_arg_rev
|
||||
|
||||
return self.rev
|
||||
|
||||
def to_args(self):
|
||||
# type: () -> List[str]
|
||||
"""
|
||||
Return the VCS-specific command arguments.
|
||||
"""
|
||||
args = [] # type: List[str]
|
||||
rev = self.arg_rev
|
||||
if rev is not None:
|
||||
args += self.vc_class.get_base_rev_args(rev)
|
||||
args += self.extra_args
|
||||
|
||||
return args
|
||||
|
||||
def to_display(self):
|
||||
# type: () -> str
|
||||
if not self.rev:
|
||||
return ''
|
||||
|
||||
return ' (to revision {})'.format(self.rev)
|
||||
|
||||
def make_new(self, rev):
|
||||
# type: (str) -> RevOptions
|
||||
"""
|
||||
Make a copy of the current instance, but with a new rev.
|
||||
|
||||
Args:
|
||||
rev: the name of the revision for the new object.
|
||||
"""
|
||||
return self.vc_class.make_rev_options(rev, extra_args=self.extra_args)
|
||||
|
||||
|
||||
class VcsSupport(object):
|
||||
_registry = {} # type: Dict[str, VersionControl]
|
||||
schemes = ['ssh', 'git', 'hg', 'bzr', 'sftp', 'svn']
|
||||
|
||||
def __init__(self):
|
||||
# type: () -> None
|
||||
# Register more schemes with urlparse for various version control
|
||||
# systems
|
||||
urllib_parse.uses_netloc.extend(self.schemes)
|
||||
# Python >= 2.7.4, 3.3 doesn't have uses_fragment
|
||||
if getattr(urllib_parse, 'uses_fragment', None):
|
||||
urllib_parse.uses_fragment.extend(self.schemes)
|
||||
super(VcsSupport, self).__init__()
|
||||
|
||||
def __iter__(self):
|
||||
return self._registry.__iter__()
|
||||
|
||||
@property
|
||||
def backends(self):
|
||||
# type: () -> List[VersionControl]
|
||||
return list(self._registry.values())
|
||||
|
||||
@property
|
||||
def dirnames(self):
|
||||
# type: () -> List[str]
|
||||
return [backend.dirname for backend in self.backends]
|
||||
|
||||
@property
|
||||
def all_schemes(self):
|
||||
# type: () -> List[str]
|
||||
schemes = [] # type: List[str]
|
||||
for backend in self.backends:
|
||||
schemes.extend(backend.schemes)
|
||||
return schemes
|
||||
|
||||
def register(self, cls):
|
||||
# type: (Type[VersionControl]) -> None
|
||||
if not hasattr(cls, 'name'):
|
||||
logger.warning('Cannot register VCS %s', cls.__name__)
|
||||
return
|
||||
if cls.name not in self._registry:
|
||||
self._registry[cls.name] = cls()
|
||||
logger.debug('Registered VCS backend: %s', cls.name)
|
||||
|
||||
def unregister(self, name):
|
||||
# type: (str) -> None
|
||||
if name in self._registry:
|
||||
del self._registry[name]
|
||||
|
||||
def get_backend_for_dir(self, location):
|
||||
# type: (str) -> Optional[VersionControl]
|
||||
"""
|
||||
Return a VersionControl object if a repository of that type is found
|
||||
at the given directory.
|
||||
"""
|
||||
for vcs_backend in self._registry.values():
|
||||
if vcs_backend.controls_location(location):
|
||||
logger.debug('Determine that %s uses VCS: %s',
|
||||
location, vcs_backend.name)
|
||||
return vcs_backend
|
||||
return None
|
||||
|
||||
def get_backend(self, name):
|
||||
# type: (str) -> Optional[VersionControl]
|
||||
"""
|
||||
Return a VersionControl object or None.
|
||||
"""
|
||||
name = name.lower()
|
||||
return self._registry.get(name)
|
||||
|
||||
|
||||
vcs = VcsSupport()
|
||||
|
||||
|
||||
class VersionControl(object):
|
||||
name = ''
|
||||
dirname = ''
|
||||
repo_name = ''
|
||||
# List of supported schemes for this Version Control
|
||||
schemes = () # type: Tuple[str, ...]
|
||||
# Iterable of environment variable names to pass to call_subprocess().
|
||||
unset_environ = () # type: Tuple[str, ...]
|
||||
default_arg_rev = None # type: Optional[str]
|
||||
|
||||
@classmethod
|
||||
def should_add_vcs_url_prefix(cls, remote_url):
|
||||
"""
|
||||
Return whether the vcs prefix (e.g. "git+") should be added to a
|
||||
repository's remote url when used in a requirement.
|
||||
"""
|
||||
return not remote_url.lower().startswith('{}:'.format(cls.name))
|
||||
|
||||
@classmethod
|
||||
def get_subdirectory(cls, repo_dir):
|
||||
"""
|
||||
Return the path to setup.py, relative to the repo root.
|
||||
"""
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_requirement_revision(cls, repo_dir):
|
||||
"""
|
||||
Return the revision string that should be used in a requirement.
|
||||
"""
|
||||
return cls.get_revision(repo_dir)
|
||||
|
||||
@classmethod
|
||||
def get_src_requirement(cls, repo_dir, project_name):
|
||||
"""
|
||||
Return the requirement string to use to redownload the files
|
||||
currently at the given repository directory.
|
||||
|
||||
Args:
|
||||
project_name: the (unescaped) project name.
|
||||
|
||||
The return value has a form similar to the following:
|
||||
|
||||
{repository_url}@{revision}#egg={project_name}
|
||||
"""
|
||||
repo_url = cls.get_remote_url(repo_dir)
|
||||
if repo_url is None:
|
||||
return None
|
||||
|
||||
if cls.should_add_vcs_url_prefix(repo_url):
|
||||
repo_url = '{}+{}'.format(cls.name, repo_url)
|
||||
|
||||
revision = cls.get_requirement_revision(repo_dir)
|
||||
subdir = cls.get_subdirectory(repo_dir)
|
||||
req = make_vcs_requirement_url(repo_url, revision, project_name,
|
||||
subdir=subdir)
|
||||
|
||||
return req
|
||||
|
||||
@staticmethod
|
||||
def get_base_rev_args(rev):
|
||||
"""
|
||||
Return the base revision arguments for a vcs command.
|
||||
|
||||
Args:
|
||||
rev: the name of a revision to install. Cannot be None.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def make_rev_options(cls, rev=None, extra_args=None):
|
||||
# type: (Optional[str], Optional[List[str]]) -> RevOptions
|
||||
"""
|
||||
Return a RevOptions object.
|
||||
|
||||
Args:
|
||||
rev: the name of a revision to install.
|
||||
extra_args: a list of extra options.
|
||||
"""
|
||||
return RevOptions(cls, rev, extra_args=extra_args)
|
||||
|
||||
@classmethod
|
||||
def _is_local_repository(cls, repo):
|
||||
# type: (str) -> bool
|
||||
"""
|
||||
posix absolute paths start with os.path.sep,
|
||||
win32 ones start with drive (like c:\\folder)
|
||||
"""
|
||||
drive, tail = os.path.splitdrive(repo)
|
||||
return repo.startswith(os.path.sep) or bool(drive)
|
||||
|
||||
def export(self, location, url):
|
||||
"""
|
||||
Export the repository at the url to the destination location
|
||||
i.e. only download the files, without vcs informations
|
||||
|
||||
:param url: the repository URL starting with a vcs prefix.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def get_netloc_and_auth(cls, netloc, scheme):
|
||||
"""
|
||||
Parse the repository URL's netloc, and return the new netloc to use
|
||||
along with auth information.
|
||||
|
||||
Args:
|
||||
netloc: the original repository URL netloc.
|
||||
scheme: the repository URL's scheme without the vcs prefix.
|
||||
|
||||
This is mainly for the Subversion class to override, so that auth
|
||||
information can be provided via the --username and --password options
|
||||
instead of through the URL. For other subclasses like Git without
|
||||
such an option, auth information must stay in the URL.
|
||||
|
||||
Returns: (netloc, (username, password)).
|
||||
"""
|
||||
return netloc, (None, None)
|
||||
|
||||
@classmethod
|
||||
def get_url_rev_and_auth(cls, url):
|
||||
# type: (str) -> Tuple[str, Optional[str], AuthInfo]
|
||||
"""
|
||||
Parse the repository URL to use, and return the URL, revision,
|
||||
and auth info to use.
|
||||
|
||||
Returns: (url, rev, (username, password)).
|
||||
"""
|
||||
scheme, netloc, path, query, frag = urllib_parse.urlsplit(url)
|
||||
if '+' not in scheme:
|
||||
raise ValueError(
|
||||
"Sorry, {!r} is a malformed VCS url. "
|
||||
"The format is <vcs>+<protocol>://<url>, "
|
||||
"e.g. svn+http://myrepo/svn/MyApp#egg=MyApp".format(url)
|
||||
)
|
||||
# Remove the vcs prefix.
|
||||
scheme = scheme.split('+', 1)[1]
|
||||
netloc, user_pass = cls.get_netloc_and_auth(netloc, scheme)
|
||||
rev = None
|
||||
if '@' in path:
|
||||
path, rev = path.rsplit('@', 1)
|
||||
url = urllib_parse.urlunsplit((scheme, netloc, path, query, ''))
|
||||
return url, rev, user_pass
|
||||
|
||||
@staticmethod
|
||||
def make_rev_args(username, password):
|
||||
"""
|
||||
Return the RevOptions "extra arguments" to use in obtain().
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_url_rev_options(self, url):
|
||||
# type: (str) -> Tuple[str, RevOptions]
|
||||
"""
|
||||
Return the URL and RevOptions object to use in obtain() and in
|
||||
some cases export(), as a tuple (url, rev_options).
|
||||
"""
|
||||
url, rev, user_pass = self.get_url_rev_and_auth(url)
|
||||
username, password = user_pass
|
||||
extra_args = self.make_rev_args(username, password)
|
||||
rev_options = self.make_rev_options(rev, extra_args=extra_args)
|
||||
|
||||
return url, rev_options
|
||||
|
||||
@staticmethod
|
||||
def normalize_url(url):
|
||||
# type: (str) -> str
|
||||
"""
|
||||
Normalize a URL for comparison by unquoting it and removing any
|
||||
trailing slash.
|
||||
"""
|
||||
return urllib_parse.unquote(url).rstrip('/')
|
||||
|
||||
@classmethod
|
||||
def compare_urls(cls, url1, url2):
|
||||
# type: (str, str) -> bool
|
||||
"""
|
||||
Compare two repo URLs for identity, ignoring incidental differences.
|
||||
"""
|
||||
return (cls.normalize_url(url1) == cls.normalize_url(url2))
|
||||
|
||||
def fetch_new(self, dest, url, rev_options):
|
||||
"""
|
||||
Fetch a revision from a repository, in the case that this is the
|
||||
first fetch from the repository.
|
||||
|
||||
Args:
|
||||
dest: the directory to fetch the repository to.
|
||||
rev_options: a RevOptions object.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def switch(self, dest, url, rev_options):
|
||||
"""
|
||||
Switch the repo at ``dest`` to point to ``URL``.
|
||||
|
||||
Args:
|
||||
rev_options: a RevOptions object.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def update(self, dest, url, rev_options):
|
||||
"""
|
||||
Update an already-existing repo to the given ``rev_options``.
|
||||
|
||||
Args:
|
||||
rev_options: a RevOptions object.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def is_commit_id_equal(cls, dest, name):
|
||||
"""
|
||||
Return whether the id of the current commit equals the given name.
|
||||
|
||||
Args:
|
||||
dest: the repository directory.
|
||||
name: a string name.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def obtain(self, dest, url):
|
||||
# type: (str, str) -> None
|
||||
"""
|
||||
Install or update in editable mode the package represented by this
|
||||
VersionControl object.
|
||||
|
||||
:param dest: the repository directory in which to install or update.
|
||||
:param url: the repository URL starting with a vcs prefix.
|
||||
"""
|
||||
url, rev_options = self.get_url_rev_options(url)
|
||||
|
||||
if not os.path.exists(dest):
|
||||
self.fetch_new(dest, url, rev_options)
|
||||
return
|
||||
|
||||
rev_display = rev_options.to_display()
|
||||
if self.is_repository_directory(dest):
|
||||
existing_url = self.get_remote_url(dest)
|
||||
if self.compare_urls(existing_url, url):
|
||||
logger.debug(
|
||||
'%s in %s exists, and has correct URL (%s)',
|
||||
self.repo_name.title(),
|
||||
display_path(dest),
|
||||
url,
|
||||
)
|
||||
if not self.is_commit_id_equal(dest, rev_options.rev):
|
||||
logger.info(
|
||||
'Updating %s %s%s',
|
||||
display_path(dest),
|
||||
self.repo_name,
|
||||
rev_display,
|
||||
)
|
||||
self.update(dest, url, rev_options)
|
||||
else:
|
||||
logger.info('Skipping because already up-to-date.')
|
||||
return
|
||||
|
||||
logger.warning(
|
||||
'%s %s in %s exists with URL %s',
|
||||
self.name,
|
||||
self.repo_name,
|
||||
display_path(dest),
|
||||
existing_url,
|
||||
)
|
||||
prompt = ('(s)witch, (i)gnore, (w)ipe, (b)ackup ',
|
||||
('s', 'i', 'w', 'b'))
|
||||
else:
|
||||
logger.warning(
|
||||
'Directory %s already exists, and is not a %s %s.',
|
||||
dest,
|
||||
self.name,
|
||||
self.repo_name,
|
||||
)
|
||||
# https://github.com/python/mypy/issues/1174
|
||||
prompt = ('(i)gnore, (w)ipe, (b)ackup ', # type: ignore
|
||||
('i', 'w', 'b'))
|
||||
|
||||
logger.warning(
|
||||
'The plan is to install the %s repository %s',
|
||||
self.name,
|
||||
url,
|
||||
)
|
||||
response = ask_path_exists('What to do? %s' % prompt[0], prompt[1])
|
||||
|
||||
if response == 'a':
|
||||
sys.exit(-1)
|
||||
|
||||
if response == 'w':
|
||||
logger.warning('Deleting %s', display_path(dest))
|
||||
rmtree(dest)
|
||||
self.fetch_new(dest, url, rev_options)
|
||||
return
|
||||
|
||||
if response == 'b':
|
||||
dest_dir = backup_dir(dest)
|
||||
logger.warning(
|
||||
'Backing up %s to %s', display_path(dest), dest_dir,
|
||||
)
|
||||
shutil.move(dest, dest_dir)
|
||||
self.fetch_new(dest, url, rev_options)
|
||||
return
|
||||
|
||||
# Do nothing if the response is "i".
|
||||
if response == 's':
|
||||
logger.info(
|
||||
'Switching %s %s to %s%s',
|
||||
self.repo_name,
|
||||
display_path(dest),
|
||||
url,
|
||||
rev_display,
|
||||
)
|
||||
self.switch(dest, url, rev_options)
|
||||
|
||||
def unpack(self, location, url):
|
||||
# type: (str, str) -> None
|
||||
"""
|
||||
Clean up current location and download the url repository
|
||||
(and vcs infos) into location
|
||||
|
||||
:param url: the repository URL starting with a vcs prefix.
|
||||
"""
|
||||
if os.path.exists(location):
|
||||
rmtree(location)
|
||||
self.obtain(location, url=url)
|
||||
|
||||
@classmethod
|
||||
def get_remote_url(cls, location):
|
||||
"""
|
||||
Return the url used at location
|
||||
|
||||
Raises RemoteNotFoundError if the repository does not have a remote
|
||||
url configured.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def get_revision(cls, location):
|
||||
"""
|
||||
Return the current commit id of the files at the given location.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def run_command(
|
||||
cls,
|
||||
cmd, # type: List[str]
|
||||
show_stdout=True, # type: bool
|
||||
cwd=None, # type: Optional[str]
|
||||
on_returncode='raise', # type: str
|
||||
extra_ok_returncodes=None, # type: Optional[Iterable[int]]
|
||||
command_desc=None, # type: Optional[str]
|
||||
extra_environ=None, # type: Optional[Mapping[str, Any]]
|
||||
spinner=None # type: Optional[SpinnerInterface]
|
||||
):
|
||||
# type: (...) -> Optional[Text]
|
||||
"""
|
||||
Run a VCS subcommand
|
||||
This is simply a wrapper around call_subprocess that adds the VCS
|
||||
command name, and checks that the VCS is available
|
||||
"""
|
||||
cmd = [cls.name] + cmd
|
||||
try:
|
||||
return call_subprocess(cmd, show_stdout, cwd,
|
||||
on_returncode=on_returncode,
|
||||
extra_ok_returncodes=extra_ok_returncodes,
|
||||
command_desc=command_desc,
|
||||
extra_environ=extra_environ,
|
||||
unset_environ=cls.unset_environ,
|
||||
spinner=spinner)
|
||||
except OSError as e:
|
||||
# errno.ENOENT = no such file or directory
|
||||
# In other words, the VCS executable isn't available
|
||||
if e.errno == errno.ENOENT:
|
||||
raise BadCommand(
|
||||
'Cannot find command %r - do you have '
|
||||
'%r installed and in your '
|
||||
'PATH?' % (cls.name, cls.name))
|
||||
else:
|
||||
raise # re-raise exception if a different error occurred
|
||||
|
||||
@classmethod
|
||||
def is_repository_directory(cls, path):
|
||||
# type: (str) -> bool
|
||||
"""
|
||||
Return whether a directory path is a repository directory.
|
||||
"""
|
||||
logger.debug('Checking in %s for %s (%s)...',
|
||||
path, cls.dirname, cls.name)
|
||||
return os.path.exists(os.path.join(path, cls.dirname))
|
||||
|
||||
@classmethod
|
||||
def controls_location(cls, location):
|
||||
# type: (str) -> bool
|
||||
"""
|
||||
Check if a location is controlled by the vcs.
|
||||
It is meant to be overridden to implement smarter detection
|
||||
mechanisms for specific vcs.
|
||||
|
||||
This can do more than is_repository_directory() alone. For example,
|
||||
the Git override checks that Git is actually available.
|
||||
"""
|
||||
return cls.is_repository_directory(location)
|
||||
# Import all vcs modules to register each VCS in the VcsSupport object.
|
||||
import pip._internal.vcs.bazaar
|
||||
import pip._internal.vcs.git
|
||||
import pip._internal.vcs.mercurial
|
||||
import pip._internal.vcs.subversion # noqa: F401
|
||||
|
|
|
|||
|
|
@ -5,9 +5,8 @@ import os
|
|||
|
||||
from pip._vendor.six.moves.urllib import parse as urllib_parse
|
||||
|
||||
from pip._internal.download import path_to_url
|
||||
from pip._internal.utils.misc import display_path, rmtree
|
||||
from pip._internal.vcs import VersionControl, vcs
|
||||
from pip._internal.utils.misc import display_path, path_to_url, rmtree
|
||||
from pip._internal.vcs.versioncontrol import VersionControl, vcs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -12,7 +12,9 @@ from pip._internal.exceptions import BadCommand
|
|||
from pip._internal.utils.compat import samefile
|
||||
from pip._internal.utils.misc import display_path, redact_password_from_url
|
||||
from pip._internal.utils.temp_dir import TempDirectory
|
||||
from pip._internal.vcs import RemoteNotFoundError, VersionControl, vcs
|
||||
from pip._internal.vcs.versioncontrol import (
|
||||
RemoteNotFoundError, VersionControl, vcs,
|
||||
)
|
||||
|
||||
urlsplit = urllib_parse.urlsplit
|
||||
urlunsplit = urllib_parse.urlunsplit
|
||||
|
|
|
|||
|
|
@ -5,10 +5,9 @@ import os
|
|||
|
||||
from pip._vendor.six.moves import configparser
|
||||
|
||||
from pip._internal.download import path_to_url
|
||||
from pip._internal.utils.misc import display_path
|
||||
from pip._internal.utils.misc import display_path, path_to_url
|
||||
from pip._internal.utils.temp_dir import TempDirectory
|
||||
from pip._internal.vcs import VersionControl, vcs
|
||||
from pip._internal.vcs.versioncontrol import VersionControl, vcs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ from pip._internal.utils.misc import (
|
|||
display_path, rmtree, split_auth_from_netloc,
|
||||
)
|
||||
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
|
||||
from pip._internal.vcs import VersionControl, vcs
|
||||
from pip._internal.vcs.versioncontrol import VersionControl, vcs
|
||||
|
||||
_svn_xml_url_re = re.compile('url="([^"]+)"')
|
||||
_svn_rev_re = re.compile(r'committed-rev="(\d+)"')
|
||||
|
|
@ -20,7 +20,7 @@ _svn_info_xml_url_re = re.compile(r'<url>(.*)</url>')
|
|||
|
||||
if MYPY_CHECK_RUNNING:
|
||||
from typing import List, Optional, Tuple
|
||||
from pip._internal.vcs import RevOptions
|
||||
from pip._internal.vcs.versioncontrol import RevOptions
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,600 @@
|
|||
"""Handles all VCS (version control) support"""
|
||||
from __future__ import absolute_import
|
||||
|
||||
import errno
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
from pip._vendor import pkg_resources
|
||||
from pip._vendor.six.moves.urllib import parse as urllib_parse
|
||||
|
||||
from pip._internal.exceptions import BadCommand
|
||||
from pip._internal.utils.misc import (
|
||||
ask_path_exists, backup_dir, call_subprocess, display_path, rmtree,
|
||||
)
|
||||
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
|
||||
|
||||
if MYPY_CHECK_RUNNING:
|
||||
from typing import (
|
||||
Any, Dict, Iterable, List, Mapping, Optional, Text, Tuple, Type
|
||||
)
|
||||
from pip._internal.utils.ui import SpinnerInterface
|
||||
|
||||
AuthInfo = Tuple[Optional[str], Optional[str]]
|
||||
|
||||
__all__ = ['vcs']
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def make_vcs_requirement_url(repo_url, rev, project_name, subdir=None):
|
||||
"""
|
||||
Return the URL for a VCS requirement.
|
||||
|
||||
Args:
|
||||
repo_url: the remote VCS url, with any needed VCS prefix (e.g. "git+").
|
||||
project_name: the (unescaped) project name.
|
||||
"""
|
||||
egg_project_name = pkg_resources.to_filename(project_name)
|
||||
req = '{}@{}#egg={}'.format(repo_url, rev, egg_project_name)
|
||||
if subdir:
|
||||
req += '&subdirectory={}'.format(subdir)
|
||||
|
||||
return req
|
||||
|
||||
|
||||
class RemoteNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class RevOptions(object):
|
||||
|
||||
"""
|
||||
Encapsulates a VCS-specific revision to install, along with any VCS
|
||||
install options.
|
||||
|
||||
Instances of this class should be treated as if immutable.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
vc_class, # type: Type[VersionControl]
|
||||
rev=None, # type: Optional[str]
|
||||
extra_args=None, # type: Optional[List[str]]
|
||||
):
|
||||
# type: (...) -> None
|
||||
"""
|
||||
Args:
|
||||
vc_class: a VersionControl subclass.
|
||||
rev: the name of the revision to install.
|
||||
extra_args: a list of extra options.
|
||||
"""
|
||||
if extra_args is None:
|
||||
extra_args = []
|
||||
|
||||
self.extra_args = extra_args
|
||||
self.rev = rev
|
||||
self.vc_class = vc_class
|
||||
|
||||
def __repr__(self):
|
||||
return '<RevOptions {}: rev={!r}>'.format(self.vc_class.name, self.rev)
|
||||
|
||||
@property
|
||||
def arg_rev(self):
|
||||
# type: () -> Optional[str]
|
||||
if self.rev is None:
|
||||
return self.vc_class.default_arg_rev
|
||||
|
||||
return self.rev
|
||||
|
||||
def to_args(self):
|
||||
# type: () -> List[str]
|
||||
"""
|
||||
Return the VCS-specific command arguments.
|
||||
"""
|
||||
args = [] # type: List[str]
|
||||
rev = self.arg_rev
|
||||
if rev is not None:
|
||||
args += self.vc_class.get_base_rev_args(rev)
|
||||
args += self.extra_args
|
||||
|
||||
return args
|
||||
|
||||
def to_display(self):
|
||||
# type: () -> str
|
||||
if not self.rev:
|
||||
return ''
|
||||
|
||||
return ' (to revision {})'.format(self.rev)
|
||||
|
||||
def make_new(self, rev):
|
||||
# type: (str) -> RevOptions
|
||||
"""
|
||||
Make a copy of the current instance, but with a new rev.
|
||||
|
||||
Args:
|
||||
rev: the name of the revision for the new object.
|
||||
"""
|
||||
return self.vc_class.make_rev_options(rev, extra_args=self.extra_args)
|
||||
|
||||
|
||||
class VcsSupport(object):
|
||||
_registry = {} # type: Dict[str, VersionControl]
|
||||
schemes = ['ssh', 'git', 'hg', 'bzr', 'sftp', 'svn']
|
||||
|
||||
def __init__(self):
|
||||
# type: () -> None
|
||||
# Register more schemes with urlparse for various version control
|
||||
# systems
|
||||
urllib_parse.uses_netloc.extend(self.schemes)
|
||||
# Python >= 2.7.4, 3.3 doesn't have uses_fragment
|
||||
if getattr(urllib_parse, 'uses_fragment', None):
|
||||
urllib_parse.uses_fragment.extend(self.schemes)
|
||||
super(VcsSupport, self).__init__()
|
||||
|
||||
def __iter__(self):
|
||||
return self._registry.__iter__()
|
||||
|
||||
@property
|
||||
def backends(self):
|
||||
# type: () -> List[VersionControl]
|
||||
return list(self._registry.values())
|
||||
|
||||
@property
|
||||
def dirnames(self):
|
||||
# type: () -> List[str]
|
||||
return [backend.dirname for backend in self.backends]
|
||||
|
||||
@property
|
||||
def all_schemes(self):
|
||||
# type: () -> List[str]
|
||||
schemes = [] # type: List[str]
|
||||
for backend in self.backends:
|
||||
schemes.extend(backend.schemes)
|
||||
return schemes
|
||||
|
||||
def register(self, cls):
|
||||
# type: (Type[VersionControl]) -> None
|
||||
if not hasattr(cls, 'name'):
|
||||
logger.warning('Cannot register VCS %s', cls.__name__)
|
||||
return
|
||||
if cls.name not in self._registry:
|
||||
self._registry[cls.name] = cls()
|
||||
logger.debug('Registered VCS backend: %s', cls.name)
|
||||
|
||||
def unregister(self, name):
|
||||
# type: (str) -> None
|
||||
if name in self._registry:
|
||||
del self._registry[name]
|
||||
|
||||
def get_backend_for_dir(self, location):
|
||||
# type: (str) -> Optional[VersionControl]
|
||||
"""
|
||||
Return a VersionControl object if a repository of that type is found
|
||||
at the given directory.
|
||||
"""
|
||||
for vcs_backend in self._registry.values():
|
||||
if vcs_backend.controls_location(location):
|
||||
logger.debug('Determine that %s uses VCS: %s',
|
||||
location, vcs_backend.name)
|
||||
return vcs_backend
|
||||
return None
|
||||
|
||||
def get_backend(self, name):
|
||||
# type: (str) -> Optional[VersionControl]
|
||||
"""
|
||||
Return a VersionControl object or None.
|
||||
"""
|
||||
name = name.lower()
|
||||
return self._registry.get(name)
|
||||
|
||||
|
||||
vcs = VcsSupport()
|
||||
|
||||
|
||||
class VersionControl(object):
|
||||
name = ''
|
||||
dirname = ''
|
||||
repo_name = ''
|
||||
# List of supported schemes for this Version Control
|
||||
schemes = () # type: Tuple[str, ...]
|
||||
# Iterable of environment variable names to pass to call_subprocess().
|
||||
unset_environ = () # type: Tuple[str, ...]
|
||||
default_arg_rev = None # type: Optional[str]
|
||||
|
||||
@classmethod
|
||||
def should_add_vcs_url_prefix(cls, remote_url):
|
||||
"""
|
||||
Return whether the vcs prefix (e.g. "git+") should be added to a
|
||||
repository's remote url when used in a requirement.
|
||||
"""
|
||||
return not remote_url.lower().startswith('{}:'.format(cls.name))
|
||||
|
||||
@classmethod
|
||||
def get_subdirectory(cls, repo_dir):
|
||||
"""
|
||||
Return the path to setup.py, relative to the repo root.
|
||||
"""
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def get_requirement_revision(cls, repo_dir):
|
||||
"""
|
||||
Return the revision string that should be used in a requirement.
|
||||
"""
|
||||
return cls.get_revision(repo_dir)
|
||||
|
||||
@classmethod
|
||||
def get_src_requirement(cls, repo_dir, project_name):
|
||||
"""
|
||||
Return the requirement string to use to redownload the files
|
||||
currently at the given repository directory.
|
||||
|
||||
Args:
|
||||
project_name: the (unescaped) project name.
|
||||
|
||||
The return value has a form similar to the following:
|
||||
|
||||
{repository_url}@{revision}#egg={project_name}
|
||||
"""
|
||||
repo_url = cls.get_remote_url(repo_dir)
|
||||
if repo_url is None:
|
||||
return None
|
||||
|
||||
if cls.should_add_vcs_url_prefix(repo_url):
|
||||
repo_url = '{}+{}'.format(cls.name, repo_url)
|
||||
|
||||
revision = cls.get_requirement_revision(repo_dir)
|
||||
subdir = cls.get_subdirectory(repo_dir)
|
||||
req = make_vcs_requirement_url(repo_url, revision, project_name,
|
||||
subdir=subdir)
|
||||
|
||||
return req
|
||||
|
||||
@staticmethod
|
||||
def get_base_rev_args(rev):
|
||||
"""
|
||||
Return the base revision arguments for a vcs command.
|
||||
|
||||
Args:
|
||||
rev: the name of a revision to install. Cannot be None.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def make_rev_options(cls, rev=None, extra_args=None):
|
||||
# type: (Optional[str], Optional[List[str]]) -> RevOptions
|
||||
"""
|
||||
Return a RevOptions object.
|
||||
|
||||
Args:
|
||||
rev: the name of a revision to install.
|
||||
extra_args: a list of extra options.
|
||||
"""
|
||||
return RevOptions(cls, rev, extra_args=extra_args)
|
||||
|
||||
@classmethod
|
||||
def _is_local_repository(cls, repo):
|
||||
# type: (str) -> bool
|
||||
"""
|
||||
posix absolute paths start with os.path.sep,
|
||||
win32 ones start with drive (like c:\\folder)
|
||||
"""
|
||||
drive, tail = os.path.splitdrive(repo)
|
||||
return repo.startswith(os.path.sep) or bool(drive)
|
||||
|
||||
def export(self, location, url):
|
||||
"""
|
||||
Export the repository at the url to the destination location
|
||||
i.e. only download the files, without vcs informations
|
||||
|
||||
:param url: the repository URL starting with a vcs prefix.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def get_netloc_and_auth(cls, netloc, scheme):
|
||||
"""
|
||||
Parse the repository URL's netloc, and return the new netloc to use
|
||||
along with auth information.
|
||||
|
||||
Args:
|
||||
netloc: the original repository URL netloc.
|
||||
scheme: the repository URL's scheme without the vcs prefix.
|
||||
|
||||
This is mainly for the Subversion class to override, so that auth
|
||||
information can be provided via the --username and --password options
|
||||
instead of through the URL. For other subclasses like Git without
|
||||
such an option, auth information must stay in the URL.
|
||||
|
||||
Returns: (netloc, (username, password)).
|
||||
"""
|
||||
return netloc, (None, None)
|
||||
|
||||
@classmethod
|
||||
def get_url_rev_and_auth(cls, url):
|
||||
# type: (str) -> Tuple[str, Optional[str], AuthInfo]
|
||||
"""
|
||||
Parse the repository URL to use, and return the URL, revision,
|
||||
and auth info to use.
|
||||
|
||||
Returns: (url, rev, (username, password)).
|
||||
"""
|
||||
scheme, netloc, path, query, frag = urllib_parse.urlsplit(url)
|
||||
if '+' not in scheme:
|
||||
raise ValueError(
|
||||
"Sorry, {!r} is a malformed VCS url. "
|
||||
"The format is <vcs>+<protocol>://<url>, "
|
||||
"e.g. svn+http://myrepo/svn/MyApp#egg=MyApp".format(url)
|
||||
)
|
||||
# Remove the vcs prefix.
|
||||
scheme = scheme.split('+', 1)[1]
|
||||
netloc, user_pass = cls.get_netloc_and_auth(netloc, scheme)
|
||||
rev = None
|
||||
if '@' in path:
|
||||
path, rev = path.rsplit('@', 1)
|
||||
url = urllib_parse.urlunsplit((scheme, netloc, path, query, ''))
|
||||
return url, rev, user_pass
|
||||
|
||||
@staticmethod
|
||||
def make_rev_args(username, password):
|
||||
"""
|
||||
Return the RevOptions "extra arguments" to use in obtain().
|
||||
"""
|
||||
return []
|
||||
|
||||
def get_url_rev_options(self, url):
|
||||
# type: (str) -> Tuple[str, RevOptions]
|
||||
"""
|
||||
Return the URL and RevOptions object to use in obtain() and in
|
||||
some cases export(), as a tuple (url, rev_options).
|
||||
"""
|
||||
url, rev, user_pass = self.get_url_rev_and_auth(url)
|
||||
username, password = user_pass
|
||||
extra_args = self.make_rev_args(username, password)
|
||||
rev_options = self.make_rev_options(rev, extra_args=extra_args)
|
||||
|
||||
return url, rev_options
|
||||
|
||||
@staticmethod
|
||||
def normalize_url(url):
|
||||
# type: (str) -> str
|
||||
"""
|
||||
Normalize a URL for comparison by unquoting it and removing any
|
||||
trailing slash.
|
||||
"""
|
||||
return urllib_parse.unquote(url).rstrip('/')
|
||||
|
||||
@classmethod
|
||||
def compare_urls(cls, url1, url2):
|
||||
# type: (str, str) -> bool
|
||||
"""
|
||||
Compare two repo URLs for identity, ignoring incidental differences.
|
||||
"""
|
||||
return (cls.normalize_url(url1) == cls.normalize_url(url2))
|
||||
|
||||
def fetch_new(self, dest, url, rev_options):
|
||||
"""
|
||||
Fetch a revision from a repository, in the case that this is the
|
||||
first fetch from the repository.
|
||||
|
||||
Args:
|
||||
dest: the directory to fetch the repository to.
|
||||
rev_options: a RevOptions object.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def switch(self, dest, url, rev_options):
|
||||
"""
|
||||
Switch the repo at ``dest`` to point to ``URL``.
|
||||
|
||||
Args:
|
||||
rev_options: a RevOptions object.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def update(self, dest, url, rev_options):
|
||||
"""
|
||||
Update an already-existing repo to the given ``rev_options``.
|
||||
|
||||
Args:
|
||||
rev_options: a RevOptions object.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def is_commit_id_equal(cls, dest, name):
|
||||
"""
|
||||
Return whether the id of the current commit equals the given name.
|
||||
|
||||
Args:
|
||||
dest: the repository directory.
|
||||
name: a string name.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def obtain(self, dest, url):
|
||||
# type: (str, str) -> None
|
||||
"""
|
||||
Install or update in editable mode the package represented by this
|
||||
VersionControl object.
|
||||
|
||||
:param dest: the repository directory in which to install or update.
|
||||
:param url: the repository URL starting with a vcs prefix.
|
||||
"""
|
||||
url, rev_options = self.get_url_rev_options(url)
|
||||
|
||||
if not os.path.exists(dest):
|
||||
self.fetch_new(dest, url, rev_options)
|
||||
return
|
||||
|
||||
rev_display = rev_options.to_display()
|
||||
if self.is_repository_directory(dest):
|
||||
existing_url = self.get_remote_url(dest)
|
||||
if self.compare_urls(existing_url, url):
|
||||
logger.debug(
|
||||
'%s in %s exists, and has correct URL (%s)',
|
||||
self.repo_name.title(),
|
||||
display_path(dest),
|
||||
url,
|
||||
)
|
||||
if not self.is_commit_id_equal(dest, rev_options.rev):
|
||||
logger.info(
|
||||
'Updating %s %s%s',
|
||||
display_path(dest),
|
||||
self.repo_name,
|
||||
rev_display,
|
||||
)
|
||||
self.update(dest, url, rev_options)
|
||||
else:
|
||||
logger.info('Skipping because already up-to-date.')
|
||||
return
|
||||
|
||||
logger.warning(
|
||||
'%s %s in %s exists with URL %s',
|
||||
self.name,
|
||||
self.repo_name,
|
||||
display_path(dest),
|
||||
existing_url,
|
||||
)
|
||||
prompt = ('(s)witch, (i)gnore, (w)ipe, (b)ackup ',
|
||||
('s', 'i', 'w', 'b'))
|
||||
else:
|
||||
logger.warning(
|
||||
'Directory %s already exists, and is not a %s %s.',
|
||||
dest,
|
||||
self.name,
|
||||
self.repo_name,
|
||||
)
|
||||
# https://github.com/python/mypy/issues/1174
|
||||
prompt = ('(i)gnore, (w)ipe, (b)ackup ', # type: ignore
|
||||
('i', 'w', 'b'))
|
||||
|
||||
logger.warning(
|
||||
'The plan is to install the %s repository %s',
|
||||
self.name,
|
||||
url,
|
||||
)
|
||||
response = ask_path_exists('What to do? %s' % prompt[0], prompt[1])
|
||||
|
||||
if response == 'a':
|
||||
sys.exit(-1)
|
||||
|
||||
if response == 'w':
|
||||
logger.warning('Deleting %s', display_path(dest))
|
||||
rmtree(dest)
|
||||
self.fetch_new(dest, url, rev_options)
|
||||
return
|
||||
|
||||
if response == 'b':
|
||||
dest_dir = backup_dir(dest)
|
||||
logger.warning(
|
||||
'Backing up %s to %s', display_path(dest), dest_dir,
|
||||
)
|
||||
shutil.move(dest, dest_dir)
|
||||
self.fetch_new(dest, url, rev_options)
|
||||
return
|
||||
|
||||
# Do nothing if the response is "i".
|
||||
if response == 's':
|
||||
logger.info(
|
||||
'Switching %s %s to %s%s',
|
||||
self.repo_name,
|
||||
display_path(dest),
|
||||
url,
|
||||
rev_display,
|
||||
)
|
||||
self.switch(dest, url, rev_options)
|
||||
|
||||
def unpack(self, location, url):
|
||||
# type: (str, str) -> None
|
||||
"""
|
||||
Clean up current location and download the url repository
|
||||
(and vcs infos) into location
|
||||
|
||||
:param url: the repository URL starting with a vcs prefix.
|
||||
"""
|
||||
if os.path.exists(location):
|
||||
rmtree(location)
|
||||
self.obtain(location, url=url)
|
||||
|
||||
@classmethod
|
||||
def get_remote_url(cls, location):
|
||||
"""
|
||||
Return the url used at location
|
||||
|
||||
Raises RemoteNotFoundError if the repository does not have a remote
|
||||
url configured.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def get_revision(cls, location):
|
||||
"""
|
||||
Return the current commit id of the files at the given location.
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def run_command(
|
||||
cls,
|
||||
cmd, # type: List[str]
|
||||
show_stdout=True, # type: bool
|
||||
cwd=None, # type: Optional[str]
|
||||
on_returncode='raise', # type: str
|
||||
extra_ok_returncodes=None, # type: Optional[Iterable[int]]
|
||||
command_desc=None, # type: Optional[str]
|
||||
extra_environ=None, # type: Optional[Mapping[str, Any]]
|
||||
spinner=None # type: Optional[SpinnerInterface]
|
||||
):
|
||||
# type: (...) -> Optional[Text]
|
||||
"""
|
||||
Run a VCS subcommand
|
||||
This is simply a wrapper around call_subprocess that adds the VCS
|
||||
command name, and checks that the VCS is available
|
||||
"""
|
||||
cmd = [cls.name] + cmd
|
||||
try:
|
||||
return call_subprocess(cmd, show_stdout, cwd,
|
||||
on_returncode=on_returncode,
|
||||
extra_ok_returncodes=extra_ok_returncodes,
|
||||
command_desc=command_desc,
|
||||
extra_environ=extra_environ,
|
||||
unset_environ=cls.unset_environ,
|
||||
spinner=spinner)
|
||||
except OSError as e:
|
||||
# errno.ENOENT = no such file or directory
|
||||
# In other words, the VCS executable isn't available
|
||||
if e.errno == errno.ENOENT:
|
||||
raise BadCommand(
|
||||
'Cannot find command %r - do you have '
|
||||
'%r installed and in your '
|
||||
'PATH?' % (cls.name, cls.name))
|
||||
else:
|
||||
raise # re-raise exception if a different error occurred
|
||||
|
||||
@classmethod
|
||||
def is_repository_directory(cls, path):
|
||||
# type: (str) -> bool
|
||||
"""
|
||||
Return whether a directory path is a repository directory.
|
||||
"""
|
||||
logger.debug('Checking in %s for %s (%s)...',
|
||||
path, cls.dirname, cls.name)
|
||||
return os.path.exists(os.path.join(path, cls.dirname))
|
||||
|
||||
@classmethod
|
||||
def controls_location(cls, location):
|
||||
# type: (str) -> bool
|
||||
"""
|
||||
Check if a location is controlled by the vcs.
|
||||
It is meant to be overridden to implement smarter detection
|
||||
mechanisms for specific vcs.
|
||||
|
||||
This can do more than is_repository_directory() alone. For example,
|
||||
the Git override checks that Git is actually available.
|
||||
"""
|
||||
return cls.is_repository_directory(location)
|
||||
|
|
@ -2,13 +2,13 @@ from unittest import TestCase
|
|||
|
||||
from mock import patch
|
||||
|
||||
from pip._internal.vcs import RevOptions
|
||||
from pip._internal.vcs.subversion import Subversion
|
||||
from pip._internal.vcs.versioncontrol import RevOptions
|
||||
|
||||
|
||||
class TestSubversionArgs(TestCase):
|
||||
def setUp(self):
|
||||
patcher = patch('pip._internal.vcs.call_subprocess')
|
||||
patcher = patch('pip._internal.vcs.versioncontrol.call_subprocess')
|
||||
self.addCleanup(patcher.stop)
|
||||
self.call_subprocess_mock = patcher.start()
|
||||
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ from tempfile import mkdtemp
|
|||
|
||||
import pytest
|
||||
from mock import Mock, patch
|
||||
from pip._vendor.six.moves.urllib import request as urllib_request
|
||||
|
||||
import pip
|
||||
from pip._internal.download import (
|
||||
|
|
@ -199,22 +198,6 @@ def test_unpack_http_url_bad_downloaded_checksum(mock_unpack_file):
|
|||
rmtree(download_dir)
|
||||
|
||||
|
||||
@pytest.mark.skipif("sys.platform == 'win32'")
|
||||
def test_path_to_url_unix():
|
||||
assert path_to_url('/tmp/file') == 'file:///tmp/file'
|
||||
path = os.path.join(os.getcwd(), 'file')
|
||||
assert path_to_url('file') == 'file://' + urllib_request.pathname2url(path)
|
||||
|
||||
|
||||
@pytest.mark.skipif("sys.platform != 'win32'")
|
||||
def test_path_to_url_win():
|
||||
assert path_to_url('c:/tmp/file') == 'file:///C:/tmp/file'
|
||||
assert path_to_url('c:\\tmp\\file') == 'file:///C:/tmp/file'
|
||||
assert path_to_url(r'\\unc\as\path') == 'file://unc/as/path'
|
||||
path = os.path.join(os.getcwd(), 'file')
|
||||
assert path_to_url('file') == 'file:' + urllib_request.pathname2url(path)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("url,win_expected,non_win_expected", [
|
||||
('file:tmp', 'tmp', 'tmp'),
|
||||
('file:c:/path/to/file', r'C:\path\to\file', 'c:/path/to/file'),
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from logging import DEBUG, ERROR, INFO, WARNING
|
|||
|
||||
import pytest
|
||||
from mock import Mock, patch
|
||||
from pip._vendor.six.moves.urllib import request as urllib_request
|
||||
|
||||
from pip._internal.exceptions import (
|
||||
HashMismatch, HashMissing, InstallationError,
|
||||
|
|
@ -28,8 +29,8 @@ from pip._internal.utils.glibc import check_glibc_version
|
|||
from pip._internal.utils.hashes import Hashes, MissingHashes
|
||||
from pip._internal.utils.misc import (
|
||||
call_subprocess, egg_link_path, ensure_dir, format_command_args,
|
||||
get_installed_distributions, get_prog, normalize_path, redact_netloc,
|
||||
redact_password_from_url, remove_auth_from_url, rmtree,
|
||||
get_installed_distributions, get_prog, normalize_path, path_to_url,
|
||||
redact_netloc, redact_password_from_url, remove_auth_from_url, rmtree,
|
||||
split_auth_from_netloc, split_auth_netloc_from_url, untar_file, unzip_file,
|
||||
)
|
||||
from pip._internal.utils.temp_dir import AdjacentTempDirectory, TempDirectory
|
||||
|
|
@ -965,6 +966,22 @@ class TestCallSubprocess(object):
|
|||
)
|
||||
|
||||
|
||||
@pytest.mark.skipif("sys.platform == 'win32'")
|
||||
def test_path_to_url_unix():
|
||||
assert path_to_url('/tmp/file') == 'file:///tmp/file'
|
||||
path = os.path.join(os.getcwd(), 'file')
|
||||
assert path_to_url('file') == 'file://' + urllib_request.pathname2url(path)
|
||||
|
||||
|
||||
@pytest.mark.skipif("sys.platform != 'win32'")
|
||||
def test_path_to_url_win():
|
||||
assert path_to_url('c:/tmp/file') == 'file:///C:/tmp/file'
|
||||
assert path_to_url('c:\\tmp\\file') == 'file:///C:/tmp/file'
|
||||
assert path_to_url(r'\\unc\as\path') == 'file://unc/as/path'
|
||||
path = os.path.join(os.getcwd(), 'file')
|
||||
assert path_to_url('file') == 'file:' + urllib_request.pathname2url(path)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('netloc, expected', [
|
||||
# Test a basic case.
|
||||
('example.com', ('example.com', (None, None))),
|
||||
|
|
|
|||
|
|
@ -5,13 +5,12 @@ from mock import patch
|
|||
from pip._vendor.packaging.version import parse as parse_version
|
||||
|
||||
from pip._internal.exceptions import BadCommand
|
||||
from pip._internal.vcs import (
|
||||
RevOptions, VersionControl, make_vcs_requirement_url,
|
||||
)
|
||||
from pip._internal.vcs import make_vcs_requirement_url
|
||||
from pip._internal.vcs.bazaar import Bazaar
|
||||
from pip._internal.vcs.git import Git, looks_like_hash
|
||||
from pip._internal.vcs.mercurial import Mercurial
|
||||
from pip._internal.vcs.subversion import Subversion
|
||||
from pip._internal.vcs.versioncontrol import RevOptions, VersionControl
|
||||
from tests.lib import is_svn_installed, pyversion
|
||||
|
||||
if pyversion >= '3':
|
||||
|
|
|
|||
Loading…
Reference in New Issue