Fix get_branch() to work if the current branch shares a name with a tag

src/pip/_internal/utils/misc.py

@@ -653,6 +653,7 @@ def call_subprocess(
     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]]
     unset_environ=None,  # type: Optional[Iterable[str]]
@@ -661,9 +662,13 @@ def call_subprocess(
     # type: (...) -> Optional[Text]
     """
     Args:
+      extra_ok_returncodes: an iterable of integer return codes that are
+        acceptable, in addition to 0. Defaults to None, which means [].
       unset_environ: an iterable of environment variable names to unset
         prior to calling subprocess.Popen().
     """
+    if extra_ok_returncodes is None:
+        extra_ok_returncodes = []
     if unset_environ is None:
         unset_environ = []
     # This function's handling of subprocess output is confusing and I
@@ -740,7 +745,7 @@ def call_subprocess(
             spinner.finish("error")
         else:
             spinner.finish("done")
-    if proc.returncode:
+    if proc.returncode and proc.returncode not in extra_ok_returncodes:
         if on_returncode == 'raise':
             if (logger.getEffectiveLevel() > std_logging.DEBUG and
                     not show_stdout):

src/pip/_internal/vcs/__init__.py

@@ -17,7 +17,7 @@
 
 if MYPY_CHECK_RUNNING:
     from typing import (  # noqa: F401
-        Dict, Optional, Tuple, List, Type, Any, Mapping, Text
+        Any, Dict, Iterable, List, Mapping, Optional, Text, Tuple, Type
     )
     from pip._internal.utils.ui import SpinnerInterface  # noqa: F401
 
@@ -467,6 +467,7 @@ def run_command(
         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]
@@ -480,8 +481,10 @@ def run_command(
         cmd = [self.name] + cmd
         try:
             return call_subprocess(cmd, show_stdout, cwd,
-                                   on_returncode,
-                                   command_desc, extra_environ,
+                                   on_returncode=on_returncode,
+                                   extra_ok_returncodes=extra_ok_returncodes,
+                                   command_desc=command_desc,
+                                   extra_environ=extra_environ,
                                    unset_environ=self.unset_environ,
                                    spinner=spinner)
         except OSError as e:

src/pip/_internal/vcs/git.py

@@ -79,19 +79,25 @@ def get_git_version(self):
         version = '.'.join(version.split('.')[:3])
         return parse_version(version)
 
-    def get_branch(self, location):
+    def get_current_branch(self, location):
         """
         Return the current branch, or None if HEAD isn't at a branch
         (e.g. detached HEAD).
         """
-        args = ['rev-parse', '--abbrev-ref', 'HEAD']
-        output = self.run_command(args, show_stdout=False, cwd=location)
-        branch = output.strip()
+        # git-symbolic-ref exits with empty stdout if "HEAD" is a detached
+        # HEAD rather than a symbolic ref.  In addition, the -q causes the
+        # command to exit with status code 1 instead of 128 in this case
+        # and to suppress the message to stderr.
+        args = ['symbolic-ref', '-q', 'HEAD']
+        output = self.run_command(
+            args, extra_ok_returncodes=(1, ), show_stdout=False, cwd=location,
+        )
+        ref = output.strip()
 
-        if branch == 'HEAD':
-            return None
+        if ref.startswith('refs/heads/'):
+            return ref[len('refs/heads/'):]
 
-        return branch
+        return None
 
     def export(self, location):
         """Export the Git repository at the url to the destination location"""
@@ -210,7 +216,7 @@ def fetch_new(self, dest, url, rev_options):
                 if not self.is_commit_id_equal(dest, rev_options.rev):
                     cmd_args = ['checkout', '-q'] + rev_options.to_args()
                     self.run_command(cmd_args, cwd=dest)
-            elif self.get_branch(dest) != branch_name:
+            elif self.get_current_branch(dest) != branch_name:
                 # Then a specific branch was requested, and that branch
                 # is not yet checked out.
                 track_branch = 'origin/{}'.format(branch_name)

tests/functional/test_vcs_git.py

@@ -16,6 +16,16 @@ def get_head_sha(script, dest):
     return sha
 
 
+def checkout_ref(script, repo_dir, ref):
+    script.run('git', 'checkout', ref, cwd=repo_dir, expect_stderr=True)
+
+
+def checkout_new_branch(script, repo_dir, branch):
+    script.run(
+        'git', 'checkout', '-b', branch, cwd=repo_dir, expect_stderr=True,
+    )
+
+
 def do_commit(script, dest):
     _git_commit(script, dest, message='test commit', args=['--allow-empty'])
     return get_head_sha(script, dest)
@@ -83,26 +93,43 @@ def test_get_remote_url(script, tmpdir):
     assert remote_url == source_url
 
 
-def test_get_branch(script):
+def test_get_current_branch(script):
     repo_dir = str(script.scratch_path)
 
     script.run('git', 'init', cwd=repo_dir)
     sha = do_commit(script, repo_dir)
 
     git = Git()
-    assert git.get_branch(repo_dir) == 'master'
+    assert git.get_current_branch(repo_dir) == 'master'
 
     # Switch to a branch with the same SHA as "master" but whose name
     # is alphabetically after.
-    script.run(
-        'git', 'checkout', '-b', 'release', cwd=repo_dir,
-        expect_stderr=True,
-    )
-    assert git.get_branch(repo_dir) == 'release'
+    checkout_new_branch(script, repo_dir, 'release')
+    assert git.get_current_branch(repo_dir) == 'release'
 
     # Also test the detached HEAD case.
-    script.run('git', 'checkout', sha, cwd=repo_dir, expect_stderr=True)
-    assert git.get_branch(repo_dir) is None
+    checkout_ref(script, repo_dir, sha)
+    assert git.get_current_branch(repo_dir) is None
+
+
+def test_get_current_branch__branch_and_tag_same_name(script, tmpdir):
+    """
+    Check calling get_current_branch() from a branch or tag when the branch
+    and tag have the same name.
+    """
+    repo_dir = str(tmpdir)
+    script.run('git', 'init', cwd=repo_dir)
+    do_commit(script, repo_dir)
+    checkout_new_branch(script, repo_dir, 'dev')
+    # Create a tag with the same name as the branch.
+    script.run('git', 'tag', 'dev', cwd=repo_dir)
+
+    git = Git()
+    assert git.get_current_branch(repo_dir) == 'dev'
+
+    # Now try with the tag checked out.
+    checkout_ref(script, repo_dir, 'refs/tags/dev')
+    assert git.get_current_branch(repo_dir) is None
 
 
 def test_get_revision_sha(script):