Home | History | Annotate | Download | only in server 
      1 #!/usr/bin/python
      2 #pylint: disable-msg=C0111
      3 __author__ = "raphtee (at] google.com (Travis Miller)"
      4 
      5 import unittest, os, tempfile, logging
      6 
      7 import common
      8 from autotest_lib.server import autotest, utils, hosts, server_job, profilers
      9 from autotest_lib.client.bin import sysinfo
     10 from autotest_lib.client.common_lib import packages
     11 from autotest_lib.client.common_lib import error
     12 from autotest_lib.client.common_lib.test_utils import mock
     13 
     14 
     15 class TestAutotest(unittest.TestCase):
     16     def setUp(self):
     17         # create god
     18         self.god = mock.mock_god()
     19 
     20         # create mock host object
     21         self.host = self.god.create_mock_class(hosts.RemoteHost, "host")
     22         self.host.hostname = "hostname"
     23         self.host.job = self.god.create_mock_class(server_job.server_job,
     24                                                    "job")
     25         self.host.job.run_test_cleanup = True
     26         self.host.job.sysinfo = self.god.create_mock_class(
     27             sysinfo.sysinfo, "sysinfo")
     28         self.host.job.profilers = self.god.create_mock_class(
     29             profilers.profilers, "profilers")
     30         self.host.job.profilers.add_log = {}
     31         self.host.job.tmpdir = "/job/tmp"
     32         self.host.job.default_profile_only = False
     33         self.host.job.args = []
     34         self.host.job.record = lambda *args: None
     35         self.host.verify_job_repo_url = lambda *args: None
     36 
     37         # stubs
     38         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(utils, "system")
     44         self.god.stub_function(tempfile, "mkstemp")
     45         self.god.stub_function(tempfile, "mktemp")
     46         self.god.stub_function(os, "getcwd")
     47         self.god.stub_function(os, "system")
     48         self.god.stub_function(os, "chdir")
     49         self.god.stub_function(os, "makedirs")
     50         self.god.stub_function(os, "remove")
     51         self.god.stub_function(os, "fdopen")
     52         self.god.stub_function(os.path, "exists")
     53         self.god.stub_function(autotest, "open")
     54         self.god.stub_function(autotest.global_config.global_config,
     55                                "get_config_value")
     56         self.god.stub_function(logging, "exception")
     57         self.god.stub_class(autotest, "_Run")
     58         self.god.stub_class(autotest, "log_collector")
     59 
     60 
     61     def tearDown(self):
     62         self.god.unstub_all()
     63 
     64 
     65     def construct(self):
     66         # setup
     67         self.serverdir = "serverdir"
     68 
     69         # record
     70         utils.get_server_dir.expect_call().and_return(self.serverdir)
     71 
     72         # create the autotest object
     73         self.autotest = autotest.Autotest(self.host)
     74         self.autotest.job = self.host.job
     75         self.god.stub_function(self.autotest, "_install_using_send_file")
     76 
     77         # stub out abspath
     78         self.god.stub_function(os.path, "abspath")
     79 
     80         # check
     81         self.god.check_playback()
     82 
     83 
     84     def record_install_prologue(self):
     85         self.construct()
     86 
     87         # setup
     88         self.god.stub_class(packages, "PackageManager")
     89         self.autotest.got = False
     90         location = os.path.join(self.serverdir, '../client')
     91         location = os.path.abspath.expect_call(location).and_return(location)
     92 
     93         # record
     94         utils.get.expect_call(os.path.join(self.serverdir,
     95             '../client')).and_return('source_material')
     96 
     97         self.host.wait_up.expect_call(timeout=30)
     98         self.host.setup.expect_call()
     99         self.host.get_autodir.expect_call().and_return("autodir")
    100         self.host.set_autodir.expect_call("autodir")
    101         self.host.run.expect_call('mkdir -p autodir')
    102         self.host.run.expect_call('rm -rf autodir/results/*',
    103                                   ignore_status=True)
    104 
    105 
    106     def test_constructor(self):
    107         self.construct()
    108 
    109         # we should check the calls
    110         self.god.check_playback()
    111 
    112 
    113     def test_full_client_install(self):
    114         self.record_install_prologue()
    115 
    116         self.host.run.expect_call('rm -f "autodir/packages.checksum"')
    117         c = autotest.global_config.global_config
    118         c.get_config_value.expect_call('PACKAGES',
    119                                        'serve_packages_from_autoserv',
    120                                        type=bool).and_return(False)
    121         self.host.send_file.expect_call('source_material', 'autodir',
    122                                         delete_dest=True)
    123         self.god.stub_function(autotest.Autotest, "_send_shadow_config")
    124         autotest.Autotest._send_shadow_config.expect_call()
    125 
    126         # run and check
    127         self.autotest.install_full_client()
    128         self.god.check_playback()
    129 
    130 
    131     def test_autoserv_install(self):
    132         self.record_install_prologue()
    133 
    134         c = autotest.global_config.global_config
    135         c.get_config_value.expect_call('PACKAGES',
    136             'fetch_location', type=list, default=[]).and_return([])
    137 
    138         os.path.exists.expect_call('/etc/cros_chroot_version').and_return(True)
    139         c.get_config_value.expect_call('PACKAGES',
    140                                        'serve_packages_from_autoserv',
    141                                        type=bool).and_return(True)
    142         self.autotest._install_using_send_file.expect_call(self.host,
    143                                                            'autodir')
    144         self.god.stub_function(autotest.Autotest, "_send_shadow_config")
    145         autotest.Autotest._send_shadow_config.expect_call()
    146         # run and check
    147         self.autotest.install()
    148         self.god.check_playback()
    149 
    150 
    151     def test_packaging_install(self):
    152         self.record_install_prologue()
    153 
    154         c = autotest.global_config.global_config
    155         c.get_config_value.expect_call('PACKAGES',
    156             'fetch_location', type=list, default=[]).and_return(['repo'])
    157         os.path.exists.expect_call('/etc/cros_chroot_version').and_return(True)
    158         pkgmgr = packages.PackageManager.expect_new('autodir',
    159             repo_urls=['repo'], hostname='hostname', do_locking=False,
    160             run_function=self.host.run, run_function_dargs=dict(timeout=600))
    161         pkg_dir = os.path.join('autodir', 'packages')
    162         cmd = ('cd autodir && ls | grep -v "^packages$" | '
    163                'grep -v "^result_tools$" | '
    164                'xargs rm -rf && rm -rf .[!.]*')
    165         self.host.run.expect_call(cmd)
    166         pkgmgr.install_pkg.expect_call('autotest', 'client', pkg_dir,
    167                                        'autodir', preserve_install_dir=True)
    168 
    169         # run and check
    170         self.autotest.install()
    171         self.god.check_playback()
    172 
    173 
    174     def test_run(self):
    175         self.construct()
    176 
    177         # setup
    178         control = "control"
    179 
    180         # stub out install
    181         self.god.stub_function(self.autotest, "install")
    182 
    183         # record
    184         self.autotest.install.expect_call(self.host, use_packaging=True)
    185         self.host.wait_up.expect_call(timeout=30)
    186         os.path.abspath.expect_call('.').and_return('.')
    187         run_obj = autotest._Run.expect_new(self.host, '.', None, False, False)
    188         tag = None
    189         run_obj.manual_control_file = os.path.join('autodir',
    190                                                    'control.%s' % tag)
    191         run_obj.remote_control_file = os.path.join('autodir',
    192                                                    'control.%s.autoserv' % tag)
    193         run_obj.tag = tag
    194         run_obj.autodir = 'autodir'
    195         run_obj.verify_machine.expect_call()
    196         run_obj.background = False
    197         debug = os.path.join('.', 'debug')
    198         os.makedirs.expect_call(debug)
    199         delete_file_list = [run_obj.remote_control_file,
    200                             run_obj.remote_control_file + '.state',
    201                             run_obj.manual_control_file,
    202                             run_obj.manual_control_file + '.state']
    203         cmd = ';'.join('rm -f ' + control for control in delete_file_list)
    204         self.host.run.expect_call(cmd, ignore_status=True)
    205 
    206         utils.get.expect_call(control, local_copy=True).and_return("temp")
    207 
    208         c = autotest.global_config.global_config
    209         c.get_config_value.expect_call("PACKAGES",
    210             'fetch_location', type=list, default=[]).and_return(['repo'])
    211 
    212         cfile = self.god.create_mock_class(file, "file")
    213         cfile_orig = "original control file"
    214         cfile_new = "args = []\njob.add_repository(['repo'])\n"
    215         cfile_new += cfile_orig
    216 
    217         os.path.exists.expect_call('/etc/cros_chroot_version').and_return(True)
    218         autotest.open.expect_call("temp").and_return(cfile)
    219         cfile.read.expect_call().and_return(cfile_orig)
    220         autotest.open.expect_call("temp", 'w').and_return(cfile)
    221         cfile.write.expect_call(cfile_new)
    222 
    223         self.host.job.preprocess_client_state.expect_call().and_return(
    224             '/job/tmp/file1')
    225         self.host.send_file.expect_call(
    226             "/job/tmp/file1", "autodir/control.None.autoserv.init.state")
    227         os.remove.expect_call("/job/tmp/file1")
    228 
    229         self.host.send_file.expect_call("temp", run_obj.remote_control_file)
    230         os.path.abspath.expect_call('temp').and_return('control_file')
    231         os.path.abspath.expect_call('control').and_return('control')
    232         os.remove.expect_call("temp")
    233 
    234         run_obj.execute_control.expect_call(timeout=30,
    235                                             client_disconnect_timeout=240)
    236 
    237         # run and check output
    238         self.autotest.run(control, timeout=30)
    239         self.god.check_playback()
    240 
    241 
    242     def _stub_get_client_autodir_paths(self):
    243         def mock_get_client_autodir_paths(cls, host):
    244             return ['/some/path', '/another/path']
    245         self.god.stub_with(autotest.Autotest, 'get_client_autodir_paths',
    246                            classmethod(mock_get_client_autodir_paths))
    247 
    248 
    249     def _expect_failed_run(self, command):
    250         (self.host.run.expect_call(command)
    251          .and_raises(error.AutoservRunError('dummy', object())))
    252 
    253 
    254     def test_get_installed_autodir(self):
    255         self._stub_get_client_autodir_paths()
    256         self.host.get_autodir.expect_call().and_return(None)
    257         self._expect_failed_run('test -x /some/path/bin/autotest')
    258         self.host.run.expect_call('test -x /another/path/bin/autotest')
    259         self.host.run.expect_call('test -w /another/path')
    260 
    261         autodir = autotest.Autotest.get_installed_autodir(self.host)
    262         self.assertEquals(autodir, '/another/path')
    263 
    264 
    265     def test_get_install_dir(self):
    266         self._stub_get_client_autodir_paths()
    267         self.host.get_autodir.expect_call().and_return(None)
    268         self._expect_failed_run('test -x /some/path/bin/autotest')
    269         self._expect_failed_run('test -x /another/path/bin/autotest')
    270         self._expect_failed_run('mkdir -p /some/path')
    271         self.host.run.expect_call('mkdir -p /another/path')
    272         self.host.run.expect_call('test -w /another/path')
    273 
    274         install_dir = autotest.Autotest.get_install_dir(self.host)
    275         self.assertEquals(install_dir, '/another/path')
    276 
    277 
    278     def test_client_logger_process_line_log_copy_collection_failure(self):
    279         collector = autotest.log_collector.expect_new(self.host, '', '')
    280         logger = autotest.client_logger(self.host, '', '')
    281         collector.collect_client_job_results.expect_call().and_raises(
    282                 Exception('log copy failure'))
    283         logging.exception.expect_call(mock.is_string_comparator())
    284         logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo1')
    285 
    286 
    287     def test_client_logger_process_line_log_copy_fifo_failure(self):
    288         collector = autotest.log_collector.expect_new(self.host, '', '')
    289         logger = autotest.client_logger(self.host, '', '')
    290         collector.collect_client_job_results.expect_call()
    291         self.host.run.expect_call('echo A > /autotest/fifo2').and_raises(
    292                 Exception('fifo failure'))
    293         logging.exception.expect_call(mock.is_string_comparator())
    294         logger._process_line('AUTOTEST_TEST_COMPLETE:/autotest/fifo2')
    295 
    296 
    297     def test_client_logger_process_line_package_install_fifo_failure(self):
    298         collector = autotest.log_collector.expect_new(self.host, '', '')
    299         logger = autotest.client_logger(self.host, '', '')
    300         self.god.stub_function(logger, '_send_tarball')
    301 
    302         c = autotest.global_config.global_config
    303         c.get_config_value.expect_call('PACKAGES',
    304                                        'serve_packages_from_autoserv',
    305                                        type=bool).and_return(True)
    306         c.get_config_value.expect_call('PACKAGES',
    307                                        'serve_packages_from_autoserv',
    308                                        type=bool).and_return(True)
    309         logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')
    310 
    311         self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(
    312                 Exception('fifo failure'))
    313         logging.exception.expect_call(mock.is_string_comparator())
    314         logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'
    315                              '/autotest/dest/:/autotest/fifo3')
    316 
    317 
    318 if __name__ == "__main__":
    319     unittest.main()
    320