src/third_party/chromite/cbuildbot/remote_try_unittest.py

   1 #!/usr/bin/python
   2
   3 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
   4 # Use of this source code is governed by a BSD-style license that can be
   5 # found in the LICENSE file.
   6
   7 """Unittests for remote_try.py."""
   8
   9 import json
  10 import mox
  11 import os
  12 import sys
  13 import shutil
  14 import time
  15
  16 import constants
  17 sys.path.insert(0, constants.SOURCE_ROOT)
  18 from chromite.lib import cros_build_lib
  19 from chromite.lib import cros_test_lib
  20 from chromite.lib import git
  21 from chromite.cbuildbot import remote_try
  22 from chromite.cbuildbot import repository
  23 from chromite.scripts import cbuildbot
  24
  25 class RemoteTryJobMock(remote_try.RemoteTryJob):
  26   """Helper for Mocking out a RemoteTryJob."""
  27
  28 # pylint: disable=W0212,R0904,E1101
  29 class RemoteTryTests(cros_test_lib.MoxTempDirTestCase):
  30   """Test cases related to remote try jobs."""
  31
  32   PATCHES = ('5555', '6666')
  33   BOTS = ('x86-generic-paladin', 'arm-generic-paladin')
  34
  35   def setUp(self):
  36     self.parser = cbuildbot._CreateParser()
  37     args = ['-r', '/tmp/test_build1', '-g', '5555', '-g',
  38             '6666', '--remote']
  39     args.extend(self.BOTS)
  40     self.options, args = cbuildbot._ParseCommandLine(self.parser, args)
  41     self.options.cache_dir = self.tempdir
  42     self.checkout_dir = os.path.join(self.tempdir, 'test_checkout')
  43     self.int_mirror, self.ext_mirror = None, None
  44
  45   def _RunGitSingleOutput(self, cwd, cmd):
  46     result = git.RunGit(cwd, cmd)
  47     out_lines = result.output.split()
  48     self.assertEqual(len(out_lines), 1)
  49     return out_lines[0]
  50
  51   def _GetNewestFile(self, dirname, basehash):
  52     newhash = git.GetGitRepoRevision(dirname)
  53     self.assertNotEqual(basehash, newhash)
  54     cmd = ['log', '--format=%H', '%s..' % basehash]
  55     # Make sure we have a single commit.
  56     self._RunGitSingleOutput(dirname, cmd)
  57     cmd = ['diff', '--name-only', 'HEAD^']
  58     # Make sure only one file per commit.
  59     return self._RunGitSingleOutput(dirname, cmd)
  60
  61   def _SubmitJob(self, checkout_dir, job, version=None):
  62     """Returns the path to the tryjob description."""
  63     self.assertTrue(isinstance(job, RemoteTryJobMock))
  64     basehash = git.GetGitRepoRevision(job.repo_url)
  65     if version is not None:
  66       self._SetMirrorVersion(version)
  67     job.Submit(workdir=checkout_dir, dryrun=True)
  68     # Get the file that was just created.
  69     created_file = self._GetNewestFile(checkout_dir, basehash)
  70     return os.path.join(checkout_dir, created_file)
  71
  72   def _SetupMirrors(self):
  73     mirror = os.path.join(self.tempdir, 'tryjobs_mirror')
  74     os.mkdir(mirror)
  75     url = '%s/%s' % (constants.EXTERNAL_GOB_URL, 'chromiumos/tryjobs')
  76     repository.CloneGitRepo(mirror, url,
  77                             bare=True)
  78     self.ext_mirror = mirror
  79     mirror = os.path.join(self.tempdir, 'tryjobs_int_mirror')
  80     os.mkdir(mirror)
  81     repository.CloneGitRepo(mirror, self.ext_mirror, reference=self.ext_mirror,
  82                             bare=True)
  83
  84     self.int_mirror = mirror
  85     RemoteTryJobMock.EXTERNAL_URL = self.ext_mirror
  86     RemoteTryJobMock.INTERNAL_URL = self.int_mirror
  87     self._SetMirrorVersion(remote_try.RemoteTryJob.TRYJOB_FORMAT_VERSION, True)
  88
  89   def _SetMirrorVersion(self, version, only_if_missing=False):
  90     for path in (self.ext_mirror, self.int_mirror):
  91       vpath = os.path.join(path, remote_try.RemoteTryJob.TRYJOB_FORMAT_FILE)
  92       if os.path.exists(vpath) and only_if_missing:
  93         continue
  94       # Get ourselves a working dir.
  95       tmp_repo = os.path.join(self.tempdir, 'tmp-repo')
  96       git.RunGit(self.tempdir, ['clone', path, tmp_repo])
  97       vpath = os.path.join(tmp_repo, remote_try.RemoteTryJob.TRYJOB_FORMAT_FILE)
  98       with open(vpath, 'w') as f:
  99         f.write(str(version))
 100       git.RunGit(tmp_repo, ['add', vpath])
 101       git.RunGit(tmp_repo, ['commit', '-m', 'setting version to %s' % version])
 102       git.RunGit(tmp_repo, ['push', path, 'master:master'])
 103       shutil.rmtree(tmp_repo)
 104
 105   def _CreateJob(self, mirror=True):
 106     job_class = remote_try.RemoteTryJob
 107     if mirror:
 108       job_class = RemoteTryJobMock
 109       self._SetupMirrors()
 110
 111     job = job_class(self.options, self.BOTS, [])
 112     return job
 113
 114   def testJobTimestamp(self):
 115     """Verify jobs have unique names."""
 116     def submit_helper(dirname):
 117       work_dir = os.path.join(self.tempdir, dirname)
 118       return os.path.basename(self._SubmitJob(work_dir, job))
 119
 120     self.mox.StubOutWithMock(repository, 'IsARepoRoot')
 121     repository.IsARepoRoot(mox.IgnoreArg()).AndReturn(False)
 122     self.mox.ReplayAll()
 123     job = self._CreateJob()
 124
 125     file1 = submit_helper('test1')
 126     # Tryjob file names are based on timestamp, so delay one second to avoid two
 127     # jobfiles having the same name.
 128     time.sleep(1)
 129     file2 = submit_helper('test2')
 130     self.assertNotEqual(file1, file2)
 131
 132   def testSimpleTryJob(self, version=None):
 133     """Test that a tryjob spec file is created and pushed properly."""
 134     self.mox.StubOutWithMock(repository, 'IsARepoRoot')
 135     repository.IsARepoRoot(mox.IgnoreArg()).AndReturn(True)
 136     self.mox.StubOutWithMock(repository, 'IsInternalRepoCheckout')
 137     repository.IsInternalRepoCheckout(mox.IgnoreArg()).AndReturn(False)
 138
 139     self.mox.ReplayAll()
 140     try:
 141       os.environ["GIT_AUTHOR_EMAIL"] = "Elmer Fudd <efudd@google.com>"
 142       os.environ["GIT_COMMITTER_EMAIL"] = "Elmer Fudd <efudd@google.com>"
 143       job = self._CreateJob()
 144     finally:
 145       os.environ.pop("GIT_AUTHOR_EMAIL", None)
 146       os.environ.pop("GIT_COMMITTER_EMAIL", None)
 147     created_file = self._SubmitJob(self.checkout_dir, job, version=version)
 148     with open(created_file, 'rb') as job_desc_file:
 149       values = json.load(job_desc_file)
 150
 151     self.assertTrue('efudd@google.com' in values['email'][0])
 152
 153     for patch in self.PATCHES:
 154       self.assertTrue(patch in values['extra_args'],
 155                       msg="expected patch %s in args %s" %
 156                           (patch, values['extra_args']))
 157
 158     self.assertTrue(set(self.BOTS).issubset(values['bot']))
 159
 160     remote_url = cros_build_lib.RunCommand(
 161         ['git', 'config', 'remote.origin.url'], redirect_stdout=True,
 162         cwd=self.checkout_dir).output.strip()
 163     self.assertEqual(remote_url, self.ext_mirror)
 164
 165   def testClientVersionAwareness(self):
 166     self.assertRaises(
 167         remote_try.ChromiteUpgradeNeeded,
 168         self.testSimpleTryJob,
 169         version=remote_try.RemoteTryJob.TRYJOB_FORMAT_VERSION + 1)
 170
 171   def testInternalTryJob(self):
 172     """Verify internal tryjobs are pushed properly."""
 173     self.mox.StubOutWithMock(repository, 'IsARepoRoot')
 174     repository.IsARepoRoot(mox.IgnoreArg()).AndReturn(True)
 175     self.mox.StubOutWithMock(repository, 'IsInternalRepoCheckout')
 176     repository.IsInternalRepoCheckout(mox.IgnoreArg()).AndReturn(True)
 177
 178     self.mox.ReplayAll()
 179     job = self._CreateJob()
 180     self._SubmitJob(self.checkout_dir, job)
 181
 182     remote_url = cros_build_lib.RunCommand(
 183         ['git', 'config', 'remote.origin.url'], redirect_stdout=True,
 184         cwd=self.checkout_dir).output.strip()
 185     self.assertEqual(remote_url, self.int_mirror)
 186
 187   def testBareTryJob(self):
 188     """Verify submitting a tryjob from just a chromite checkout works."""
 189     self.mox.StubOutWithMock(repository, 'IsARepoRoot')
 190     repository.IsARepoRoot(mox.IgnoreArg()).AndReturn(False)
 191     self.mox.StubOutWithMock(repository, 'IsInternalRepoCheckout')
 192
 193     self.mox.ReplayAll()
 194     job = self._CreateJob(mirror=False)
 195     self.assertEqual(job.repo_url, remote_try.RemoteTryJob.EXTERNAL_URL)
 196
 197
 198 if __name__ == '__main__':
 199   cros_test_lib.main()