#!/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, unittest
import mox
import common
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
from six.moves import range
from six.moves import zip


class StartsWithList(mox.Comparator):
    def __init__(self, start_of_list):
        """Mox comparator which returns True if the argument
        to the mocked function is a list that begins with the elements
        in start_of_list.
        """
        self._lhs = start_of_list

    def equals(self, rhs):
        if len(rhs)<len(self._lhs):
            return False
        for (x, y) in zip(self._lhs, rhs):
            if x != y:
                return False
        return True


class ContainsSublist(mox.Comparator):
    def __init__(self, sublist):
        """Mox comparator which returns True if the argument
        to the mocked function is a list that contains sublist
        as a sub-list.
        """
        self._sublist = sublist

    def equals(self, rhs):
        n = len(self._sublist)
        if len(rhs)<n:
            return False
        return any((self._sublist == rhs[i:i+n])
                   for i in range(len(rhs) - n + 1))

class DummyJob(object):
    def __init__(self, id=1):
        self.id = id


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


class TestRunnerUnittests(mox.MoxTestBase):
    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 setUp(self):
        mox.MoxTestBase.setUp(self)


    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):
        getter = self.mox.CreateMock(control_file_getter.DevServerGetter)
        getter.get_control_file_list(suite_name=mox.IgnoreArg()).AndReturn([])
        job = DummyJob()
        test = self.mox.CreateMock(control_data.ControlData)
        test.job_retries = 5
        self.mox.StubOutWithMock(test_runner_utils.LocalSuite,
                                 '_retry_local_result')
        self.mox.ReplayAll()
        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):
        getter = self.mox.CreateMock(control_file_getter.DevServerGetter)
        getter.get_control_file_list(suite_name=mox.IgnoreArg()).AndReturn([])
        job = DummyJob()
        test = self.mox.CreateMock(control_data.ControlData)
        test.job_retries = 5
        self.mox.StubOutWithMock(test_runner_utils.LocalSuite,
                                 '_retry_local_result')
        test_runner_utils.LocalSuite._retry_local_result(
            job.id, mox.IgnoreArg()).AndReturn(42)
        self.mox.ReplayAll()
        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."""
        self.mox.StubOutWithMock(test_runner_utils, '_auto_detect_labels')
        self.mox.StubOutWithMock(test_runner_utils, 'get_all_control_files')

        test_runner_utils._auto_detect_labels(self.remote).AndReturn(
                ['os:cros', 'has_chameleon:True'])

        test_runner_utils.get_all_control_files(
                self.test_arg, self.autotest_path).AndReturn([
                        fake_tests(test, deps=['has_chameleon:True'])
                        for test in self.suite_control_files
                ])

        self.mox.StubOutWithMock(test_runner_utils, 'run_job')
        for control_file in self.suite_control_files:
            test_runner_utils.run_job(
                    mox.ContainsAttributeValue('control_file', control_file),
                    self.remote,
                    mox.IsA(host_info.HostInfo),
                    self.autotest_path,
                    self.results_dir,
                    self.fast_mode,
                    self.id_digits,
                    self.ssh_verbosity,
                    self.ssh_options,
                    mox.StrContains(self.args),
                    False,
                    False,
                    None,
            ).AndReturn((0, '/fake/dir'))

        self.mox.ReplayAll()
        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)

    def test_perform_local_run_missing_deps(self):
        """Test a local run with missing dependencies. No tests should run."""
        self.mox.StubOutWithMock(test_runner_utils, '_auto_detect_labels')
        self.mox.StubOutWithMock(test_runner_utils, 'get_all_control_files')

        test_runner_utils._auto_detect_labels(self.remote).AndReturn(
                ['os:cros', 'has_chameleon:True'])

        test_runner_utils.get_all_control_files(
                self.test_arg, self.autotest_path).AndReturn([
                        fake_tests(test, deps=['has_chameleon:False'])
                        for test in self.suite_control_files
                ])

        self.mox.ReplayAll()

        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, [])


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