| # -*- coding: utf-8 -*- |
| # Copyright (c) 2013 The Chromium OS 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 chromite.lib.git and helpers for testing that module.""" |
| |
| from __future__ import print_function |
| |
| import functools |
| import mock |
| import os |
| |
| from chromite.lib import cros_build_lib |
| from chromite.lib import cros_test_lib |
| from chromite.lib import git |
| from chromite.lib import osutils |
| from chromite.lib import partial_mock |
| from chromite.lib import patch_unittest |
| |
| |
| class ManifestMock(partial_mock.PartialMock): |
| """Partial mock for git.Manifest.""" |
| TARGET = 'chromite.lib.git.Manifest' |
| ATTRS = ('_RunParser',) |
| |
| def _RunParser(self, *_args): |
| pass |
| |
| |
| class ManifestCheckoutMock(partial_mock.PartialMock): |
| """Partial mock for git.ManifestCheckout.""" |
| TARGET = 'chromite.lib.git.ManifestCheckout' |
| ATTRS = ('_GetManifestsBranch',) |
| |
| def _GetManifestsBranch(self, _root): |
| return 'default' |
| |
| |
| class NormalizeRefTest(cros_test_lib.TestCase): |
| """Test the Normalize*Ref functions.""" |
| |
| def _TestNormalize(self, functor, tests): |
| """Helper function for testing Normalize*Ref functions. |
| |
| Args: |
| functor: Normalize*Ref functor that only needs the input |
| ref argument. |
| tests: Dict of test inputs to expected test outputs. |
| """ |
| for test_input, test_output in tests.iteritems(): |
| result = functor(test_input) |
| msg = ('Expected %s to translate %r to %r, but got %r.' % |
| (functor.__name__, test_input, test_output, result)) |
| self.assertEquals(test_output, result, msg) |
| |
| def testNormalizeRef(self): |
| """Test git.NormalizeRef function.""" |
| tests = { |
| # These should all get 'refs/heads/' prefix. |
| 'foo': 'refs/heads/foo', |
| 'foo-bar-123': 'refs/heads/foo-bar-123', |
| |
| # If input starts with 'refs/' it should be left alone. |
| 'refs/foo/bar': 'refs/foo/bar', |
| 'refs/heads/foo': 'refs/heads/foo', |
| |
| # Plain 'refs' is nothing special. |
| 'refs': 'refs/heads/refs', |
| |
| None: None, |
| } |
| self._TestNormalize(git.NormalizeRef, tests) |
| |
| def testNormalizeRemoteRef(self): |
| """Test git.NormalizeRemoteRef function.""" |
| remote = 'TheRemote' |
| tests = { |
| # These should all get 'refs/remotes/TheRemote' prefix. |
| 'foo': 'refs/remotes/%s/foo' % remote, |
| 'foo-bar-123': 'refs/remotes/%s/foo-bar-123' % remote, |
| |
| # These should be translated from local to remote ref. |
| 'refs/heads/foo': 'refs/remotes/%s/foo' % remote, |
| 'refs/heads/foo-bar-123': 'refs/remotes/%s/foo-bar-123' % remote, |
| |
| # These should be moved from one remote to another. |
| 'refs/remotes/OtherRemote/foo': 'refs/remotes/%s/foo' % remote, |
| |
| # These should be left alone. |
| 'refs/remotes/%s/foo' % remote: 'refs/remotes/%s/foo' % remote, |
| 'refs/foo/bar': 'refs/foo/bar', |
| |
| # Plain 'refs' is nothing special. |
| 'refs': 'refs/remotes/%s/refs' % remote, |
| |
| None: None, |
| } |
| |
| # Add remote arg to git.NormalizeRemoteRef. |
| functor = functools.partial(git.NormalizeRemoteRef, remote) |
| functor.__name__ = git.NormalizeRemoteRef.__name__ |
| |
| self._TestNormalize(functor, tests) |
| |
| |
| class GitWrappersTest(cros_test_lib.RunCommandTempDirTestCase): |
| """Tests for small git wrappers""" |
| |
| CHANGE_ID = 'I0da12ef6d2c670305f0281641bc53db22faf5c1a' |
| COMMIT_LOG = ''' |
| foo: Change to foo. |
| |
| Change-Id: %s |
| ''' % CHANGE_ID |
| |
| PUSH_REMOTE = 'fake_remote' |
| PUSH_BRANCH = 'fake_branch' |
| PUSH_LOCAL = 'fake_local_branch' |
| |
| def setUp(self): |
| self.fake_git_dir = os.path.join(self.tempdir, 'foo/bar') |
| self.fake_file = 'baz' |
| self.fake_path = os.path.join(self.fake_git_dir, self.fake_file) |
| |
| def testInit(self): |
| git.Init(self.fake_path) |
| |
| # Should have created the git repo directory, if it didn't exist. |
| self.assertExists(self.fake_git_dir) |
| self.assertCommandContains(['init']) |
| |
| def testClone(self): |
| url = 'http://happy/git/repo' |
| |
| git.Clone(self.fake_git_dir, url) |
| |
| # Should have created the git repo directory, if it didn't exist. |
| self.assertExists(self.fake_git_dir) |
| self.assertCommandContains(['git', 'clone', url, self.fake_git_dir]) |
| |
| def testCloneComplex(self): |
| url = 'http://happy/git/repo' |
| ref = 'other/git/repo' |
| |
| git.Clone(self.fake_git_dir, url, |
| reference=ref, branch='feature', single_branch=True) |
| |
| self.assertCommandContains(['git', 'clone', url, self.fake_git_dir, |
| '--reference', ref, |
| '--branch', 'feature', '--single-branch']) |
| |
| def testShallowFetch(self): |
| url = 'http://happy/git/repo' |
| |
| sparse_checkout = os.path.join(self.fake_git_dir, |
| '.git', 'info', 'sparse-checkout') |
| osutils.SafeMakedirs(os.path.dirname(sparse_checkout)) |
| |
| git.ShallowFetch(self.fake_git_dir, url, |
| sparse_checkout=['dir1/file1', 'dir2/file2']) |
| |
| # Should have created the git repo directory, if it didn't exist. |
| self.assertExists(self.fake_git_dir) |
| self.assertCommandContains(['init']) |
| self.assertCommandContains(['config', 'core.sparsecheckout', 'true']) |
| self.assertCommandContains(['remote', 'add', 'origin', url]) |
| self.assertCommandContains(['fetch', '--depth=1']) |
| self.assertCommandContains(['pull', 'origin', 'master']) |
| self.assertEquals(osutils.ReadFile(sparse_checkout), |
| 'dir1/file1\ndir2/file2') |
| |
| def testFindGitTopLevel(self): |
| git.FindGitTopLevel(self.fake_path) |
| self.assertCommandContains(['--show-toplevel']) |
| |
| def testAddPath(self): |
| git.AddPath(self.fake_path) |
| self.assertCommandContains(['add']) |
| self.assertCommandContains([self.fake_file]) |
| |
| def testRmPath(self): |
| git.RmPath(self.fake_path) |
| self.assertCommandContains(['rm']) |
| self.assertCommandContains([self.fake_file]) |
| |
| def testGetObjectAtRev(self): |
| git.GetObjectAtRev(self.fake_git_dir, '.', '1234') |
| self.assertCommandContains(['show']) |
| |
| def testRevertPath(self): |
| git.RevertPath(self.fake_git_dir, self.fake_file, '1234') |
| self.assertCommandContains(['checkout']) |
| self.assertCommandContains([self.fake_file]) |
| |
| def testCommit(self): |
| self.rc.AddCmdResult(partial_mock.In('log'), output=self.COMMIT_LOG) |
| git.Commit(self.fake_git_dir, 'bar') |
| self.assertCommandContains(['--amend'], expected=False) |
| cid = git.Commit(self.fake_git_dir, 'bar', amend=True) |
| self.assertCommandContains(['--amend']) |
| self.assertCommandContains(['--allow-empty'], expected=False) |
| self.assertEqual(cid, self.CHANGE_ID) |
| cid = git.Commit(self.fake_git_dir, 'new', allow_empty=True) |
| self.assertCommandContains(['--allow-empty']) |
| |
| def testUploadCLNormal(self): |
| git.UploadCL(self.fake_git_dir, self.PUSH_REMOTE, self.PUSH_BRANCH, |
| local_branch=self.PUSH_LOCAL) |
| self.assertCommandContains(['%s:refs/for/%s' % (self.PUSH_LOCAL, |
| self.PUSH_BRANCH)], |
| capture_output=False) |
| |
| def testUploadCLDraft(self): |
| git.UploadCL(self.fake_git_dir, self.PUSH_REMOTE, self.PUSH_BRANCH, |
| local_branch=self.PUSH_LOCAL, draft=True) |
| self.assertCommandContains(['%s:refs/drafts/%s' % (self.PUSH_LOCAL, |
| self.PUSH_BRANCH)], |
| capture_output=False) |
| |
| def testUploadCLCaptured(self): |
| git.UploadCL(self.fake_git_dir, self.PUSH_REMOTE, self.PUSH_BRANCH, |
| local_branch=self.PUSH_LOCAL, draft=True, capture_output=True) |
| self.assertCommandContains(['%s:refs/drafts/%s' % (self.PUSH_LOCAL, |
| self.PUSH_BRANCH)], |
| capture_output=True) |
| |
| def testGetGitRepoRevision(self): |
| git.GetGitRepoRevision(self.fake_git_dir) |
| self.assertCommandContains(['rev-parse', 'HEAD']) |
| git.GetGitRepoRevision(self.fake_git_dir, branch='branch') |
| self.assertCommandContains(['rev-parse', 'branch']) |
| git.GetGitRepoRevision(self.fake_git_dir, short=True) |
| self.assertCommandContains(['rev-parse', '--short', 'HEAD']) |
| git.GetGitRepoRevision(self.fake_git_dir, branch='branch', short=True) |
| self.assertCommandContains(['rev-parse', '--short', 'branch']) |
| |
| def testGetGitGitdir(self): |
| git.Init(self.fake_git_dir) |
| os.makedirs(os.path.join(self.fake_git_dir, '.git', 'refs', 'heads')) |
| os.makedirs(os.path.join(self.fake_git_dir, '.git', 'objects')) |
| other_file = os.path.join(self.fake_git_dir, 'other_file') |
| osutils.Touch(other_file) |
| |
| ret = git.GetGitGitdir(self.fake_git_dir) |
| self.assertEqual(ret, os.path.join(self.fake_git_dir, ".git")) |
| |
| def testGetGitGitdir_bare(self): |
| git.Init(self.fake_git_dir) |
| os.makedirs(os.path.join(self.fake_git_dir, 'refs', 'heads')) |
| os.makedirs(os.path.join(self.fake_git_dir, 'objects')) |
| config_file = os.path.join(self.fake_git_dir, 'config') |
| osutils.Touch(config_file) |
| |
| ret = git.GetGitGitdir(self.fake_git_dir) |
| self.assertEqual(ret, self.fake_git_dir) |
| |
| def testGetGitGitdir_negative(self): |
| ret = git.GetGitGitdir(self.tempdir) |
| self.assertFalse(ret) |
| |
| def testDeleteStaleLocks(self): |
| git.Init(self.fake_git_dir) |
| refs_heads = os.path.join(self.fake_git_dir, '.git', 'refs', 'heads') |
| os.makedirs(refs_heads) |
| objects = os.path.join(self.fake_git_dir, '.git', 'objects') |
| os.makedirs(objects) |
| fake_lock = os.path.join(refs_heads, 'master.lock') |
| osutils.Touch(fake_lock) |
| os.makedirs(self.fake_path) |
| dot_lock_not_in_dot_git = os.path.join(self.fake_git_dir, 'some.lock') |
| osutils.Touch(dot_lock_not_in_dot_git) |
| other_file = os.path.join(self.fake_path, 'other_file') |
| osutils.Touch(other_file) |
| |
| git.DeleteStaleLocks(self.fake_git_dir) |
| self.assertExists(os.path.join(self.fake_git_dir, '.git')) |
| self.assertExists(refs_heads) |
| self.assertExists(objects) |
| self.assertExists(dot_lock_not_in_dot_git) |
| self.assertExists(other_file) |
| self.assertNotExists(fake_lock) |
| |
| def testDeleteStaleLocks_bare(self): |
| git.Init(self.fake_git_dir) |
| refs_heads = os.path.join(self.fake_git_dir, 'refs', 'heads') |
| os.makedirs(refs_heads) |
| objects = os.path.join(self.fake_git_dir, 'objects') |
| os.makedirs(objects) |
| fake_lock = os.path.join(refs_heads, 'master.lock') |
| osutils.Touch(fake_lock) |
| os.makedirs(self.fake_path) |
| other_file = os.path.join(self.fake_path, 'other_file') |
| osutils.Touch(other_file) |
| |
| git.DeleteStaleLocks(self.fake_git_dir) |
| self.assertExists(refs_heads) |
| self.assertExists(objects) |
| self.assertExists(other_file) |
| self.assertNotExists(fake_lock) |
| |
| |
| class ProjectCheckoutTest(cros_test_lib.TestCase): |
| """Tests for git.ProjectCheckout""" |
| |
| def setUp(self): |
| self.fake_unversioned_patchable = git.ProjectCheckout( |
| dict(name='chromite', |
| path='src/chromite', |
| revision='remotes/for/master')) |
| self.fake_unversioned_unpatchable = git.ProjectCheckout( |
| dict(name='chromite', |
| path='src/platform/somethingsomething/chromite', |
| # Pinned to a SHA1. |
| revision='1deadbeeaf1deadbeeaf1deadbeeaf1deadbeeaf')) |
| self.fake_versioned_patchable = git.ProjectCheckout( |
| dict(name='chromite', |
| path='src/chromite', |
| revision='1deadbeeaf1deadbeeaf1deadbeeaf1deadbeeaf', |
| upstream='remotes/for/master')) |
| self.fake_versioned_unpatchable = git.ProjectCheckout( |
| dict(name='chromite', |
| path='src/chromite', |
| revision='1deadbeeaf1deadbeeaf1deadbeeaf1deadbeeaf', |
| upstream='1deadbeeaf1deadbeeaf1deadbeeaf1deadbeeaf')) |
| |
| |
| class RawDiffTest(cros_test_lib.MockTestCase): |
| """Tests for git.RawDiff function.""" |
| |
| def testRawDiff(self): |
| """Test the parsing of the git.RawDiff function.""" |
| |
| diff_output = ''' |
| :100644 100644 ac234b2... 077d1f8... M\tchromeos-base/chromeos-chrome/Manifest |
| :100644 100644 9e5d11b... 806bf9b... R099\tchromeos-base/chromeos-chrome/chromeos-chrome-40.0.2197.0_rc-r1.ebuild\tchromeos-base/chromeos-chrome/chromeos-chrome-40.0.2197.2_rc-r1.ebuild |
| :100644 100644 70d6e94... 821c642... M\tchromeos-base/chromeos-chrome/chromeos-chrome-9999.ebuild |
| :100644 100644 be445f9... be445f9... R100\tchromeos-base/chromium-source/chromium-source-40.0.2197.0_rc-r1.ebuild\tchromeos-base/chromium-source/chromium-source-40.0.2197.2_rc-r1.ebuild |
| ''' |
| result = cros_build_lib.CommandResult(output=diff_output) |
| self.PatchObject(git, 'RunGit', return_value=result) |
| |
| entries = git.RawDiff('foo', 'bar') |
| self.assertEqual(entries, [ |
| ('100644', '100644', 'ac234b2', '077d1f8', 'M', None, |
| 'chromeos-base/chromeos-chrome/Manifest', None), |
| ('100644', '100644', '9e5d11b', '806bf9b', 'R', '099', |
| 'chromeos-base/chromeos-chrome/' |
| 'chromeos-chrome-40.0.2197.0_rc-r1.ebuild', |
| 'chromeos-base/chromeos-chrome/' |
| 'chromeos-chrome-40.0.2197.2_rc-r1.ebuild'), |
| ('100644', '100644', '70d6e94', '821c642', 'M', None, |
| 'chromeos-base/chromeos-chrome/chromeos-chrome-9999.ebuild', None), |
| ('100644', '100644', 'be445f9', 'be445f9', 'R', '100', |
| 'chromeos-base/chromium-source/' |
| 'chromium-source-40.0.2197.0_rc-r1.ebuild', |
| 'chromeos-base/chromium-source/' |
| 'chromium-source-40.0.2197.2_rc-r1.ebuild') |
| ]) |
| |
| |
| class GitPushTest(cros_test_lib.RunCommandTestCase): |
| """Tests for git.GitPush function.""" |
| |
| # Non fast-forward push error message. |
| NON_FF_PUSH_ERROR = ( |
| 'To https://localhost/repo.git\n' |
| '! [remote rejected] master -> master (non-fast-forward)\n' |
| 'error: failed to push some refs to \'https://localhost/repo.git\'\n') |
| |
| # List of possible GoB transient errors. |
| TRANSIENT_ERRORS = ( |
| # Hook error when creating a new branch from SHA1 ref. |
| ('remote: Processing changes: (-)To https://localhost/repo.git\n' |
| '! [remote rejected] 6c78ca083c3a9d64068c945fd9998eb1e0a3e739 -> ' |
| 'stabilize-4636.B (error in hook)\n' |
| 'error: failed to push some refs to \'https://localhost/repo.git\'\n'), |
| |
| # 'failed to lock' error when creating a new branch from SHA1 ref. |
| ('remote: Processing changes: done\nTo https://localhost/repo.git\n' |
| '! [remote rejected] 4ea09c129b5fedb261bae2431ce2511e35ac3923 -> ' |
| 'stabilize-daisy-4319.96.B (failed to lock)\n' |
| 'error: failed to push some refs to \'https://localhost/repo.git\'\n'), |
| |
| # Hook error when pushing branch. |
| ('remote: Processing changes: (\\)To https://localhost/repo.git\n' |
| '! [remote rejected] temp_auto_checkin_branch -> ' |
| 'master (error in hook)\n' |
| 'error: failed to push some refs to \'https://localhost/repo.git\'\n'), |
| |
| # Another kind of error when pushing a branch. |
| 'fatal: remote error: Internal Server Error', |
| |
| # crbug.com/298189 |
| ('error: gnutls_handshake() failed: A TLS packet with unexpected length ' |
| 'was received. while accessing ' |
| 'http://localhost/repo.git/info/refs?service=git-upload-pack\n' |
| 'fatal: HTTP request failed'), |
| |
| # crbug.com/298189 |
| ('fatal: unable to access \'https://localhost/repo.git\': GnuTLS recv ' |
| 'error (-9): A TLS packet with unexpected length was received.'), |
| ) |
| |
| def setUp(self): |
| self.StartPatcher(mock.patch('time.sleep')) |
| |
| @staticmethod |
| def _RunGitPush(): |
| """Runs git.GitPush with some default arguments.""" |
| git.GitPush('some_repo_path', 'local-ref', |
| git.RemoteRef('some-remote', 'remote-ref')) |
| |
| def testGitPushSimple(self): |
| """Test GitPush with minimal arguments.""" |
| git.GitPush('git_path', 'HEAD', git.RemoteRef('origin', 'master')) |
| self.assertCommandCalled(['git', 'push', 'origin', 'HEAD:master'], |
| capture_output=True, print_cmd=False, |
| cwd='git_path') |
| |
| def testGitPushComplix(self): |
| """Test GitPush with some arguments.""" |
| git.GitPush('git_path', 'HEAD', git.RemoteRef('origin', 'master'), |
| force=True, dry_run=True) |
| self.assertCommandCalled(['git', 'push', 'origin', 'HEAD:master', |
| '--force', '--dry-run'], |
| capture_output=True, print_cmd=False, |
| cwd='git_path') |
| |
| def testNonFFPush(self): |
| """Non fast-forward push error propagates to the caller.""" |
| self.rc.AddCmdResult(partial_mock.In('push'), returncode=128, |
| error=self.NON_FF_PUSH_ERROR) |
| self.assertRaises(cros_build_lib.RunCommandError, self._RunGitPush) |
| |
| def testPersistentTransientError(self): |
| """GitPush fails if transient error occurs multiple times.""" |
| for error in self.TRANSIENT_ERRORS: |
| self.rc.AddCmdResult(partial_mock.In('push'), returncode=128, |
| error=error) |
| self.assertRaises(cros_build_lib.RunCommandError, self._RunGitPush) |
| |
| |
| class GitIntegrationTest(patch_unittest.GitRepoPatchTestCase): |
| """Tests that git library functions work with actual git repos.""" |
| |
| def testIsReachableTrue(self): |
| git1 = self._MakeRepo('git1', self.source) |
| patch1 = self.CommitFile(git1, 'foo', 'foo') |
| patch2 = self.CommitFile(git1, 'bar', 'bar') |
| self.assertTrue(git.IsReachable(git1, patch1.sha1, patch2.sha1)) |
| |
| def testIsReachableFalse(self): |
| git1 = self._MakeRepo('git1', self.source) |
| patch1 = self.CommitFile(git1, 'foo', 'foo') |
| patch2 = self.CommitFile(git1, 'bar', 'bar') |
| self.assertFalse(git.IsReachable(git1, patch2.sha1, patch1.sha1)) |
| |
| def testDoesCommitExistInRepoWithAmbiguousBranchName(self): |
| git1 = self._MakeRepo('git1', self.source) |
| git.CreateBranch(git1, 'peach', track=True) |
| self.CommitFile(git1, 'peach', 'Keep me.') |
| self.assertTrue(git.DoesCommitExistInRepo(git1, 'peach')) |
