#!/usr/bin/python
|
# Copyright (c) 2014 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.
|
|
import collections
|
import datetime as datetime_base
|
from datetime import datetime
|
import mock
|
import time
|
import unittest
|
|
import common
|
|
from autotest_lib.server.cros.dynamic_suite import constants
|
from autotest_lib.site_utils import run_suite
|
from autotest_lib.site_utils import run_suite_common
|
from autotest_lib.site_utils import diagnosis_utils
|
|
|
class ReturnResultUnittest(unittest.TestCase):
|
"""_ReturnResult tests."""
|
|
def setUp(self):
|
super(ReturnResultUnittest, self).setUp()
|
patcher = mock.patch.object(run_suite, '_RETURN_RESULTS',
|
collections.OrderedDict())
|
self.results = results = patcher.start()
|
self.addCleanup(patcher.stop)
|
results['small'] = run_suite._ReturnResult(0, 'small')
|
results['big'] = run_suite._ReturnResult(1, 'big')
|
|
patcher = mock.patch.object(run_suite, '_RETURN_RESULTS_LIST',
|
list(results.values()))
|
patcher.start()
|
self.addCleanup(patcher.stop)
|
|
def test_equal(self):
|
"""Test _ReturnResult equal."""
|
self.assertEqual(self.results['small'], self.results['small'])
|
|
def test_unequal(self):
|
"""Test _ReturnResult unequal."""
|
self.assertNotEqual(self.results['big'], self.results['small'])
|
|
def test_greater_than(self):
|
"""Test _ReturnResult greater than."""
|
self.assertGreater(self.results['big'], self.results['small'])
|
|
def test_bitwise_or(self):
|
"""Test _ReturnResult bitwise or."""
|
self.assertEqual(self.results['big'],
|
self.results['big'] | self.results['small'])
|
|
|
class ResultCollectorUnittest(unittest.TestCase):
|
"""Runsuite unittest"""
|
|
JOB_MAX_RUNTIME_MINS = 10
|
|
def setUp(self):
|
"""Set up test."""
|
self.afe = mock.MagicMock()
|
self.tko = mock.MagicMock()
|
|
|
def _build_view(self, test_idx, test_name, subdir, status, afe_job_id,
|
job_name='fake_job_name', reason='fake reason',
|
job_keyvals=None, test_started_time=None,
|
test_finished_time=None, invalidates_test_idx=None,
|
job_started_time=None, job_finished_time=None):
|
"""Build a test view using the given fields.
|
|
@param test_idx: An integer representing test_idx.
|
@param test_name: A string, e.g. 'dummy_Pass'
|
@param subdir: A string representing the subdir field of the test view.
|
e.g. 'dummy_Pass'.
|
@param status: A string representing the test status.
|
e.g. 'FAIL', 'PASS'
|
@param afe_job_id: An integer representing the afe job id.
|
@param job_name: A string representing the job name.
|
@param reason: A string representing the reason field of the test view.
|
@param job_keyvals: A dictionary stroing the job keyvals.
|
@param test_started_time: A string, e.g. '2014-04-12 12:35:33'
|
@param test_finished_time: A string, e.g. '2014-04-12 12:35:33'
|
@param invalidates_test_idx: An integer, representing the idx of the
|
test that has been retried.
|
@param job_started_time: A string, e.g. '2014-04-12 12:35:33'
|
@param job_finished_time: A string, e.g. '2014-04-12 12:35:33'
|
|
@reutrn: A dictionary representing a test view.
|
|
"""
|
if job_keyvals is None:
|
job_keyvals = {}
|
return {'test_idx': test_idx, 'test_name': test_name, 'subdir':subdir,
|
'status': status, 'afe_job_id': afe_job_id,
|
'job_name': job_name, 'reason': reason,
|
'job_keyvals': job_keyvals,
|
'test_started_time': test_started_time,
|
'test_finished_time': test_finished_time,
|
'invalidates_test_idx': invalidates_test_idx,
|
'job_started_time': job_started_time,
|
'job_finished_time': job_finished_time}
|
|
|
def _mock_tko_get_detailed_test_views(self, test_views,
|
missing_results=[]):
|
"""Mock tko method get_detailed_test_views call.
|
|
@param test_views: A list of test views that will be returned
|
by get_detailed_test_views.
|
"""
|
return_values = {}
|
for v in test_views:
|
views_of_job = return_values.setdefault(
|
('get_detailed_test_views', v['afe_job_id']), [])
|
views_of_job.append(v)
|
for job_id in missing_results:
|
views_of_job = return_values.setdefault(
|
('get_detailed_test_views', job_id), [])
|
|
def side_effect(*args, **kwargs):
|
"""Maps args and kwargs to the mocked return values."""
|
key = (kwargs['call'], kwargs['afe_job_id'])
|
return return_values[key]
|
|
self.tko.run = mock.MagicMock(side_effect=side_effect)
|
|
|
def _mock_afe_get_jobs(self, suite_job_id, child_job_ids):
|
"""Mock afe get_jobs call.
|
|
@param suite_job_id: The afe job id of the suite job.
|
@param child_job_ids: A list of job ids of the child jobs.
|
|
"""
|
suite_job = mock.MagicMock()
|
suite_job.id = suite_job_id
|
suite_job.max_runtime_mins = 10
|
suite_job.parent_job = None
|
|
return_values = {suite_job_id: []}
|
for job_id in child_job_ids:
|
new_job = mock.MagicMock()
|
new_job.id = job_id
|
new_job.name = 'test.%d' % job_id
|
new_job.max_runtime_mins = self.JOB_MAX_RUNTIME_MINS
|
new_job.parent_job = suite_job
|
return_values[suite_job_id].append(new_job)
|
|
def side_effect(*args, **kwargs):
|
"""Maps args and kwargs to the mocked return values."""
|
if kwargs.get('id') == suite_job_id:
|
return [suite_job]
|
return return_values[kwargs['parent_job_id']]
|
|
self.afe.get_jobs = mock.MagicMock(side_effect=side_effect)
|
|
|
def testFetchSuiteTestView(self):
|
"""Test that it fetches the correct suite test views."""
|
suite_job_id = 100
|
suite_name = 'dummy'
|
build = 'R23-1.1.1.1'
|
server_job_view = self._build_view(
|
10, 'SERVER_JOB', '----', 'GOOD', suite_job_id)
|
test_to_ignore = self._build_view(
|
11, 'dummy_Pass', '101-user/host/dummy_Pass',
|
'GOOD', suite_job_id)
|
test_to_include = self._build_view(
|
12, 'dummy_Pass.bluetooth', None, 'TEST_NA', suite_job_id)
|
test_missing = self._build_view(
|
13, 'dummy_Missing', None, 'ABORT', suite_job_id)
|
self._mock_afe_get_jobs(suite_job_id, [])
|
self._mock_tko_get_detailed_test_views(
|
[server_job_view, test_to_ignore, test_to_include,
|
test_missing])
|
collector = run_suite.ResultCollector(
|
'fake_server', self.afe, self.tko,
|
build=build, suite_name=suite_name,
|
suite_job_id=suite_job_id,
|
return_code_function=run_suite._ReturnCodeComputer())
|
collector._missing_results = {
|
test_missing['test_name']: [14, 15],
|
}
|
suite_views = collector._fetch_relevant_test_views_of_suite()
|
suite_views = sorted(suite_views, key=lambda view: view['test_idx'])
|
# Verify that SERVER_JOB is renamed to 'Suite job'
|
self.assertEqual(suite_views[0].get_testname(),
|
run_suite.TestView.SUITE_JOB)
|
# Verify that the test with a subidr is not included.
|
self.assertEqual(suite_views[0]['test_idx'], 10)
|
self.assertEqual(suite_views[1]['test_idx'], 12)
|
self.assertEqual(suite_views[1]['afe_job_id'], suite_job_id)
|
# Verify that the test with missing results had it's AFE job id
|
# replaced.
|
self.assertEqual(suite_views[2]['test_idx'], 13)
|
self.assertEqual(suite_views[2]['afe_job_id'], 14)
|
|
|
def testFetchTestViewOfChildJobs(self):
|
"""Test that it fetches the correct child test views."""
|
build = 'lumpy-release/R36-5788.0.0'
|
board = 'lumpy'
|
suite_name = 'my_suite'
|
suite_job_id = 100
|
invalid_job_id = 101
|
invalid_job_name = '%s/%s/test_Pass' % (build, suite_name)
|
good_job_id = 102
|
good_job_name = '%s/%s/test_Pass' % (build, suite_name)
|
bad_job_id = 103
|
bad_job_name = '%s/%s/test_ServerJobFail' % (build, suite_name)
|
missing_job_id = 104
|
|
invalid_test = self._build_view(
|
19, 'test_Pass_Old', 'fake/subdir',
|
'FAIL', invalid_job_id, invalid_job_name)
|
good_job_server_job = self._build_view(
|
20, 'SERVER_JOB', '----', 'GOOD', good_job_id, good_job_name)
|
good_job_test = self._build_view(
|
21, 'test_Pass', 'fake/subdir', 'GOOD',
|
good_job_id, good_job_name,
|
job_keyvals={'retry_original_job_id': invalid_job_id})
|
bad_job_server_job = self._build_view(
|
22, 'SERVER_JOB', '----', 'FAIL', bad_job_id, bad_job_name)
|
bad_job_test = self._build_view(
|
23, 'test_ServerJobFail', 'fake/subdir', 'GOOD',
|
bad_job_id, bad_job_name)
|
self._mock_tko_get_detailed_test_views(
|
[good_job_server_job, good_job_test,
|
bad_job_server_job, bad_job_test, invalid_test],
|
[missing_job_id])
|
self._mock_afe_get_jobs(suite_job_id,
|
[good_job_id, bad_job_id, missing_job_id])
|
collector = run_suite.ResultCollector(
|
'fake_server', self.afe, self.tko,
|
build, suite_name, suite_job_id,
|
return_code_function=run_suite._ReturnCodeComputer())
|
child_views, retry_counts, missing_results = (
|
collector._fetch_test_views_of_child_jobs())
|
# child_views should contain tests 21, 22, 23
|
child_views = sorted(child_views, key=lambda view: view['test_idx'])
|
# Verify that the SERVER_JOB has been renamed properly
|
self.assertEqual(child_views[1].get_testname(),
|
'test_ServerJobFail_SERVER_JOB')
|
self.assertEqual(missing_results, {'test.104': [104]})
|
# Verify that failed SERVER_JOB and actual invalid tests are included,
|
expected = [good_job_test['test_idx'], bad_job_server_job['test_idx'],
|
bad_job_test['test_idx']]
|
child_view_ids = [v['test_idx'] for v in child_views]
|
self.assertEqual(child_view_ids, expected)
|
self.afe.get_jobs.assert_called_once_with(
|
parent_job_id=suite_job_id)
|
# Verify the retry_counts is calculated correctly
|
self.assertEqual(len(retry_counts), 1)
|
self.assertEqual(retry_counts[21], 1)
|
|
|
def testGenerateLinks(self):
|
"""Test that it generates correct web and buildbot links."""
|
suite_job_id = 100
|
suite_name = 'my_suite'
|
build = 'lumpy-release/R36-5788.0.0'
|
board = 'lumpy'
|
fake_job = mock.MagicMock()
|
fake_job.parent = suite_job_id
|
test_sponge_url = 'http://test_url'
|
job_keyvals = {'sponge_url': test_sponge_url}
|
suite_job_view = run_suite.TestView(
|
self._build_view(
|
20, 'Suite job', '----', 'GOOD', suite_job_id,
|
job_keyvals=job_keyvals),
|
fake_job, suite_name, build, 'chromeos-test')
|
good_test = run_suite.TestView(
|
self._build_view(
|
21, 'test_Pass', 'fake/subdir', 'GOOD', 101,
|
job_keyvals=job_keyvals),
|
fake_job, suite_name, build, 'chromeos-test')
|
bad_test = run_suite.TestView(
|
self._build_view(
|
23, 'test_Fail', 'fake/subdir', 'FAIL', 102,
|
job_keyvals=job_keyvals),
|
fake_job, suite_name, build, 'chromeos-test')
|
|
collector = run_suite.ResultCollector(
|
'fake_server', self.afe, self.tko,
|
build, suite_name, suite_job_id, user='chromeos-test',
|
return_code_function=run_suite._ReturnCodeComputer())
|
collector._suite_views = [suite_job_view]
|
collector._test_views = [suite_job_view, good_test, bad_test]
|
collector._max_testname_width = max(
|
[len(v.get_testname()) for v in collector._test_views]) + 3
|
collector._generate_web_and_buildbot_links()
|
URL_PATTERN = run_suite._URL_PATTERN
|
# expected_web_links is list of (anchor, url) tuples we
|
# are expecting.
|
expected_web_links = [
|
(v.get_testname(),
|
URL_PATTERN % ('http://fake_server',
|
'%s-%s' % (v['afe_job_id'], 'chromeos-test')),
|
test_sponge_url)
|
for v in collector._test_views]
|
# Verify web links are generated correctly.
|
for i in range(len(collector._web_links)):
|
expect = expected_web_links[i]
|
self.assertEqual(collector._web_links[i].anchor, expect[0])
|
self.assertEqual(collector._web_links[i].url, expect[1])
|
self.assertEqual(collector._web_links[i].sponge_url, expect[2])
|
|
expected_buildbot_links = [
|
(v.get_testname(),
|
URL_PATTERN % ('http://fake_server',
|
'%s-%s' % (v['afe_job_id'], 'chromeos-test')))
|
for v in collector._test_views if v['status'] != 'GOOD']
|
# Verify buildbot links are generated correctly.
|
for i in range(len(collector.buildbot_links)):
|
expect = expected_buildbot_links[i]
|
self.assertEqual(collector.buildbot_links[i].anchor, expect[0])
|
self.assertEqual(collector.buildbot_links[i].url, expect[1])
|
self.assertEqual(collector.buildbot_links[i].retry_count, 0)
|
# Assert that a retry dashboard link is created.
|
self.assertNotEqual(
|
collector.buildbot_links[i].GenerateRetryLink(), '')
|
self.assertNotEqual(
|
collector.buildbot_links[i].GenerateHistoryLink(), '')
|
|
|
def _end_to_end_test_helper(
|
self, include_bad_test=False, include_warn_test=False,
|
include_timeout_test=False,
|
include_self_aborted_test=False,
|
include_aborted_by_suite_test=False,
|
include_good_retry=False, include_bad_retry=False,
|
include_good_test=True,
|
suite_job_timed_out=False, suite_job_status='GOOD'):
|
"""A helper method for testing ResultCollector end-to-end.
|
|
This method mocks the retrieving of required test views,
|
and call ResultCollector.run() to collect the results.
|
|
@param include_bad_test:
|
If True, include a view of a test which has status 'FAIL'.
|
@param include_warn_test:
|
If True, include a view of a test which has status 'WARN'
|
@param include_timeout_test:
|
If True, include a view of a test which was aborted before
|
started.
|
@param include_self_aborted_test:
|
If True, include a view of test which was aborted after
|
started and hit hits own timeout.
|
@param include_self_aborted_by_suite_test:
|
If True, include a view of test which was aborted after
|
started but has not hit its own timeout.
|
@param include_good_retry:
|
If True, include a test that passed after retry.
|
@param include_bad_retry:
|
If True, include a test that failed after retry.
|
@param include_good_test:
|
If True, include a test that passed. If False, pretend no tests
|
(including the parent suite job) came back with any test
|
results.
|
@param suite_job_status: One of 'GOOD' 'FAIL' 'ABORT' 'RUNNING'
|
|
@returns: A ResultCollector instance.
|
"""
|
suite_job_id = 100
|
good_job_id = 101
|
bad_job_id = 102
|
warn_job_id = 102
|
timeout_job_id = 100
|
self_aborted_job_id = 104
|
aborted_by_suite_job_id = 105
|
good_retry_job_id = 106
|
bad_retry_job_id = 107
|
invalid_job_id_1 = 90
|
invalid_job_id_2 = 91
|
suite_name = 'dummy'
|
build = 'lumpy-release/R27-3888.0.0'
|
suite_job_keyvals = {
|
constants.DOWNLOAD_STARTED_TIME: '2014-04-29 13:14:20',
|
constants.PAYLOAD_FINISHED_TIME: '2014-04-29 13:14:25',
|
constants.ARTIFACT_FINISHED_TIME: '2014-04-29 13:14:30'}
|
|
suite_job_started_time = '2014-04-29 13:14:37'
|
if suite_job_timed_out:
|
suite_job_keyvals['aborted_by'] = 'test_user'
|
suite_job_finished_time = '2014-04-29 13:25:37'
|
suite_job_status = 'ABORT'
|
else:
|
suite_job_finished_time = '2014-04-29 13:23:37'
|
|
server_job_view = self._build_view(
|
10, 'SERVER_JOB', '----', suite_job_status, suite_job_id,
|
'lumpy-release/R27-3888.0.0-test_suites/control.dummy',
|
'', suite_job_keyvals, '2014-04-29 13:14:37',
|
'2014-04-29 13:20:27', job_started_time=suite_job_started_time,
|
job_finished_time=suite_job_finished_time)
|
good_test = self._build_view(
|
11, 'dummy_Pass', '101-user/host/dummy_Pass', 'GOOD',
|
good_job_id, 'lumpy-release/R27-3888.0.0/dummy/dummy_Pass',
|
'', {}, '2014-04-29 13:15:35', '2014-04-29 13:15:36')
|
bad_test = self._build_view(
|
12, 'dummy_Fail.Fail', '102-user/host/dummy_Fail.Fail', 'FAIL',
|
bad_job_id, 'lumpy-release/R27-3888.0.0/dummy/dummy_Fail.Fail',
|
'always fail', {}, '2014-04-29 13:16:00',
|
'2014-04-29 13:16:02')
|
warn_test = self._build_view(
|
13, 'dummy_Fail.Warn', '102-user/host/dummy_Fail.Warn', 'WARN',
|
warn_job_id, 'lumpy-release/R27-3888.0.0/dummy/dummy_Fail.Warn',
|
'always warn', {}, '2014-04-29 13:16:00',
|
'2014-04-29 13:16:02')
|
timeout_test = self._build_view(
|
15, 'dummy_Timeout', '', 'ABORT',
|
timeout_job_id,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_Timeout',
|
'child job did not run', {}, '2014-04-29 13:15:37',
|
'2014-04-29 13:15:38')
|
self_aborted_test = self._build_view(
|
16, 'dummy_Abort', '104-user/host/dummy_Abort', 'ABORT',
|
self_aborted_job_id,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_Abort',
|
'child job aborted', {'aborted_by': 'test_user'},
|
'2014-04-29 13:15:39', '2014-04-29 13:15:40',
|
job_started_time='2014-04-29 13:15:39',
|
job_finished_time='2014-04-29 13:25:40')
|
aborted_by_suite = self._build_view(
|
17, 'dummy_AbortBySuite', '105-user/host/dummy_AbortBySuite',
|
'RUNNING', aborted_by_suite_job_id,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_Abort',
|
'aborted by suite', {'aborted_by': 'test_user'},
|
'2014-04-29 13:15:39', '2014-04-29 13:15:40',
|
job_started_time='2014-04-29 13:15:39',
|
job_finished_time='2014-04-29 13:15:40')
|
good_retry = self._build_view(
|
18, 'dummy_RetryPass', '106-user/host/dummy_RetryPass', 'GOOD',
|
good_retry_job_id,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_RetryPass',
|
'', {'retry_original_job_id': invalid_job_id_1},
|
'2014-04-29 13:15:37',
|
'2014-04-29 13:15:38', invalidates_test_idx=1)
|
bad_retry = self._build_view(
|
19, 'dummy_RetryFail', '107-user/host/dummy_RetryFail', 'FAIL',
|
bad_retry_job_id,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_RetryFail',
|
'retry failed', {'retry_original_job_id': invalid_job_id_2},
|
'2014-04-29 13:15:39', '2014-04-29 13:15:40',
|
invalidates_test_idx=2)
|
invalid_test_1 = self._build_view(
|
1, 'dummy_RetryPass', '90-user/host/dummy_RetryPass', 'GOOD',
|
invalid_job_id_1,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_RetryPass',
|
'original test failed', {}, '2014-04-29 13:10:00',
|
'2014-04-29 13:10:01')
|
invalid_test_2 = self._build_view(
|
2, 'dummy_RetryFail', '91-user/host/dummy_RetryFail', 'FAIL',
|
invalid_job_id_2,
|
'lumpy-release/R27-3888.0.0/dummy/dummy_RetryFail',
|
'original test failed', {},
|
'2014-04-29 13:10:03', '2014-04-29 13:10:04')
|
|
test_views = []
|
child_jobs = set()
|
missing_results = []
|
if include_good_test:
|
test_views.append(server_job_view)
|
test_views.append(good_test)
|
child_jobs.add(good_job_id)
|
# Emulate missing even the parent/suite job.
|
else:
|
missing_results.append(suite_job_id)
|
if include_bad_test:
|
test_views.append(bad_test)
|
child_jobs.add(bad_job_id)
|
if include_warn_test:
|
test_views.append(warn_test)
|
child_jobs.add(warn_job_id)
|
if include_timeout_test:
|
test_views.append(timeout_test)
|
if include_self_aborted_test:
|
test_views.append(self_aborted_test)
|
child_jobs.add(self_aborted_job_id)
|
if include_good_retry:
|
test_views.extend([good_retry, invalid_test_1])
|
child_jobs.add(good_retry_job_id)
|
if include_bad_retry:
|
test_views.extend([bad_retry, invalid_test_2])
|
child_jobs.add(bad_retry_job_id)
|
if include_aborted_by_suite_test:
|
test_views.append(aborted_by_suite)
|
child_jobs.add(aborted_by_suite_job_id)
|
self._mock_tko_get_detailed_test_views(test_views,
|
missing_results=missing_results)
|
self._mock_afe_get_jobs(suite_job_id, child_jobs)
|
collector = run_suite.ResultCollector(
|
'fake_server', self.afe, self.tko,
|
'lumpy-release/R36-5788.0.0', 'dummy', suite_job_id,
|
return_code_function=run_suite._ReturnCodeComputer())
|
collector.run()
|
collector.output_results()
|
return collector
|
|
|
def testEndToEndSuiteEmpty(self):
|
"""Test it returns code INFRA_FAILURE when no tests report back."""
|
collector = self._end_to_end_test_helper(include_good_test=False)
|
self.assertEqual(collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.INFRA_FAILURE)
|
|
|
def testEndToEndSuitePass(self):
|
"""Test it returns code OK when all test pass."""
|
collector = self._end_to_end_test_helper()
|
self.assertEqual(collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.OK)
|
|
|
def testEndToEndSuiteWarn(self):
|
"""Test it returns code WARNING when there is a test that warns."""
|
collector = self._end_to_end_test_helper(include_warn_test=True)
|
self.assertEqual(collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.WARNING)
|
|
|
def testEndToEndSuiteFail(self):
|
"""Test it returns code ERROR when there is a test that fails."""
|
collector = self._end_to_end_test_helper(include_bad_test=True)
|
self.assertEqual(collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.ERROR)
|
|
|
def testEndToEndSuiteJobFail(self):
|
"""Test it returns code SUITE_FAILURE when only the suite job failed."""
|
collector = self._end_to_end_test_helper(suite_job_status='ABORT')
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.INFRA_FAILURE)
|
|
collector = self._end_to_end_test_helper(suite_job_status='ERROR')
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.INFRA_FAILURE)
|
|
|
def testEndToEndRetry(self):
|
"""Test it returns correct code when a test was retried."""
|
collector = self._end_to_end_test_helper(include_good_retry=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.WARNING)
|
|
collector = self._end_to_end_test_helper(include_good_retry=True,
|
include_self_aborted_test=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.ERROR)
|
|
collector = self._end_to_end_test_helper(include_good_retry=True,
|
include_bad_test=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.ERROR)
|
|
collector = self._end_to_end_test_helper(include_bad_retry=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.ERROR)
|
|
|
def testEndToEndSuiteTimeout(self):
|
"""Test it returns correct code when a child job timed out."""
|
# a child job timed out before started, none failed.
|
collector = self._end_to_end_test_helper(include_timeout_test=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.SUITE_TIMEOUT)
|
|
# a child job timed out before started, and one test failed.
|
collector = self._end_to_end_test_helper(
|
include_bad_test=True, include_timeout_test=True)
|
self.assertEqual(collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.ERROR)
|
|
# a child job timed out before started, and one test warned.
|
collector = self._end_to_end_test_helper(
|
include_warn_test=True, include_timeout_test=True)
|
self.assertEqual(collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.SUITE_TIMEOUT)
|
|
# a child job timed out before started, and one test was retried.
|
collector = self._end_to_end_test_helper(include_good_retry=True,
|
include_timeout_test=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.SUITE_TIMEOUT)
|
|
# a child jot was aborted because suite timed out.
|
collector = self._end_to_end_test_helper(
|
include_aborted_by_suite_test=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.SUITE_TIMEOUT)
|
|
# suite job timed out.
|
collector = self._end_to_end_test_helper(suite_job_timed_out=True)
|
self.assertEqual(
|
collector.return_result.return_code,
|
run_suite_common.RETURN_CODES.SUITE_TIMEOUT)
|
|
|
class LogLinkUnittests(unittest.TestCase):
|
"""Test the LogLink"""
|
|
def testGenerateBuildbotLinks(self):
|
"""Test LogLink GenerateBuildbotLinks"""
|
log_link_a = run_suite.LogLink('mock_anchor', 'mock_server',
|
'mock_job_string',
|
bug_info=('mock_bug_id', 1),
|
reason='mock_reason',
|
retry_count=1,
|
testname='mock_testname')
|
# Generate a bug link and a log link when bug_info is present
|
self.assertTrue(len(list(log_link_a.GenerateBuildbotLinks())) == 2)
|
|
log_link_b = run_suite.LogLink('mock_anchor', 'mock_server',
|
'mock_job_string_b',
|
reason='mock_reason',
|
retry_count=1,
|
testname='mock_testname')
|
# Generate a log link when there is no bug_info
|
self.assertTrue(len(list(log_link_b.GenerateBuildbotLinks())) == 1)
|
|
|
class SimpleTimerUnittests(unittest.TestCase):
|
"""Test the simple timer."""
|
|
def testPoll(self):
|
"""Test polling the timer."""
|
interval_hours = 0.0001
|
t = diagnosis_utils.SimpleTimer(interval_hours=interval_hours)
|
deadline = t.deadline
|
self.assertTrue(deadline is not None and
|
t.interval_hours == interval_hours)
|
min_deadline = (datetime.now() +
|
datetime_base.timedelta(hours=interval_hours))
|
time.sleep(interval_hours * 3600)
|
self.assertTrue(t.poll())
|
self.assertTrue(t.deadline >= min_deadline)
|
|
|
def testBadInterval(self):
|
"""Test a bad interval."""
|
t = diagnosis_utils.SimpleTimer(interval_hours=-1)
|
self.assertTrue(t.deadline is None and t.poll() == False)
|
t._reset()
|
self.assertTrue(t.deadline is None and t.poll() == False)
|
|
|
if __name__ == '__main__':
|
unittest.main()
|
