git-cl: Use scm.GIT.FetchUpstreamTuple.

scm:
- Add methods to deal with git configs and tests for them.
  Will make it easier to mock git config calls as opposed
  to all git calls.
- Add tests to FetchUpstreamTuple

git-cl:
- Mock out settings.GetChangeRoot()
- Use scm.GIT.FetchUpstreamTuple() to reduce code duplication,
  and mock it out on tests.

Bug: 1051631
Change-Id: I1a3220b4c0f3e6221b3c00550259a433c2775798
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/tools/depot_tools/+/2052552
Commit-Queue: Edward Lesmes <ehmaldonado@chromium.org>
Reviewed-by: Anthony Polito <apolito@google.com>

git_cl.py

@@ -1194,30 +1194,15 @@ class Changelist(object):
     """Returns a tuple containing remote and remote ref,
        e.g. 'origin', 'refs/heads/master'
     """
-    remote = '.'
-    upstream_branch = _git_get_branch_config_value('merge', branch=branch)
-
-    if upstream_branch:
-      remote = _git_get_branch_config_value('remote', branch=branch)
-    else:
-      upstream_branch = RunGit(['config', 'rietveld.upstream-branch'],
-                               error_ok=True).strip()
-      if upstream_branch:
-        remote = RunGit(['config', 'rietveld.upstream-remote']).strip()
-      else:
-        # Else, try to guess the origin remote.
-        remote_branches = RunGit(['branch', '-r']).split()
-        if 'origin/master' in remote_branches:
-          # Fall back on origin/master if it exits.
-          remote = 'origin'
-          upstream_branch = 'refs/heads/master'
-        else:
-          DieWithError(
-             'Unable to determine default branch to diff against.\n'
-             'Either pass complete "git diff"-style arguments, like\n'
-             '  git cl upload origin/master\n'
-             'or verify this branch is set up to track another \n'
-             '(via the --track argument to "git checkout -b ...").')
+    remote, upstream_branch = scm.GIT.FetchUpstreamTuple(
+        settings.GetRoot(), branch)
+    if not remote or not upstream_branch:
+      DieWithError(
+         'Unable to determine default branch to diff against.\n'
+         'Either pass complete "git diff"-style arguments, like\n'
+         '  git cl upload origin/master\n'
+         'or verify this branch is set up to track another \n'
+         '(via the --track argument to "git checkout -b ...").')
 
     return remote, upstream_branch
 

scm.py

@@ -143,6 +143,33 @@ class GIT(object):
         results.append(('%s      ' % m.group(1)[0], m.group(2)))
     return results
 
+  @staticmethod
+  def GetConfig(cwd, key, default=None):
+    try:
+      return GIT.Capture(['config', key], cwd=cwd)
+    except subprocess2.CalledProcessError:
+      return default
+
+  @staticmethod
+  def GetBranchConfig(cwd, branch, key, default=None):
+    assert branch, 'A branch must be given'
+    key = 'branch.%s.%s' % (branch, key)
+    return GIT.GetConfig(cwd, key, default)
+
+  @staticmethod
+  def SetConfig(cwd, key, value=None):
+    if value is None:
+      args = ['config', '--unset', key]
+    else:
+      args = ['config', key, value]
+    GIT.Capture(args, cwd=cwd)
+
+  @staticmethod
+  def SetBranchConfig(cwd, branch, key, value=None):
+    assert branch, 'A branch must be given'
+    key = 'branch.%s.%s' % (branch, key)
+    GIT.SetConfig(cwd, key, value)
+
   @staticmethod
   def IsWorkTreeDirty(cwd):
     return GIT.Capture(['status', '-s'], cwd=cwd) != ''
@@ -150,10 +177,7 @@ class GIT(object):
   @staticmethod
   def GetEmail(cwd):
     """Retrieves the user email address if known."""
-    try:
-      return GIT.Capture(['config', 'user.email'], cwd=cwd)
-    except subprocess2.CalledProcessError:
-      return ''
+    return GIT.GetConfig(cwd, 'user.email', '')
 
   @staticmethod
   def ShortBranchName(branch):
@@ -171,47 +195,35 @@ class GIT(object):
     return GIT.ShortBranchName(GIT.GetBranchRef(cwd))
 
   @staticmethod
-  def FetchUpstreamTuple(cwd):
+  def GetRemoteBranches(cwd):
+    return GIT.Capture(['branch', '-r'], cwd=cwd).split()
+
+  @staticmethod
+  def FetchUpstreamTuple(cwd, branch=None):
     """Returns a tuple containg remote and remote ref,
        e.g. 'origin', 'refs/heads/master'
     """
-    remote = '.'
-    branch = GIT.GetBranch(cwd)
     try:
