How to use _server_system_wide_install method in autotest

Best Python code snippet using autotest_python

autotest_remote_unittest.py

Source:autotest_remote_unittest.py

1#!/usr/bin/python2__author__ = "raphtee@google.com (Travis Miller)"3import logging4import os5import tempfile6import unittest7try:8    import autotest.common as common  # pylint: disable=W06119except ImportError:10    import common  # pylint: disable=W061111from autotest.client import sysinfo12from autotest.client import utils as client_utils13from autotest.client.shared import error14from autotest.client.shared import packages15from autotest.client.shared.test_utils import mock16from autotest.server import autotest_remote, utils, hosts, server_job, profilers17class TestBaseAutotest(unittest.TestCase):18    def setUp(self):19        # create god20        self.god = mock.mock_god()21        # create mock host object22        self.host = self.god.create_mock_class(hosts.RemoteHost, "host")23        self.host.hostname = "hostname"24        self.host.job = self.god.create_mock_class(server_job.server_job,25                                                   "job")26        self.host.job.run_test_cleanup = True27        self.host.job.last_boot_tag = 'Autotest'28        self.host.job.sysinfo = self.god.create_mock_class(29            sysinfo.sysinfo, "sysinfo")30        self.host.job.profilers = self.god.create_mock_class(31            profilers.profilers, "profilers")32        self.host.job.profilers.add_log = {}33        self.host.job.tmpdir = "/job/tmp"34        self.host.job.default_profile_only = False35        self.host.job.args = []36        self.host.job.record = lambda *args: None37        # stubs38        self.god.stub_function(utils, "get_server_dir")39        self.god.stub_function(utils, "run")40        self.god.stub_function(utils, "get")41        self.god.stub_function(utils, "read_keyval")42        self.god.stub_function(utils, "write_keyval")43        self.god.stub_function(client_utils, "get_os_vendor")44        self.god.stub_function(utils, "system")45        self.god.stub_function(tempfile, "mkstemp")46        self.god.stub_function(tempfile, "mktemp")47        self.god.stub_function(os, "getcwd")48        self.god.stub_function(os, "system")49        self.god.stub_function(os, "chdir")50        self.god.stub_function(os, "makedirs")51        self.god.stub_function(os, "remove")52        self.god.stub_function(os, "fdopen")53        self.god.stub_function(os.path, "exists")54        self.god.stub_function(autotest_remote, "open")55        self.god.stub_function(autotest_remote.settings, "get_value")56        self.god.stub_function(autotest_remote, "_server_system_wide_install")57        self.god.stub_function(autotest_remote, "_client_system_wide_install")58        self.god.stub_function(logging, "exception")59        self.god.stub_class(autotest_remote, "_Run")60        self.god.stub_class(autotest_remote, "log_collector")61    def tearDown(self):62        self.god.unstub_all()63    def construct(self):64        # setup65        self.serverdir = "serverdir"66        # record67        utils.get_server_dir.expect_call().and_return(self.serverdir)68        client_utils.get_os_vendor.expect_call().and_return('Debian')69        autotest_remote._server_system_wide_install.expect_call().and_return(False)70        # create the autotest object71        self.base_autotest = autotest_remote.BaseAutotest(self.host)72        self.base_autotest.job = self.host.job73        self.god.stub_function(self.base_autotest, "_install_using_send_file")74        # stub out abspath75        self.god.stub_function(os.path, "abspath")76        # check77        self.god.check_playback()78    def record_install_prologue(self):79        self.construct()80        # setup81        self.god.stub_class(packages, "PackageManager")82        self.base_autotest.got = False83        location = os.path.join(self.serverdir, '../client')84        location = os.path.abspath.expect_call(location).and_return(location)85        # record86        os.getcwd.expect_call().and_return('cwd')87        os.chdir.expect_call(os.path.join(self.serverdir, '../client'))88        utils.system.expect_call('tools/make_clean', ignore_status=True)89        os.chdir.expect_call('cwd')90        utils.get.expect_call(os.path.join(self.serverdir,91                                           '../client')).and_return('source_material')92        self.host.wait_up.expect_call(timeout=30)93        self.host.setup.expect_call()94        self.host.get_autodir.expect_call().and_return("autodir")95        self.host.set_autodir.expect_call("autodir")96        self.host.run.expect_call('mkdir -p autodir')97        self.host.run.expect_call('rm -rf autodir/results/*',98                                  ignore_status=True)99    def test_constructor(self):100        self.construct()101        # we should check the calls102        self.god.check_playback()103    def test_full_client_install(self):104        self.record_install_prologue()105        autotest_remote.settings.get_value.expect_call('PACKAGES',106                                                       'serve_packages_from_autoserv',107                                                       type=bool).and_return(False)108        self.host.send_file.expect_call('source_material', 'autodir',109                                        delete_dest=True)110        tmpdir = 'autodir/tmp'111        autotest_remote.settings.get_value.expect_call('COMMON',112                                                       'test_output_dir',113                                                       default=tmpdir).and_return(tmpdir)114        self.host.run.expect_call('mkdir -p %s' % tmpdir)115        # run and check116        self.base_autotest.install_full_client()117        self.god.check_playback()118    def test_autoserv_install(self):119        self.record_install_prologue()120        autotest_remote.settings.get_value.expect_call('PACKAGES',121                                                       'fetch_location', type=list, default=[]).and_return([])122        autotest_remote.settings.get_value.expect_call('PACKAGES',123                                                       'serve_packages_from_autoserv',124                                                       type=bool).and_return(True)125        self.base_autotest._install_using_send_file.expect_call(self.host,126                                                                'autodir')127        tmpdir = 'autodir/tmp'128        autotest_remote.settings.get_value.expect_call('COMMON',129                                                       'test_output_dir',130                                                       default=tmpdir).and_return(tmpdir)131        self.host.run.expect_call('mkdir -p %s' % tmpdir)132        # run and check133        self.base_autotest.install()134        self.god.check_playback()135    def test_packaging_install(self):136        self.record_install_prologue()137        autotest_remote.settings.get_value.expect_call('PACKAGES',138                                                       'fetch_location', type=list, default=[]).and_return(['repo'])139        pkgmgr = packages.PackageManager.expect_new('autodir',140                                                    repo_urls=['repo'], hostname='hostname', do_locking=False,141                                                    run_function=self.host.run, run_function_dargs=dict(timeout=600))142        pkg_dir = os.path.join('autodir', 'packages')143        cmd = ('cd autodir && ls | grep -v "^packages$"'144               ' | xargs rm -rf && rm -rf .[^.]*')145        self.host.run.expect_call(cmd)146        pkgmgr.install_pkg.expect_call('autotest', 'client', pkg_dir,147                                       'autodir', preserve_install_dir=True)148        tmpdir = 'autodir/tmp'149        autotest_remote.settings.get_value.expect_call('COMMON',150                                                       'test_output_dir',151                                                       default=tmpdir).and_return(tmpdir)152        self.host.run.expect_call('mkdir -p %s' % tmpdir)153        autotest_remote.settings.get_value.expect_call('COMMON',154                                                       'test_output_dir',155                                                       default=tmpdir).and_return(tmpdir)156        self.host.run.expect_call('mkdir -p %s' % tmpdir)157        # run and check158        self.base_autotest.install()159        self.god.check_playback()160    def test_run(self):161        self.construct()162        # setup163        control = "control"164        # stub out install165        self.god.stub_function(self.base_autotest, "install")166        self.god.stub_class(packages, "PackageManager")167        # record168        self.base_autotest.install.expect_call(self.host)169        self.host.wait_up.expect_call(timeout=30)170        os.path.abspath.expect_call('.').and_return('.')171        run_obj = autotest_remote._Run.expect_new(self.host, '.', None, False, False)172        tag = None173        run_obj.manual_control_file = os.path.join('autodir',174                                                   'control.%s' % tag)175        run_obj.manual_control_state = os.path.join('autodir',176                                                    'control.%s.state' % tag)177        run_obj.remote_control_file = os.path.join('autodir',178                                                   'control.%s.autoserv' % tag)179        run_obj.remote_control_state = os.path.join('autodir',180                                                    'control.%s.autoserv.state' % tag)181        run_obj.remote_control_init_state = os.path.join('autodir',182                                                         'control.%s.autoserv.init.state' % tag)183        run_obj.tag = tag184        run_obj.autodir = 'autodir'185        run_obj.verify_machine.expect_call()186        run_obj.verify_machine.expect_call()187        run_obj.background = False188        debug = os.path.join('.', 'debug')189        os.makedirs.expect_call(debug)190        delete_file_list = [run_obj.remote_control_file,191                            run_obj.remote_control_state,192                            run_obj.manual_control_file,193                            run_obj.manual_control_state]194        cmd = ';'.join('rm -f ' + control for control in delete_file_list)195        self.host.run.expect_call(cmd, ignore_status=True)196        utils.get.expect_call(control).and_return("temp")197        autotest_remote.settings.get_value.expect_call("PACKAGES",198                                                       'fetch_location', type=list, default=[]).and_return(['repo'])199        pkgmgr = packages.PackageManager.expect_new('autotest',200                                                    repo_urls=['repo'],201                                                    hostname='hostname')202        cfile = self.god.create_mock_class(file, "file")203        cfile_orig = "original control file"204        cfile_new = "args = []\njob.add_repository(['repo'])\n"205        cfile_new += cfile_orig206        autotest_remote.open.expect_call("temp").and_return(cfile)207        cfile.read.expect_call().and_return(cfile_orig)208        autotest_remote.open.expect_call("temp", 'w').and_return(cfile)209        cfile.write.expect_call(cfile_new)210        self.host.job.preprocess_client_state.expect_call().and_return(211            '/job/tmp/file1')212        self.host.send_file.expect_call(213            "/job/tmp/file1", "autodir/control.None.autoserv.init.state")214        os.remove.expect_call("/job/tmp/file1")215        self.host.send_file.expect_call("temp", run_obj.remote_control_file)216        os.path.abspath.expect_call('temp').and_return('control_file')217        os.path.abspath.expect_call('control').and_return('control')218        os.remove.expect_call("temp")219        run_obj.execute_control.expect_call(timeout=30,220                                            client_disconnect_timeout=1800)221        # run and check output222        self.base_autotest.run(control, timeout=30)223        self.god.check_playback()224    def _stub_get_client_autodir_paths(self):225        def mock_get_client_autodir_paths(cls, host):226            return ['/some/path', '/another/path']227        self.god.stub_with(autotest_remote.Autotest, 'get_client_autodir_paths',228                           classmethod(mock_get_client_autodir_paths))229    def _expect_failed_run(self, command):230        (self.host.run.expect_call(command)231         .and_raises(error.AutoservRunError('dummy', object())))232    def test_get_installed_autodir(self):233        self._stub_get_client_autodir_paths()234        self.host.get_autodir.expect_call().and_return(None)235        autotest_remote._server_system_wide_install.expect_call().and_return(False)236        self._expect_failed_run('test -x /some/path/autotest-local-streamhandler')237        self.host.run.expect_call('test -x /another/path/autotest-local-streamhandler')238        self.host.run.expect_call('test -w /another/path')239        autodir = autotest_remote.Autotest.get_installed_autodir(self.host)240        self.assertEquals(autodir, '/another/path')241    def test_get_install_dir(self):242        self._stub_get_client_autodir_paths()243        self.host.get_autodir.expect_call().and_return(None)244        autotest_remote._server_system_wide_install.expect_call().and_return(False)245        self._expect_failed_run('test -x /some/path/autotest-local-streamhandler')246        self._expect_failed_run('test -x /another/path/autotest-local-streamhandler')247        self._expect_failed_run('mkdir -p /some/path')248        self.host.run.expect_call('mkdir -p /another/path')249        self.host.run.expect_call('test -w /another/path')250        install_dir = autotest_remote.Autotest.get_install_dir(self.host)251        self.assertEquals(install_dir, '/another/path')252    def test_client_logger_process_line_log_copy_collection_failure(self):253        collector = autotest_remote.log_collector.expect_new(self.host, '', '')254        logger = autotest_remote.client_logger(self.host, '', '')255        collector.collect_client_job_results.expect_call().and_raises(256            Exception('log copy failure'))257        logging.exception.expect_call(mock.is_string_comparator())258        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo1')259    def test_client_logger_process_line_log_copy_fifo_failure(self):260        collector = autotest_remote.log_collector.expect_new(self.host, '', '')261        logger = autotest_remote.client_logger(self.host, '', '')262        collector.collect_client_job_results.expect_call()263        self.host.run.expect_call('echo A > /autotest/fifo2').and_raises(264            Exception('fifo failure'))265        logging.exception.expect_call(mock.is_string_comparator())266        logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo2')267    def test_client_logger_process_line_package_install_fifo_failure(self):268        collector = autotest_remote.log_collector.expect_new(self.host, '', '')269        logger = autotest_remote.client_logger(self.host, '', '')270        self.god.stub_function(logger, '_send_tarball')271        autotest_remote.settings.get_value.expect_call('PACKAGES',272                                                       'serve_packages_from_autoserv',273                                                       type=bool).and_return(True)274        logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')275        self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(276            Exception('fifo failure'))277        logging.exception.expect_call(mock.is_string_comparator())278        logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'279                             '/autotest/dest/:/autotest/fifo3')280class test_autotest_mixin(unittest.TestCase):281    def setUp(self):282        # a dummy Autotest and job class for use in the mixin283        class stub_autotest(object):284            class job(object):285                state_dict = {}286                def get_state(self, var, default):287                    return self.state_dict.get(var, default)288            job = job()289            @staticmethod290            def run(control_file, host=None):291                self.control_file = control_file292                self.host = host293        self.mixin = autotest_remote.AutotestHostMixin()294        self.mixin._Autotest = stub_autotest295        self.job = self.mixin._Autotest.job296    def test_passes(self):297        self.job.state_dict['test_result'] = True298        self.assertEqual(True, self.mixin.run_test('sleeptest', seconds=1))299        self.assert_("job.run_test('sleeptest', seconds=1)\n"300                     in self.control_file)301        self.assertEqual(self.mixin, self.host)302    def test_fails_clean(self):303        self.job.state_dict['test_result'] = False304        self.assertEqual(False, self.mixin.run_test('sleeptest', seconds='2'))305        self.assert_("job.run_test('sleeptest', seconds='2')\n"306                     in self.control_file)307        self.assertEqual(self.mixin, self.host)308    def test_fails_with_exception(self):309        self.assertEqual(False, self.mixin.run_test('sleeptest'))310        self.assert_("job.run_test('sleeptest')\n" in self.control_file)311        self.assertEqual(self.mixin, self.host)312if __name__ == "__main__":...

