server/autotest_unittest.py

   1 #!/usr/bin/python
   2
   3 __author__ = "raphtee@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 utils as client_utils, 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 TestBaseAutotest(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.last_boot_tag = 'Autotest'
  27         self.host.job.sysinfo = self.god.create_mock_class(
  28             sysinfo.sysinfo, "sysinfo")
  29         self.host.job.profilers = self.god.create_mock_class(
  30             profilers.profilers, "profilers")
  31         self.host.job.profilers.add_log = {}
  32         self.host.job.tmpdir = "/job/tmp"
  33         self.host.job.default_profile_only = False
  34         self.host.job.args = []
  35         self.host.job.record = 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.base_autotest = autotest.BaseAutotest(self.host)
  74         self.base_autotest.job = self.host.job
  75         self.god.stub_function(self.base_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.base_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         os.getcwd.expect_call().and_return('cwd')
  95         os.chdir.expect_call(os.path.join(self.serverdir, '../client'))
  96         utils.system.expect_call('tools/make_clean', ignore_status=True)
  97         os.chdir.expect_call('cwd')
  98         utils.get.expect_call(os.path.join(self.serverdir,
  99             '../client')).and_return('source_material')
 100
 101         self.host.wait_up.expect_call(timeout=30)
 102         self.host.setup.expect_call()
 103         self.host.get_autodir.expect_call().and_return("autodir")
 104         self.host.set_autodir.expect_call("autodir")
 105         self.host.run.expect_call('mkdir -p autodir')
 106         self.host.run.expect_call('rm -rf autodir/results/*',
 107                                   ignore_status=True)
 108
 109
 110     def test_constructor(self):
 111         self.construct()
 112
 113         # we should check the calls
 114         self.god.check_playback()
 115
 116
 117     def test_full_client_install(self):
 118         self.record_install_prologue()
 119
 120         c = autotest.global_config.global_config
 121         c.get_config_value.expect_call('PACKAGES',
 122                                        'serve_packages_from_autoserv',
 123                                        type=bool).and_return(False)
 124         self.host.send_file.expect_call('source_material', 'autodir',
 125                                         delete_dest=True)
 126
 127         # run and check
 128         self.base_autotest.install_full_client()
 129         self.god.check_playback()
 130
 131
 132     def test_autoserv_install(self):
 133         self.record_install_prologue()
 134
 135         c = autotest.global_config.global_config
 136         c.get_config_value.expect_call('PACKAGES',
 137             'fetch_location', type=list, default=[]).and_return([])
 138
 139         c.get_config_value.expect_call('PACKAGES',
 140                                        'serve_packages_from_autoserv',
 141                                        type=bool).and_return(True)
 142         self.base_autotest._install_using_send_file.expect_call(self.host,
 143                                                                 'autodir')
 144         # run and check
 145         self.base_autotest.install()
 146         self.god.check_playback()
 147
 148
 149     def test_packaging_install(self):
 150         self.record_install_prologue()
 151
 152         c = autotest.global_config.global_config
 153         c.get_config_value.expect_call('PACKAGES',
 154             'fetch_location', type=list, default=[]).and_return(['repo'])
 155         pkgmgr = packages.PackageManager.expect_new('autodir',
 156             repo_urls=['repo'], hostname='hostname', do_locking=False,
 157             run_function=self.host.run, run_function_dargs=dict(timeout=600))
 158         pkg_dir = os.path.join('autodir', 'packages')
 159         cmd = ('cd autodir && ls | grep -v "^packages$"'
 160                ' | xargs rm -rf && rm -rf .[^.]*')
 161         self.host.run.expect_call(cmd)
 162         pkgmgr.install_pkg.expect_call('autotest', 'client', pkg_dir,
 163                                        'autodir', preserve_install_dir=True)
 164
 165         # run and check
 166         self.base_autotest.install()
 167         self.god.check_playback()
 168
 169
 170     def test_run(self):
 171         self.construct()
 172
 173         # setup
 174         control = "control"
 175
 176         # stub out install
 177         self.god.stub_function(self.base_autotest, "install")
 178         self.god.stub_class(packages, "PackageManager")
 179
 180         # record
 181         self.base_autotest.install.expect_call(self.host)
 182         self.host.wait_up.expect_call(timeout=30)
 183         os.path.abspath.expect_call('.').and_return('.')
 184         run_obj = autotest._Run.expect_new(self.host, '.', None, False, False)
 185         tag = None
 186         run_obj.manual_control_file = os.path.join('autodir',
 187                                                    'control.%s' % tag)
 188         run_obj.remote_control_file = os.path.join('autodir',
 189                                                    'control.%s.autoserv' % tag)
 190         run_obj.tag = tag
 191         run_obj.autodir = 'autodir'
 192         run_obj.verify_machine.expect_call()
 193         run_obj.verify_machine.expect_call()
 194         run_obj.background = False
 195         debug = os.path.join('.', 'debug')
 196         os.makedirs.expect_call(debug)
 197         delete_file_list = [run_obj.remote_control_file,
 198                             run_obj.remote_control_file + '.state',
 199                             run_obj.manual_control_file,
 200                             run_obj.manual_control_file + '.state']
 201         cmd = ';'.join('rm -f ' + control for control in delete_file_list)
 202         self.host.run.expect_call(cmd, ignore_status=True)
 203
 204         utils.get.expect_call(control).and_return("temp")
 205
 206         c = autotest.global_config.global_config
 207         c.get_config_value.expect_call("PACKAGES",
 208             'fetch_location', type=list, default=[]).and_return(['repo'])
 209         pkgmgr = packages.PackageManager.expect_new('autotest',
 210                                                      repo_urls=['repo'],
 211                                                      hostname='hostname')
 212
 213         cfile = self.god.create_mock_class(file, "file")
 214         cfile_orig = "original control file"
 215         cfile_new = "args = []\njob.add_repository(['repo'])\n"
 216         cfile_new += cfile_orig
 217
 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=1800)
 236
 237         # run and check output
 238         self.base_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         logger._send_tarball.expect_call('pkgname.tar.bz2', '/autotest/dest/')
 307
 308         self.host.run.expect_call('echo B > /autotest/fifo3').and_raises(
 309                 Exception('fifo failure'))
 310         logging.exception.expect_call(mock.is_string_comparator())
 311         logger._process_line('AUTOTEST_FETCH_PACKAGE:pkgname.tar.bz2:'
 312                              '/autotest/dest/:/autotest/fifo3')
 313
 314
 315 class test_autotest_mixin(unittest.TestCase):
 316     def setUp(self):
 317         # a dummy Autotest and job class for use in the mixin
 318         class stub_autotest(object):
 319             class job(object):
 320                 state_dict = {}
 321                 def get_state(self, var, default):
 322                     return self.state_dict.get(var, default)
 323             job = job()
 324
 325             @staticmethod
 326             def run(control_file, host=None):
 327                 self.control_file = control_file
 328                 self.host = host
 329
 330         self.mixin = autotest.AutotestHostMixin()
 331         self.mixin._Autotest = stub_autotest
 332         self.job = self.mixin._Autotest.job
 333
 334
 335     def test_passes(self):
 336         self.job.state_dict['test_result'] = True
 337         self.assertEqual(True, self.mixin.run_test('sleeptest', seconds=1))
 338         self.assert_("job.run_test('sleeptest', seconds=1)\n"
 339                      in self.control_file)
 340         self.assertEqual(self.mixin, self.host)
 341
 342
 343     def test_fails_clean(self):
 344         self.job.state_dict['test_result'] = False
 345         self.assertEqual(False, self.mixin.run_test('sleeptest', seconds='2'))
 346         self.assert_("job.run_test('sleeptest', seconds='2')\n"
 347                      in self.control_file)
 348         self.assertEqual(self.mixin, self.host)
 349
 350
 351     def test_fails_with_exception(self):
 352         self.assertEqual(False, self.mixin.run_test('sleeptest'))
 353         self.assert_("job.run_test('sleeptest')\n" in self.control_file)
 354         self.assertEqual(self.mixin, self.host)
 355
 356
 357 if __name__ == "__main__":
 358     unittest.main()