#!/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 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 diagnosis_utils


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='fake/build', board='fake', suite_name='dummy',
                suite_job_id=suite_job_id)
        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, board, suite_name, suite_job_id)
        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
        suite_job_view = run_suite.TestView(
                self._build_view(
                    20, 'Suite job', '----', 'GOOD', suite_job_id),
                fake_job, suite_name, build, 'chromeos-test')
        good_test = run_suite.TestView(
                self._build_view(
                    21, 'test_Pass', 'fake/subdir', 'GOOD', 101),
                fake_job, suite_name, build, 'chromeos-test')
        bad_test = run_suite.TestView(
                self._build_view(
                    23, 'test_Fail', 'fake/subdir', 'FAIL', 102),
                fake_job, suite_name, build, 'chromeos-test')

        collector = run_suite.ResultCollector(
                'fake_server', self.afe, self.tko,
                build, board, suite_name, suite_job_id, user='chromeos-test')
        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 % ('fake_server',
                                '%s-%s' % (v['afe_job_id'], 'chromeos-test')))
                 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])

        expected_buildbot_links = [
                 (v.get_testname(),
                  URL_PATTERN % ('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 wmatrix retry dashboard link is created.
            self.assertNotEqual(
                    collector._buildbot_links[i].GenerateWmatrixRetryLink(),'')


    def _end_to_end_test_helper(
            self, include_bad_test=False, include_warn_test=False,
            include_experimental_bad_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,
            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_experimental_bad_test:
                If True, include a view of an experimental test
                which has status 'FAIL'.
        @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 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
        experimental_bad_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')
        experimental_bad_test = self._build_view(
                14, 'experimental_dummy_Fail.Fail',
                '102-user/host/dummy_Fail.Fail', 'FAIL',
                experimental_bad_job_id,
                'lumpy-release/R27-3888.0.0/dummy/experimental_dummy_Fail.Fail',
                'always fail', {'experimental': 'True'}, '2014-04-29 13:16:06',
                '2014-04-29 13:16:07')
        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 = [server_job_view, good_test]
        child_jobs = set([good_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_experimental_bad_test:
            test_views.append(experimental_bad_test)
            child_jobs.add(experimental_bad_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)
        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', 'lumpy', 'dummy', suite_job_id)
        collector.run()
        return collector


    def testEndToEndSuitePass(self):
        """Test it returns code OK when all test pass."""
        collector = self._end_to_end_test_helper()
        self.assertEqual(collector.return_code, run_suite.RETURN_CODES.OK)


    def testEndToEndExperimentalTestFails(self):
        """Test that it returns code OK when only experimental test fails."""
        collector = self._end_to_end_test_helper(
                include_experimental_bad_test=True)
        self.assertEqual(collector.return_code, run_suite.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_code, run_suite.RETURN_CODES.WARNING)


    def testEndToEndSuiteFail(self):
        """Test it returns code ERROR when there is a test that fails."""
        # Test that it returns ERROR when there is test that fails.
        collector = self._end_to_end_test_helper(include_bad_test=True)
        self.assertEqual(collector.return_code, run_suite.RETURN_CODES.ERROR)

        # Test that it returns ERROR when both experimental and non-experimental
        # test fail.
        collector = self._end_to_end_test_helper(
                include_bad_test=True, include_warn_test=True,
                include_experimental_bad_test=True)
        self.assertEqual(collector.return_code, run_suite.RETURN_CODES.ERROR)

        collector = self._end_to_end_test_helper(include_self_aborted_test=True)
        self.assertEqual(collector.return_code, run_suite.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_code, run_suite.RETURN_CODES.INFRA_FAILURE)

        collector = self._end_to_end_test_helper(suite_job_status='ERROR')
        self.assertEqual(
                collector.return_code, run_suite.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_code, run_suite.RETURN_CODES.WARNING)

        collector = self._end_to_end_test_helper(include_good_retry=True,
                include_self_aborted_test=True)
        self.assertEqual(
                collector.return_code, run_suite.RETURN_CODES.ERROR)

        collector = self._end_to_end_test_helper(include_good_retry=True,
                include_bad_test=True)
        self.assertEqual(
                collector.return_code, run_suite.RETURN_CODES.ERROR)

        collector = self._end_to_end_test_helper(include_bad_retry=True)
        self.assertEqual(
                collector.return_code, run_suite.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_code, run_suite.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_code, run_suite.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_code,
                         run_suite.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_code, run_suite.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_code, run_suite.RETURN_CODES.OK)

        # suite job timed out.
        collector = self._end_to_end_test_helper(suite_job_timed_out=True)
        self.assertEqual(
                collector.return_code, run_suite.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)


class ArgumentParserUnittests(unittest.TestCase):
    """Tests for argument parser."""

    @unittest.expectedFailure
    def test_crbug_658013(self):
        """crbug.com/658013

        Expected failure due to http://bugs.python.org/issue9334
        """
        parser = run_suite.make_parser()
        args = [
            '--board', 'heli',
            '--build', 'trybot-heli-paladin/R56-8918.0.0-b1601',
            '--suite_name', 'test_that_wrapper',
            '--pool', 'suites',
            '--max_runtime_mins', '20',
            '--suite_args', '-b heli -i trybot-heli-paladin/R56-8918.0.0-b1601 :lab: suite:bvt-inline',
        ]

        def error_handler(msg):  # pylint: disable=missing-docstring
            self.fail('Argument parsing failed: ' + msg)

        parser.error = error_handler
        got = parser.parse_args(args)
        self.assertEqual(
            got.board, 'heli')
        self.assertEqual(
            got.build, 'trybot-heli-paladin/R56-8918.0.0-b1601')
        self.assertEqual(
            got.suite_args,
            '-b heli -i trybot-heli-paladin/R56-8918.0.0-b1601 :lab: suite:bvt-inline')

    def test_crbug_658013b(self):
        """crbug.com/658013

        Unambiguous behavior.
        """
        parser = run_suite.make_parser()
        args = [
            '--board=heli',
            '--build=trybot-heli-paladin/R56-8918.0.0-b1601',
            '--suite_name=test_that_wrapper',
            '--pool=suites',
            '--max_runtime_mins=20',
            '--suite_args=-b heli -i trybot-heli-paladin/R56-8918.0.0-b1601 :lab: suite:bvt-inline',
        ]

        def error_handler(msg):  # pylint: disable=missing-docstring
            self.fail('Argument parsing failed: ' + msg)

        parser.error = error_handler
        got = parser.parse_args(args)
        self.assertEqual(
            got.board, 'heli')
        self.assertEqual(
            got.build, 'trybot-heli-paladin/R56-8918.0.0-b1601')
        self.assertEqual(
            got.suite_args,
            '-b heli -i trybot-heli-paladin/R56-8918.0.0-b1601 :lab: suite:bvt-inline')


if __name__ == '__main__':
    unittest.main()