| #!/usr/bin/python |
| # Copyright (c) 2013 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. |
| |
| """Unit tests for upload_command_stats_unittest.py.""" |
| |
| from __future__ import print_function |
| |
| import os |
| import sys |
| |
| sys.path.insert(0, os.path.abspath('%s/../..' % os.path.dirname(__file__))) |
| from chromite.cbuildbot import constants |
| from chromite.lib import cros_build_lib |
| from chromite.lib import cros_test_lib |
| from chromite.lib import osutils |
| from chromite.lib import stats |
| from chromite.lib import stats_unittest |
| from chromite.scripts import upload_command_stats |
| |
| TEST_FILE = """\ |
| Chromium OS Build Command Stats - Version 1 |
| cpu_count 32 |
| cmd_args --board=lumpy |
| host typewriter.mtv.corp.google.com |
| run_time 0 |
| cmd_line ./build_packages --board=lumpy |
| username monkey@chromium.org |
| cmd_base build_packages |
| cpu_type Intel(R) Xeon(R) CPU E5-2690 0 @ 2.90GHz |
| board lumpy |
| """ |
| |
| class RunScriptTest(cros_test_lib.MockTempDirTestCase, |
| cros_test_lib.LoggingTestCase): |
| """Test the main functionality.""" |
| # pylint: disable=W0212 |
| |
| def setUp(self): |
| self.upload_file = os.path.join(self.tempdir, 'upload_File') |
| osutils.WriteFile(self.upload_file, TEST_FILE) |
| self.argv = [self.upload_file] |
| self.PatchObject(cros_build_lib, 'GetHostDomain', autospec=True, |
| return_value='noname.com') |
| self.StartPatcher(stats_unittest.StatsUploaderMock()) |
| |
| def testNormalRun(self): |
| """Going for code coverage.""" |
| upload_command_stats.main(self.argv) |
| self.assertEquals(stats.StatsUploader._Upload.call_count, 1) |
| |
| def testStatsDebugMsg(self, golo=False): |
| """We hide debug messages from stats module when not in golo.""" |
| stats.StatsUploader._Upload.side_effect = EnvironmentError() |
| with cros_test_lib.LoggingCapturer() as logs: |
| upload_command_stats.main(self.argv) |
| self.AssertLogsContain( |
| logs, stats.StatsUploader.ENVIRONMENT_ERROR, inverted=(not golo)) |
| |
| def testGoloRun(self): |
| """Test when running in the golo.""" |
| cros_build_lib.GetHostDomain.return_value = constants.GOLO_DOMAIN |
| upload_command_stats.main(self.argv) |
| self.assertEquals(stats.StatsUploader._Upload.call_count, 1) |
| self.testStatsDebugMsg(golo=True) |
| |
| def LogContainsOnError(self, msg): |
| """Verifies a logging.error() message is printed.""" |
| with cros_test_lib.LoggingCapturer() as logs: |
| self.assertRaises2(SystemExit, upload_command_stats.main, self.argv, |
| check_attrs={'code': 1}) |
| self.AssertLogsContain(logs, msg) |
| |
| def testLoadFileErrorIgnore(self): |
| """We don't propagate timeouts during upload.""" |
| self.PatchObject( |
| upload_command_stats.StatsLoader, 'LoadFile', |
| side_effect=upload_command_stats.LoadError(), autospec=True) |
| self.LogContainsOnError( |
| upload_command_stats.FILE_LOAD_ERROR % self.upload_file) |
| |
| def testUploadErrorIgnore(self): |
| """We don't propagate timeouts during upload.""" |
| stats.StatsUploader._Upload.side_effect = Exception() |
| # Logging level for the error is logging.ERROR. |
| self.LogContainsOnError(upload_command_stats.UNCAUGHT_ERROR) |
| |
| |
| if __name__ == '__main__': |
| cros_test_lib.main() |