job_unittest.py revision 6f731364e6207c198ac8de748a043eba1e9223b3
1#!/usr/bin/python 2 3import os, unittest, pickle, shutil, sys 4import common 5 6from autotest_lib.client.bin import job, boottool, config, sysinfo, harness 7from autotest_lib.client.bin import test 8from autotest_lib.client.common_lib import utils, error 9from autotest_lib.client.common_lib.test_utils import mock 10 11 12class first_line_comparator(mock.argument_comparator): 13 def __init__(self, first_line): 14 self.first_line = first_line 15 16 17 def is_satisfied_by(self, parameter): 18 return self.first_line == parameter.splitlines()[0] 19 20 21class TestBaseJob(unittest.TestCase): 22 def setUp(self): 23 # make god 24 self.god = mock.mock_god() 25 26 # need to set some environ variables 27 self.autodir = "autodir" 28 os.environ['AUTODIR'] = self.autodir 29 30 # set up some variables 31 self.control = "control" 32 self.jobtag = "jobtag" 33 34 # stub out some stuff 35 self.god.stub_function(os.path, 'exists') 36 self.god.stub_function(os.path, 'isdir') 37 self.god.stub_function(os, 'mkdir') 38 self.god.stub_function(shutil, 'copyfile') 39 self.god.stub_function(job, 'open') 40 self.god.stub_function(utils, 'system') 41 self.god.stub_function(harness, 'select') 42 self.god.stub_function(sysinfo, 'log_per_reboot_data') 43 self.god.stub_function(pickle, 'load') 44 self.god.stub_function(sysinfo, 'log_per_reboot_data') 45 46 self.god.stub_class(config, 'config') 47 self.god.stub_class(boottool, 'boottool') 48 49 50 def tearDown(self): 51 self.god.unstub_all() 52 53 54 def construct_job(self, cont): 55 # will construct class instance using __new__ 56 self.job = job.base_job.__new__(job.base_job) 57 58 # now some specific stubs 59 self.god.stub_function(self.job, '_load_state') 60 self.god.stub_function(self.job, '_init_group_level') 61 self.god.stub_function(self.job, 'config_get') 62 self.god.stub_function(self.job, 'record') 63 self.god.stub_function(self.job, '_increment_group_level') 64 self.god.stub_function(self.job, '_decrement_group_level') 65 self.god.stub_function(self.job, 'get_state') 66 67 # other setup 68 tmpdir = os.path.join(self.autodir, 'tmp') 69 results = os.path.join(self.autodir, 'results') 70 download = os.path.join(self.autodir, 'tests', 'download') 71 resultdir = os.path.join(self.autodir, 'results', self.jobtag) 72 sysinfodir = os.path.join(resultdir, 'sysinfo') 73 74 # record 75 self.job._load_state.expect_call() 76 if not cont: 77 os.path.exists.expect_call(tmpdir).and_return(False) 78 os.mkdir.expect_call(tmpdir) 79 os.path.exists.expect_call(results).and_return(False) 80 os.mkdir.expect_call(results) 81 os.path.exists.expect_call(download).and_return(False) 82 os.mkdir.expect_call(download) 83 os.path.exists.expect_call(resultdir).and_return(True) 84 utils.system.expect_call('rm -rf ' + resultdir) 85 os.mkdir.expect_call(resultdir) 86 os.mkdir.expect_call(sysinfodir) 87 os.mkdir.expect_call(os.path.join(resultdir, 'debug')) 88 os.mkdir.expect_call(os.path.join(resultdir, 89 'analysis')) 90 shutil.copyfile.expect_call(mock.is_string_comparator(), 91 os.path.join(resultdir, 'control')) 92 93 self.job._init_group_level.expect_call() 94 self.config = config.config.expect_new(self.job) 95 my_harness = self.god.create_mock_class(harness.harness, 96 'my_harness') 97 harness.select.expect_call(None, 98 self.job).and_return(my_harness) 99 self.job.config_get.expect_call( 100 'boottool.executable').and_return(None) 101 bootloader = boottool.boottool.expect_new(None) 102 sysinfo.log_per_reboot_data.expect_call(sysinfodir) 103 if not cont: 104 self.job.record.expect_call('START', None, None) 105 self.job._increment_group_level.expect_call() 106 107 my_harness.run_start.expect_call() 108 self.job.get_state.expect_call('__monitor_disk', 109 default=0.0).and_return(0.0) 110 111 # finish constructor 112 self.job.__init__(self.control, self.jobtag, cont) 113 114 # check 115 self.god.check_playback() 116 117 118 def test_constructor(self): 119 self.construct_job(False) 120 121 122 def test_monitor_disk_usage(self): 123 self.construct_job(True) 124 125 # setup 126 self.god.stub_function(self.job, 'set_state') 127 128 # record 129 max_rate = 10.0 130 self.job.set_state.expect_call('__monitor_disk', max_rate) 131 132 # test 133 self.job.monitor_disk_usage(max_rate) 134 self.god.check_playback() 135 136 137 def test_relitive_path(self): 138 self.construct_job(True) 139 dummy = "asdf" 140 ret = self.job.relative_path(os.path.join(self.job.resultdir, dummy)) 141 self.assertEquals(ret, dummy) 142 143 144 def test_control_functions(self): 145 self.construct_job(True) 146 control_file = "blah" 147 self.job.control_set(control_file) 148 self.assertEquals(self.job.control_get(), os.path.abspath(control_file)) 149 150 151 def test_harness_select(self): 152 self.construct_job(True) 153 154 # record 155 which = "which" 156 harness.select.expect_call(which, self.job).and_return(None) 157 158 # run and test 159 self.job.harness_select(which) 160 self.god.check_playback() 161 162 163 def test_config_set(self): 164 self.construct_job(True) 165 166 # record 167 name = "foo" 168 val = 10 169 self.config.set.expect_call(name, val) 170 171 # run and test 172 self.job.config_set(name, val) 173 self.god.check_playback() 174 175 176 def test_config_get(self): 177 self.construct_job(True) 178 179 # unstub config_get 180 self.god.unstub(self.job, 'config_get') 181 # record 182 name = "foo" 183 val = 10 184 self.config.get.expect_call(name).and_return(val) 185 186 # run and test 187 self.job.config_get(name) 188 self.god.check_playback() 189 190 191 def test_setup_dirs_raise(self): 192 self.construct_job(True) 193 194 # setup 195 results_dir = 'foo' 196 tmp_dir = 'bar' 197 198 # record 199 os.path.exists.expect_call(tmp_dir).and_return(True) 200 os.path.isdir.expect_call(tmp_dir).and_return(False) 201 202 # test 203 self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir) 204 self.god.check_playback() 205 206 207 def test_setup_dirs(self): 208 self.construct_job(True) 209 210 # setup 211 results_dir1 = os.path.join(self.job.resultdir, 'build') 212 results_dir2 = os.path.join(self.job.resultdir, 'build.2') 213 results_dir3 = os.path.join(self.job.resultdir, 'build.3') 214 tmp_dir = 'bar' 215 216 # record 217 os.path.exists.expect_call(tmp_dir).and_return(False) 218 os.mkdir.expect_call(tmp_dir) 219 os.path.isdir.expect_call(tmp_dir).and_return(True) 220 os.path.exists.expect_call(results_dir1).and_return(True) 221 os.path.exists.expect_call(results_dir2).and_return(True) 222 os.path.exists.expect_call(results_dir3).and_return(False) 223 os.path.exists.expect_call(results_dir3).and_return(False) 224 os.mkdir.expect_call(results_dir3) 225 226 # test 227 self.assertEqual(self.job.setup_dirs(None, tmp_dir), 228 (results_dir3, tmp_dir)) 229 self.god.check_playback() 230 231 232 def test_run_test_logs_test_error_from_unhandled_error(self): 233 self.construct_job(True) 234 235 # set up stubs 236 self.god.stub_function(test, "testname") 237 self.god.stub_function(self.job, "_runtest") 238 239 # create an unhandled error object 240 class MyError(error.TestError): 241 pass 242 real_error = MyError("this is the real error message") 243 unhandled_error = error.UnhandledError(real_error) 244 245 # set up the recording 246 testname = "error_test" 247 outputdir = os.path.join(self.job.resultdir, testname) 248 test.testname.expect_call(testname).and_return(("", testname)) 249 os.path.exists.expect_call(outputdir).and_return(False) 250 os.mkdir.expect_call(outputdir) 251 self.job.record.expect_call("START", testname, testname) 252 self.job._increment_group_level.expect_call() 253 self.job._runtest.expect_call(testname, "", (), {}).and_raises( 254 unhandled_error) 255 self.job.record.expect_call("FAIL", testname, testname, 256 first_line_comparator(str(real_error))) 257 self.job._decrement_group_level.expect_call() 258 self.job.record.expect_call("END FAIL", testname, testname, 259 first_line_comparator(str(real_error))) 260 261 # run and check 262 self.job.run_test(testname) 263 self.god.check_playback() 264 265 266 def test_run_test_logs_non_test_error_from_unhandled_error(self): 267 self.construct_job(True) 268 269 # set up stubs 270 self.god.stub_function(test, "testname") 271 self.god.stub_function(self.job, "_runtest") 272 273 # create an unhandled error object 274 class MyError(Exception): 275 pass 276 real_error = MyError("this is the real error message") 277 unhandled_error = error.UnhandledError(real_error) 278 reason = first_line_comparator("Unhandled MyError: %s" % real_error) 279 280 # set up the recording 281 testname = "error_test" 282 outputdir = os.path.join(self.job.resultdir, testname) 283 test.testname.expect_call(testname).and_return(("", testname)) 284 os.path.exists.expect_call(outputdir).and_return(False) 285 os.mkdir.expect_call(outputdir) 286 self.job.record.expect_call("START", testname, testname) 287 self.job._increment_group_level.expect_call() 288 self.job._runtest.expect_call(testname, "", (), {}).and_raises( 289 unhandled_error) 290 self.job.record.expect_call("FAIL", testname, testname, reason) 291 self.job._decrement_group_level.expect_call() 292 self.job.record.expect_call("END FAIL", testname, testname, reason) 293 294 # run and check 295 self.job.run_test(testname) 296 self.god.check_playback() 297 298if __name__ == "__main__": 299 unittest.main() 300