-      upstream_branch = GIT.Capture(
-          ['config', '--local', 'branch.%s.merge' % branch], cwd=cwd)
+      branch = branch or GIT.GetBranch(cwd)
     except subprocess2.CalledProcessError:
-      upstream_branch = None
-    if upstream_branch:
-      try:
-        remote = GIT.Capture(
-            ['config', '--local', 'branch.%s.remote' % branch], cwd=cwd)
-      except subprocess2.CalledProcessError:
-        pass
-    else:
-      try:
-        upstream_branch = GIT.Capture(
-            ['config', '--local', 'rietveld.upstream-branch'], cwd=cwd)
-      except subprocess2.CalledProcessError:
-        upstream_branch = None
+      pass
+    if branch:
+      upstream_branch = GIT.GetBranchConfig(cwd, branch, 'merge')
       if upstream_branch:
-        try:
-          remote = GIT.Capture(
-              ['config', '--local', 'rietveld.upstream-remote'], cwd=cwd)
-        except subprocess2.CalledProcessError:
-          pass
-      else:
-        # Else, try to guess the origin remote.
-        remote_branches = GIT.Capture(['branch', '-r'], cwd=cwd).split()
-        if 'origin/master' in remote_branches:
-          # Fall back on origin/master if it exits.
-          remote = 'origin'
-          upstream_branch = 'refs/heads/master'
-        else:
-          # Give up.
-          remote = None
-          upstream_branch = None
-    return remote, upstream_branch
+        remote = GIT.GetBranchConfig(cwd, branch, 'remote', '.')
+        return remote, upstream_branch
+
+    upstream_branch = GIT.GetConfig(cwd, 'rietveld.upstream-branch')
+    if upstream_branch:
+      remote = GIT.GetConfig(cwd, 'rietveld.upstream-remote', '.')
+      return remote, upstream_branch
+
+    # Else, try to guess the origin remote.
+    if 'origin/master' in GIT.GetRemoteBranches(cwd):
+      # Fall back on origin/master if it exits.
+      return 'origin', 'refs/heads/master'
+
+    return None, None
 
   @staticmethod
   def RefToRemoteRef(ref, remote):

tests/scm_unittest.py

@@ -23,6 +23,10 @@ import scm
 import subprocess2
 
 
+def callError(code=1, cmd='', cwd='', stdout=b'', stderr=b''):
+  return subprocess2.CalledProcessError(code, cmd, cwd, stdout, stderr)
+
+
 class GitWrapperTestCase(unittest.TestCase):
   def setUp(self):
     super(GitWrapperTestCase, self).setUp()
@@ -84,29 +88,115 @@ class RealGitTest(fake_repos.FakeReposTestBase):
     super(RealGitTest, self).setUp()
     self.enabled = self.FAKE_REPOS.set_up_git()
     if self.enabled:
-      self.clone_dir = scm.os.path.join(self.FAKE_REPOS.git_base, 'repo_1')
+      self.cwd = scm.os.path.join(self.FAKE_REPOS.git_base, 'repo_1')
     else:
       self.skipTest('git fake repos not available')
 
   def testIsValidRevision(self):
     # Sha1's are [0-9a-z]{32}, so starting with a 'z' or 'r' should always fail.
-    self.assertFalse(scm.GIT.IsValidRevision(cwd=self.clone_dir, rev='zebra'))
-    self.assertFalse(scm.GIT.IsValidRevision(cwd=self.clone_dir, rev='r123456'))
+    self.assertFalse(scm.GIT.IsValidRevision(cwd=self.cwd, rev='zebra'))
+    self.assertFalse(scm.GIT.IsValidRevision(cwd=self.cwd, rev='r123456'))
     # Valid cases
     first_rev = self.githash('repo_1', 1)
-    self.assertTrue(scm.GIT.IsValidRevision(cwd=self.clone_dir, rev=first_rev))
-    self.assertTrue(scm.GIT.IsValidRevision(cwd=self.clone_dir, rev='HEAD'))
+    self.assertTrue(scm.GIT.IsValidRevision(cwd=self.cwd, rev=first_rev))
+    self.assertTrue(scm.GIT.IsValidRevision(cwd=self.cwd, rev='HEAD'))
 
   def testIsAncestor(self):
     self.assertTrue(scm.GIT.IsAncestor(
-        self.clone_dir, self.githash('repo_1', 1), self.githash('repo_1', 2)))
+        self.cwd, self.githash('repo_1', 1), self.githash('repo_1', 2)))
     self.assertFalse(scm.GIT.IsAncestor(
-        self.clone_dir, self.githash('repo_1', 2), self.githash('repo_1', 1)))
+        self.cwd, self.githash('repo_1', 2), self.githash('repo_1', 1)))
     self.assertFalse(scm.GIT.IsAncestor(
-        self.clone_dir, self.githash('repo_1', 1), 'zebra'))
+        self.cwd, self.githash('repo_1', 1), 'zebra'))
 
   def testGetAllFiles(self):