auto_remote.py

Source:auto_remote.py

...9import logging10from caliper.server import utils11from caliper.server.hosts import installable_object12from caliper.client.shared.settings import settings13#def _server_system_wide_install():14#    for path in SYSYTEM_WIDE_PATHS:15#        try:16#            os_dep.command(path)17#        except ValueError:18#            return False19#    return True20class BaseAutotest(installable_object.InstallableObject):21    """22    This class represents the auto program. It can be used to run test automatically23    and collect the results.24    """25    def __init__(self, host=None):26        self.host = host27        self.got = False28        self.installed = False29        self.serverdir = utils.get_server_dir()30        #self.os_vendor = client_utils.get_os_vendor()31        #self.server_system_wide_install = _server_system_wide_install()32        super(BaseAutotest, self).__init__()33    install_in_tmpdir = False34    @classmethod35    def get_client_autodir_paths(cls, host):36        return settings.get_value('auto', 'client_autodir_paths', type=list)37    def install(self, host=None, autodir=None):38        self._install(host=host, autodir=autodir)39    def _install(self, host=None, autodir=None, use_auto=True, use_packaging=True):40        """41        Install Caliper.42        :param host: A host instance on which caliper will be installed43        :param autodir: Location on the remote host to install to44        :param use_auto: Enable install modes that depend on the client running with 'autoserv harness'45        :param use_packaging: Enable install modes that use the packaging system...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.