server/autotest_unittest.py

#!/usr/bin/python
#pylint: disable-msg=C0111
__author__ = "raphtee@google.com (Travis Miller)"

import unittest, os, tempfile, logging

import common
from autotest_lib.server import autotest, utils, hosts, server_job, profilers
from autotest_lib.client.bin import sysinfo
from autotest_lib.client.common_lib import packages
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.test_utils import mock


class TestBaseAutotest(unittest.TestCase):
    def setUp(self):
        # create god
        self.god = mock.mock_god()

        # create mock host object
        self.host = self.god.create_mock_class(hosts.RemoteHost, "host")
        self.host.hostname = "hostname"
        self.host.job = self.god.create_mock_class(server_job.server_job,
                                                   "job")
        self.host.job.run_test_cleanup = True
        self.host.job.last_boot_tag = 'Autotest'
        self.host.job.sysinfo = self.god.create_mock_class(
            sysinfo.sysinfo, "sysinfo")
        self.host.job.profilers = self.god.create_mock_class(
            profilers.profilers, "profilers")
        self.host.job.profilers.add_log = {}
        self.host.job.tmpdir = "/job/tmp"
        self.host.job.default_profile_only = False
        self.host.job.args = []
        self.host.job.record = lambda *args: None

        # stubs
        self.god.stub_function(utils, "get_server_dir")
        self.god.stub_function(utils, "run")
        self.god.stub_function(utils, "get")
        self.god.stub_function(utils, "read_keyval")
        self.god.stub_function(utils, "write_keyval")
        self.god.stub_function(utils, "system")
        self.god.stub_function(tempfile, "mkstemp")
        self.god.stub_function(tempfile, "mktemp")
        self.god.stub_function(os, "getcwd")
        self.god.stub_function(os, "system")
        self.god.stub_function(os, "chdir")
        self.god.stub_function(os, "makedirs")
        self.god.stub_function(os, "remove")
        self.god.stub_function(os, "fdopen")
        self.god.stub_function(os.path, "exists")
        self.god.stub_function(autotest, "open")
        self.god.stub_function(autotest.global_config.global_config,
                               "get_config_value")
        self.god.stub_function(logging, "exception")
        self.god.stub_class(autotest, "_Run")
        self.god.stub_class(autotest, "log_collector")


    def tearDown(self):
        self.god.unstub_all()


    def construct(self):
        # setup
        self.serverdir = "serverdir"

        # record
        utils.get_server_dir.expect_call().and_return(self.serverdir)

        # create the autotest object
        self.base_autotest = autotest.BaseAutotest(self.host)
        self.base_autotest.job = self.host.job
        self.god.stub_function(self.base_autotest, "_install_using_send_file")

        # stub out abspath
        self.god.stub_function(os.path, "abspath")

        # check
        self.god.check_playback()


    def record_install_prologue(self):
        self.construct()

        # setup
        self.god.stub_class(packages, "PackageManager")
        self.base_autotest.got = False
        location = os.path.join(self.serverdir, '../client')
        location = os.path.abspath.expect_call(location).and_return(location)

        # record
        os.getcwd.expect_call().and_return('cwd')
        os.chdir.expect_call(os.path.join(self.serverdir, '../client'))
        utils.system.expect_call('tools/make_clean', ignore_status=True)
        os.chdir.expect_call('cwd')
        utils.get.expect_call(os.path.join(self.serverdir,
            '../client')).and_return('source_material')

        self.host.wait_up.expect_call(timeout=30)
        self.host.setup.expect_call()
        self.host.get_autodir.expect_call().and_return("autodir")
        self.host.set_autodir.expect_call("autodir")
        self.host.run.expect_call('mkdir -p autodir')
        self.host.run.expect_call('rm -rf autodir/results/*',
                                  ignore_status=True)


    def test_constructor(self):
        self.construct()

        # we should check the calls
        self.god.check_playback()


    def test_full_client_install(self):
        self.record_install_prologue()

        c = autotest.global_config.global_config
        c.get_config_value.expect_call('PACKAGES',
                                       'serve_packages_from_autoserv',
                                       type=bool).and_return(False)
        self.host.send_file.expect_call('source_material', 'autodir',
                                        delete_dest=True)

        # run and check
        self.base_autotest.install_full_client()
        self.god.check_playback()


    def test_autoserv_install(self):
        self.record_install_prologue()

        c = autotest.global_config.global_config
        c.get_config_value.expect_call('PACKAGES',
            'fetch_location', type=list, default=[]).and_return([])

        c.get_config_value.expect_call('PACKAGES',
                                       'serve_packages_from_autoserv',
                                       type=bool).and_return(True)
        self.base_autotest._install_using_send_file.expect_call(self.host,
                                                                'autodir')
        # run and check
        self.base_autotest.install()
        self.god.check_playback()


    def test_packaging_install(self):
        self.record_install_prologue()

        c = autotest.global_config.global_config
        c.get_config_value.expect_call('PACKAGES',
            'fetch_location', type=list, default=[]).and_return(['repo'])
        pkgmgr = packages.PackageManager.expect_new('autodir',
            repo_urls=['repo'], hostname='hostname', do_locking=False,
            run_function=self.host.run, run_function_dargs=dict(timeout=600))
        pkg_dir = os.path.join('autodir', 'packages')
        cmd = ('cd autodir && ls | grep -v "^packages$"'
               ' | xargs rm -rf && rm -rf .[^.]*')
        self.host.run.expect_call(cmd)
        pkgmgr.install_pkg.expect_call('autotest', 'client', pkg_dir,
                                       'autodir', preserve_install_dir=True)

        # run and check
        self.base_autotest.install()
        self.god.check_playback()


    def test_run(self):
        self.construct()

        # setup
        control = "control"

        # stub out install
        self.god.stub_function(self.base_autotest, "install")
        self.god.stub_class(packages, "PackageManager")

        # record
        self.base_autotest.install.expect_call(self.host)
        self.host.wait_up.expect_call(timeout=30)
        os.path.abspath.expect_call('.').and_return('.')
        run_obj = autotest._Run.expect_new(self.host, '.', None, False, False)
        tag = None
        run_obj.manual_control_file = os.path.join('autodir',
                                                   'control.%s' % tag)
        run_obj.remote_control_file = os.path.join('autodir',
                                                   'control.%s.autoserv' % tag)
        run_obj.tag = tag
        run_obj.autodir = 'autodir'
        run_obj.verify_machine.expect_call()
        run_obj.verify_machine.expect_call()
        run_obj.background = False
        debug = os.path.join('.', 'debug')
        os.makedirs.expect_call(debug)
        delete_file_list = [run_obj.remote_control_file,
                            run_obj.remote_control_file + '.state',
                            run_obj.manual_control_file,
                            run_obj.manual_control_file + '.state']
        cmd = ';'.join('rm -f ' + control for control in delete_file_list)
        self.host.run.expect_call(cmd, ignore_status=True)

        utils.get.expect_call(control, local_copy=True).and_return("temp")

        c = autotest.global_config.global_config
        c.get_config_value.expect_call("PACKAGES",
            'fetch_location', type=list, default=[]).and_return(['repo'])
        pkgmgr = packages.PackageManager.expect_new('autotest',
                                                     repo_urls=['repo'],
                                                     hostname='hostname')

        cfile = self.god.create_mock_class(file, "file")
        cfile_orig = "original control file"
        cfile_new = "args = []\njob.add_repository(['repo'])\n"
        cfile_new += cfile_orig

        autotest.open.expect_call("temp").and_return(cfile)
        cfile.read.expect_call().and_return(cfile_orig)
        autotest.open.expect_call("temp", 'w').and_return(cfile)
        cfile.write.expect_call(cfile_new)

        self.host.job.preprocess_client_state.expect_call().and_return(
            '/job/tmp/file1')
        self.host.send_file.expect_call(
            "/job/tmp/file1", "autodir/control.None.autoserv.init.state")
        os.remove.expect_call("/job/tmp/file1")

        self.host.send_file.expect_call("temp", run_obj.remote_control_file)
        os.path.abspath.expect_call('temp').and_return('control_file')
        os.path.abspath.expect_call('control').and_return('control')
        os.remove.expect_call("temp")

        run_obj.execute_control.expect_call(timeout=30,
                                            client_disconnect_timeout=240)

        # run and check output
        self.base_autotest.run(control, timeout=30)
        self.god.check_playback()


    def _stub_get_client_autodir_paths(self):
        def mock_get_client_autodir_paths(cls, host):
            return ['/some/path', '/another/path']
        self.god.stub_with(autotest.Autotest, 'get_client_autodir_paths',
                           classmethod(mock_get_client_autodir_paths))


    def _expect_failed_run(self, command):
        (self.host.run.expect_call(command)
         .and_raises(error.AutoservRunError('dummy', object())))


    def test_get_installed_autodir(self):
        self._stub_get_client_autodir_paths()
        self.host.get_autodir.expect_call().and_return(None)
        self._expect_failed_run('test -x /some/path/bin/autotest')
        self.host.run.expect_call('test -x /another/path/bin/autotest')
        self.host.run.expect_call('test -w /another/path')

        autodir = autotest.Autotest.get_installed_autodir(self.host)
        self.assertEquals(autodir, '/another/path')


    def test_get_install_dir(self):
        self._stub_get_client_autodir_paths()
        self.host.get_autodir.expect_call().and_return(None)
        self._expect_failed_run('test -x /some/path/bin/autotest')
        self._expect_failed_run('test -x /another/path/bin/autotest')
        self._expect_failed_run('mkdir -p /some/path')
        self.host.run.expect_call('mkdir -p /another/path')
        self.host.run.expect_call('test -w /another/path')

        install_dir = autotest.Autotest.get_install_dir(self.host)
        self.assertEquals(install_dir, '/another/path')


    def test_client_logger_process_line_log_copy_collection_failure(self):
        collector = autotest.log_collector.expect_new(self.host, '', '')
        logger = autotest.client_logger(self.host, '', '')
        collector.collect_client_job_results.expect_call().and_raises(
                Exception('log copy failure'))
        logging.exception.expect_call(mock.is_string_comparator())
        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo1')


    def test_client_logger_process_line_log_copy_fifo_failure(self):
        collector = autotest.log_collector.expect_new(self.host, '', '')
        logger = autotest.client_logger(self.host, '', '')
        collector.collect_client_job_results.expect_call()
        self.host.run.expect_call('echo A > /autotest/fifo2').and_raises(
                Exception('fifo failure'))
        logging.exception.expect_call(mock.is_string_comparator())
        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo2')


    def test_client_logger_process_line_package_install_fifo_failure(self):
        collector = autotest.log_collector.expect_new(self.host, '', '')
        logger = autotest.client_logger(self.host, '', '')
        self.god.stub_function(logger, '_send_tarball')

        c = autotest.global_config.global_config
        c.get_config_value.expect_call('PACKAGES',
                                       'serve_packages_from_autoserv',
                                       type=bool).and_return(True)
        c.get_config_value.expect_call('PACKAGES',
                                       'serve_packages_from_autoserv',
                                       type=bool).and_return(True)
        logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')

        self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(
                Exception('fifo failure'))
        logging.exception.expect_call(mock.is_string_comparator())
        logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'
                             '/autotest/dest/:/autotest/fifo3')


