src/third_party/chromite/cbuildbot/archive_lib_unittest.py

   1 #!/usr/bin/python
   2
   3 # Copyright (c) 2014 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 """Test the archive_lib module."""
   8
   9 import logging
  10 import os
  11 import sys
  12
  13 sys.path.insert(0, os.path.abspath('%s/../..' % os.path.dirname(__file__)))
  14 from chromite.cbuildbot import archive_lib
  15 from chromite.cbuildbot import cbuildbot_config
  16 from chromite.cbuildbot import cbuildbot_run
  17 from chromite.lib import cros_test_lib
  18
  19 import mock
  20
  21 DEFAULT_ARCHIVE_PREFIX = 'bogus_bucket/TheArchiveBase'
  22 DEFAULT_ARCHIVE_BASE = 'gs://%s' % DEFAULT_ARCHIVE_PREFIX
  23 DEFAULT_BUILDROOT = '/tmp/foo/bar/buildroot'
  24 DEFAULT_BUILDNUMBER = 12345
  25 DEFAULT_BRANCH = 'TheBranch'
  26 DEFAULT_CHROME_BRANCH = 'TheChromeBranch'
  27 DEFAULT_VERSION_STRING = 'TheVersionString'
  28 DEFAULT_BOARD = 'TheBoard'
  29 DEFAULT_BOT_NAME = 'TheCoolBot'
  30
  31 # Access to protected member.
  32 # pylint: disable=W0212
  33
  34 DEFAULT_OPTIONS = cros_test_lib.EasyAttr(
  35     archive_base=DEFAULT_ARCHIVE_BASE,
  36     buildroot=DEFAULT_BUILDROOT,
  37     buildnumber=DEFAULT_BUILDNUMBER,
  38     buildbot=True,
  39     branch=DEFAULT_BRANCH,
  40     remote_trybot=False,
  41     debug=False,
  42 )
  43 DEFAULT_CONFIG = cbuildbot_config._config(
  44     name=DEFAULT_BOT_NAME,
  45     master=True,
  46     boards=[DEFAULT_BOARD],
  47     child_configs=[cbuildbot_config._config(name='foo'),
  48                    cbuildbot_config._config(name='bar'),
  49                   ],
  50 )
  51
  52
  53 def _ExtendDefaultOptions(**kwargs):
  54   """Extend DEFAULT_OPTIONS with keys/values in kwargs."""
  55   options_kwargs = DEFAULT_OPTIONS.copy()
  56   options_kwargs.update(kwargs)
  57   return cros_test_lib.EasyAttr(**options_kwargs)
  58
  59
  60 def _ExtendDefaultConfig(**kwargs):
  61   """Extend DEFAULT_CONFIG with keys/values in kwargs."""
  62   config_kwargs = DEFAULT_CONFIG.copy()
  63   config_kwargs.update(kwargs)
  64   return cbuildbot_config._config(**config_kwargs)
  65
  66
  67 def _NewBuilderRun(options=None, config=None):
  68   """Create a BuilderRun objection from options and config values.
  69
  70   Args:
  71     options: Specify options or default to DEFAULT_OPTIONS.
  72     config: Specify build config or default to DEFAULT_CONFIG.
  73
  74   Returns:
  75     BuilderRun object.
  76   """
  77   # Make up a fake object with a Queue() method.
  78   class _FakeMultiprocessManager(object):
  79     """This just needs to not crash when various methods are called."""
  80     def Queue(self):
  81       return 'SomeQueue'
  82     def RLock(self):
  83       return 'SomeLock'
  84     def dict(self):
  85       return {}
  86     def list(self):
  87       return []
  88
  89   manager = _FakeMultiprocessManager()
  90
  91   options = options or DEFAULT_OPTIONS
  92   config = config or DEFAULT_CONFIG
  93   return cbuildbot_run.BuilderRun(options, config, manager)
  94
  95
  96 class GetBaseUploadURITest(cros_test_lib.TestCase):
  97   """Test the GetBaseUploadURI function."""
  98
  99   ARCHIVE_BASE = '/tmp/the/archive/base'
 100   BOT_ID = 'TheNewBotId'
 101
 102   def setUp(self):
 103     self.cfg = DEFAULT_CONFIG
 104
 105   def _GetBaseUploadURI(self, *args, **kwargs):
 106     """Test GetBaseUploadURI with archive_base and no bot_id."""
 107     return archive_lib.GetBaseUploadURI(self.cfg, *args, **kwargs)
 108
 109   def testArchiveBaseRemoteTrybotFalse(self):
 110     expected_result = '%s/%s' % (self.ARCHIVE_BASE, DEFAULT_BOT_NAME)
 111     result = self._GetBaseUploadURI(archive_base=self.ARCHIVE_BASE,
 112                                     remote_trybot=False)
 113     self.assertEqual(expected_result, result)
 114
 115   def testArchiveBaseRemoteTrybotTrue(self):
 116     expected_result = '%s/trybot-%s' % (self.ARCHIVE_BASE, DEFAULT_BOT_NAME)
 117     result = self._GetBaseUploadURI(archive_base=self.ARCHIVE_BASE,
 118                                     remote_trybot=True)
 119     self.assertEqual(expected_result, result)
 120
 121   def testArchiveBaseBotIdRemoteTrybotFalse(self):
 122     expected_result = '%s/%s' % (self.ARCHIVE_BASE, self.BOT_ID)
 123     result = self._GetBaseUploadURI(archive_base=self.ARCHIVE_BASE,
 124                                     bot_id=self.BOT_ID, remote_trybot=False)
 125     self.assertEqual(expected_result, result)
 126
 127   def testArchiveBaseBotIdRemoteTrybotTrue(self):
 128     expected_result = '%s/%s' % (self.ARCHIVE_BASE, self.BOT_ID)
 129     result = self._GetBaseUploadURI(archive_base=self.ARCHIVE_BASE,
 130                                     bot_id=self.BOT_ID, remote_trybot=True)
 131     self.assertEqual(expected_result, result)
 132
 133   def testRemoteTrybotTrue(self):
 134     """Test GetBaseUploadURI with no archive base but remote_trybot is True."""
 135     expected_result = ('%s/trybot-%s' %
 136                        (archive_lib.constants.DEFAULT_ARCHIVE_BUCKET,
 137                         DEFAULT_BOT_NAME))
 138     result = self._GetBaseUploadURI(remote_trybot=True)
 139     self.assertEqual(expected_result, result)
 140
 141   def testBotIdRemoteTrybotTrue(self):
 142     expected_result = ('%s/%s' %
 143                        (archive_lib.constants.DEFAULT_ARCHIVE_BUCKET,
 144                         self.BOT_ID))
 145     result = self._GetBaseUploadURI(bot_id=self.BOT_ID, remote_trybot=True)
 146     self.assertEqual(expected_result, result)
 147
 148   def testDefaultGSPathRemoteTrybotFalse(self):
 149     """Test GetBaseUploadURI with default gs_path value in config."""
 150     self.cfg = _ExtendDefaultConfig(gs_path=cbuildbot_config.GS_PATH_DEFAULT)
 151
 152     # Test without bot_id.
 153     expected_result = ('%s/%s' %
 154                        (archive_lib.constants.DEFAULT_ARCHIVE_BUCKET,
 155                         DEFAULT_BOT_NAME))
 156     result = self._GetBaseUploadURI(remote_trybot=False)
 157     self.assertEqual(expected_result, result)
 158
 159     # Test with bot_id.
 160     expected_result = ('%s/%s' %
 161                        (archive_lib.constants.DEFAULT_ARCHIVE_BUCKET,
 162                         self.BOT_ID))
 163     result = self._GetBaseUploadURI(bot_id=self.BOT_ID, remote_trybot=False)
 164     self.assertEqual(expected_result, result)
 165
 166   def testOverrideGSPath(self):
 167     """Test GetBaseUploadURI with default gs_path value in config."""
 168     self.cfg = _ExtendDefaultConfig(gs_path='gs://funkytown/foo/bar')
 169
 170     # Test without bot_id.
 171     expected_result = self.cfg.gs_path
 172     result = self._GetBaseUploadURI(remote_trybot=False)
 173     self.assertEqual(expected_result, result)
 174
 175     # Test with bot_id.
 176     expected_result = self.cfg.gs_path
 177     result = self._GetBaseUploadURI(bot_id=self.BOT_ID, remote_trybot=False)
 178     self.assertEqual(expected_result, result)
 179
 180
 181 class ArchiveTest(cros_test_lib.TestCase):
 182   """Test the Archive class."""
 183   _VERSION = '6543.2.1'
 184
 185   def _GetAttributeValue(self, attr, options=None, config=None):
 186     with mock.patch.object(cbuildbot_run._BuilderRunBase, 'GetVersion') as m:
 187       m.return_value = self._VERSION
 188
 189       run = _NewBuilderRun(options, config)
 190       return getattr(run.GetArchive(), attr)
 191
 192   def testVersion(self):
 193     value = self._GetAttributeValue('version')
 194     self.assertEqual(self._VERSION, value)
 195
 196   def testVersionNotReady(self):
 197     run = _NewBuilderRun()
 198     self.assertRaises(AttributeError, getattr, run, 'version')
 199
 200   def testArchivePathTrybot(self):
 201     options = _ExtendDefaultOptions(buildbot=False)
 202     value = self._GetAttributeValue('archive_path', options=options)
 203     expected_value = ('%s/%s/%s/%s' %
 204                       (DEFAULT_BUILDROOT,
 205                        archive_lib.Archive._TRYBOT_ARCHIVE,
 206                        DEFAULT_BOT_NAME,
 207                        self._VERSION))
 208     self.assertEqual(expected_value, value)
 209
 210   def testArchivePathBuildbot(self):
 211     value = self._GetAttributeValue('archive_path')
 212     expected_value = ('%s/%s/%s/%s' %
 213                       (DEFAULT_BUILDROOT,
 214                        archive_lib.Archive._BUILDBOT_ARCHIVE,
 215                        DEFAULT_BOT_NAME,
 216                        self._VERSION))
 217     self.assertEqual(expected_value, value)
 218
 219   def testUploadUri(self):
 220     value = self._GetAttributeValue('upload_url')
 221     expected_value = '%s/%s/%s' % (DEFAULT_ARCHIVE_BASE,
 222                                    DEFAULT_BOT_NAME,
 223                                    self._VERSION)
 224     self.assertEqual(expected_value, value)
 225
 226   def testDownloadURLBuildbot(self):
 227     value = self._GetAttributeValue('download_url')
 228     expected_value = ('%s%s/%s/%s' %
 229                       (archive_lib.gs.PRIVATE_BASE_HTTPS_URL,
 230                        DEFAULT_ARCHIVE_PREFIX,
 231                        DEFAULT_BOT_NAME,
 232                        self._VERSION))
 233     self.assertEqual(expected_value, value)
 234
 235
 236
 237
 238 if __name__ == '__main__':
 239   cros_test_lib.main(level=logging.DEBUG)