| #!/usr/bin/env vpython3 |
| # Copyright (c) 2012 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """Unit tests for gclient_scm.py.""" |
| |
| # pylint: disable=E1103 |
| |
| from io import StringIO |
| import json |
| import logging |
| import os |
| import re |
| from subprocess import Popen, PIPE, STDOUT |
| import sys |
| import tempfile |
| import unittest |
| from unittest import mock |
| |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
| |
| import gclient_scm |
| import gclient_utils |
| import git_cache |
| import subprocess2 |
| from testing_support import fake_repos |
| from testing_support import test_case_utils |
| |
| # TODO: Should fix these warnings. |
| # pylint: disable=line-too-long |
| |
| GIT = 'git' if sys.platform != 'win32' else 'git.bat' |
| |
| # Disable global git cache |
| git_cache.Mirror.SetCachePath(None) |
| |
| # Shortcut since this function is used often |
| join = gclient_scm.os.path.join |
| |
| TIMESTAMP_RE = re.compile(r'\[[0-9]{1,2}:[0-9]{2}:[0-9]{2}\] (.*)', re.DOTALL) |
| |
| |
| def strip_timestamps(value): |
| lines = value.splitlines(True) |
| for i in range(len(lines)): |
| m = TIMESTAMP_RE.match(lines[i]) |
| if m: |
| lines[i] = m.group(1) |
| return ''.join(lines) |
| |
| |
| class BasicTests(unittest.TestCase): |
| @mock.patch('gclient_scm.scm.GIT.Capture') |
| def testGetFirstRemoteUrl(self, mockCapture): |
| REMOTE_STRINGS = [ |
| ('remote.origin.url\nE:\\foo\\bar\x00', 'E:\\foo\\bar'), |
| ('remote.origin.url\n/b/foo/bar\x00', '/b/foo/bar'), |
| ('remote.origin.url\nhttps://foo/bar\x00', 'https://foo/bar'), |
| ('remote.origin.url\nE:\\Fo Bar\\bax\x00', 'E:\\Fo Bar\\bax'), |
| ('remote.origin.url\ngit://what/"do\x00', 'git://what/"do') |
| ] |
| FAKE_PATH = '/fake/path' |
| mockCapture.side_effect = [question for question, _ in REMOTE_STRINGS] |
| |
| for _, answer in REMOTE_STRINGS: |
| self.assertEqual( |
| gclient_scm.SCMWrapper._get_first_remote_url(FAKE_PATH), answer) |
| gclient_scm.scm.GIT._clear_config(FAKE_PATH) |
| |
| expected_calls = [ |
| mock.call(['config', '--list', '-z'], |
| cwd=FAKE_PATH, |
| strip_out=False) for _ in REMOTE_STRINGS |
| ] |
| self.assertEqual(mockCapture.mock_calls, expected_calls) |
| |
| |
| class BaseGitWrapperTestCase(unittest.TestCase, test_case_utils.TestCaseUtils): |
| """This class doesn't use pymox.""" |
| class OptionsObject(object): |
| def __init__(self, verbose=False, revision=None): |
| self.auto_rebase = False |
| self.verbose = verbose |
| self.revision = revision |
| self.deps_os = None |
| self.force = False |
| self.reset = False |
| self.nohooks = False |
| self.no_history = False |
| self.upstream = False |
| self.cache_dir = None |
| self.merge = False |
| self.jobs = 1 |
| self.break_repo_locks = False |
| self.delete_unversioned_trees = False |
| self.patch_ref = None |
| self.patch_repo = None |
| self.rebase_patch_ref = True |
| self.reset_patch_ref = True |
| |
| sample_git_import = """blob |
| mark :1 |
| data 6 |
| Hello |
| |
| blob |
| mark :2 |
| data 4 |
| Bye |
| |
| reset refs/heads/main |
| commit refs/heads/main |
| mark :3 |
| author Bob <bob@example.com> 1253744361 -0700 |
| committer Bob <bob@example.com> 1253744361 -0700 |
| data 8 |
| A and B |
| M 100644 :1 a |
| M 100644 :2 b |
| |
| blob |
| mark :4 |
| data 10 |
| Hello |
| You |
| |
| blob |
| mark :5 |
| data 8 |
| Bye |
| You |
| |
| commit refs/heads/origin |
| mark :6 |
| author Alice <alice@example.com> 1253744424 -0700 |
| committer Alice <alice@example.com> 1253744424 -0700 |
| data 13 |
| Personalized |
| from :3 |
| M 100644 :4 a |
| M 100644 :5 b |
| |
| blob |
| mark :7 |
| data 5 |
| Mooh |
| |
| commit refs/heads/feature |
| mark :8 |
| author Bob <bob@example.com> 1390311986 -0000 |
| committer Bob <bob@example.com> 1390311986 -0000 |
| data 6 |
| Add C |
| from :3 |
| M 100644 :7 c |
| |
| reset refs/heads/main |
| from :3 |
| """ |
| |
| def Options(self, *args, **kwargs): |
| return self.OptionsObject(*args, **kwargs) |
| |
| def checkstdout(self, expected): |
| # pylint: disable=no-member |
| value = sys.stdout.getvalue() |
| sys.stdout.close() |
| # Check that the expected output appears. |
| self.assertIn(expected, strip_timestamps(value)) |
| |
| @staticmethod |
| def CreateGitRepo(git_import, path): |
| """Do it for real.""" |
| try: |
| Popen([GIT, 'init', '-q'], stdout=PIPE, stderr=STDOUT, |
| cwd=path).communicate() |
| except OSError: |
| # git is not available, skip this test. |
| return False |
| Popen([GIT, 'fast-import', '--quiet'], |
| stdin=PIPE, |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate(input=git_import.encode()) |
| Popen([GIT, 'checkout', '-q'], stdout=PIPE, stderr=STDOUT, |
| cwd=path).communicate() |
| Popen([GIT, 'remote', 'add', '-f', 'origin', '.'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| Popen([GIT, 'checkout', '-b', 'new', 'origin/main', '-q'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| Popen([GIT, 'push', 'origin', 'origin/origin:origin/main', '-q'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| Popen([GIT, 'config', '--unset', 'remote.origin.fetch'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| Popen([GIT, 'config', 'user.email', 'someuser@chromium.org'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| Popen([GIT, 'config', 'user.name', 'Some User'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| # Set HEAD back to main |
| Popen([GIT, 'checkout', 'main', '-q'], |
| stdout=PIPE, |
| stderr=STDOUT, |
| cwd=path).communicate() |
| return True |
| |
| def _GetAskForDataCallback(self, expected_prompt, return_value): |
| def AskForData(prompt, options): |
| self.assertEqual(prompt, expected_prompt) |
| return return_value |
| |
| return AskForData |
| |
| def setUp(self): |
| unittest.TestCase.setUp(self) |
| test_case_utils.TestCaseUtils.setUp(self) |
| self.url = 'git://foo' |
| # The .git suffix allows gclient_scm to recognize the dir as a git repo |
| # when cloning it locally |
| self.root_dir = tempfile.mkdtemp('.git') |
| self.relpath = '.' |
| self.base_path = join(self.root_dir, self.relpath) |
| self.enabled = self.CreateGitRepo(self.sample_git_import, |
| self.base_path) |
| mock.patch('sys.stdout', StringIO()).start() |
| self.addCleanup(mock.patch.stopall) |
| self.addCleanup(gclient_utils.rmtree, self.root_dir) |
| |
| |
| class ManagedGitWrapperTestCase(BaseGitWrapperTestCase): |
| def testRevertMissing(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| file_path = join(self.base_path, 'a') |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| gclient_scm.os.remove(file_path) |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEqual(file_list, [file_path]) |
| file_list = [] |
| scm.diff(options, self.args, file_list) |
| self.assertEqual(file_list, []) |
| sys.stdout.close() |
| |
| def testRevertNone(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEqual(file_list, []) |
| self.assertEqual(scm.revinfo(options, self.args, None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| sys.stdout.close() |
| |
| def testRevertModified(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| file_path = join(self.base_path, 'a') |
| with open(file_path, 'a') as f: |
| f.writelines('touched\n') |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEqual(file_list, [file_path]) |
| file_list = [] |
| scm.diff(options, self.args, file_list) |
| self.assertEqual(file_list, []) |
| self.assertEqual(scm.revinfo(options, self.args, None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| sys.stdout.close() |
| |
| def testRevertNew(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| file_path = join(self.base_path, 'c') |
| with open(file_path, 'w') as f: |
| f.writelines('new\n') |
| Popen([GIT, 'add', 'c'], stdout=PIPE, stderr=STDOUT, |
| cwd=self.base_path).communicate() |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEqual(file_list, [file_path]) |
| file_list = [] |
| scm.diff(options, self.args, file_list) |
| self.assertEqual(file_list, []) |
| self.assertEqual(scm.revinfo(options, self.args, None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| sys.stdout.close() |
| |
| def testStatusRef(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| file_paths = [join(self.base_path, 'a')] |
| with open(file_paths[0], 'a') as f: |
| f.writelines('touched\n') |
| scm = gclient_scm.GitWrapper(self.url + '@refs/heads/feature', |
| self.root_dir, self.relpath) |
| file_paths.append(join(self.base_path, 'c')) # feature branch touches c |
| file_list = [] |
| scm.status(options, self.args, file_list) |
| self.assertEqual(file_list, file_paths) |
| self.checkstdout(( |
| '\n________ running \'git -c core.quotePath=false diff --name-status ' |
| 'refs/remotes/origin/feature\' in \'%s\'\n\nM\ta\n') % |
| join(self.root_dir, '.')) |
| |
| def testStatusNew(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| file_path = join(self.base_path, 'a') |
| with open(file_path, 'a') as f: |
| f.writelines('touched\n') |
| scm = gclient_scm.GitWrapper( |
| self.url + '@069c602044c5388d2d15c3f875b057c852003458', |
| self.root_dir, self.relpath) |
| file_list = [] |
| scm.status(options, self.args, file_list) |
| self.assertEqual(file_list, [file_path]) |
| self.checkstdout(( |
| '\n________ running \'git -c core.quotePath=false diff --name-status ' |
| '069c602044c5388d2d15c3f875b057c852003458\' in \'%s\'\n\nM\ta\n') % |
| join(self.root_dir, '.')) |
| |
| def testStatusNewNoBaseRev(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| file_path = join(self.base_path, 'a') |
| with open(file_path, 'a') as f: |
| f.writelines('touched\n') |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.status(options, self.args, file_list) |
| self.assertEqual(file_list, [file_path]) |
| self.checkstdout(( |
| '\n________ running \'git -c core.quotePath=false diff --name-status' |
| '\' in \'%s\'\n\nM\ta\n') % join(self.root_dir, '.')) |
| |
| def testStatus2New(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| expected_file_list = [] |
| for f in ['a', 'b']: |
| file_path = join(self.base_path, f) |
| with open(file_path, 'a') as f: |
| f.writelines('touched\n') |
| expected_file_list.extend([file_path]) |
| scm = gclient_scm.GitWrapper( |
| self.url + '@069c602044c5388d2d15c3f875b057c852003458', |
| self.root_dir, self.relpath) |
| file_list = [] |
| scm.status(options, self.args, file_list) |
| expected_file_list = [join(self.base_path, x) for x in ['a', 'b']] |
| self.assertEqual(sorted(file_list), expected_file_list) |
| self.checkstdout(( |
| '\n________ running \'git -c core.quotePath=false diff --name-status ' |
| '069c602044c5388d2d15c3f875b057c852003458\' in \'%s\'\n\nM\ta\nM\tb\n' |
| ) % join(self.root_dir, '.')) |
| |
| def testUpdateUpdate(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| expected_file_list = [join(self.base_path, x) for x in ['a', 'b']] |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| |
| scm.update(options, (), file_list) |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| self.assertEqual( |
| scm._Capture(['config', '--get', 'diff.ignoreSubmodules']), 'dirty') |
| self.assertEqual( |
| scm._Capture(['config', '--get', 'fetch.recurseSubmodules']), 'off') |
| self.assertEqual( |
| scm._Capture(['config', '--get', 'push.recurseSubmodules']), 'off') |
| sys.stdout.close() |
| |
| def testUpdateMerge(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| options.merge = True |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| # This sets correct remote HEAD |
| scm.update(options, (), []) |
| |
| scm._Run(['checkout', '-q', 'feature'], options) |
| rev = scm.revinfo(options, (), None) |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assertEqual(file_list, |
| [join(self.base_path, x) for x in ['a', 'b', 'c']]) |
| # The actual commit that is created is unstable, so we verify its tree |
| # and parents instead. |
| self.assertEqual(scm._Capture(['rev-parse', 'HEAD:']), |
| 'd2e35c10ac24d6c621e14a1fcadceb533155627d') |
| parent = 'HEAD^' if sys.platform != 'win32' else 'HEAD^^' |
| self.assertEqual(scm._Capture(['rev-parse', parent + '1']), rev) |
| self.assertEqual(scm._Capture(['rev-parse', parent + '2']), |
| scm._Capture(['rev-parse', 'origin/main'])) |
| sys.stdout.close() |
| |
| def testUpdateRebase(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| # This sets correct remote HEAD |
| scm.update(options, (), []) |
| |
| scm._Run(['checkout', '-q', 'feature'], options) |
| # Fake a 'y' key press. |
| scm._AskForData = self._GetAskForDataCallback( |
| 'Cannot fast-forward merge, attempt to rebase? ' |
| '(y)es / (q)uit / (s)kip : ', 'y') |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assertEqual(file_list, |
| [join(self.base_path, x) for x in ['a', 'b', 'c']]) |
| # The actual commit that is created is unstable, so we verify its tree |
| # and parent instead. |
| self.assertEqual(scm._Capture(['rev-parse', 'HEAD:']), |
| 'd2e35c10ac24d6c621e14a1fcadceb533155627d') |
| parent = 'HEAD^' if sys.platform != 'win32' else 'HEAD^^' |
| self.assertEqual(scm._Capture(['rev-parse', parent + '1']), |
| scm._Capture(['rev-parse', 'origin/main'])) |
| sys.stdout.close() |
| |
| def testUpdateReset(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| options.reset = True |
| |
| dir_path = join(self.base_path, 'c') |
| os.mkdir(dir_path) |
| with open(join(dir_path, 'nested'), 'w') as f: |
| f.writelines('new\n') |
| |
| file_path = join(self.base_path, 'file') |
| with open(file_path, 'w') as f: |
| f.writelines('new\n') |
| |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assert_(gclient_scm.os.path.isdir(dir_path)) |
| self.assert_(gclient_scm.os.path.isfile(file_path)) |
| sys.stdout.close() |
| |
| def testUpdateResetUnsetsFetchConfig(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| options.reset = True |
| |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| scm._Run([ |
| 'config', 'remote.origin.fetch', |
| '+refs/heads/bad/ref:refs/remotes/origin/bad/ref' |
| ], options) |
| |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| '069c602044c5388d2d15c3f875b057c852003458') |
| sys.stdout.close() |
| |
| def testUpdateResetDeleteUnversionedTrees(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| options.reset = True |
| options.delete_unversioned_trees = True |
| |
| dir_path = join(self.base_path, 'dir') |
| os.mkdir(dir_path) |
| with open(join(dir_path, 'nested'), 'w') as f: |
| f.writelines('new\n') |
| |
| file_path = join(self.base_path, 'file') |
| with open(file_path, 'w') as f: |
| f.writelines('new\n') |
| |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assert_(not gclient_scm.os.path.isdir(dir_path)) |
| self.assert_(gclient_scm.os.path.isfile(file_path)) |
| sys.stdout.close() |
| |
| def testUpdateUnstagedConflict(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_path = join(self.base_path, 'b') |
| with open(file_path, 'w') as f: |
| f.writelines('conflict\n') |
| try: |
| scm.update(options, (), []) |
| self.fail() |
| except (gclient_scm.gclient_utils.Error, |
| subprocess2.CalledProcessError): |
| # The exact exception text varies across git versions so it's not |
| # worth verifying it. It's fine as long as it throws. |
| pass |
| # Manually flush stdout since we can't verify it's content accurately |
| # across git versions. |
| sys.stdout.getvalue() |
| sys.stdout.close() |
| |
| @unittest.skip('Skipping until crbug.com/670884 is resolved.') |
| def testUpdateLocked(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_path = join(self.base_path, '.git', 'index.lock') |
| with open(file_path, 'w'): |
| pass |
| with self.assertRaises(subprocess2.CalledProcessError): |
| scm.update(options, (), []) |
| sys.stdout.close() |
| |
| def testUpdateLockedBreak(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| options.break_repo_locks = True |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_path = join(self.base_path, '.git', 'index.lock') |
| with open(file_path, 'w'): |
| pass |
| scm.update(options, (), []) |
| self.assertRegexpMatches(sys.stdout.getvalue(), |
| r'breaking lock.*\.git[/|\\]index\.lock') |
| self.assertFalse(os.path.exists(file_path)) |
| sys.stdout.close() |
| |
| def testUpdateConflict(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_path = join(self.base_path, 'b') |
| with open(file_path, 'w') as f: |
| f.writelines('conflict\n') |
| scm._Run(['commit', '-am', 'test'], options) |
| scm._AskForData = self._GetAskForDataCallback( |
| 'Cannot fast-forward merge, attempt to rebase? ' |
| '(y)es / (q)uit / (s)kip : ', 'y') |
| |
| with self.assertRaises(gclient_scm.gclient_utils.Error) as e: |
| scm.update(options, (), []) |
| self.assertEqual( |
| e.exception.args[0], 'Conflict while rebasing this branch.\n' |
| 'Fix the conflict and run gclient again.\n' |
| 'See \'man git-rebase\' for details.\n') |
| |
| with self.assertRaises(gclient_scm.gclient_utils.Error) as e: |
| scm.update(options, (), []) |
| self.assertEqual( |
| e.exception.args[0], '\n____ . at refs/remotes/origin/main\n' |
| '\tYou have uncommitted changes.\n' |
| '\tcd into ., run git status to see changes,\n' |
| '\tand commit, stash, or reset.\n') |
| |
| sys.stdout.close() |
| |
| def testRevinfo(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| rev_info = scm.revinfo(options, (), None) |
| self.assertEqual(rev_info, '069c602044c5388d2d15c3f875b057c852003458') |
| |
| |
| class ManagedGitWrapperTestCaseMock(unittest.TestCase): |
| class OptionsObject(object): |
| def __init__(self, verbose=False, revision=None, force=False): |
| self.verbose = verbose |
| self.revision = revision |
| self.deps_os = None |
| self.force = force |
| self.reset = False |
| self.nohooks = False |
| self.break_repo_locks = False |
| # TODO(maruel): Test --jobs > 1. |
| self.jobs = 1 |
| self.patch_ref = None |
| self.patch_repo = None |
| self.rebase_patch_ref = True |
| |
| def Options(self, *args, **kwargs): |
| return self.OptionsObject(*args, **kwargs) |
| |
| def checkstdout(self, expected): |
| # pylint: disable=no-member |
| value = sys.stdout.getvalue() |
| sys.stdout.close() |
| # Check that the expected output appears. |
| self.assertIn(expected, strip_timestamps(value)) |
| |
| def setUp(self): |
| self.fake_hash_1 = 't0ta11yf4k3' |
| self.fake_hash_2 = '3v3nf4k3r' |
| self.url = 'git://foo' |
| self.root_dir = '/tmp' if sys.platform != 'win32' else 't:\\tmp' |
| self.relpath = 'fake' |
| self.base_path = os.path.join(self.root_dir, self.relpath) |
| self.backup_base_path = os.path.join(self.root_dir, |
| 'old_%s.git' % self.relpath) |
| mock.patch('gclient_scm.scm.GIT.ApplyEnvVars').start() |
| mock.patch('gclient_scm.GitWrapper._Fetch').start() |
| mock.patch('gclient_scm.GitWrapper._DeleteOrMove').start() |
| mock.patch('sys.stdout', StringIO()).start() |
| self.addCleanup(mock.patch.stopall) |
| |
| @mock.patch('scm.GIT.IsValidRevision') |
| @mock.patch('os.path.isdir', lambda _: True) |
| def testGetUsableRevGit(self, mockIsValidRevision): |
| # pylint: disable=no-member |
| options = self.Options(verbose=True) |
| |
| mockIsValidRevision.side_effect = lambda cwd, rev: rev != '1' |
| |
| git_scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| # A [fake] git sha1 with a git repo should work (this is in the case |
| # that the LKGR gets flipped to git sha1's some day). |
| self.assertEqual(git_scm.GetUsableRev(self.fake_hash_1, options), |
| self.fake_hash_1) |
| # An SVN rev with an existing purely git repo should raise an exception. |
| self.assertRaises(gclient_scm.gclient_utils.Error, git_scm.GetUsableRev, |
| '1', options) |
| |
| @mock.patch('gclient_scm.GitWrapper._Clone') |
| @mock.patch('os.path.isdir') |
| @mock.patch('os.path.exists') |
| @mock.patch('subprocess2.check_output') |
| def testUpdateNoDotGit(self, mockCheckOutput, mockExists, mockIsdir, |
| mockClone): |
| mockIsdir.side_effect = lambda path: path == self.base_path |
| mockExists.side_effect = lambda path: path == self.base_path |
| mockCheckOutput.side_effect = [b'refs/remotes/origin/main', b'', b''] |
| |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| scm.update(options, None, []) |
| |
| env = gclient_scm.scm.GIT.ApplyEnvVars({}) |
| self.assertEqual(mockCheckOutput.mock_calls, [ |
| mock.call(['git', 'symbolic-ref', 'refs/remotes/origin/HEAD'], |
| cwd=self.base_path, |
| env=env, |
| stderr=-1), |
| mock.call(['git', '-c', 'core.quotePath=false', 'ls-files'], |
| cwd=self.base_path, |
| env=env, |
| stderr=-1), |
| mock.call(['git', 'rev-parse', '--verify', 'HEAD'], |
| cwd=self.base_path, |
| env=env, |
| stderr=-1), |
| ]) |
| mockClone.assert_called_with('refs/remotes/origin/main', self.url, |
| options) |
| self.checkstdout('\n') |
| |
| @mock.patch('gclient_scm.GitWrapper._Clone') |
| @mock.patch('os.path.isdir') |
| @mock.patch('os.path.exists') |
| @mock.patch('subprocess2.check_output') |
| def testUpdateConflict(self, mockCheckOutput, mockExists, mockIsdir, |
| mockClone): |
| mockIsdir.side_effect = lambda path: path == self.base_path |
| mockExists.side_effect = lambda path: path == self.base_path |
| mockCheckOutput.side_effect = [b'refs/remotes/origin/main', b'', b''] |
| mockClone.side_effect = [ |
| gclient_scm.subprocess2.CalledProcessError(None, None, None, None, |
| None), |
| None, |
| ] |
| |
| options = self.Options() |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| scm.update(options, None, []) |
| |
| env = gclient_scm.scm.GIT.ApplyEnvVars({}) |
| self.assertEqual(mockCheckOutput.mock_calls, [ |
| mock.call(['git', 'symbolic-ref', 'refs/remotes/origin/HEAD'], |
| cwd=self.base_path, |
| env=env, |
| stderr=-1), |
| mock.call(['git', '-c', 'core.quotePath=false', 'ls-files'], |
| cwd=self.base_path, |
| env=env, |
| stderr=-1), |
| mock.call(['git', 'rev-parse', '--verify', 'HEAD'], |
| cwd=self.base_path, |
| env=env, |
| stderr=-1), |
| ]) |
| mockClone.assert_called_with('refs/remotes/origin/main', self.url, |
| options) |
| self.checkstdout('\n') |
| |
| |
| class UnmanagedGitWrapperTestCase(BaseGitWrapperTestCase): |
| def checkInStdout(self, expected): |
| # pylint: disable=no-member |
| value = sys.stdout.getvalue() |
| sys.stdout.close() |
| self.assertIn(expected, value) |
| |
| def checkNotInStdout(self, expected): |
| # pylint: disable=no-member |
| value = sys.stdout.getvalue() |
| sys.stdout.close() |
| self.assertNotIn(expected, value) |
| |
| def getCurrentBranch(self): |
| # Returns name of current branch or HEAD for detached HEAD |
| branch = gclient_scm.scm.GIT.Capture( |
| ['rev-parse', '--abbrev-ref', 'HEAD'], cwd=self.base_path) |
| if branch == 'HEAD': |
| return None |
| return branch |
| |
| def testUpdateClone(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| |
| origin_root_dir = self.root_dir |
| self.addCleanup(gclient_utils.rmtree, origin_root_dir) |
| |
| self.root_dir = tempfile.mkdtemp() |
| self.relpath = '.' |
| self.base_path = join(self.root_dir, self.relpath) |
| |
| scm = gclient_scm.GitWrapper(origin_root_dir, self.root_dir, |
| self.relpath) |
| |
| expected_file_list = [ |
| join(self.base_path, "a"), |
| join(self.base_path, "b") |
| ] |
| file_list = [] |
| options.revision = 'unmanaged' |
| scm.update(options, (), file_list) |
| |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| '069c602044c5388d2d15c3f875b057c852003458') |
| # indicates detached HEAD |
| self.assertEqual(self.getCurrentBranch(), None) |
| self.checkInStdout( |
| 'Checked out refs/remotes/origin/main to a detached HEAD') |
| |
| def testUpdateCloneOnCommit(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| |
| origin_root_dir = self.root_dir |
| self.addCleanup(gclient_utils.rmtree, origin_root_dir) |
| |
| self.root_dir = tempfile.mkdtemp() |
| self.relpath = '.' |
| self.base_path = join(self.root_dir, self.relpath) |
| url_with_commit_ref = origin_root_dir +\ |
| '@a7142dc9f0009350b96a11f372b6ea658592aa95' |
| |
| scm = gclient_scm.GitWrapper(url_with_commit_ref, self.root_dir, |
| self.relpath) |
| |
| expected_file_list = [ |
| join(self.base_path, "a"), |
| join(self.base_path, "b") |
| ] |
| file_list = [] |
| options.revision = 'unmanaged' |
| scm.update(options, (), file_list) |
| |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| # indicates detached HEAD |
| self.assertEqual(self.getCurrentBranch(), None) |
| self.checkInStdout( |
| 'Checked out a7142dc9f0009350b96a11f372b6ea658592aa95 to a detached HEAD' |
| ) |
| |
| def testUpdateCloneOnBranch(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| |
| origin_root_dir = self.root_dir |
| self.addCleanup(gclient_utils.rmtree, origin_root_dir) |
| |
| self.root_dir = tempfile.mkdtemp() |
| self.relpath = '.' |
| self.base_path = join(self.root_dir, self.relpath) |
| url_with_branch_ref = origin_root_dir + '@feature' |
| |
| scm = gclient_scm.GitWrapper(url_with_branch_ref, self.root_dir, |
| self.relpath) |
| |
| expected_file_list = [ |
| join(self.base_path, "a"), |
| join(self.base_path, "b"), |
| join(self.base_path, "c") |
| ] |
| file_list = [] |
| options.revision = 'unmanaged' |
| scm.update(options, (), file_list) |
| |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| '9a51244740b25fa2ded5252ca00a3178d3f665a9') |
| # indicates detached HEAD |
| self.assertEqual(self.getCurrentBranch(), None) |
| self.checkInStdout( |
| 'Checked out 9a51244740b25fa2ded5252ca00a3178d3f665a9 ' |
| 'to a detached HEAD') |
| |
| def testUpdateCloneOnFetchedRemoteBranch(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| |
| origin_root_dir = self.root_dir |
| self.addCleanup(gclient_utils.rmtree, origin_root_dir) |
| |
| self.root_dir = tempfile.mkdtemp() |
| self.relpath = '.' |
| self.base_path = join(self.root_dir, self.relpath) |
| url_with_branch_ref = origin_root_dir + '@refs/remotes/origin/feature' |
| |
| scm = gclient_scm.GitWrapper(url_with_branch_ref, self.root_dir, |
| self.relpath) |
| |
| expected_file_list = [ |
| join(self.base_path, "a"), |
| join(self.base_path, "b"), |
| join(self.base_path, "c") |
| ] |
| file_list = [] |
| options.revision = 'unmanaged' |
| scm.update(options, (), file_list) |
| |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| '9a51244740b25fa2ded5252ca00a3178d3f665a9') |
| # indicates detached HEAD |
| self.assertEqual(self.getCurrentBranch(), None) |
| self.checkInStdout( |
| 'Checked out refs/remotes/origin/feature to a detached HEAD') |
| |
| def testUpdateCloneOnTrueRemoteBranch(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| |
| origin_root_dir = self.root_dir |
| self.addCleanup(gclient_utils.rmtree, origin_root_dir) |
| |
| self.root_dir = tempfile.mkdtemp() |
| self.relpath = '.' |
| self.base_path = join(self.root_dir, self.relpath) |
| url_with_branch_ref = origin_root_dir + '@refs/heads/feature' |
| |
| scm = gclient_scm.GitWrapper(url_with_branch_ref, self.root_dir, |
| self.relpath) |
| |
| expected_file_list = [ |
| join(self.base_path, "a"), |
| join(self.base_path, "b"), |
| join(self.base_path, "c") |
| ] |
| file_list = [] |
| options.revision = 'unmanaged' |
| scm.update(options, (), file_list) |
| |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| '9a51244740b25fa2ded5252ca00a3178d3f665a9') |
| # @refs/heads/feature is AKA @refs/remotes/origin/feature in the clone, |
| # so should be treated as such by gclient. TODO(mmoss): Though really, |
| # we should only allow DEPS to specify branches as they are known in the |
| # upstream repo, since the mapping into the local repo can be modified |
| # by users (or we might even want to change the gclient defaults at some |
| # point). But that will take more work to stop using refs/remotes/ |
| # everywhere that we do (and to stop assuming a DEPS ref will always |
| # resolve locally, like when passing them to show-ref or rev-list). |
| self.assertEqual(self.getCurrentBranch(), None) |
| self.checkInStdout( |
| 'Checked out refs/remotes/origin/feature to a detached HEAD') |
| |
| def testUpdateUpdate(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| expected_file_list = [] |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, self.relpath) |
| file_list = [] |
| options.revision = 'unmanaged' |
| scm.update(options, (), file_list) |
| self.assertEqual(file_list, expected_file_list) |
| self.assertEqual(scm.revinfo(options, (), None), |
| '069c602044c5388d2d15c3f875b057c852003458') |
| self.checkstdout('________ unmanaged solution; skipping .\n') |
| |
| |
| class CipdWrapperTestCase(unittest.TestCase): |
| def setUp(self): |
| # Create this before setting up mocks. |
| self._cipd_root_dir = tempfile.mkdtemp() |
| self._workdir = tempfile.mkdtemp() |
| |
| self._cipd_instance_url = 'https://chrome-infra-packages.appspot.com' |
| self._cipd_root = gclient_scm.CipdRoot(self._cipd_root_dir, |
| self._cipd_instance_url) |
| self._cipd_packages = [ |
| self._cipd_root.add_package('f', 'foo_package', 'foo_version'), |
| self._cipd_root.add_package('b', 'bar_package', 'bar_version'), |
| self._cipd_root.add_package('b', 'baz_package', 'baz_version'), |
| ] |
| mock.patch('tempfile.mkdtemp', lambda: self._workdir).start() |
| mock.patch('gclient_scm.CipdRoot.add_package').start() |
| mock.patch('gclient_scm.CipdRoot.clobber').start() |
| mock.patch('gclient_scm.CipdRoot.ensure_file_resolve').start() |
| mock.patch('gclient_scm.CipdRoot.ensure').start() |
| self.addCleanup(mock.patch.stopall) |
| self.addCleanup(gclient_utils.rmtree, self._cipd_root_dir) |
| self.addCleanup(gclient_utils.rmtree, self._workdir) |
| |
| def createScmWithPackageThatSatisfies(self, condition): |
| return gclient_scm.CipdWrapper( |
| url=self._cipd_instance_url, |
| root_dir=self._cipd_root_dir, |
| relpath='fake_relpath', |
| root=self._cipd_root, |
| package=self.getPackageThatSatisfies(condition)) |
| |
| def getPackageThatSatisfies(self, condition): |
| for p in self._cipd_packages: |
| if condition(p): |
| return p |
| |
| self.fail('Unable to find a satisfactory package.') |
| |
| def testRevert(self): |
| """Checks that revert does nothing.""" |
| scm = self.createScmWithPackageThatSatisfies(lambda _: True) |
| scm.revert(None, (), []) |
| |
| @mock.patch('gclient_scm.gclient_utils.CheckCallAndFilter') |
| @mock.patch('gclient_scm.gclient_utils.rmtree') |
| def testRevinfo(self, mockRmtree, mockCheckCallAndFilter): |
| """Checks that revinfo uses the JSON from cipd describe.""" |
| scm = self.createScmWithPackageThatSatisfies(lambda _: True) |
| |
| expected_revinfo = '0123456789abcdef0123456789abcdef01234567' |
| json_contents = { |
| 'result': { |
| 'pin': { |
| 'instance_id': expected_revinfo, |
| } |
| } |
| } |
| describe_json_path = join(self._workdir, 'describe.json') |
| with open(describe_json_path, 'w') as describe_json: |
| json.dump(json_contents, describe_json) |
| |
| revinfo = scm.revinfo(None, (), []) |
| self.assertEqual(revinfo, expected_revinfo) |
| |
| mockRmtree.assert_called_with(self._workdir) |
| mockCheckCallAndFilter.assert_called_with([ |
| 'cipd', |
| 'describe', |
| 'foo_package', |
| '-log-level', |
| 'error', |
| '-version', |
| 'foo_version', |
| '-json-output', |
| describe_json_path, |
| ]) |
| |
| def testUpdate(self): |
| """Checks that update does nothing.""" |
| scm = self.createScmWithPackageThatSatisfies(lambda _: True) |
| scm.update(None, (), []) |
| |
| |
| class GcsWrapperTestCase(unittest.TestCase): |
| |
| def setUp(self): |
| self.workdir = tempfile.mkdtemp() |
| self.url = 'gs://123bucket/path_to_tar.gz' |
| |
| def createScm(self): |
| return gclient_scm.GcsWrapper(url=self.url, |
| root_dir=self.workdir, |
| relpath='fake_relpath') |
| |
| def testRevert(self): |
| """Checks that revert does nothing.""" |
| scm = self.createScm() |
| scm.revert(None, (), []) |
| |
| def testRevinfo(self): |
| """Checks that revinfo does nothing.""" |
| scm = self.createScm() |
| scm.revinfo(None, (), []) |
| |
| def testUpdate(self): |
| """Checks that update does nothing.""" |
| scm = self.createScm() |
| scm.update(None, (), []) |
| |
| |
| class BranchHeadsFakeRepo(fake_repos.FakeReposBase): |
| def populateGit(self): |
| # Creates a tree that looks like this: |
| # |
| # 5 refs/branch-heads/5 |
| # | |
| # 4 |
| # | |
| # 1--2--3 refs/heads/main |
| self._commit_git('repo_1', {'commit 1': 'touched'}) |
| self._commit_git('repo_1', {'commit 2': 'touched'}) |
| self._commit_git('repo_1', {'commit 3': 'touched'}) |
| self._create_ref('repo_1', 'refs/heads/main', 3) |
| |
| self._commit_git('repo_1', {'commit 4': 'touched'}, base=2) |
| self._commit_git('repo_1', {'commit 5': 'touched'}, base=2) |
| self._create_ref('repo_1', 'refs/branch-heads/5', 5) |
| |
| |
| class BranchHeadsTest(fake_repos.FakeReposTestBase): |
| FAKE_REPOS_CLASS = BranchHeadsFakeRepo |
| |
| def setUp(self): |
| super(BranchHeadsTest, self).setUp() |
| self.enabled = self.FAKE_REPOS.set_up_git() |
| self.options = BaseGitWrapperTestCase.OptionsObject() |
| self.url = self.git_base + 'repo_1' |
| self.mirror = None |
| mock.patch('sys.stdout', StringIO()).start() |
| self.addCleanup(mock.patch.stopall) |
| |
| def setUpMirror(self): |
| self.mirror = tempfile.mkdtemp('mirror') |
| git_cache.Mirror.SetCachePath(self.mirror) |
| self.addCleanup(gclient_utils.rmtree, self.mirror) |
| self.addCleanup(git_cache.Mirror.SetCachePath, None) |
| |
| def testCheckoutBranchHeads(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = 'refs/branch-heads/5' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCheckoutUpdatedBranchHeads(self): |
| # Travel back in time, and set refs/branch-heads/5 to its parent. |
| subprocess2.check_call([ |
| 'git', 'update-ref', 'refs/branch-heads/5', |
| self.githash('repo_1', 4) |
| ], |
| cwd=self.url) |
| |
| # Sync to refs/branch-heads/5 |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| self.options.revision = 'refs/branch-heads/5' |
| scm.update(self.options, None, []) |
| |
| # Set refs/branch-heads/5 back to its original value. |
| subprocess2.check_call([ |
| 'git', 'update-ref', 'refs/branch-heads/5', |
| self.githash('repo_1', 5) |
| ], |
| cwd=self.url) |
| |
| # Attempt to sync to refs/branch-heads/5 again. |
| self.testCheckoutBranchHeads() |
| |
| def testCheckoutBranchHeadsMirror(self): |
| self.setUpMirror() |
| self.testCheckoutBranchHeads() |
| |
| def testCheckoutUpdatedBranchHeadsMirror(self): |
| self.setUpMirror() |
| self.testCheckoutUpdatedBranchHeads() |
| |
| |
| class GerritChangesFakeRepo(fake_repos.FakeReposBase): |
| def populateGit(self): |
| # Creates a tree that looks like this: |
| # |
| # 6 refs/changes/35/1235/1 |
| # | |
| # 5 refs/changes/34/1234/1 |
| # | |
| # 1--2--3--4 refs/heads/main |
| # | | |
| # | 11(5)--12 refs/heads/main-with-5 |
| # | |
| # 7--8--9 refs/heads/feature |
| # | |
| # 10 refs/changes/36/1236/1 |
| # |
| |
| self._commit_git('repo_1', {'commit 1': 'touched'}) |
| self._commit_git('repo_1', {'commit 2': 'touched'}) |
| self._commit_git('repo_1', {'commit 3': 'touched'}) |
| self._commit_git('repo_1', {'commit 4': 'touched'}) |
| self._create_ref('repo_1', 'refs/heads/main', 4) |
| |
| # Create a change on top of commit 3 that consists of two commits. |
| self._commit_git('repo_1', { |
| 'commit 5': 'touched', |
| 'change': '1234' |
| }, |
| base=3) |
| self._create_ref('repo_1', 'refs/changes/34/1234/1', 5) |
| self._commit_git('repo_1', {'commit 6': 'touched', 'change': '1235'}) |
| self._create_ref('repo_1', 'refs/changes/35/1235/1', 6) |
| |
| # Create a refs/heads/feature branch on top of commit 2, consisting of |
| # three commits. |
| self._commit_git('repo_1', {'commit 7': 'touched'}, base=2) |
| self._commit_git('repo_1', {'commit 8': 'touched'}) |
| self._commit_git('repo_1', {'commit 9': 'touched'}) |
| self._create_ref('repo_1', 'refs/heads/feature', 9) |
| |
| # Create a change of top of commit 8. |
| self._commit_git('repo_1', { |
| 'commit 10': 'touched', |
| 'change': '1236' |
| }, |
| base=8) |
| self._create_ref('repo_1', 'refs/changes/36/1236/1', 10) |
| |
| # Create a refs/heads/main-with-5 on top of commit 3 which is a branch |
| # where refs/changes/34/1234/1 (commit 5) has already landed as commit |
| # 11. |
| self._commit_git( |
| 'repo_1', |
| # This is really commit 11, but has the changes of commit 5 |
| { |
| 'commit 5': 'touched', |
| 'change': '1234' |
| }, |
| base=3) |
| self._commit_git('repo_1', {'commit 12': 'touched'}) |
| self._create_ref('repo_1', 'refs/heads/main-with-5', 12) |
| |
| |
| class GerritChangesTest(fake_repos.FakeReposTestBase): |
| FAKE_REPOS_CLASS = GerritChangesFakeRepo |
| |
| def setUp(self): |
| super(GerritChangesTest, self).setUp() |
| self.enabled = self.FAKE_REPOS.set_up_git() |
| self.options = BaseGitWrapperTestCase.OptionsObject() |
| self.url = self.git_base + 'repo_1' |
| self.mirror = None |
| mock.patch('sys.stdout', StringIO()).start() |
| self.addCleanup(mock.patch.stopall) |
| |
| def setUpMirror(self): |
| self.mirror = tempfile.mkdtemp() |
| git_cache.Mirror.SetCachePath(self.mirror) |
| self.addCleanup(gclient_utils.rmtree, self.mirror) |
| self.addCleanup(git_cache.Mirror.SetCachePath, None) |
| |
| def assertCommits(self, commits): |
| """Check that all, and only |commits| are present in the current checkout. |
| """ |
| for i in commits: |
| name = os.path.join(self.root_dir, 'commit ' + str(i)) |
| self.assertTrue(os.path.exists(name), 'Commit not found: %s' % name) |
| |
| all_commits = set(range(1, len(self.FAKE_REPOS.git_hashes['repo_1']))) |
| for i in all_commits - set(commits): |
| name = os.path.join(self.root_dir, 'commit ' + str(i)) |
| self.assertFalse(os.path.exists(name), |
| 'Unexpected commit: %s' % name) |
| |
| def testCanCloneGerritChange(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = 'refs/changes/35/1235/1' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 6), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCanSyncToGerritChange(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = self.githash('repo_1', 1) |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 1), |
| self.gitrevparse(self.root_dir)) |
| |
| self.options.revision = 'refs/changes/35/1235/1' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 6), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCanCloneGerritChangeMirror(self): |
| self.setUpMirror() |
| self.testCanCloneGerritChange() |
| |
| def testCanSyncToGerritChangeMirror(self): |
| self.setUpMirror() |
| self.testCanSyncToGerritChange() |
| |
| def testMirrorPushUrl(self): |
| self.setUpMirror() |
| |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| self.assertIsNotNone(scm._GetMirror(self.url, self.options)) |
| |
| scm.update(self.options, None, file_list) |
| |
| fetch_url = scm._Capture(['remote', 'get-url', 'origin']) |
| self.assertTrue( |
| fetch_url.startswith(self.mirror), |
| msg='\n'.join([ |
| 'Repository fetch url should be in the git cache mirror directory.', |
| ' fetch_url: %s' % fetch_url, |
| ' mirror: %s' % self.mirror |
| ])) |
| push_url = scm._Capture(['remote', 'get-url', '--push', 'origin']) |
| self.assertEqual(push_url, self.url) |
| |
| def testAppliesPatchOnTopOfMasterByDefault(self): |
| """Test the default case, where we apply a patch on top of main.""" |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Make sure we don't specify a revision. |
| self.options.revision = None |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| self.assertCommits([1, 2, 3, 4, 5, 6]) |
| self.assertEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCheckoutOlderThanPatchBase(self): |
| """Test applying a patch on an old checkout. |
| |
| We first checkout commit 1, and try to patch refs/changes/35/1235/1, which |
| contains commits 5 and 6, and is based on top of commit 3. |
| The final result should contain commits 1, 5 and 6, but not commits 2 or 3. |
| """ |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Sync to commit 1 |
| self.options.revision = self.githash('repo_1', 1) |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 1), |
| self.gitrevparse(self.root_dir)) |
| |
| # Apply the change on top of that. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| self.assertCommits([1, 5, 6]) |
| self.assertEqual(self.githash('repo_1', 1), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCheckoutOriginFeature(self): |
| """Tests that we can apply a patch on a branch other than main.""" |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Sync to remote's refs/heads/feature |
| self.options.revision = 'refs/heads/feature' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 9), |
| self.gitrevparse(self.root_dir)) |
| |
| # Apply the change on top of that. |
| scm.apply_patch_ref(self.url, 'refs/changes/36/1236/1', |
| 'refs/heads/feature', self.options, file_list) |
| |
| self.assertCommits([1, 2, 7, 8, 9, 10]) |
| self.assertEqual(self.githash('repo_1', 9), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCheckoutOriginFeatureOnOldRevision(self): |
| """Tests that we can apply a patch on an old checkout, on a branch other |
| than main.""" |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Sync to remote's refs/heads/feature on an old revision |
| self.options.revision = self.githash('repo_1', 7) |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 7), |
| self.gitrevparse(self.root_dir)) |
| |
| # Apply the change on top of that. |
| scm.apply_patch_ref(self.url, 'refs/changes/36/1236/1', |
| 'refs/heads/feature', self.options, file_list) |
| |
| # We shouldn't have rebased on top of 2 (which is the merge base between |
| # remote's main branch and the change) but on top of 7 (which is the |
| # merge base between remote's feature branch and the change). |
| self.assertCommits([1, 2, 7, 10]) |
| self.assertEqual(self.githash('repo_1', 7), |
| self.gitrevparse(self.root_dir)) |
| |
| def testCheckoutOriginFeaturePatchBranch(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Sync to the hash instead of remote's refs/heads/feature. |
| self.options.revision = self.githash('repo_1', 9) |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 9), |
| self.gitrevparse(self.root_dir)) |
| |
| # Apply refs/changes/34/1234/1, created for remote's main branch on top |
| # of remote's feature branch. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| # Commits 5 and 6 are part of the patch, and commits 1, 2, 7, 8 and 9 |
| # are part of remote's feature branch. |
| self.assertCommits([1, 2, 5, 6, 7, 8, 9]) |
| self.assertEqual(self.githash('repo_1', 9), |
| self.gitrevparse(self.root_dir)) |
| |
| def testDoesntRebasePatchMaster(self): |
| """Tests that we can apply a patch without rebasing it.""" |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.rebase_patch_ref = False |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| # Apply the change on top of that. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| self.assertCommits([1, 2, 3, 5, 6]) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| def testDoesntRebasePatchOldCheckout(self): |
| """Tests that we can apply a patch without rebasing it on an old checkout. |
| """ |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Sync to commit 1 |
| self.options.revision = self.githash('repo_1', 1) |
| self.options.rebase_patch_ref = False |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 1), |
| self.gitrevparse(self.root_dir)) |
| |
| # Apply the change on top of that. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| self.assertCommits([1, 2, 3, 5, 6]) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| def testDoesntSoftResetIfNotAskedTo(self): |
| """Test that we can apply a patch without doing a soft reset.""" |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.reset_patch_ref = False |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| self.assertCommits([1, 2, 3, 4, 5, 6]) |
| # The commit hash after cherry-picking is not known, but it must be |
| # different from what the repo was synced at before patching. |
| self.assertNotEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| @mock.patch('gerrit_util.GetChange', return_value={'topic': 'test_topic'}) |
| @mock.patch('gerrit_util.QueryChanges', |
| return_value=[{ |
| '_number': 1234 |
| }, { |
| '_number': 1235, |
| 'current_revision': 'abc', |
| 'revisions': { |
| 'abc': { |
| 'ref': 'refs/changes/35/1235/1' |
| } |
| } |
| }]) |
| def testDownloadTopics(self, query_changes_mock, get_change_mock): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = 'refs/changes/34/1234/1' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| # pylint: disable=attribute-defined-outside-init |
| self.options.download_topics = True |
| scm.url = 'https://test-repo.googlesource.com/repo_1.git' |
| scm.apply_patch_ref(self.url, 'refs/changes/34/1234/1', |
| 'refs/heads/main', self.options, file_list) |
| |
| get_change_mock.assert_called_once_with(mock.ANY, '1234') |
| query_changes_mock.assert_called_once_with(mock.ANY, |
| [('topic', 'test_topic'), |
| ('status', 'open'), |
| ('repo', 'repo_1')], |
| o_params=['ALL_REVISIONS']) |
| |
| self.assertCommits([1, 2, 3, 5, 6]) |
| # The commit hash after the two cherry-picks is not known, but it must |
| # be different from what the repo was synced at before patching. |
| self.assertNotEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| def testRecoversAfterPatchFailure(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = 'refs/changes/34/1234/1' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| # Checkout 'refs/changes/34/1234/1' modifies the 'change' file, so |
| # trying to patch 'refs/changes/36/1236/1' creates a patch failure. |
| with self.assertRaises(subprocess2.CalledProcessError) as cm: |
| scm.apply_patch_ref(self.url, 'refs/changes/36/1236/1', |
| 'refs/heads/main', self.options, file_list) |
| self.assertEqual(cm.exception.cmd[:2], ['git', 'cherry-pick']) |
| self.assertIn(b'error: could not apply', cm.exception.stderr) |
| |
| # Try to apply 'refs/changes/35/1235/1', which doesn't have a merge |
| # conflict. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| self.assertCommits([1, 2, 3, 5, 6]) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| def testIgnoresAlreadyMergedCommits(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = 'refs/heads/main-with-5' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 12), |
| self.gitrevparse(self.root_dir)) |
| |
| # When we try 'refs/changes/35/1235/1' on top of 'refs/heads/feature', |
| # 'refs/changes/34/1234/1' will be an empty commit, since the changes |
| # were already present in the tree as commit 11. Make sure we deal with |
| # this gracefully. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/feature', self.options, file_list) |
| self.assertCommits([1, 2, 3, 5, 6, 12]) |
| self.assertEqual(self.githash('repo_1', 12), |
| self.gitrevparse(self.root_dir)) |
| |
| def testRecoversFromExistingCherryPick(self): |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| self.options.revision = 'refs/changes/34/1234/1' |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| # Checkout 'refs/changes/34/1234/1' modifies the 'change' file, so |
| # trying to cherry-pick 'refs/changes/36/1236/1' raises an error. |
| scm._Run(['fetch', 'origin', 'refs/changes/36/1236/1'], self.options) |
| with self.assertRaises(subprocess2.CalledProcessError) as cm: |
| scm._Run(['cherry-pick', 'FETCH_HEAD'], self.options) |
| self.assertEqual(cm.exception.cmd[:2], ['git', 'cherry-pick']) |
| |
| # Try to apply 'refs/changes/35/1235/1', which doesn't have a merge |
| # conflict. |
| scm.apply_patch_ref(self.url, 'refs/changes/35/1235/1', |
| 'refs/heads/main', self.options, file_list) |
| self.assertCommits([1, 2, 3, 5, 6]) |
| self.assertEqual(self.githash('repo_1', 5), |
| self.gitrevparse(self.root_dir)) |
| |
| |
| class DepsChangesFakeRepo(fake_repos.FakeReposBase): |
| def populateGit(self): |
| self._commit_git('repo_1', {'DEPS': 'versionA', 'doesnotmatter': 'B'}) |
| self._commit_git('repo_1', {'DEPS': 'versionA', 'doesnotmatter': 'C'}) |
| |
| self._commit_git('repo_1', {'DEPS': 'versionB'}) |
| self._commit_git('repo_1', {'DEPS': 'versionA', 'doesnotmatter': 'C'}) |
| self._create_ref('repo_1', 'refs/heads/main', 4) |
| |
| |
| class CheckDiffTest(fake_repos.FakeReposTestBase): |
| FAKE_REPOS_CLASS = DepsChangesFakeRepo |
| |
| def setUp(self): |
| super(CheckDiffTest, self).setUp() |
| self.enabled = self.FAKE_REPOS.set_up_git() |
| self.options = BaseGitWrapperTestCase.OptionsObject() |
| self.url = self.git_base + 'repo_1' |
| self.mirror = None |
| mock.patch('sys.stdout', StringIO()).start() |
| self.addCleanup(mock.patch.stopall) |
| |
| def setUpMirror(self): |
| self.mirror = tempfile.mkdtemp() |
| git_cache.Mirror.SetCachePath(self.mirror) |
| self.addCleanup(gclient_utils.rmtree, self.mirror) |
| self.addCleanup(git_cache.Mirror.SetCachePath, None) |
| |
| def testCheckDiff(self): |
| """Correctly check for diffs.""" |
| scm = gclient_scm.GitWrapper(self.url, self.root_dir, '.') |
| file_list = [] |
| |
| # Make sure we don't specify a revision. |
| self.options.revision = None |
| scm.update(self.options, None, file_list) |
| self.assertEqual(self.githash('repo_1', 4), |
| self.gitrevparse(self.root_dir)) |
| |
| self.assertFalse( |
| scm.check_diff(self.githash('repo_1', 1), files=['DEPS'])) |
| self.assertTrue(scm.check_diff(self.githash('repo_1', 1))) |
| self.assertTrue( |
| scm.check_diff(self.githash('repo_1', 3), files=['DEPS'])) |
| |
| self.assertFalse( |
| scm.check_diff(self.githash('repo_1', 2), |
| files=['DEPS', 'doesnotmatter'])) |
| self.assertFalse(scm.check_diff(self.githash('repo_1', 2))) |
| |
| |
| if 'unittest.util' in __import__('sys').modules: |
| # Show full diff in self.assertEqual. |
| __import__('sys').modules['unittest.util']._MAX_LENGTH = 999999999 |
| |
| if __name__ == '__main__': |
| level = logging.DEBUG if '-v' in sys.argv else logging.FATAL |
| logging.basicConfig(level=level, |
| format='%(asctime).19s %(levelname)s %(filename)s:' |
| '%(lineno)s %(message)s') |
| unittest.main() |
| |
| # vim: ts=2:sw=2:tw=80:et: |
