| #!/usr/bin/python |
| |
| # Copyright (c) 2011 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. |
| |
| """Unittests for lkgm_manager. Needs to be run inside of chroot for mox.""" |
| |
| import mox |
| import os |
| import shutil |
| import sys |
| import tempfile |
| import threading |
| import time |
| import unittest |
| from xml.dom import minidom |
| |
| if __name__ == '__main__': |
| import constants |
| sys.path.append(constants.SOURCE_ROOT) |
| |
| from chromite.buildbot import lkgm_manager |
| from chromite.buildbot import manifest_version |
| from chromite.buildbot import manifest_version_unittest |
| from chromite.buildbot import patch |
| from chromite.lib import cros_build_lib as cros_lib |
| |
| |
| # pylint: disable=W0212,R0904 |
| FAKE_VERSION_STRING = '1.2.4-rc3' |
| FAKE_VERSION_STRING_NEXT = '1.2.4-rc4' |
| CHROME_BRANCH = '13' |
| |
| FAKE_VERSION = """ |
| CHROMEOS_BUILD=1 |
| CHROMEOS_BRANCH=2 |
| CHROMEOS_PATCH=4 |
| CHROME_BRANCH=13 |
| """ |
| |
| |
| class LKGMCandidateInfoTest(mox.MoxTestBase): |
| """Test methods testing methods in _LKGMCandidateInfo class.""" |
| |
| def setUp(self): |
| mox.MoxTestBase.setUp(self) |
| self.tmpdir = tempfile.mkdtemp() |
| |
| @classmethod |
| def CreateFakeVersionFile(cls, tmpdir): |
| """Helper method to create a version file from FAKE_VERSION.""" |
| if not os.path.exists(tmpdir): os.makedirs(tmpdir) |
| (version_file_fh, version_file) = tempfile.mkstemp(dir=tmpdir) |
| os.write(version_file_fh, FAKE_VERSION) |
| os.close(version_file_fh) |
| return version_file |
| |
| def testLoadFromString(self): |
| """Tests whether we can load from a string.""" |
| info = lkgm_manager._LKGMCandidateInfo(version_string=FAKE_VERSION_STRING, |
| chrome_branch=CHROME_BRANCH) |
| self.assertEqual(info.VersionString(), FAKE_VERSION_STRING) |
| |
| def testIncrementVersionPatch(self): |
| """Tests whether we can increment a lkgm info.""" |
| info = lkgm_manager._LKGMCandidateInfo(version_string=FAKE_VERSION_STRING, |
| chrome_branch=CHROME_BRANCH) |
| info.IncrementVersion() |
| self.assertEqual(info.VersionString(), FAKE_VERSION_STRING_NEXT) |
| |
| def testVersionCompare(self): |
| """Tests whether our comparision method works.""" |
| info1 = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc1') |
| info2 = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc2') |
| info3 = lkgm_manager._LKGMCandidateInfo('1.2.200.4-rc1') |
| info4 = lkgm_manager._LKGMCandidateInfo('1.4.3.4-rc1') |
| |
| self.assertTrue(info2 > info1) |
| self.assertTrue(info3 > info1) |
| self.assertTrue(info3 > info2) |
| self.assertTrue(info4 > info1) |
| self.assertTrue(info4 > info2) |
| self.assertTrue(info4 > info3) |
| |
| def tearDown(self): |
| shutil.rmtree(self.tmpdir) |
| |
| |
| class LKGMManagerTest(mox.MoxTestBase): |
| """Tests for the BuildSpecs manager.""" |
| |
| def setUp(self): |
| mox.MoxTestBase.setUp(self) |
| |
| self.tmpdir = tempfile.mkdtemp() |
| self.source_repo = 'ssh://source/repo' |
| self.manifest_repo = 'ssh://manifest/repo' |
| self.version_file = 'version-file.sh' |
| self.branch = 'master' |
| self.build_name = 'x86-generic' |
| self.incr_type = 'patch' |
| |
| # Change default to something we clean up. |
| self.tmpmandir = tempfile.mkdtemp() |
| manifest_version.BuildSpecsManager._TMP_MANIFEST_DIR = self.tmpmandir |
| |
| self.manager = lkgm_manager.LKGMManager( |
| self.tmpdir, self.source_repo, self.manifest_repo, self.branch, |
| self.build_name, 'binary', dry_run=True) |
| |
| self.manager.all_specs_dir = '/LKGM/path' |
| self.manager.specs_for_builder = os.path.join(self.manager.GetManifestDir(), |
| self.manager.lkgm_subdir, |
| 'build-name', '%(builder)s') |
| self.manager.SLEEP_TIMEOUT = 1 |
| |
| def _CommonTestLatestCandidateByVersion(self, version, expected_candidate, |
| no_all=False): |
| """Common helper function for latest candidate tests. |
| |
| Helper function to test given a version whether we can the right candidate |
| back. |
| |
| Args: |
| version: The Chrome OS version to look for the latest candidate of. |
| expected_candidate: What we expect to come back. |
| """ |
| if not no_all: |
| self.manager.all = ['1.2.4-rc1', |
| '1.2.4-rc2', |
| '1.2.4-rc9', |
| '1.2.5-rc1', |
| '1.2.6-rc2', |
| '1.3.4-rc1', |
| ] |
| |
| info_for_test = manifest_version.VersionInfo(version) |
| candidate = self.manager._GetLatestCandidateByVersion(info_for_test) |
| self.assertEqual(candidate.VersionString(), expected_candidate) |
| |
| def testGetLatestCandidateByVersionCommonCase(self): |
| """Tests whether we can get the latest candidate under the common case. |
| |
| This test tests whether or not we get the right candidate when we have |
| many of the same candidate version around but with different rc's. |
| """ |
| self._CommonTestLatestCandidateByVersion('1.2.4', '1.2.4-rc9') |
| |
| def testGetLatestCandidateByVersionOnlyOne(self): |
| """Tests whether we can get the latest candidate with only one rc.""" |
| self._CommonTestLatestCandidateByVersion('1.2.6', '1.2.6-rc2') |
| |
| def testGetLatestCandidateByVersionNone(self): |
| """Tests whether we can get the latest candidate with no rc's.""" |
| self._CommonTestLatestCandidateByVersion('1.2.7', '1.2.7-rc1') |
| |
| def testGetLatestCandidateByVersionNoneNoAll(self): |
| """Tests whether we can get the latest candidate with no rc's at all.""" |
| self._CommonTestLatestCandidateByVersion('10.0.1', '10.0.1-rc1', True) |
| |
| def _GetPathToManifest(self, info): |
| return os.path.join(self.manager.all_specs_dir, '%s.xml' % |
| info.VersionString()) |
| |
| def testCreateNewCandidate(self): |
| """Tests that we can create a new candidate and uprev and old rc.""" |
| # Let's stub out other LKGMManager calls cause they're already |
| # unit tested. |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, |
| '_GetLatestCandidateByVersion') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_CreateNewBuildSpec') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PrepSpecChanges') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PushSpecChanges') |
| |
| my_info = manifest_version.VersionInfo('1.2.3.4') |
| most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc12') |
| new_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc13') |
| |
| lkgm_manager.LKGMManager._GetCurrentVersionInfo().AndReturn(my_info) |
| lkgm_manager.LKGMManager._LoadSpecs(my_info) |
| lkgm_manager.LKGMManager._GetLatestCandidateByVersion(my_info).AndReturn( |
| most_recent_candidate) |
| lkgm_manager.LKGMManager._PrepSpecChanges() |
| lkgm_manager.LKGMManager._CreateNewBuildSpec( |
| most_recent_candidate).AndReturn(new_candidate.VersionString()) |
| lkgm_manager.LKGMManager._SetInFlight() |
| lkgm_manager.LKGMManager._PushSpecChanges( |
| mox.StrContains(new_candidate.VersionString())) |
| |
| self.mox.ReplayAll() |
| candidate_path = self.manager.CreateNewCandidate() |
| self.assertEqual(candidate_path, self._GetPathToManifest(new_candidate)) |
| self.mox.VerifyAll() |
| |
| def testCreateNewCandidateReturnNoneIfNoWorkToDo(self): |
| """Tests that we return nothing if there is nothing to create.""" |
| # Let's stub out other LKGMManager calls cause they're already |
| # unit tested. |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, |
| '_GetLatestCandidateByVersion') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_CreateNewBuildSpec') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PrepSpecChanges') |
| |
| |
| my_info = manifest_version.VersionInfo('1.2.3.4') |
| most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3.4-rc12') |
| |
| lkgm_manager.LKGMManager._GetCurrentVersionInfo().AndReturn(my_info) |
| lkgm_manager.LKGMManager._LoadSpecs(my_info) |
| lkgm_manager.LKGMManager._GetLatestCandidateByVersion(my_info).AndReturn( |
| most_recent_candidate) |
| lkgm_manager.LKGMManager._PrepSpecChanges() |
| lkgm_manager.LKGMManager._CreateNewBuildSpec( |
| most_recent_candidate).AndReturn(None) |
| |
| self.mox.ReplayAll() |
| candidate = self.manager.CreateNewCandidate() |
| self.assertEqual(candidate, None) |
| self.mox.VerifyAll() |
| |
| def testGetLatestCandidate(self): |
| """Makes sure we can get the latest created candidate manifest.""" |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PrepSpecChanges') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PushSpecChanges') |
| |
| my_info = manifest_version.VersionInfo('1.2.3') |
| most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.3-rc12') |
| |
| lkgm_manager.LKGMManager._GetCurrentVersionInfo().AndReturn(my_info) |
| lkgm_manager.LKGMManager._LoadSpecs(my_info) |
| lkgm_manager.LKGMManager._PrepSpecChanges() |
| lkgm_manager.LKGMManager._SetInFlight() |
| lkgm_manager.LKGMManager._PushSpecChanges( |
| mox.StrContains(most_recent_candidate.VersionString())) |
| |
| self.mox.ReplayAll() |
| self.manager.latest_unprocessed = '1.2.3-rc12' |
| candidate = self.manager.GetLatestCandidate() |
| self.assertEqual(candidate, self._GetPathToManifest(most_recent_candidate)) |
| self.mox.VerifyAll() |
| |
| def testGetLatestCandidateOneRetry(self): |
| """Makes sure we can get the latest candidate even on retry.""" |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_SetInFlight') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PrepSpecChanges') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_PushSpecChanges') |
| |
| my_info = manifest_version.VersionInfo('1.2.4') |
| most_recent_candidate = lkgm_manager._LKGMCandidateInfo('1.2.4-rc12', |
| CHROME_BRANCH) |
| |
| lkgm_manager.LKGMManager._GetCurrentVersionInfo().AndReturn(my_info) |
| lkgm_manager.LKGMManager._LoadSpecs(my_info) |
| lkgm_manager.LKGMManager._PrepSpecChanges() |
| lkgm_manager.LKGMManager._SetInFlight() |
| lkgm_manager.LKGMManager._PushSpecChanges( |
| mox.StrContains(most_recent_candidate.VersionString())).AndRaise( |
| manifest_version.GitCommandException('Push failed')) |
| |
| lkgm_manager.LKGMManager._PrepSpecChanges() |
| lkgm_manager.LKGMManager._SetInFlight() |
| lkgm_manager.LKGMManager._PushSpecChanges( |
| mox.StrContains(most_recent_candidate.VersionString())) |
| |
| self.mox.ReplayAll() |
| self.manager.latest_unprocessed = '1.2.4-rc12' |
| candidate = self.manager.GetLatestCandidate() |
| self.assertEqual(candidate, self._GetPathToManifest(most_recent_candidate)) |
| self.mox.VerifyAll() |
| |
| def testGetLatestCandidateNone(self): |
| """Makes sure we get nothing if there is no work to be done.""" |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_GetCurrentVersionInfo') |
| self.mox.StubOutWithMock(lkgm_manager.LKGMManager, '_LoadSpecs') |
| |
| my_info = manifest_version.VersionInfo('1.2.3.4') |
| lkgm_manager.LKGMManager._GetCurrentVersionInfo().AndReturn(my_info) |
| lkgm_manager.LKGMManager._LoadSpecs(my_info) |
| |
| self.mox.ReplayAll() |
| self.manager.MAX_TIMEOUT_SECONDS = 1 # Only run once. |
| candidate = self.manager.GetLatestCandidate() |
| self.assertEqual(candidate, None) |
| self.mox.VerifyAll() |
| |
| def _CreateManifest(self): |
| """Returns a created test manifest in tmpdir with its dir_pfx.""" |
| self.manager.current_version = '1.2.4-rc21' |
| dir_pfx = CHROME_BRANCH |
| manifest = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, 'buildspecs', |
| dir_pfx, '1.2.4-rc21.xml') |
| manifest_version_unittest.TouchFile(manifest) |
| return manifest, dir_pfx |
| |
| @staticmethod |
| def _FinishBuilderHelper(manifest, path_for_builder, dir_pfx, status, wait=0): |
| time.sleep(wait) |
| manifest_version.CreateSymlink( |
| manifest, os.path.join(path_for_builder, status, dir_pfx, |
| os.path.basename(manifest))) |
| |
| def _FinishBuild(self, manifest, path_for_builder, dir_pfx, status, wait=0): |
| """Finishes a build by marking a status with optional delay.""" |
| if wait > 0: |
| thread = threading.Thread(target=self._FinishBuilderHelper, |
| args=(manifest, path_for_builder, dir_pfx, |
| status, wait)) |
| thread.start() |
| return thread |
| else: |
| self._FinishBuilderHelper(manifest, path_for_builder, dir_pfx, status, 0) |
| |
| def testGetBuildersStatusBothFinished(self): |
| """Tests GetBuilderStatus where both builds have finished.""" |
| fake_version_file = LKGMCandidateInfoTest.CreateFakeVersionFile(self.tmpdir) |
| self.mox.StubOutWithMock(lkgm_manager, '_SyncGitRepo') |
| |
| manifest, dir_pfx = self._CreateManifest() |
| print manifest, dir_pfx |
| for_build1 = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, |
| 'build-name', 'build1') |
| for_build2 = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, |
| 'build-name', 'build2') |
| |
| self._FinishBuild(manifest, for_build1, dir_pfx, 'fail') |
| self._FinishBuild(manifest, for_build2, dir_pfx, 'pass') |
| |
| lkgm_manager._SyncGitRepo(self.manager._TMP_MANIFEST_DIR) |
| self.mox.ReplayAll() |
| statuses = self.manager.GetBuildersStatus(['build1', 'build2'], |
| fake_version_file) |
| self.assertEqual(statuses['build1'], 'fail') |
| self.assertEqual(statuses['build2'], 'pass') |
| self.mox.VerifyAll() |
| |
| def testGetBuildersStatusWaitForOne(self): |
| """Tests GetBuilderStatus where both builds have finished with one delay.""" |
| fake_version_file = LKGMCandidateInfoTest.CreateFakeVersionFile(self.tmpdir) |
| self.mox.StubOutWithMock(lkgm_manager, '_SyncGitRepo') |
| |
| manifest, dir_pfx = self._CreateManifest() |
| for_build1 = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, |
| 'build-name', 'build1') |
| for_build2 = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, |
| 'build-name', 'build2') |
| |
| self._FinishBuild(manifest, for_build1, dir_pfx, 'fail') |
| self._FinishBuild(manifest, for_build2, dir_pfx, 'pass', wait=3) |
| |
| lkgm_manager._SyncGitRepo(self.manager._TMP_MANIFEST_DIR).MultipleTimes() |
| self.mox.ReplayAll() |
| |
| statuses = self.manager.GetBuildersStatus(['build1', 'build2'], |
| fake_version_file) |
| self.assertEqual(statuses['build1'], 'fail') |
| self.assertEqual(statuses['build2'], 'pass') |
| self.mox.VerifyAll() |
| |
| def testGetBuildersStatusReachTimeout(self): |
| """Tests GetBuilderStatus where one build finishes and one never does.""" |
| fake_version_file = LKGMCandidateInfoTest.CreateFakeVersionFile(self.tmpdir) |
| self.mox.StubOutWithMock(lkgm_manager, '_SyncGitRepo') |
| |
| manifest, dir_pfx = self._CreateManifest() |
| for_build1 = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, |
| 'build-name', 'build1') |
| for_build2 = os.path.join(self.manager._TMP_MANIFEST_DIR, |
| self.manager.lkgm_subdir, |
| 'build-name', 'build2') |
| |
| self._FinishBuild(manifest, for_build1, dir_pfx, 'fail', wait=3) |
| thread = self._FinishBuild(manifest, for_build2, dir_pfx, 'pass', wait=10) |
| |
| lkgm_manager._SyncGitRepo(self.manager._TMP_MANIFEST_DIR).MultipleTimes() |
| self.mox.ReplayAll() |
| # Let's reduce this. |
| self.manager.LONG_MAX_TIMEOUT_SECONDS = 5 |
| statuses = self.manager.GetBuildersStatus(['build1', 'build2'], |
| fake_version_file) |
| self.assertEqual(statuses['build1'], 'fail') |
| self.assertEqual(statuses['build2'], None) |
| thread.join() |
| self.mox.VerifyAll() |
| |
| def testGenerateBlameListSinceLKGM(self): |
| """Tests that we can generate a blamelist from one commit message.""" |
| fake_git_log = """Author: Sammy Sosa <fake@fake.com> |
| |
| Date: Mon Aug 8 14:52:06 2011 -0700 |
| |
| Add in a test for cbuildbot |
| |
| TEST=So much testing |
| BUG=chromium-os:99999 |
| |
| Change-Id: Ib72a742fd2cee3c4a5223b8easwasdgsdgfasdf |
| Reviewed-on: http://gerrit.chromium.org/gerrit/1234 |
| Reviewed-by: Fake person <fake@fake.org> |
| Tested-by: Sammy Sosa <fake@fake.com> |
| """ |
| fake_revision = '1234567890' |
| fake_project_handler = self.mox.CreateMock(cros_lib.ManifestHandler) |
| fake_project_handler.projects = { 'fake/repo': { 'name': 'fake/repo', |
| 'path': 'fake/path', |
| 'revision': fake_revision, |
| } |
| } |
| fake_result = self.mox.CreateMock(cros_lib.CommandResult) |
| fake_result.output = fake_git_log |
| |
| self.mox.StubOutWithMock(cros_lib.ManifestHandler, 'ParseManifest') |
| |
| cros_lib.ManifestHandler.ParseManifest( |
| self.tmpmandir + '/LKGM/lkgm.xml').AndReturn(fake_project_handler) |
| self.mox.ReplayAll() |
| self.manager.GenerateBlameListSinceLKGM() |
| self.mox.VerifyAll() |
| |
| def testAddPatchesToManifest(self): |
| """Tests whether we can add a fake patch to an empty manifest file. |
| |
| This test creates an empty xml file with just manifest/ tag in it then |
| runs the AddPatchesToManifest with one mocked out GerritPatch and ensures |
| the newly generated manifest has the correct patch information afterwards. |
| """ |
| tmp_manifest = tempfile.mktemp('manifest') |
| try: |
| # Create fake but empty manifest file. |
| new_doc = minidom.getDOMImplementation().createDocument(None, 'manifest', |
| None) |
| with open(tmp_manifest, 'w+') as manifest_file: |
| print new_doc.toxml() |
| new_doc.writexml(manifest_file) |
| |
| gerrit_patch = self.mox.CreateMock(patch.GerritPatch) |
| gerrit_patch.project = 'chromite/tacos' |
| gerrit_patch.id = '1234567890' |
| gerrit_patch.commit = '0987654321' |
| self.manager.AddPatchesToManifest(tmp_manifest, [gerrit_patch]) |
| |
| new_doc = minidom.parse(tmp_manifest) |
| element = new_doc.getElementsByTagName( |
| lkgm_manager.PALADIN_COMMIT_ELEMENT)[0] |
| self.assertEqual(element.getAttribute( |
| lkgm_manager.PALADIN_CHANGE_ID_ATTR), '1234567890') |
| self.assertEqual(element.getAttribute( |
| lkgm_manager.PALADIN_COMMIT_ATTR), '0987654321') |
| self.assertEqual(element.getAttribute(lkgm_manager.PALADIN_PROJECT_ATTR), |
| 'chromite/tacos') |
| finally: |
| os.remove(tmp_manifest) |
| |
| |
| def tearDown(self): |
| if os.path.exists(self.tmpdir): shutil.rmtree(self.tmpdir) |
| shutil.rmtree(self.tmpmandir) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
