src/third_party/chromite/scripts/upload_command_stats_unittest.py

   1 #!/usr/bin/python
   2 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """Unit tests for upload_command_stats_unittest.py."""
   7
   8 import os
   9 import sys
  10
  11 sys.path.insert(0, os.path.abspath('%s/../..' % os.path.dirname(__file__)))
  12 from chromite.buildbot import constants
  13 from chromite.lib import cros_build_lib
  14 from chromite.lib import cros_test_lib
  15 from chromite.lib import osutils
  16 from chromite.lib import stats
  17 from chromite.lib import stats_unittest
  18 from chromite.scripts import upload_command_stats
  19
  20 TEST_FILE = """\
  21 Chromium OS Build Command Stats - Version 1
  22 cpu_count 32
  23 cmd_args --board=lumpy
  24 host typewriter.mtv.corp.google.com
  25 run_time 0
  26 cmd_line ./build_packages --board=lumpy
  27 username monkey@chromium.org
  28 cmd_base build_packages
  29 cpu_type Intel(R) Xeon(R) CPU E5-2690 0 @ 2.90GHz
  30 board lumpy
  31 """
  32
  33 class RunScriptTest(cros_test_lib.MockTempDirTestCase,
  34                     cros_test_lib.LoggingTestCase):
  35   """Test the main functionality."""
  36   # pylint: disable=W0212
  37
  38   def setUp(self):
  39     self.upload_file = os.path.join(self.tempdir, 'upload_File')
  40     osutils.WriteFile(self.upload_file, TEST_FILE)
  41     self.argv = [self.upload_file]
  42     self.PatchObject(cros_build_lib, 'GetHostDomain', autospec=True,
  43                      return_value='noname.com')
  44     self.StartPatcher(stats_unittest.StatsUploaderMock())
  45
  46   def testNormalRun(self):
  47     """Going for code coverage."""
  48     upload_command_stats.main(self.argv)
  49     self.assertEquals(stats.StatsUploader._Upload.call_count, 1)
  50
  51   def testStatsDebugMsg(self, golo=False):
  52     """We hide debug messages from stats module when not in golo."""
  53     stats.StatsUploader._Upload.side_effect = EnvironmentError()
  54     with cros_test_lib.LoggingCapturer() as logs:
  55       upload_command_stats.main(self.argv)
  56       self.AssertLogsContain(
  57           logs, stats.StatsUploader.ENVIRONMENT_ERROR, inverted=(not golo))
  58
  59   def testGoloRun(self):
  60     """Test when running in the golo."""
  61     cros_build_lib.GetHostDomain.return_value = constants.GOLO_DOMAIN
  62     upload_command_stats.main(self.argv)
  63     self.assertEquals(stats.StatsUploader._Upload.call_count, 1)
  64     self.testStatsDebugMsg(golo=True)
  65
  66   def LogContainsOnError(self, msg):
  67     """Verifies a logging.error() message is printed."""
  68     with cros_test_lib.LoggingCapturer() as logs:
  69       self.assertRaises2(SystemExit, upload_command_stats.main, self.argv,
  70                          check_attrs={'code': 1})
  71       self.AssertLogsContain(logs, msg)
  72
  73   def testLoadFileErrorIgnore(self):
  74     """We don't propagate timeouts during upload."""
  75     self.PatchObject(
  76         upload_command_stats.StatsLoader, 'LoadFile',
  77         side_effect=upload_command_stats.LoadError(), autospec=True)
  78     self.LogContainsOnError(
  79         upload_command_stats.FILE_LOAD_ERROR % self.upload_file)
  80
  81   def testUploadErrorIgnore(self):
  82     """We don't propagate timeouts during upload."""
  83     stats.StatsUploader._Upload.side_effect = Exception()
  84     # Logging level for the error is logging.ERROR.
  85     self.LogContainsOnError(upload_command_stats.UNCAUGHT_ERROR)
  86
  87
  88 if __name__ == '__main__':
  89   cros_test_lib.main()