| #!/usr/bin/python |
| # Copyright (c) 2010 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.""" |
| |
| # Import before super_mox to keep valid references. |
| from os import rename |
| from shutil import rmtree |
| from subprocess import Popen, PIPE, STDOUT |
| import tempfile |
| import __builtin__ |
| |
| # Fixes include path. |
| from super_mox import mox, SuperMoxBaseTestBase, SuperMoxTestBase |
| |
| import gclient_scm |
| |
| |
| class GCBaseTestCase(SuperMoxTestBase): |
| # Like unittest's assertRaises, but checks for Gclient.Error. |
| def assertRaisesError(self, msg, fn, *args, **kwargs): |
| try: |
| fn(*args, **kwargs) |
| except gclient_scm.gclient_utils.Error, e: |
| self.assertEquals(e.args[0], msg) |
| else: |
| self.fail('%s not raised' % msg) |
| |
| |
| class BaseTestCase(GCBaseTestCase): |
| def setUp(self): |
| GCBaseTestCase.setUp(self) |
| self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'CheckCallAndFilter') |
| self.mox.StubOutWithMock(gclient_scm.gclient_utils, |
| 'CheckCallAndFilterAndHeader') |
| self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'FileRead') |
| self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'FileWrite') |
| self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'RemoveDirectory') |
| self._CaptureSVNInfo = gclient_scm.scm.SVN.CaptureInfo |
| self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'Capture') |
| self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'CaptureInfo') |
| self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'CaptureStatus') |
| self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'RunAndGetFileList') |
| self._scm_wrapper = gclient_scm.CreateSCM |
| gclient_scm.sys.stdout.flush = lambda: None |
| gclient_scm.scm.SVN.current_version = None |
| |
| |
| class SVNWrapperTestCase(BaseTestCase): |
| class OptionsObject(object): |
| def __init__(self, test_case, verbose=False, revision=None): |
| self.verbose = verbose |
| self.revision = revision |
| self.manually_grab_svn_rev = True |
| self.deps_os = None |
| self.force = False |
| self.reset = False |
| self.nohooks = False |
| self.stdout = gclient_scm.sys.stdout |
| |
| def Options(self, *args, **kwargs): |
| return self.OptionsObject(self, *args, **kwargs) |
| |
| def setUp(self): |
| BaseTestCase.setUp(self) |
| self.root_dir = self.Dir() |
| self.args = self.Args() |
| self.url = self.Url() |
| self.relpath = 'asf' |
| |
| def testDir(self): |
| members = [ |
| 'FullUrlForRelativeUrl', 'RunCommand', |
| 'cleanup', 'diff', 'export', 'pack', 'relpath', 'revert', |
| 'revinfo', 'runhooks', 'status', 'update', |
| 'updatesingle', 'url', |
| ] |
| |
| # If you add a member, be sure to add the relevant test! |
| self.compareMembers(self._scm_wrapper('svn://a'), members) |
| |
| def testUnsupportedSCM(self): |
| args = ['gopher://foo', self.root_dir, self.relpath] |
| exception_msg = 'No SCM found for url gopher://foo' |
| self.assertRaisesError(exception_msg, self._scm_wrapper, *args) |
| |
| def testSVNFullUrlForRelativeUrl(self): |
| self.url = 'svn://a/b/c/d' |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| self.assertEqual(scm.FullUrlForRelativeUrl('/crap'), 'svn://a/b/crap') |
| |
| def testGITFullUrlForRelativeUrl(self): |
| self.url = 'git://a/b/c/d' |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| self.assertEqual(scm.FullUrlForRelativeUrl('/crap'), 'git://a/b/c/crap') |
| |
| def testRunCommandException(self): |
| options = self.Options(verbose=False) |
| file_path = gclient_scm.os.path.join(self.root_dir, self.relpath, '.git') |
| gclient_scm.os.path.exists(file_path).AndReturn(False) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| exception = "Unsupported argument(s): %s" % ','.join(self.args) |
| self.assertRaisesError(exception, scm.RunCommand, |
| 'update', options, self.args) |
| |
| def testRunCommandUnknown(self): |
| # TODO(maruel): if ever used. |
| pass |
| |
| def testRevertMissing(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| gclient_scm.os.path.isdir(base_path).AndReturn(False) |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.5.1 (r32289)') |
| # It'll to a checkout instead. |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git') |
| ).AndReturn(False) |
| print("\n_____ %s is missing, synching instead" % self.relpath) |
| # Checkout. |
| gclient_scm.os.path.exists(base_path).AndReturn(False) |
| files_list = self.mox.CreateMockAnything() |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, |
| ['checkout', self.url, base_path, |
| '--force'], |
| cwd=self.root_dir, |
| file_list=files_list, |
| stdout=options.stdout) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.revert(options, self.args, files_list) |
| |
| def testRevertNone(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| gclient_scm.os.path.isdir(base_path).AndReturn(True) |
| gclient_scm.scm.SVN.CaptureStatus(base_path).AndReturn([]) |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, |
| ['update', '--revision', 'BASE'], |
| cwd=base_path, |
| file_list=mox.IgnoreArg(), |
| stdout=options.stdout) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| |
| def testRevert2Files(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| gclient_scm.os.path.isdir(base_path).AndReturn(True) |
| items = [ |
| ('M ', 'a'), |
| ('A ', 'b'), |
| ] |
| file_path1 = gclient_scm.os.path.join(base_path, 'a') |
| file_path2 = gclient_scm.os.path.join(base_path, 'b') |
| gclient_scm.scm.SVN.CaptureStatus(base_path).AndReturn(items) |
| gclient_scm.os.path.exists(file_path1).AndReturn(True) |
| gclient_scm.os.path.isfile(file_path1).AndReturn(True) |
| gclient_scm.os.remove(file_path1) |
| gclient_scm.os.path.exists(file_path2).AndReturn(True) |
| gclient_scm.os.path.isfile(file_path2).AndReturn(True) |
| gclient_scm.os.remove(file_path2) |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, |
| ['update', '--revision', 'BASE'], |
| cwd=base_path, |
| file_list=mox.IgnoreArg(), |
| stdout=options.stdout) |
| print(gclient_scm.os.path.join(base_path, 'a')) |
| print(gclient_scm.os.path.join(base_path, 'b')) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| |
| def testRevertDirectory(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| gclient_scm.os.path.isdir(base_path).AndReturn(True) |
| items = [ |
| ('~ ', 'a'), |
| ] |
| gclient_scm.scm.SVN.CaptureStatus(base_path).AndReturn(items) |
| file_path = gclient_scm.os.path.join(base_path, 'a') |
| print(file_path) |
| gclient_scm.os.path.exists(file_path).AndReturn(True) |
| gclient_scm.os.path.isfile(file_path).AndReturn(False) |
| gclient_scm.os.path.islink(file_path).AndReturn(False) |
| gclient_scm.os.path.isdir(file_path).AndReturn(True) |
| gclient_scm.gclient_utils.RemoveDirectory(file_path) |
| file_list1 = [] |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, |
| ['update', '--revision', 'BASE'], |
| cwd=base_path, |
| file_list=mox.IgnoreArg(), |
| stdout=options.stdout) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list2 = [] |
| scm.revert(options, self.args, file_list2) |
| |
| def testStatus(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| gclient_scm.os.path.isdir(base_path).AndReturn(True) |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, |
| ['status'] + self.args, |
| cwd=base_path, file_list=[], |
| stdout=options.stdout).AndReturn(None) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| self.assertEqual(scm.status(options, self.args, file_list), None) |
| |
| |
| # TODO(maruel): TEST REVISIONS!!! |
| # TODO(maruel): TEST RELOCATE!!! |
| def testUpdateCheckout(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| file_info = gclient_scm.gclient_utils.PrintableObject() |
| file_info.root = 'blah' |
| file_info.url = self.url |
| file_info.uuid = 'ABC' |
| file_info.revision = 42 |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git') |
| ).AndReturn(False) |
| # Checkout. |
| gclient_scm.os.path.exists(base_path).AndReturn(False) |
| files_list = self.mox.CreateMockAnything() |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.5.1 (r32289)') |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, |
| ['checkout', self.url, base_path, |
| '--force'], |
| cwd=self.root_dir, |
| file_list=files_list, |
| stdout=options.stdout) |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.update(options, (), files_list) |
| |
| def testUpdateUpdate(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| options.force = True |
| options.nohooks = False |
| file_info = { |
| 'Repository Root': 'blah', |
| 'URL': self.url, |
| 'UUID': 'ABC', |
| 'Revision': 42, |
| } |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git') |
| ).AndReturn(False) |
| |
| # Verify no locked files. |
| gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.') |
| ).AndReturn([]) |
| |
| # Checkout or update. |
| gclient_scm.os.path.exists(base_path).AndReturn(True) |
| gclient_scm.scm.SVN.CaptureInfo( |
| gclient_scm.os.path.join(base_path, "."), '.' |
| ).AndReturn(file_info) |
| # Cheat a bit here. |
| gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info) |
| additional_args = [] |
| if options.manually_grab_svn_rev: |
| additional_args = ['--revision', str(file_info['Revision'])] |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.5.1 (r32289)') |
| additional_args.append('--force') |
| files_list = [] |
| gclient_scm.scm.SVN.RunAndGetFileList( |
| options.verbose, |
| ['update', base_path] + additional_args, |
| cwd=self.root_dir, file_list=files_list, stdout=options.stdout) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.update(options, (), files_list) |
| |
| def testUpdateSingleCheckout(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| file_info = { |
| 'URL': self.url, |
| 'Revision': 42, |
| } |
| |
| # Checks to make sure that we support svn co --depth. |
| gclient_scm.scm.SVN.current_version = None |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.5.1 (r32289)') |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.svn') |
| ).AndReturn(False) |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, 'DEPS') |
| ).AndReturn(False) |
| |
| # Verify no locked files. |
| gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.') |
| ).AndReturn([]) |
| |
| # When checking out a single file, we issue an svn checkout and svn update. |
| files_list = self.mox.CreateMockAnything() |
| gclient_scm.gclient_utils.CheckCallAndFilterAndHeader( |
| ['svn', 'checkout', '--depth', 'empty', self.url, base_path], |
| always=True, cwd=self.root_dir, stdout=options.stdout) |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, ['update', 'DEPS'], |
| cwd=base_path, file_list=files_list, stdout=options.stdout) |
| |
| # Now we fall back on scm.update(). |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git') |
| ).AndReturn(False) |
| gclient_scm.os.path.exists(base_path).AndReturn(True) |
| gclient_scm.scm.SVN.CaptureInfo( |
| gclient_scm.os.path.join(base_path, "."), '.' |
| ).AndReturn(file_info) |
| gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info) |
| options.stdout.write("\n_____ %s at 42" % self.relpath) |
| options.stdout.write('\n') |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.updatesingle(options, ['DEPS'], files_list) |
| |
| def testUpdateSingleCheckoutSVN14(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| file_info = { |
| 'URL': self.url, |
| 'Revision': 42, |
| } |
| |
| # Checks to make sure that we support svn co --depth. |
| gclient_scm.scm.SVN.current_version = None |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.4.4 (r25188)') |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path) |
| ).AndReturn(True) |
| |
| # When checking out a single file with svn 1.4, we use svn export |
| files_list = self.mox.CreateMockAnything() |
| gclient_scm.gclient_utils.CheckCallAndFilterAndHeader( |
| ['svn', 'export', gclient_scm.os.path.join(self.url, 'DEPS'), |
| gclient_scm.os.path.join(base_path, 'DEPS')], |
| always=True, cwd=self.root_dir, stdout=options.stdout) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.updatesingle(options, ['DEPS'], files_list) |
| |
| def testUpdateSingleCheckoutSVNUpgrade(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| file_info = { |
| 'URL': self.url, |
| 'Revision': 42, |
| } |
| |
| # Checks to make sure that we support svn co --depth. |
| gclient_scm.scm.SVN.current_version = None |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.5.1 (r32289)') |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.svn') |
| ).AndReturn(False) |
| # If DEPS already exists, assume we're upgrading from svn1.4, so delete |
| # the old DEPS file. |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, 'DEPS') |
| ).AndReturn(True) |
| gclient_scm.os.remove(gclient_scm.os.path.join(base_path, 'DEPS')) |
| |
| # Verify no locked files. |
| gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.') |
| ).AndReturn([]) |
| |
| # When checking out a single file, we issue an svn checkout and svn update. |
| files_list = self.mox.CreateMockAnything() |
| gclient_scm.gclient_utils.CheckCallAndFilterAndHeader( |
| ['svn', 'checkout', '--depth', 'empty', self.url, base_path], |
| always=True, cwd=self.root_dir, stdout=options.stdout) |
| gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, ['update', 'DEPS'], |
| cwd=base_path, file_list=files_list, stdout=options.stdout) |
| |
| # Now we fall back on scm.update(). |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git') |
| ).AndReturn(False) |
| gclient_scm.os.path.exists(base_path).AndReturn(True) |
| gclient_scm.scm.SVN.CaptureInfo( |
| gclient_scm.os.path.join(base_path, "."), '.' |
| ).AndReturn(file_info) |
| gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info) |
| print("\n_____ %s at 42" % self.relpath) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.updatesingle(options, ['DEPS'], files_list) |
| |
| def testUpdateSingleUpdate(self): |
| options = self.Options(verbose=True) |
| base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| file_info = { |
| 'URL': self.url, |
| 'Revision': 42, |
| } |
| # Checks to make sure that we support svn co --depth. |
| gclient_scm.scm.SVN.current_version = None |
| gclient_scm.scm.SVN.Capture(['--version'] |
| ).AndReturn('svn, version 1.5.1 (r32289)') |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.svn') |
| ).AndReturn(True) |
| |
| # Verify no locked files. |
| gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.') |
| ).AndReturn([]) |
| |
| # Now we fall back on scm.update(). |
| files_list = self.mox.CreateMockAnything() |
| gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git') |
| ).AndReturn(False) |
| gclient_scm.os.path.exists(base_path).AndReturn(True) |
| gclient_scm.scm.SVN.CaptureInfo( |
| gclient_scm.os.path.join(base_path, "."), '.' |
| ).AndReturn(file_info) |
| gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info) |
| print("\n_____ %s at 42" % self.relpath) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| scm.updatesingle(options, ['DEPS'], files_list) |
| |
| def testUpdateGit(self): |
| options = self.Options(verbose=True) |
| file_path = gclient_scm.os.path.join(self.root_dir, self.relpath, '.git') |
| gclient_scm.os.path.exists(file_path).AndReturn(True) |
| print("________ found .git directory; skipping %s" % self.relpath) |
| |
| self.mox.ReplayAll() |
| scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.update(options, self.args, file_list) |
| |
| |
| class GitWrapperTestCase(BaseTestCase): |
| """This class doesn't use pymox.""" |
| class OptionsObject(object): |
| def __init__(self, test_case, verbose=False, revision=None): |
| self.verbose = verbose |
| self.revision = revision |
| self.manually_grab_svn_rev = True |
| self.deps_os = None |
| self.force = False |
| self.reset = False |
| self.nohooks = False |
| |
| sample_git_import = """blob |
| mark :1 |
| data 6 |
| Hello |
| |
| blob |
| mark :2 |
| data 4 |
| Bye |
| |
| reset refs/heads/master |
| commit refs/heads/master |
| 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 |
| |
| reset refs/heads/master |
| from :3 |
| """ |
| def Options(self, *args, **kwargs): |
| return self.OptionsObject(self, *args, **kwargs) |
| |
| def CreateGitRepo(self, git_import, path): |
| """Do it for real.""" |
| try: |
| Popen(['git', 'init'], stdout=PIPE, stderr=STDOUT, |
| cwd=path).communicate() |
| except OSError: |
| # git is not available, skip this test. |
| return False |
| Popen(['git', 'fast-import'], stdin=PIPE, stdout=PIPE, stderr=STDOUT, |
| cwd=path).communicate(input=git_import) |
| Popen(['git', 'checkout'], 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/master'], stdout=PIPE, |
| stderr=STDOUT, cwd=path).communicate() |
| Popen(['git', 'push', 'origin', 'origin/origin:origin/master'], stdout=PIPE, |
| stderr=STDOUT, cwd=path).communicate() |
| Popen(['git', 'config', '--unset', 'remote.origin.fetch'], stdout=PIPE, |
| stderr=STDOUT, cwd=path).communicate() |
| return True |
| |
| def setUp(self): |
| self.args = self.Args() |
| self.url = 'git://foo' |
| self.root_dir = tempfile.mkdtemp() |
| self.relpath = '.' |
| self.base_path = gclient_scm.os.path.join(self.root_dir, self.relpath) |
| self.enabled = self.CreateGitRepo(self.sample_git_import, self.base_path) |
| SuperMoxBaseTestBase.setUp(self) |
| |
| def tearDown(self): |
| SuperMoxBaseTestBase.tearDown(self) |
| rmtree(self.root_dir) |
| |
| def testDir(self): |
| members = [ |
| 'FullUrlForRelativeUrl', 'RunCommand', |
| 'cleanup', 'diff', 'export', 'pack', 'relpath', 'revert', |
| 'revinfo', 'runhooks', 'status', 'update', 'url', |
| ] |
| |
| # If you add a member, be sure to add the relevant test! |
| self.compareMembers(gclient_scm.CreateSCM(url=self.url), members) |
| |
| def testRevertMissing(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| file_path = gclient_scm.os.path.join(self.base_path, 'a') |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=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.assertEquals(file_list, [file_path]) |
| file_list = [] |
| scm.diff(options, self.args, file_list) |
| self.assertEquals(file_list, []) |
| |
| def testRevertNone(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEquals(file_list, []) |
| self.assertEquals(scm.revinfo(options, self.args, None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| |
| |
| def testRevertModified(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| file_path = gclient_scm.os.path.join(self.base_path, 'a') |
| open(file_path, 'a').writelines('touched\n') |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEquals(file_list, [file_path]) |
| file_list = [] |
| scm.diff(options, self.args, file_list) |
| self.assertEquals(file_list, []) |
| self.assertEquals(scm.revinfo(options, self.args, None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| |
| def testRevertNew(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.update(options, None, file_list) |
| file_path = gclient_scm.os.path.join(self.base_path, 'c') |
| f = open(file_path, 'w') |
| f.writelines('new\n') |
| f.close() |
| Popen(['git', 'add', 'c'], stdout=PIPE, |
| stderr=STDOUT, cwd=self.base_path).communicate() |
| file_list = [] |
| scm.revert(options, self.args, file_list) |
| self.assertEquals(file_list, [file_path]) |
| file_list = [] |
| scm.diff(options, self.args, file_list) |
| self.assertEquals(file_list, []) |
| self.assertEquals(scm.revinfo(options, self.args, None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| |
| def testStatusNew(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| file_path = gclient_scm.os.path.join(self.base_path, 'a') |
| open(file_path, 'a').writelines('touched\n') |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.status(options, self.args, file_list) |
| self.assertEquals(file_list, [file_path]) |
| |
| def testStatus2New(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| expected_file_list = [] |
| for f in ['a', 'b']: |
| file_path = gclient_scm.os.path.join(self.base_path, f) |
| open(file_path, 'a').writelines('touched\n') |
| expected_file_list.extend([file_path]) |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.status(options, self.args, file_list) |
| expected_file_list = [gclient_scm.os.path.join(self.base_path, x) |
| for x in ['a', 'b']] |
| self.assertEquals(sorted(file_list), expected_file_list) |
| |
| def testUpdateCheckout(self): |
| if not self.enabled: |
| return |
| options = self.Options(verbose=True) |
| root_dir = tempfile.mkdtemp() |
| relpath = 'foo' |
| base_path = gclient_scm.os.path.join(root_dir, relpath) |
| url = gclient_scm.os.path.join(self.root_dir, self.relpath, '.git') |
| try: |
| scm = gclient_scm.CreateSCM(url=url, root_dir=root_dir, |
| relpath=relpath) |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assertEquals(len(file_list), 2) |
| self.assert_(gclient_scm.os.path.isfile( |
| gclient_scm.os.path.join(base_path, 'a'))) |
| self.assertEquals(scm.revinfo(options, (), None), |
| '069c602044c5388d2d15c3f875b057c852003458') |
| finally: |
| rmtree(root_dir) |
| |
| def testUpdateUpdate(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| expected_file_list = [gclient_scm.os.path.join(self.base_path, x) |
| for x in ['a', 'b']] |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_list = [] |
| scm.update(options, (), file_list) |
| self.assertEquals(file_list, expected_file_list) |
| self.assertEquals(scm.revinfo(options, (), None), |
| 'a7142dc9f0009350b96a11f372b6ea658592aa95') |
| |
| def testUpdateUnstagedConflict(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_path = gclient_scm.os.path.join(self.base_path, 'b') |
| f = open(file_path, 'w').writelines('conflict\n') |
| exception = ( |
| "error: Your local changes to 'b' would be overwritten by merge. " |
| "Aborting.\n" |
| "Please, commit your changes or stash them before you can merge.\n") |
| self.assertRaisesError(exception, scm.update, options, (), []) |
| |
| def testUpdateConflict(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| file_path = gclient_scm.os.path.join(self.base_path, 'b') |
| f = open(file_path, 'w').writelines('conflict\n') |
| scm._Run(['commit', '-am', 'test']) |
| self.mox.StubOutWithMock(__builtin__, 'raw_input') |
| __builtin__.raw_input.__call__(mox.StrContains('Cannot fast-forward merge, ' |
| 'attempt to rebase? (y)es / ' |
| '(q)uit / (s)kip : ') |
| ).AndReturn('y') |
| self.mox.ReplayAll() |
| exception = \ |
| 'Conflict while rebasing this branch.\n' \ |
| 'Fix the conflict and run gclient again.\n' \ |
| "See 'man git-rebase' for details.\n" |
| self.assertRaisesError(exception, scm.update, options, (), []) |
| exception = \ |
| '\n____ . at refs/heads/master\n' \ |
| '\tYou have unstaged changes.\n' \ |
| '\tPlease commit, stash, or reset.\n' |
| self.assertRaisesError(exception, scm.update, options, (), []) |
| |
| def testUpdateNotGit(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| git_path = gclient_scm.os.path.join(self.base_path, '.git') |
| rename(git_path, git_path + 'foo') |
| exception = \ |
| '\n____ . at refs/heads/master\n' \ |
| '\tPath is not a git repo. No .git dir.\n' \ |
| '\tTo resolve:\n' \ |
| '\t\trm -rf .\n' \ |
| '\tAnd run gclient sync again\n' |
| self.assertRaisesError(exception, scm.update, options, (), []) |
| |
| def testRevinfo(self): |
| if not self.enabled: |
| return |
| options = self.Options() |
| scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir, |
| relpath=self.relpath) |
| rev_info = scm.revinfo(options, (), None) |
| self.assertEquals(rev_info, '069c602044c5388d2d15c3f875b057c852003458') |
| |
| |
| if __name__ == '__main__': |
| import unittest |
| unittest.main() |
| |
| # vim: ts=2:sw=2:tw=80:et: |