1#!/usr/bin/python 2# Copyright (c) 2014 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 6import datetime as datetime_base 7from datetime import datetime 8import mock 9import time 10import unittest 11 12import common 13 14from autotest_lib.server.cros.dynamic_suite import constants 15from autotest_lib.site_utils import run_suite 16from autotest_lib.site_utils import diagnosis_utils 17 18 19class ResultCollectorUnittest(unittest.TestCase): 20 """Runsuite unittest""" 21 22 JOB_MAX_RUNTIME_MINS = 10 23 24 def setUp(self): 25 """Set up test.""" 26 self.afe = mock.MagicMock() 27 self.tko = mock.MagicMock() 28 29 30 def _build_view(self, test_idx, test_name, subdir, status, afe_job_id, 31 job_name='fake_job_name', reason='fake reason', 32 job_keyvals=None, test_started_time=None, 33 test_finished_time=None, invalidates_test_idx=None, 34 job_started_time=None, job_finished_time=None): 35 """Build a test view using the given fields. 36 37 @param test_idx: An integer representing test_idx. 38 @param test_name: A string, e.g. 'dummy_Pass' 39 @param subdir: A string representing the subdir field of the test view. 40 e.g. 'dummy_Pass'. 41 @param status: A string representing the test status. 42 e.g. 'FAIL', 'PASS' 43 @param afe_job_id: An integer representing the afe job id. 44 @param job_name: A string representing the job name. 45 @param reason: A string representing the reason field of the test view. 46 @param job_keyvals: A dictionary stroing the job keyvals. 47 @param test_started_time: A string, e.g. '2014-04-12 12:35:33' 48 @param test_finished_time: A string, e.g. '2014-04-12 12:35:33' 49 @param invalidates_test_idx: An integer, representing the idx of the 50 test that has been retried. 51 @param job_started_time: A string, e.g. '2014-04-12 12:35:33' 52 @param job_finished_time: A string, e.g. '2014-04-12 12:35:33' 53 54 @reutrn: A dictionary representing a test view. 55 56 """ 57 if job_keyvals is None: 58 job_keyvals = {} 59 return {'test_idx': test_idx, 'test_name': test_name, 'subdir':subdir, 60 'status': status, 'afe_job_id': afe_job_id, 61 'job_name': job_name, 'reason': reason, 62 'job_keyvals': job_keyvals, 63 'test_started_time': test_started_time, 64 'test_finished_time': test_finished_time, 65 'invalidates_test_idx': invalidates_test_idx, 66 'job_started_time': job_started_time, 67 'job_finished_time': job_finished_time} 68 69 70 def _mock_tko_get_detailed_test_views(self, test_views): 71 """Mock tko method get_detailed_test_views call. 72 73 @param test_views: A list of test views that will be returned 74 by get_detailed_test_views. 75 """ 76 return_values = {} 77 for v in test_views: 78 views_of_job = return_values.setdefault( 79 ('get_detailed_test_views', v['afe_job_id']), []) 80 views_of_job.append(v) 81 82 def side_effect(*args, **kwargs): 83 """Maps args and kwargs to the mocked return values.""" 84 key = (kwargs['call'], kwargs['afe_job_id']) 85 return return_values[key] 86 87 self.tko.run = mock.MagicMock(side_effect=side_effect) 88 89 90 def _mock_afe_get_jobs(self, suite_job_id, child_job_ids): 91 """Mock afe get_jobs call. 92 93 @param suite_job_id: The afe job id of the suite job. 94 @param child_job_ids: A list of job ids of the child jobs. 95 96 """ 97 suite_job = mock.MagicMock() 98 suite_job.id = suite_job_id 99 suite_job.max_runtime_mins = 10 100 suite_job.parent_job = None 101 102 return_values = {suite_job_id: []} 103 for job_id in child_job_ids: 104 new_job = mock.MagicMock() 105 new_job.id = job_id 106 new_job.max_runtime_mins = self.JOB_MAX_RUNTIME_MINS 107 new_job.parent_job = suite_job 108 return_values[suite_job_id].append(new_job) 109 110 def side_effect(*args, **kwargs): 111 """Maps args and kwargs to the mocked return values.""" 112 if kwargs.get('id') == suite_job_id: 113 return [suite_job] 114 return return_values[kwargs['parent_job_id']] 115 116 self.afe.get_jobs = mock.MagicMock(side_effect=side_effect) 117 118 119 def testFetchSuiteTestView(self): 120 """Test that it fetches the correct suite test views.""" 121 suite_job_id = 100 122 suite_name = 'dummy' 123 build = 'R23-1.1.1.1' 124 server_job_view = self._build_view( 125 10, 'SERVER_JOB', '----', 'GOOD', suite_job_id) 126 test_to_ignore = self._build_view( 127 11, 'dummy_Pass', '101-user/host/dummy_Pass', 128 'GOOD', suite_job_id) 129 test_to_include = self._build_view( 130 12, 'dummy_Pass.bluetooth', None, 'TEST_NA', suite_job_id) 131 self._mock_afe_get_jobs(suite_job_id, []) 132 self._mock_tko_get_detailed_test_views( 133 [server_job_view, test_to_ignore, test_to_include]) 134 collector = run_suite.ResultCollector( 135 'fake_server', self.afe, self.tko, 136 build='fake/build', board='fake', suite_name='dummy', 137 suite_job_id=suite_job_id) 138 suite_views = collector._fetch_relevant_test_views_of_suite() 139 suite_views = sorted(suite_views, key=lambda view: view['test_idx']) 140 # Verify that SERVER_JOB is renamed to 'Suite Prep' 141 self.assertEqual(suite_views[0].get_testname(), 142 run_suite.TestView.SUITE_PREP) 143 # Verify that the test with a subidr is not included. 144 self.assertEqual(suite_views[0]['test_idx'], 10) 145 self.assertEqual(suite_views[1]['test_idx'], 12) 146 147 148 def testFetchTestViewOfChildJobs(self): 149 """Test that it fetches the correct child test views.""" 150 build = 'lumpy-release/R36-5788.0.0' 151 board = 'lumpy' 152 suite_name = 'my_suite' 153 suite_job_id = 100 154 invalid_job_id = 101 155 invalid_job_name = '%s/%s/test_Pass' % (build, suite_name) 156 good_job_id = 102 157 good_job_name = '%s/%s/test_Pass' % (build, suite_name) 158 bad_job_id = 103 159 bad_job_name = '%s/%s/test_ServerJobFail' % (build, suite_name) 160 161 invalid_test = self._build_view( 162 19, 'test_Pass_Old', 'fake/subdir', 163 'FAIL', invalid_job_id, invalid_job_name) 164 good_job_server_job = self._build_view( 165 20, 'SERVER_JOB', '----', 'GOOD', good_job_id, good_job_name) 166 good_job_test = self._build_view( 167 21, 'test_Pass', 'fake/subdir', 'GOOD', 168 good_job_id, good_job_name, 169 job_keyvals={'retry_original_job_id': invalid_job_id}) 170 bad_job_server_job = self._build_view( 171 22, 'SERVER_JOB', '----', 'FAIL', bad_job_id, bad_job_name) 172 bad_job_test = self._build_view( 173 23, 'test_ServerJobFail', 'fake/subdir', 'GOOD', 174 bad_job_id, bad_job_name) 175 self._mock_tko_get_detailed_test_views( 176 [good_job_server_job, good_job_test, 177 bad_job_server_job, bad_job_test, invalid_test]) 178 self._mock_afe_get_jobs(suite_job_id, [good_job_id, bad_job_id]) 179 collector = run_suite.ResultCollector( 180 'fake_server', self.afe, self.tko, 181 build, board, suite_name, suite_job_id) 182 child_views, retry_counts = collector._fetch_test_views_of_child_jobs() 183 # child_views should contain tests 21, 22, 23 184 child_views = sorted(child_views, key=lambda view: view['test_idx']) 185 # Verify that the SERVER_JOB has been renamed properly 186 self.assertEqual(child_views[1].get_testname(), 187 'test_ServerJobFail_SERVER_JOB') 188 # Verify that failed SERVER_JOB and actual invalid tests are included, 189 expected = [good_job_test['test_idx'], bad_job_server_job['test_idx'], 190 bad_job_test['test_idx']] 191 child_view_ids = [v['test_idx'] for v in child_views] 192 self.assertEqual(child_view_ids, expected) 193 self.afe.get_jobs.assert_called_once_with( 194 parent_job_id=suite_job_id) 195 # Verify the retry_counts is calculated correctly 196 self.assertEqual(len(retry_counts), 1) 197 self.assertEqual(retry_counts[21], 1) 198 199 200 def testGenerateLinks(self): 201 """Test that it generates correct web and buildbot links.""" 202 suite_job_id = 100 203 suite_name = 'my_suite' 204 build = 'lumpy-release/R36-5788.0.0' 205 board = 'lumpy' 206 fake_job = mock.MagicMock() 207 fake_job.parent = suite_job_id 208 suite_job_view = run_suite.TestView( 209 self._build_view( 210 20, 'Suite prep', '----', 'GOOD', suite_job_id), 211 fake_job, suite_name, build, 'chromeos-test') 212 good_test = run_suite.TestView( 213 self._build_view( 214 21, 'test_Pass', 'fake/subdir', 'GOOD', 101), 215 fake_job, suite_name, build, 'chromeos-test') 216 bad_test = run_suite.TestView( 217 self._build_view( 218 23, 'test_Fail', 'fake/subdir', 'FAIL', 102), 219 fake_job, suite_name, build, 'chromeos-test') 220 221 collector = run_suite.ResultCollector( 222 'fake_server', self.afe, self.tko, 223 build, board, suite_name, suite_job_id, user='chromeos-test') 224 collector._suite_views = [suite_job_view] 225 collector._test_views = [suite_job_view, good_test, bad_test] 226 collector._max_testname_width = max( 227 [len(v.get_testname()) for v in collector._test_views]) + 3 228 collector._generate_web_and_buildbot_links() 229 URL_PATTERN = run_suite.LogLink._URL_PATTERN 230 # expected_web_links is list of (anchor, url) tuples we 231 # are expecting. 232 expected_web_links = [ 233 (v.get_testname().ljust(collector._max_testname_width), 234 URL_PATTERN % ('fake_server', 235 '%s-%s' % (v['afe_job_id'], 'chromeos-test'))) 236 for v in collector._test_views] 237 # Verify web links are generated correctly. 238 for i in range(len(collector._web_links)): 239 expect = expected_web_links[i] 240 self.assertEqual(collector._web_links[i].anchor, expect[0]) 241 self.assertEqual(collector._web_links[i].url, expect[1]) 242 243 expected_buildbot_links = [ 244 (v.get_testname().ljust(collector._max_testname_width), 245 URL_PATTERN % ('fake_server', 246 '%s-%s' % (v['afe_job_id'], 'chromeos-test'))) 247 for v in collector._test_views if v['status'] != 'GOOD'] 248 # Verify buildbot links are generated correctly. 249 for i in range(len(collector._buildbot_links)): 250 expect = expected_buildbot_links[i] 251 self.assertEqual(collector._buildbot_links[i].anchor, expect[0]) 252 self.assertEqual(collector._buildbot_links[i].url, expect[1]) 253 self.assertEqual(collector._buildbot_links[i].retry_count, 0) 254 # Assert that a wmatrix retry dashboard link is created. 255 self.assertNotEqual( 256 collector._buildbot_links[i].GenerateWmatrixRetryLink(),'') 257 258 259 def _end_to_end_test_helper( 260 self, include_bad_test=False, include_warn_test=False, 261 include_experimental_bad_test=False, include_timeout_test=False, 262 include_self_aborted_test=False, 263 include_aborted_by_suite_test=False, 264 include_good_retry=False, include_bad_retry=False, 265 suite_job_timed_out=False, suite_job_status='GOOD'): 266 """A helper method for testing ResultCollector end-to-end. 267 268 This method mocks the retrieving of required test views, 269 and call ResultCollector.run() to collect the results. 270 271 @param include_bad_test: 272 If True, include a view of a test which has status 'FAIL'. 273 @param include_warn_test: 274 If True, include a view of a test which has status 'WARN' 275 @param include_experimental_bad_test: 276 If True, include a view of an experimental test 277 which has status 'FAIL'. 278 @param include_timeout_test: 279 If True, include a view of a test which was aborted before 280 started. 281 @param include_self_aborted_test: 282 If True, include a view of test which was aborted after 283 started and hit hits own timeout. 284 @param include_self_aborted_by_suite_test: 285 If True, include a view of test which was aborted after 286 started but has not hit its own timeout. 287 @param include_good_retry: 288 If True, include a test that passed after retry. 289 @param include_bad_retry: 290 If True, include a test that failed after retry. 291 @param suite_job_status: One of 'GOOD' 'FAIL' 'ABORT' 'RUNNING' 292 293 @returns: A ResultCollector instance. 294 """ 295 suite_job_id = 100 296 good_job_id = 101 297 bad_job_id = 102 298 warn_job_id = 102 299 experimental_bad_job_id = 102 300 timeout_job_id = 100 301 self_aborted_job_id = 104 302 aborted_by_suite_job_id = 105 303 good_retry_job_id = 106 304 bad_retry_job_id = 107 305 invalid_job_id_1 = 90 306 invalid_job_id_2 = 91 307 suite_name = 'dummy' 308 build = 'lumpy-release/R27-3888.0.0' 309 suite_job_keyvals = { 310 constants.DOWNLOAD_STARTED_TIME: '2014-04-29 13:14:20', 311 constants.PAYLOAD_FINISHED_TIME: '2014-04-29 13:14:25', 312 constants.ARTIFACT_FINISHED_TIME: '2014-04-29 13:14:30'} 313 314 suite_job_started_time = '2014-04-29 13:14:37' 315 if suite_job_timed_out: 316 suite_job_keyvals['aborted_by'] = 'test_user' 317 suite_job_finished_time = '2014-04-29 13:25:37' 318 suite_job_status = 'ABORT' 319 else: 320 suite_job_finished_time = '2014-04-29 13:23:37' 321 322 server_job_view = self._build_view( 323 10, 'SERVER_JOB', '----', suite_job_status, suite_job_id, 324 'lumpy-release/R27-3888.0.0-test_suites/control.dummy', 325 '', suite_job_keyvals, '2014-04-29 13:14:37', 326 '2014-04-29 13:20:27', job_started_time=suite_job_started_time, 327 job_finished_time=suite_job_finished_time) 328 good_test = self._build_view( 329 11, 'dummy_Pass', '101-user/host/dummy_Pass', 'GOOD', 330 good_job_id, 'lumpy-release/R27-3888.0.0/dummy/dummy_Pass', 331 '', {}, '2014-04-29 13:15:35', '2014-04-29 13:15:36') 332 bad_test = self._build_view( 333 12, 'dummy_Fail.Fail', '102-user/host/dummy_Fail.Fail', 'FAIL', 334 bad_job_id, 'lumpy-release/R27-3888.0.0/dummy/dummy_Fail.Fail', 335 'always fail', {}, '2014-04-29 13:16:00', 336 '2014-04-29 13:16:02') 337 warn_test = self._build_view( 338 13, 'dummy_Fail.Warn', '102-user/host/dummy_Fail.Warn', 'WARN', 339 warn_job_id, 'lumpy-release/R27-3888.0.0/dummy/dummy_Fail.Warn', 340 'always warn', {}, '2014-04-29 13:16:00', 341 '2014-04-29 13:16:02') 342 experimental_bad_test = self._build_view( 343 14, 'experimental_dummy_Fail.Fail', 344 '102-user/host/dummy_Fail.Fail', 'FAIL', 345 experimental_bad_job_id, 346 'lumpy-release/R27-3888.0.0/dummy/experimental_dummy_Fail.Fail', 347 'always fail', {'experimental': 'True'}, '2014-04-29 13:16:06', 348 '2014-04-29 13:16:07') 349 timeout_test = self._build_view( 350 15, 'dummy_Timeout', '', 'ABORT', 351 timeout_job_id, 352 'lumpy-release/R27-3888.0.0/dummy/dummy_Timeout', 353 'child job did not run', {}, '2014-04-29 13:15:37', 354 '2014-04-29 13:15:38') 355 self_aborted_test = self._build_view( 356 16, 'dummy_Abort', '104-user/host/dummy_Abort', 'ABORT', 357 self_aborted_job_id, 358 'lumpy-release/R27-3888.0.0/dummy/dummy_Abort', 359 'child job aborted', {'aborted_by': 'test_user'}, 360 '2014-04-29 13:15:39', '2014-04-29 13:15:40', 361 job_started_time='2014-04-29 13:15:39', 362 job_finished_time='2014-04-29 13:25:40') 363 aborted_by_suite = self._build_view( 364 17, 'dummy_AbortBySuite', '105-user/host/dummy_AbortBySuite', 365 'RUNNING', aborted_by_suite_job_id, 366 'lumpy-release/R27-3888.0.0/dummy/dummy_Abort', 367 'aborted by suite', {'aborted_by': 'test_user'}, 368 '2014-04-29 13:15:39', '2014-04-29 13:15:40', 369 job_started_time='2014-04-29 13:15:39', 370 job_finished_time='2014-04-29 13:15:40') 371 good_retry = self._build_view( 372 18, 'dummy_RetryPass', '106-user/host/dummy_RetryPass', 'GOOD', 373 good_retry_job_id, 374 'lumpy-release/R27-3888.0.0/dummy/dummy_RetryPass', 375 '', {'retry_original_job_id': invalid_job_id_1}, 376 '2014-04-29 13:15:37', 377 '2014-04-29 13:15:38', invalidates_test_idx=1) 378 bad_retry = self._build_view( 379 19, 'dummy_RetryFail', '107-user/host/dummy_RetryFail', 'FAIL', 380 bad_retry_job_id, 381 'lumpy-release/R27-3888.0.0/dummy/dummy_RetryFail', 382 'retry failed', {'retry_original_job_id': invalid_job_id_2}, 383 '2014-04-29 13:15:39', '2014-04-29 13:15:40', 384 invalidates_test_idx=2) 385 invalid_test_1 = self._build_view( 386 1, 'dummy_RetryPass', '90-user/host/dummy_RetryPass', 'GOOD', 387 invalid_job_id_1, 388 'lumpy-release/R27-3888.0.0/dummy/dummy_RetryPass', 389 'original test failed', {}, '2014-04-29 13:10:00', 390 '2014-04-29 13:10:01') 391 invalid_test_2 = self._build_view( 392 2, 'dummy_RetryFail', '91-user/host/dummy_RetryFail', 'FAIL', 393 invalid_job_id_2, 394 'lumpy-release/R27-3888.0.0/dummy/dummy_RetryFail', 395 'original test failed', {}, 396 '2014-04-29 13:10:03', '2014-04-29 13:10:04') 397 398 test_views = [server_job_view, good_test] 399 child_jobs = set([good_job_id]) 400 if include_bad_test: 401 test_views.append(bad_test) 402 child_jobs.add(bad_job_id) 403 if include_warn_test: 404 test_views.append(warn_test) 405 child_jobs.add(warn_job_id) 406 if include_experimental_bad_test: 407 test_views.append(experimental_bad_test) 408 child_jobs.add(experimental_bad_job_id) 409 if include_timeout_test: 410 test_views.append(timeout_test) 411 if include_self_aborted_test: 412 test_views.append(self_aborted_test) 413 child_jobs.add(self_aborted_job_id) 414 if include_good_retry: 415 test_views.extend([good_retry, invalid_test_1]) 416 child_jobs.add(good_retry_job_id) 417 if include_bad_retry: 418 test_views.extend([bad_retry, invalid_test_2]) 419 child_jobs.add(bad_retry_job_id) 420 if include_aborted_by_suite_test: 421 test_views.append(aborted_by_suite) 422 child_jobs.add(aborted_by_suite_job_id) 423 self._mock_tko_get_detailed_test_views(test_views) 424 self._mock_afe_get_jobs(suite_job_id, child_jobs) 425 collector = run_suite.ResultCollector( 426 'fake_server', self.afe, self.tko, 427 'lumpy-release/R36-5788.0.0', 'lumpy', 'dummy', suite_job_id) 428 collector.run() 429 return collector 430 431 432 def testEndToEndSuitePass(self): 433 """Test it returns code OK when all test pass.""" 434 collector = self._end_to_end_test_helper() 435 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.OK) 436 437 438 def testEndToEndExperimentalTestFails(self): 439 """Test that it returns code OK when only experimental test fails.""" 440 collector = self._end_to_end_test_helper( 441 include_experimental_bad_test=True) 442 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.OK) 443 444 445 def testEndToEndSuiteWarn(self): 446 """Test it returns code WARNING when there is a test that warns.""" 447 collector = self._end_to_end_test_helper(include_warn_test=True) 448 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.WARNING) 449 450 451 def testEndToEndSuiteFail(self): 452 """Test it returns code ERROR when there is a test that fails.""" 453 # Test that it returns ERROR when there is test that fails. 454 collector = self._end_to_end_test_helper(include_bad_test=True) 455 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.ERROR) 456 457 # Test that it returns ERROR when both experimental and non-experimental 458 # test fail. 459 collector = self._end_to_end_test_helper( 460 include_bad_test=True, include_warn_test=True, 461 include_experimental_bad_test=True) 462 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.ERROR) 463 464 collector = self._end_to_end_test_helper(include_self_aborted_test=True) 465 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.ERROR) 466 467 468 def testEndToEndSuiteJobFail(self): 469 """Test it returns code SUITE_FAILURE when only the suite job failed.""" 470 collector = self._end_to_end_test_helper(suite_job_status='ABORT') 471 self.assertEqual( 472 collector.return_code, run_suite.RETURN_CODES.INFRA_FAILURE) 473 474 collector = self._end_to_end_test_helper(suite_job_status='ERROR') 475 self.assertEqual( 476 collector.return_code, run_suite.RETURN_CODES.INFRA_FAILURE) 477 478 479 def testEndToEndRetry(self): 480 """Test it returns correct code when a test was retried.""" 481 collector = self._end_to_end_test_helper(include_good_retry=True) 482 self.assertEqual( 483 collector.return_code, run_suite.RETURN_CODES.WARNING) 484 485 collector = self._end_to_end_test_helper(include_good_retry=True, 486 include_self_aborted_test=True) 487 self.assertEqual( 488 collector.return_code, run_suite.RETURN_CODES.ERROR) 489 490 collector = self._end_to_end_test_helper(include_good_retry=True, 491 include_bad_test=True) 492 self.assertEqual( 493 collector.return_code, run_suite.RETURN_CODES.ERROR) 494 495 collector = self._end_to_end_test_helper(include_bad_retry=True) 496 self.assertEqual( 497 collector.return_code, run_suite.RETURN_CODES.ERROR) 498 499 500 def testEndToEndSuiteTimeout(self): 501 """Test it returns correct code when a child job timed out.""" 502 # a child job timed out before started, none failed. 503 collector = self._end_to_end_test_helper(include_timeout_test=True) 504 self.assertEqual( 505 collector.return_code, run_suite.RETURN_CODES.SUITE_TIMEOUT) 506 507 # a child job timed out before started, and one test failed. 508 collector = self._end_to_end_test_helper( 509 include_bad_test=True, include_timeout_test=True) 510 self.assertEqual(collector.return_code, run_suite.RETURN_CODES.ERROR) 511 512 # a child job timed out before started, and one test warned. 513 collector = self._end_to_end_test_helper( 514 include_warn_test=True, include_timeout_test=True) 515 self.assertEqual(collector.return_code, 516 run_suite.RETURN_CODES.SUITE_TIMEOUT) 517 518 # a child job timed out before started, and one test was retried. 519 collector = self._end_to_end_test_helper(include_good_retry=True, 520 include_timeout_test=True) 521 self.assertEqual( 522 collector.return_code, run_suite.RETURN_CODES.SUITE_TIMEOUT) 523 524 # a child jot was aborted because suite timed out. 525 collector = self._end_to_end_test_helper( 526 include_aborted_by_suite_test=True) 527 self.assertEqual( 528 collector.return_code, run_suite.RETURN_CODES.OK) 529 530 # suite job timed out. 531 collector = self._end_to_end_test_helper(suite_job_timed_out=True) 532 self.assertEqual( 533 collector.return_code, run_suite.RETURN_CODES.SUITE_TIMEOUT) 534 535 536class SimpleTimerUnittests(unittest.TestCase): 537 """Test the simple timer.""" 538 539 def testPoll(self): 540 """Test polling the timer.""" 541 interval_hours = 0.0001 542 t = diagnosis_utils.SimpleTimer(interval_hours=interval_hours) 543 deadline = t.deadline 544 self.assertTrue(deadline is not None and 545 t.interval_hours == interval_hours) 546 min_deadline = (datetime.now() + 547 datetime_base.timedelta(hours=interval_hours)) 548 time.sleep(interval_hours * 3600) 549 self.assertTrue(t.poll()) 550 self.assertTrue(t.deadline >= min_deadline) 551 552 553 def testBadInterval(self): 554 """Test a bad interval.""" 555 t = diagnosis_utils.SimpleTimer(interval_hours=-1) 556 self.assertTrue(t.deadline is None and t.poll() == False) 557 t._reset() 558 self.assertTrue(t.deadline is None and t.poll() == False) 559 560 561if __name__ == '__main__': 562 unittest.main() 563