-    self.assertEqual(['DEPS','origin'], scm.GIT.GetAllFiles(self.clone_dir))
+    self.assertEqual(['DEPS','origin'], scm.GIT.GetAllFiles(self.cwd))
+
+  def testGetSetConfig(self):
+    key = 'scm.test-key'
+
+    self.assertIsNone(scm.GIT.GetConfig(self.cwd, key))
+    self.assertEqual(
+        'default-value', scm.GIT.GetConfig(self.cwd, key, 'default-value'))
+
+    scm.GIT.SetConfig(self.cwd, key, 'set-value')
+    self.assertEqual('set-value', scm.GIT.GetConfig(self.cwd, key))
+    self.assertEqual(
+        'set-value', scm.GIT.GetConfig(self.cwd, key, 'default-value'))
+
+    scm.GIT.SetConfig(self.cwd, key)
+    self.assertIsNone(scm.GIT.GetConfig(self.cwd, key))
+    self.assertEqual(
+        'default-value', scm.GIT.GetConfig(self.cwd, key, 'default-value'))
+
+  def testGetSetBranchConfig(self):
+    branch = scm.GIT.GetBranch(self.cwd)
+    key = 'scm.test-key'
+
+    self.assertIsNone(scm.GIT.GetBranchConfig(self.cwd, branch, key))
+    self.assertEqual(
+        'default-value',
+        scm.GIT.GetBranchConfig(self.cwd, branch, key, 'default-value'))
+
+    scm.GIT.SetBranchConfig(self.cwd, branch, key, 'set-value')
+    self.assertEqual(
+        'set-value', scm.GIT.GetBranchConfig(self.cwd, branch, key))
+    self.assertEqual(
+        'set-value',
+        scm.GIT.GetBranchConfig(self.cwd, branch, key, 'default-value'))
+    self.assertEqual(
+        'set-value',
+        scm.GIT.GetConfig(self.cwd, 'branch.%s.%s' % (branch, key)))
+
+    scm.GIT.SetBranchConfig(self.cwd, branch, key)
+    self.assertIsNone(scm.GIT.GetBranchConfig(self.cwd, branch, key))
+
+  def testFetchUpstreamTuple_NoUpstreamFound(self):
+    self.assertEqual(
+        (None, None), scm.GIT.FetchUpstreamTuple(self.cwd))
+
+  @mock.patch('scm.GIT.GetRemoteBranches', return_value=['origin/master'])
+  def testFetchUpstreamTuple_GuessOriginMaster(self, _mockGetRemoteBranches):
+    self.assertEqual(
+        ('origin', 'refs/heads/master'), scm.GIT.FetchUpstreamTuple(self.cwd))
+
+  def testFetchUpstreamTuple_RietveldUpstreamConfig(self):
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-branch', 'rietveld-upstream')
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-remote', 'rietveld-remote')
+    self.assertEqual(
+        ('rietveld-remote', 'rietveld-upstream'),
+        scm.GIT.FetchUpstreamTuple(self.cwd))
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-branch')
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-remote')
+
+  @mock.patch('scm.GIT.GetBranch', side_effect=callError())
+  def testFetchUpstreamTuple_NotOnBranch(self, _mockGetBranch):
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-branch', 'rietveld-upstream')
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-remote', 'rietveld-remote')
+    self.assertEqual(
+        ('rietveld-remote', 'rietveld-upstream'),
+        scm.GIT.FetchUpstreamTuple(self.cwd))
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-branch')
+    scm.GIT.SetConfig(self.cwd, 'rietveld.upstream-remote')
+
+  def testFetchUpstreamTuple_BranchConfig(self):
+    branch = scm.GIT.GetBranch(self.cwd)
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'merge', 'branch-merge')
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'remote', 'branch-remote')
+    self.assertEqual(
+        ('branch-remote', 'branch-merge'), scm.GIT.FetchUpstreamTuple(self.cwd))
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'merge')
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'remote')
+
+  def testFetchUpstreamTuple_AnotherBranchConfig(self):
+    branch = 'scm-test-branch'
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'merge', 'other-merge')
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'remote', 'other-remote')
+    self.assertEqual(
+        ('other-remote', 'other-merge'),
+        scm.GIT.FetchUpstreamTuple(self.cwd, branch))
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'merge')
+    scm.GIT.SetBranchConfig(self.cwd, branch, 'remote')
 
 
 if __name__ == '__main__':