class test_autotest_mixin(unittest.TestCase):
    def setUp(self):
        # a dummy Autotest and job class for use in the mixin
        class stub_autotest(object):
            class job(object):
                state_dict = {}
                def get_state(self, var, default):
                    return self.state_dict.get(var, default)
            job = job()

            @staticmethod
            def run(control_file, host=None):
                self.control_file = control_file
                self.host = host

        self.mixin = autotest.AutotestHostMixin()
        self.mixin._Autotest = stub_autotest
        self.job = self.mixin._Autotest.job


    def test_passes(self):
        self.job.state_dict['test_result'] = True
        self.assertEqual(True, self.mixin.run_test('sleeptest', seconds=1))
        self.assert_("job.run_test('sleeptest', seconds=1)\n"
                     in self.control_file)
        self.assertEqual(self.mixin, self.host)


    def test_fails_clean(self):
        self.job.state_dict['test_result'] = False
        self.assertEqual(False, self.mixin.run_test('sleeptest', seconds='2'))
        self.assert_("job.run_test('sleeptest', seconds='2')\n"
                     in self.control_file)
        self.assertEqual(self.mixin, self.host)


    def test_fails_with_exception(self):
        self.assertEqual(False, self.mixin.run_test('sleeptest'))
        self.assert_("job.run_test('sleeptest')\n" in self.control_file)
        self.assertEqual(self.mixin, self.host)


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