#!/usr/bin/python2
# Copyright 2015 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.
# pylint: disable-msg=C0111

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import os
import unittest

import common

from mock import patch
import shutil
import tempfile
import types
from autotest_lib.client.common_lib import control_data
from autotest_lib.server.cros.dynamic_suite import control_file_getter
from autotest_lib.server.cros.dynamic_suite import suite as suite_module
from autotest_lib.server.hosts import host_info
from autotest_lib.site_utils import test_runner_utils


class TypeMatcher(object):
    """Matcher for object is of type."""

    def __init__(self, expected_type):
        self.expected_type = expected_type

    def __eq__(self, other):
        return isinstance(other, self.expected_type)


class ContainsMatcher:
    """Matcher for object contains attr."""

    def __init__(self, key, value):
        self.key = key
        self.value = value

    def __eq__(self, rhs):
        try:
            return getattr(rhs, self._key) == self._value
        except Exception:
            return False


class SampleJob(object):
    """Sample to be used for mocks."""

    def __init__(self, id=1):
        self.id = id


class FakeTests(object):
    """A fake test to be used for mocks."""

    def __init__(self, text, deps=[], py_version=None):
        self.text = text
        self.test_type = 'client'
        self.dependencies = deps
        self.name = None
        self.py_version = py_version


class TestRunnerUnittests(unittest.TestCase):
    """Test test_runner_utils."""

    autotest_path = 'ottotest_path'
    suite_name = 'sweet_name'
    test_arg = 'suite:' + suite_name
    remote = 'remoat'
    build = 'bild'
    board = 'bored'
    fast_mode = False
    suite_control_files = ['c1', 'c2', 'c3', 'c4']
    results_dir = '/tmp/test_that_results_fake'
    id_digits = 1
    ssh_verbosity = 2
    ssh_options = '-F /dev/null -i /dev/null'
    args = 'matey'
    retry = True

    def _results_directory_from_results_list(self, results_list):
        """Generate a temp directory filled with provided test results.

        @param results_list: List of results, each result is a tuple of strings
                             (test_name, test_status_message).
        @returns: Absolute path to the results directory.
        """
        global_dir = tempfile.mkdtemp()
        for index, (test_name, test_status_message) in enumerate(results_list):
            dir_name = '-'.join(['results',
                                 "%02.f" % (index + 1),
                                 test_name])
            local_dir = os.path.join(global_dir, dir_name)
            os.mkdir(local_dir)
            os.mkdir('%s/debug' % local_dir)
            with open("%s/status.log" % local_dir, mode='w+') as status:
                status.write(test_status_message)
                status.flush()
        return global_dir

    def test_handle_local_result_for_good_test(self):
        patcher = patch.object(control_file_getter, 'DevServerGetter')
        getter = patcher.start()
        self.addCleanup(patcher.stop)
        getter.get_control_file_list.return_value = []
        job = SampleJob()

        test_patcher = patch.object(control_data, 'ControlData')
        test = test_patcher.start()
        self.addCleanup(test_patcher.stop)
        test.job_retries = 5

        suite = test_runner_utils.LocalSuite([], "tag", [], None, getter,
                                             job_retry=True)
        suite._retry_handler = suite_module.RetryHandler({job.id: test})

        #No calls, should not be retried
        directory = self._results_directory_from_results_list([
            ("dummy_Good", "GOOD: nonexistent test completed successfully")])
        new_id = suite.handle_local_result(
            job.id, directory,
            lambda log_entry, log_in_subdir=False: None)
        self.assertIsNone(new_id)
        shutil.rmtree(directory)

    def test_handle_local_result_for_bad_test(self):
        patcher = patch.object(control_file_getter, 'DevServerGetter')
        getter = patcher.start()
        self.addCleanup(patcher.stop)
        getter.get_control_file_list.return_value = []

        job = SampleJob()

        test_patcher = patch.object(control_data, 'ControlData')
        test = test_patcher.start()
        self.addCleanup(test_patcher.stop)
        test.job_retries = 5

        utils_mock = patch.object(test_runner_utils.LocalSuite,
                                  '_retry_local_result')
        test_runner_utils_mock = utils_mock.start()
        self.addCleanup(utils_mock.stop)
        test_runner_utils_mock._retry_local_result.return_value = 42

        suite = test_runner_utils.LocalSuite([], "tag", [], None, getter,
                                             job_retry=True)
        suite._retry_handler = suite_module.RetryHandler({job.id: test})

        directory = self._results_directory_from_results_list([
            ("dummy_Bad", "FAIL")])
        new_id = suite.handle_local_result(
            job.id, directory,
            lambda log_entry, log_in_subdir=False: None)
        self.assertIsNotNone(new_id)
        shutil.rmtree(directory)


    def test_generate_report_status_code_success_with_retries(self):
        global_dir = self._results_directory_from_results_list([
            ("dummy_Flaky", "FAIL"),
            ("dummy_Flaky", "GOOD: nonexistent test completed successfully")])
        status_code = test_runner_utils.generate_report(
            global_dir, just_status_code=True)
        self.assertEquals(status_code, 0)
        shutil.rmtree(global_dir)


    def test_generate_report_status_code_failure_with_retries(self):
        global_dir = self._results_directory_from_results_list([
            ("dummy_Good", "GOOD: nonexistent test completed successfully"),
            ("dummy_Bad", "FAIL"),
            ("dummy_Bad", "FAIL")])
        status_code = test_runner_utils.generate_report(
            global_dir, just_status_code=True)
        self.assertNotEquals(status_code, 0)
        shutil.rmtree(global_dir)


    def test_get_predicate_for_test_arg(self):
        # Assert the type signature of get_predicate_for_test(...)
        # Because control.test_utils_wrapper calls this function,
        # it is imperative for backwards compatilbility that
        # the return type of the tested function does not change.
        tests = ['dummy_test', 'e:name_expression', 'f:expression',
                 'suite:suitename']
        for test in tests:
            pred, desc = test_runner_utils.get_predicate_for_test_arg(test)
            self.assertTrue(isinstance(pred, types.FunctionType))
            self.assertTrue(isinstance(desc, str))

    def test_perform_local_run(self):
        """Test a local run that should pass."""
        patcher = patch.object(test_runner_utils, '_auto_detect_labels')
        _auto_detect_labels_mock = patcher.start()
        self.addCleanup(patcher.stop)

        patcher2 = patch.object(test_runner_utils, 'get_all_control_files')
        get_all_control_files_mock = patcher2.start()
        self.addCleanup(patcher2.stop)

        _auto_detect_labels_mock.return_value = [
                'os:cros', 'has_chameleon:True'
        ]

        get_all_control_files_mock.return_value = [
                FakeTests(test, deps=['has_chameleon:True'])
                for test in self.suite_control_files
        ]

        patcher3 = patch.object(test_runner_utils, 'run_job')
        run_job_mock = patcher3.start()
        self.addCleanup(patcher3.stop)

        for control_file in self.suite_control_files:
            run_job_mock.return_value = (0, '/fake/dir')
        test_runner_utils.perform_local_run(self.autotest_path,
                                            ['suite:' + self.suite_name],
                                            self.remote,
                                            self.fast_mode,
                                            build=self.build,
                                            board=self.board,
                                            ssh_verbosity=self.ssh_verbosity,
                                            ssh_options=self.ssh_options,
                                            args=self.args,
                                            results_directory=self.results_dir,
                                            job_retry=self.retry,
                                            ignore_deps=False)
        run_job_mock.assert_called_with(
                TypeMatcher(test_runner_utils.SimpleJob),
                self.remote,
                TypeMatcher(host_info.HostInfo),
                self.autotest_path,
                self.results_dir,
                self.fast_mode,
                self.id_digits,
                self.ssh_verbosity,
                self.ssh_options,
                TypeMatcher(str),
                False,
                False,
                None,
        )

    def test_perform_local_run_missing_deps(self):
        """Test a local run with missing dependencies. No tests should run."""
        patcher = patch.object(test_runner_utils, '_auto_detect_labels')
        getter = patcher.start()
        self.addCleanup(patcher.stop)

        getter.return_value = ['os:cros', 'has_chameleon:True']

        patcher2 = patch.object(test_runner_utils, 'get_all_control_files')
        test_runner_utils_mock = patcher2.start()
        self.addCleanup(patcher2.stop)
        test_runner_utils_mock.return_value = [
                FakeTests(test, deps=['has_chameleon:False'])
                for test in self.suite_control_files
        ]

        res = test_runner_utils.perform_local_run(
                self.autotest_path, ['suite:' + self.suite_name],
                self.remote,
                self.fast_mode,
                build=self.build,
                board=self.board,
                ssh_verbosity=self.ssh_verbosity,
                ssh_options=self.ssh_options,
                args=self.args,
                results_directory=self.results_dir,
                job_retry=self.retry,
                ignore_deps=False)

        # Verify when the deps are not met, the tests are not run.
        self.assertEquals(res, [])

    def test_set_pyversion(self):
        """Test the tests can properly set the python version."""

        # When a test is missing a version, use the current setting.
        starting_version = os.getenv('PY_VERSION')

        try:
            fake_test1 = FakeTests('foo')
            fake_test2 = FakeTests('foo', py_version=2)
            fake_test3 = FakeTests('foo', py_version=3)

            test_runner_utils._set_pyversion(
                    [fake_test1, fake_test2, fake_test3])
            self.assertEqual(os.getenv('PY_VERSION'), starting_version)

            # When there is a mix, use the current setting.
            starting_version = os.getenv('PY_VERSION')
            fake_test1 = FakeTests('foo', py_version=2)
            fake_test2 = FakeTests('foo', py_version=2)
            fake_test3 = FakeTests('foo', py_version=3)

            test_runner_utils._set_pyversion(
                    [fake_test1, fake_test2, fake_test3])
            self.assertEqual(os.getenv('PY_VERSION'), starting_version)

            # When all agree, but still 1 missing, use the current setting.
            fake_test1 = FakeTests('foo')
            fake_test2 = FakeTests('foo', py_version=3)
            fake_test3 = FakeTests('foo', py_version=3)

            test_runner_utils._set_pyversion(
                    [fake_test1, fake_test2, fake_test3])
            self.assertEqual(os.getenv('PY_VERSION'), starting_version)

            # When all are set to 3, use 3.
            fake_test1 = FakeTests('foo', py_version=3)
            fake_test2 = FakeTests('foo', py_version=3)
            fake_test3 = FakeTests('foo', py_version=3)

            test_runner_utils._set_pyversion(
                    [fake_test1, fake_test2, fake_test3])
            self.assertEqual(os.getenv('PY_VERSION'), '3')

            # When all are set to 2, use 2.
            fake_test1 = FakeTests('foo', py_version=2)
            fake_test2 = FakeTests('foo', py_version=2)
            fake_test3 = FakeTests('foo', py_version=2)

            test_runner_utils._set_pyversion(
                    [fake_test1, fake_test2, fake_test3])
            self.assertEqual(os.getenv('PY_VERSION'), '2')
        finally:
            # In the event something breaks, reset the pre-test version.
            os.environ['PY_VERSION'] = starting_version


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