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