Merge pull request #6887 from cjerdonek/add-vcs-url-type-annotations

Add type annotations to VCS url-passing methods

src/pip/_internal/vcs/bazaar.py

@@ -6,8 +6,14 @@
 from pip._vendor.six.moves.urllib import parse as urllib_parse
 
 from pip._internal.utils.misc import display_path, path_to_url, rmtree
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 from pip._internal.vcs.versioncontrol import VersionControl, vcs
 
+if MYPY_CHECK_RUNNING:
+    from typing import Optional, Tuple
+    from pip._internal.vcs.versioncontrol import AuthInfo, RevOptions
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -32,6 +38,7 @@ def get_base_rev_args(rev):
         return ['-r', rev]
 
     def export(self, location, url):
+        # type: (str, str) -> None
         """
         Export the Bazaar repository at the url to the destination location
         """
@@ -46,6 +53,7 @@ def export(self, location, url):
         )
 
     def fetch_new(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         rev_display = rev_options.to_display()
         logger.info(
             'Checking out %s%s to %s',
@@ -57,14 +65,17 @@ def fetch_new(self, dest, url, rev_options):
         self.run_command(cmd_args)
 
     def switch(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         self.run_command(['switch', url], cwd=dest)
 
     def update(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         cmd_args = ['pull', '-q'] + rev_options.to_args()
         self.run_command(cmd_args, cwd=dest)
 
     @classmethod
     def get_url_rev_and_auth(cls, url):
+        # type: (str) -> Tuple[str, Optional[str], AuthInfo]
         # hotfix the URL scheme after removing bzr+ from bzr+ssh:// readd it
         url, rev, user_pass = super(Bazaar, cls).get_url_rev_and_auth(url)
         if url.startswith('ssh://'):

src/pip/_internal/vcs/git.py

@@ -12,12 +12,18 @@
 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.utils.typing import MYPY_CHECK_RUNNING
 from pip._internal.vcs.versioncontrol import (
     RemoteNotFoundError,
     VersionControl,
     vcs,
 )
 
+if MYPY_CHECK_RUNNING:
+    from typing import Optional, Tuple
+    from pip._internal.vcs.versioncontrol import AuthInfo, RevOptions
+
+
 urlsplit = urllib_parse.urlsplit
 urlunsplit = urllib_parse.urlunsplit
 
@@ -83,6 +89,7 @@ def get_current_branch(cls, location):
         return None
 
     def export(self, location, url):
+        # type: (str, str) -> None
         """Export the Git repository at the url to the destination location"""
         if not location.endswith('/'):
             location = location + '/'
@@ -131,6 +138,7 @@ def get_revision_sha(cls, dest, rev):
 
     @classmethod
     def resolve_revision(cls, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> RevOptions
         """
         Resolve a revision to a new RevOptions object with the SHA1 of the
         branch, tag, or ref if found.
@@ -139,6 +147,10 @@ def resolve_revision(cls, dest, url, rev_options):
           rev_options: a RevOptions object.
         """
         rev = rev_options.arg_rev
+        # The arg_rev property's implementation for Git ensures that the
+        # rev return value is always non-None.
+        assert rev is not None
+
         sha, is_branch = cls.get_revision_sha(dest, rev)
 
         if sha is not None:
@@ -185,6 +197,7 @@ def is_commit_id_equal(cls, dest, name):
         return cls.get_revision(dest) == name
 
     def fetch_new(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         rev_display = rev_options.to_display()
         logger.info(
             'Cloning %s%s to %s', redact_password_from_url(url),
@@ -215,13 +228,15 @@ def fetch_new(self, dest, url, rev_options):
         self.update_submodules(dest)
 
     def switch(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         self.run_command(['config', 'remote.origin.url', url], cwd=dest)
         cmd_args = ['checkout', '-q'] + rev_options.to_args()
         self.run_command(cmd_args, cwd=dest)
 
         self.update_submodules(dest)
 
     def update(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         # First fetch changes from the default remote
         if self.get_git_version() >= parse_version('1.9.0'):
             # fetch tags in addition to everything else
@@ -300,6 +315,7 @@ def get_subdirectory(cls, location):
 
     @classmethod
     def get_url_rev_and_auth(cls, url):
+        # type: (str) -> Tuple[str, Optional[str], AuthInfo]
         """
         Prefixes stub URLs like 'user@hostname:user/repo.git' with 'ssh://'.
         That's required because although they use SSH they sometimes don't

src/pip/_internal/vcs/mercurial.py

@@ -7,8 +7,13 @@
 
 from pip._internal.utils.misc import display_path, path_to_url
 from pip._internal.utils.temp_dir import TempDirectory
+from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 from pip._internal.vcs.versioncontrol import VersionControl, vcs
 
+if MYPY_CHECK_RUNNING:
+    from pip._internal.vcs.versioncontrol import RevOptions
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -23,6 +28,7 @@ def get_base_rev_args(rev):
         return [rev]
 
     def export(self, location, url):
+        # type: (str, str) -> None
         """Export the Hg repository at the url to the destination location"""
         with TempDirectory(kind="export") as temp_dir:
             self.unpack(temp_dir.path, url=url)
@@ -32,6 +38,7 @@ def export(self, location, url):
             )
 
     def fetch_new(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         rev_display = rev_options.to_display()
         logger.info(
             'Cloning hg %s%s to %s',
@@ -44,6 +51,7 @@ def fetch_new(self, dest, url, rev_options):
         self.run_command(cmd_args, cwd=dest)
 
     def switch(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         repo_config = os.path.join(dest, self.dirname, 'hgrc')
         config = configparser.RawConfigParser()
         try:
@@ -60,6 +68,7 @@ def switch(self, dest, url, rev_options):
             self.run_command(cmd_args, cwd=dest)
 
     def update(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         self.run_command(['pull', '-q'], cwd=dest)
         cmd_args = ['update', '-q'] + rev_options.to_args()
         self.run_command(cmd_args, cwd=dest)

src/pip/_internal/vcs/subversion.py

@@ -22,7 +22,8 @@
 
 if MYPY_CHECK_RUNNING:
     from typing import List, Optional, Tuple
-    from pip._internal.vcs.versioncontrol import RevOptions
+    from pip._internal.vcs.versioncontrol import AuthInfo, RevOptions
+
 
 logger = logging.getLogger(__name__)
 
@@ -84,6 +85,7 @@ def get_netloc_and_auth(cls, netloc, scheme):
 
     @classmethod
     def get_url_rev_and_auth(cls, url):
+        # type: (str) -> Tuple[str, Optional[str], AuthInfo]
         # hotfix the URL scheme after removing svn+ from svn+ssh:// readd it
         url, rev, user_pass = super(Subversion, cls).get_url_rev_and_auth(url)
         if url.startswith('ssh://'):
@@ -92,7 +94,8 @@ def get_url_rev_and_auth(cls, url):
 
     @staticmethod
     def make_rev_args(username, password):
-        extra_args = []
+        # type: (Optional[str], Optional[str]) -> List[str]
+        extra_args = []  # type: List[str]
         if username:
             extra_args += ['--username', username]
         if password:
@@ -273,6 +276,7 @@ def get_remote_call_options(self):
         return []
 
     def export(self, location, url):
+        # type: (str, str) -> None
         """Export the svn repository at the url to the destination location"""
         url, rev_options = self.get_url_rev_options(url)
 

src/pip/_internal/vcs/versioncontrol.py

@@ -82,6 +82,7 @@ def __init__(
         self.extra_args = extra_args
         self.rev = rev
         self.vc_class = vc_class
+        self.branch_name = None  # type: Optional[str]
 
     def __repr__(self):
         return '<RevOptions {}: rev={!r}>'.format(self.vc_class.name, self.rev)
@@ -291,6 +292,7 @@ def _is_local_repository(cls, repo):
         return repo.startswith(os.path.sep) or bool(drive)
 
     def export(self, location, url):
+        # type: (str, str) -> None
         """
         Export the repository at the url to the destination location
         i.e. only download the files, without vcs informations
@@ -345,6 +347,7 @@ def get_url_rev_and_auth(cls, url):
 
     @staticmethod
     def make_rev_args(username, password):
+        # type: (Optional[str], Optional[str]) -> List[str]
         """
         Return the RevOptions "extra arguments" to use in obtain().
         """
@@ -381,6 +384,7 @@ def compare_urls(cls, url1, url2):
         return (cls.normalize_url(url1) == cls.normalize_url(url2))
 
     def fetch_new(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         """
         Fetch a revision from a repository, in the case that this is the
         first fetch from the repository.
@@ -392,6 +396,7 @@ def fetch_new(self, dest, url, rev_options):
         raise NotImplementedError
 
     def switch(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         """
         Switch the repo at ``dest`` to point to ``URL``.
 
@@ -401,6 +406,7 @@ def switch(self, dest, url, rev_options):
         raise NotImplementedError
 
     def update(self, dest, url, rev_options):
+        # type: (str, str, RevOptions) -> None
         """
         Update an already-existing repo to the given ``rev_options``.