site_utils/test_runner_utils_unittest.py

#!/usr/bin/python3
# Copyright 2015 The ChromiumOS Authors
# 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
from unittest.mock import patch

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


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 JobMatcher(object):
    """Matcher for JobObject + Name."""

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

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


class hostinfoMatcher(object):
    """Match hostinfo stuff"""

    def __init__(self, labels, attributes):
        self.labels = labels.split(' ')
        self.attributes = attributes

    def __eq__(self, other):
        return self.labels == other.labels and self.attributes == other.attributes


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

        run_job_mock.assert_called_with(job=TypeMatcher(
                test_runner_utils.SimpleJob),
                                        host=self.remote,
                                        info=TypeMatcher(host_info.HostInfo),
                                        autotest_path=self.autotest_path,
                                        results_directory=self.results_dir,
                                        fast_mode=self.fast_mode,
                                        id_digits=self.id_digits,
                                        ssh_verbosity=self.ssh_verbosity,
                                        ssh_options=self.ssh_options,
                                        args=TypeMatcher(str),
                                        pretend=False,
                                        autoserv_verbose=False,
                                        companion_hosts=None,
                                        dut_servers=None,
                                        libs_server=None,
                                        is_cft=False,
                                        ch_info={})

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

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

    def test_minus_flag(self):
        """Verify the minus flag skips tests."""
        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)

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

        minus_tests = [FakeTests(self.suite_control_files[0])]
        all_tests = [
                FakeTests(test, deps=[]) for test in self.suite_control_files
        ]

        test_runner_utils_mock.side_effect = [minus_tests, all_tests]
        run_job_mock.side_effect = [(0, 'fakedir') for _ in range(3)]
        test_labels = "'a' 'test' 'label'"
        test_attributes = {"servo": "yes"}

        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,
                host_attributes=test_attributes,
                job_retry=self.retry,
                ignore_deps=False,
                minus=[self.suite_control_files[0]],
                is_cft=True,
                host_labels=test_labels,
                label=None)

        from unittest.mock import call

        calls = []
        for name in self.suite_control_files[1:]:
            calls.append(
                    call(job=JobMatcher(test_runner_utils.SimpleJob,
                                        name=name),
                         host=self.remote,
                         info=hostinfoMatcher(labels=test_labels,
                                              attributes=test_attributes),
                         autotest_path=self.autotest_path,
                         results_directory=self.results_dir,
                         fast_mode=self.fast_mode,
                         id_digits=self.id_digits,
                         ssh_verbosity=self.ssh_verbosity,
                         ssh_options=self.ssh_options,
                         args=TypeMatcher(str),
                         pretend=False,
                         autoserv_verbose=False,
                         companion_hosts=None,
                         dut_servers=None,
                         libs_server=None,
                         is_cft=True,
                         ch_info={}))

        run_job_mock.assert_has_calls(calls, any_order=True)
        assert run_job_mock.call_count == len(calls)

    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

    def test_host_info_write(self):

        dirpath = tempfile.mkdtemp()

        info = host_info.HostInfo(['some', 'labels'], {'attrib1': '1'})
        import pathlib
        expected_path = os.path.join(
                pathlib.Path(__file__).parent.absolute(),
                'host_info_store_testfile')
        try:

            test_runner_utils._write_host_info(dirpath, 'host_info_store',
                                               'localhost:1234', info)
            test_path = os.path.join(dirpath, 'host_info_store',
                                     'localhost:1234.store')
            with open(test_path, 'r') as rf:
                test_data = rf.read()
            with open(expected_path, 'r') as rf:
                expected_data = rf.read()
            self.assertEqual(test_data, expected_data)

        finally:
            shutil.rmtree(dirpath)


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