| # Copyright 2012 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unittests for lkgm_manager""" |
| |
| import contextlib |
| import os |
| import tempfile |
| from unittest import mock |
| from xml.dom import minidom |
| |
| from chromite.cbuildbot import cbuildbot_alerts |
| from chromite.cbuildbot import lkgm_manager |
| from chromite.cbuildbot import manifest_version |
| from chromite.cbuildbot import repository |
| from chromite.lib import config_lib |
| from chromite.lib import constants |
| 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.buildstore import FakeBuildStore |
| |
| |
| 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 |
| """ |
| |
| |
| # pylint: disable=protected-access |
| |
| |
| class LKGMCandidateInfoTest(cros_test_lib.TestCase): |
| """Test methods testing methods in _LKGMCandidateInfo class.""" |
| |
| 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.""" |
| info0 = lkgm_manager._LKGMCandidateInfo("5.2.3-rc100") |
| info1 = lkgm_manager._LKGMCandidateInfo("1.2.3-rc1") |
| info2 = lkgm_manager._LKGMCandidateInfo("1.2.3-rc2") |
| info3 = lkgm_manager._LKGMCandidateInfo("1.2.200-rc1") |
| info4 = lkgm_manager._LKGMCandidateInfo("1.4.3-rc1") |
| |
| self.assertGreater(info0, info1) |
| self.assertGreater(info0, info2) |
| self.assertGreater(info0, info3) |
| self.assertGreater(info0, info4) |
| self.assertGreater(info2, info1) |
| self.assertGreater(info3, info1) |
| self.assertGreater(info3, info2) |
| self.assertGreater(info4, info1) |
| self.assertGreater(info4, info2) |
| self.assertGreater(info4, info3) |
| self.assertEqual(info0, info0) |
| self.assertEqual(info1, info1) |
| self.assertEqual(info2, info2) |
| self.assertEqual(info3, info3) |
| self.assertEqual(info4, info4) |
| self.assertNotEqual(info0, info1) |
| self.assertNotEqual(info0, info2) |
| self.assertNotEqual(info0, info3) |
| self.assertNotEqual(info0, info4) |
| self.assertNotEqual(info1, info0) |
| self.assertNotEqual(info1, info2) |
| self.assertNotEqual(info1, info3) |
| self.assertNotEqual(info1, info4) |
| self.assertNotEqual(info2, info0) |
| self.assertNotEqual(info2, info1) |
| self.assertNotEqual(info2, info3) |
| self.assertNotEqual(info2, info4) |
| self.assertNotEqual(info3, info0) |
| self.assertNotEqual(info3, info1) |
| self.assertNotEqual(info3, info2) |
| self.assertNotEqual(info3, info4) |
| self.assertNotEqual(info4, info0) |
| self.assertNotEqual(info4, info1) |
| self.assertNotEqual(info4, info1) |
| self.assertNotEqual(info4, info3) |
| |
| |
| @contextlib.contextmanager |
| def TemporaryManifest(): |
| with tempfile.NamedTemporaryFile(mode="w") as f: |
| # Create fake but empty manifest file. |
| new_doc = minidom.getDOMImplementation().createDocument( |
| None, "manifest", None |
| ) |
| print(new_doc.toxml()) |
| new_doc.writexml(f) |
| f.flush() |
| yield f |
| |
| |
| class LKGMManagerTest(cros_test_lib.MockTempDirTestCase): |
| """Tests for the BuildSpecs manager.""" |
| |
| def setUp(self): |
| self.manager = None |
| self.push_mock = self.PatchObject(git, "CreatePushBranch") |
| |
| self.source_repo = "ssh://source/repo" |
| self.manifest_repo = "ssh://manifest/repo" |
| self.version_file = "version-file.sh" |
| self.branch = "master" |
| self.build_name = "amd64-generic" |
| self.incr_type = "branch" |
| self.buildstore = FakeBuildStore() |
| |
| # Create tmp subdirs based on the one provided TempDirMixin. |
| self.tmpdir = os.path.join(self.tempdir, "base") |
| osutils.SafeMakedirs(self.tmpdir) |
| self.tmpmandir = os.path.join(self.tempdir, "man") |
| osutils.SafeMakedirs(self.tmpmandir) |
| |
| def _LKGMManager(self, build_type=constants.PFQ_TYPE, config=None): |
| repo = repository.RepoRepository( |
| self.source_repo, self.tmpdir, self.branch, depth=1 |
| ) |
| manager = lkgm_manager.LKGMManager( |
| repo, |
| self.manifest_repo, |
| self.build_name, |
| build_type, |
| "branch", |
| force=False, |
| branch=self.branch, |
| config=config, |
| buildstore=self.buildstore, |
| dry_run=True, |
| ) |
| manager.manifest_dir = self.tmpmandir |
| manager.lkgm_path = os.path.join( |
| self.tmpmandir, constants.LKGM_MANIFEST |
| ) |
| |
| manager.all_specs_dir = "/LKGM/path" |
| manifest_dir = manager.manifest_dir |
| manager.specs_for_builder = os.path.join( |
| manifest_dir, manager.rel_working_dir, "build-name", "%(builder)s" |
| ) |
| manager.SLEEP_TIMEOUT = 0 |
| return manager |
| |
| def _GetPathToManifest(self, info): |
| return os.path.join( |
| self.manager.all_specs_dir, "%s.xml" % info.VersionString() |
| ) |
| |
| def testCreateFromManifest(self): |
| """Tests that we can create a new candidate from another manifest.""" |
| self.manager = self._LKGMManager() |
| # Let's stub out other LKGMManager calls cause they're already |
| # unit tested. |
| |
| version = "2010.0.0-rc7" |
| my_info = lkgm_manager._LKGMCandidateInfo("2010.0.0") |
| new_candidate = lkgm_manager._LKGMCandidateInfo(version) |
| manifest = ( |
| "/tmp/manifest-versions-internal/paladin/buildspecs/" |
| "20/%s.xml" % version |
| ) |
| new_manifest = "/path/to/tmp/file.xml" |
| |
| build_id = 20162 |
| |
| site_params = config_lib.GetSiteParams() |
| # Patch out our RepoRepository to make sure we don't corrupt real repo. |
| self.PatchObject(self.manager, "cros_source") |
| filter_mock = self.PatchObject( |
| manifest_version, "FilterManifest", return_value=new_manifest |
| ) |
| |
| # Do manifest refresh work. |
| self.PatchObject( |
| lkgm_manager.LKGMManager, |
| "GetCurrentVersionInfo", |
| return_value=my_info, |
| ) |
| self.PatchObject(lkgm_manager.LKGMManager, "RefreshManifestCheckout") |
| init_mock = self.PatchObject( |
| lkgm_manager.LKGMManager, "InitializeManifestVariables" |
| ) |
| |
| # Publish new candidate. |
| publish_mock = self.PatchObject( |
| lkgm_manager.LKGMManager, "PublishManifest" |
| ) |
| |
| candidate_path = self.manager.CreateFromManifest( |
| manifest, build_id=build_id |
| ) |
| self.assertEqual(candidate_path, self._GetPathToManifest(new_candidate)) |
| self.assertEqual(self.manager.current_version, version) |
| |
| filter_mock.assert_called_once_with( |
| manifest, whitelisted_remotes=site_params.EXTERNAL_REMOTES |
| ) |
| publish_mock.assert_called_once_with( |
| new_manifest, version, build_id=build_id |
| ) |
| init_mock.assert_called_once_with(my_info) |
| self.push_mock.assert_called_once_with(mock.ANY, mock.ANY, sync=False) |
| |
| def testCreateNewCandidateReturnNoneIfNoWorkToDo(self): |
| """Tests that we return nothing if there is nothing to create.""" |
| self.manager = self._LKGMManager() |
| new_manifest = "some_manifest" |
| my_info = lkgm_manager._LKGMCandidateInfo("1.2.3") |
| |
| # Patch out our RepoRepository to make sure we don't corrupt real repo. |
| cros_source_mock = self.PatchObject(self.manager, "cros_source") |
| cros_source_mock.branch = "master" |
| cros_source_mock.directory = "/foo/repo" |
| |
| self.PatchObject(lkgm_manager.LKGMManager, "CheckoutSourceCode") |
| self.PatchObject( |
| lkgm_manager.LKGMManager, |
| "CreateManifest", |
| return_value=new_manifest, |
| ) |
| self.PatchObject(lkgm_manager.LKGMManager, "RefreshManifestCheckout") |
| self.PatchObject( |
| lkgm_manager.LKGMManager, |
| "GetCurrentVersionInfo", |
| return_value=my_info, |
| ) |
| init_mock = self.PatchObject( |
| lkgm_manager.LKGMManager, "InitializeManifestVariables" |
| ) |
| self.PatchObject( |
| lkgm_manager.LKGMManager, "HasCheckoutBeenBuilt", return_value=True |
| ) |
| |
| candidate = self.manager.CreateNewCandidate() |
| self.assertEqual(candidate, None) |
| init_mock.assert_called_once_with(my_info) |
| |
| 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.manifest_dir, |
| self.manager.rel_working_dir, |
| "buildspecs", |
| dir_pfx, |
| "1.2.4-rc21.xml", |
| ) |
| osutils.Touch(manifest) |
| return manifest, dir_pfx |
| |
| def _MockParseGitLog(self, fake_git_log, project): |
| exists_mock = self.PatchObject(os.path, "exists", return_value=True) |
| link_mock = self.PatchObject(cbuildbot_alerts, "PrintBuildbotLink") |
| fake_project_handler = mock.Mock(spec=git.Manifest) |
| fake_project_handler.checkouts_by_path = {project["path"]: project} |
| self.PatchObject(git, "Manifest", return_value=fake_project_handler) |
| |
| fake_result = cros_build_lib.CompletedProcess(stdout=fake_git_log) |
| self.PatchObject(git, "RunGit", return_value=fake_result) |
| |
| return exists_mock, link_mock |
| |
| def testAddChromeVersionToManifest(self): |
| """Tests whether we can write the chrome version to the manifest file.""" |
| self.manager = self._LKGMManager() |
| with TemporaryManifest() as f: |
| chrome_version = "35.0.1863.0" |
| # Write the chrome element to manifest. |
| self.manager._AddChromeVersionToManifest(f.name, chrome_version) |
| |
| # Read the manifest file. |
| new_doc = minidom.parse(f.name) |
| elements = new_doc.getElementsByTagName(lkgm_manager.CHROME_ELEMENT) |
| self.assertEqual(len(elements), 1) |
| self.assertEqual( |
| elements[0].getAttribute(lkgm_manager.CHROME_VERSION_ATTR), |
| chrome_version, |
| ) |
