job_unittest.py revision e39c3c0dbe16caed55f4538af78dbf057deb7fad 
1#!/usr/bin/python
2
3import os, unittest, shutil, sys, time
4import common
5
6from autotest_lib.client.bin import job, boottool, config, sysinfo, harness
7from autotest_lib.client.bin import test, xen, kernel, autotest_utils, cpuset
8from autotest_lib.client.bin import autotest_utils
9from autotest_lib.client.common_lib import packages, utils, error, log
10from autotest_lib.client.common_lib.test_utils import mock
11
12
13class first_line_comparator(mock.argument_comparator):
14    def __init__(self, first_line):
15        self.first_line = first_line
16
17
18    def is_satisfied_by(self, parameter):
19        return self.first_line == parameter.splitlines()[0]
20
21
22class TestBaseJob(unittest.TestCase):
23    def setUp(self):
24        # make god
25        self.god = mock.mock_god()
26
27        # need to set some environ variables
28        self.autodir = "autodir"
29        os.environ['AUTODIR'] = self.autodir
30
31        # set up some variables
32        self.control = "control"
33        self.jobtag = "jobtag"
34
35        # stub out some stuff
36        self.god.stub_function(os.path, 'exists')
37        self.god.stub_function(os.path, 'isdir')
38        self.god.stub_function(os, 'mkdir')
39        self.god.stub_function(shutil, 'copyfile')
40        self.god.stub_function(job, 'open')
41        self.god.stub_function(utils, 'system')
42        self.god.stub_function(autotest_utils, 'drop_caches')
43        self.god.stub_function(harness, 'select')
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        self.god.stub_class(sysinfo, 'sysinfo')
49
50
51    def tearDown(self):
52        self.god.unstub_all()
53
54
55    def construct_job(self, cont):
56        # will construct class instance using __new__
57        self.job = job.base_job.__new__(job.base_job)
58
59        # now some specific stubs
60        self.god.stub_function(self.job, '_load_state')
61        self.god.stub_function(self.job, '_init_group_level')
62        self.god.stub_function(self.job, 'config_get')
63        self.god.stub_function(self.job, 'record')
64        self.god.stub_function(self.job, '_increment_group_level')
65        self.god.stub_function(self.job, '_decrement_group_level')
66        self.god.stub_function(self.job, 'get_state')
67
68        # other setup
69        tmpdir = os.path.join(self.autodir, 'tmp')
70        results = os.path.join(self.autodir, 'results')
71        download = os.path.join(self.autodir, 'tests', 'download')
72        resultdir = os.path.join(self.autodir, 'results', self.jobtag)
73        pkgdir = os.path.join(self.autodir, 'packages')
74
75        # record
76        autotest_utils.drop_caches.expect_call()
77        self.job._load_state.expect_call()
78        self.job.get_state.expect_call("__run_test_cleanup",
79                                       default=True).and_return(True)
80        job_sysinfo = sysinfo.sysinfo.expect_new(resultdir)
81        self.job.get_state.expect_call("__sysinfo",
82                                       None).and_return(None)
83        self.job.get_state.expect_call("__last_boot_tag",
84                                       default=None).and_return(None)
85        if not cont:
86            os.path.exists.expect_call(tmpdir).and_return(False)
87            os.mkdir.expect_call(tmpdir)
88            os.path.exists.expect_call(pkgdir).and_return(False)
89            os.mkdir.expect_call(pkgdir)
90            os.path.exists.expect_call(results).and_return(False)
91            os.mkdir.expect_call(results)
92            os.path.exists.expect_call(download).and_return(False)
93            os.mkdir.expect_call(download)
94            os.path.exists.expect_call(resultdir).and_return(True)
95            utils.system.expect_call('rm -rf ' + resultdir)
96            os.mkdir.expect_call(resultdir)
97            os.mkdir.expect_call(os.path.join(resultdir, 'debug'))
98            os.mkdir.expect_call(os.path.join(resultdir,
99                                              'analysis'))
100            shutil.copyfile.expect_call(mock.is_string_comparator(),
101                                 os.path.join(resultdir, 'control'))
102
103        self.job._init_group_level.expect_call()
104        self.config = config.config.expect_new(self.job)
105        my_harness = self.god.create_mock_class(harness.harness,
106                                                'my_harness')
107        harness.select.expect_call(None,
108                                   self.job).and_return(my_harness)
109        self.job.config_get.expect_call(
110                'boottool.executable').and_return(None)
111        bootloader = boottool.boottool.expect_new(None)
112        job_sysinfo.log_per_reboot_data.expect_call()
113        if not cont:
114            self.job.record.expect_call('START', None, None)
115            self.job._increment_group_level.expect_call()
116
117        my_harness.run_start.expect_call()
118        self.job.get_state.expect_call('__monitor_disk',
119                                       default=0.0).and_return(0.0)
120
121        # finish constructor
122        self.job.__init__(self.control, self.jobtag, cont)
123
124        # check
125        self.god.check_playback()
126
127
128    def test_constructor(self):
129        self.construct_job(False)
130
131
132    def test_monitor_disk_usage(self):
133        self.construct_job(True)
134
135        # setup
136        self.god.stub_function(self.job, 'set_state')
137
138        # record
139        max_rate = 10.0
140        self.job.set_state.expect_call('__monitor_disk', max_rate)
141
142        # test
143        self.job.monitor_disk_usage(max_rate)
144        self.god.check_playback()
145
146
147    def test_relative_path(self):
148        self.construct_job(True)
149        dummy = "asdf"
150        ret = self.job.relative_path(os.path.join(self.job.resultdir, dummy))
151        self.assertEquals(ret, dummy)
152
153
154    def test_control_functions(self):
155        self.construct_job(True)
156        control_file = "blah"
157        self.job.control_set(control_file)
158        self.assertEquals(self.job.control_get(), os.path.abspath(control_file))
159
160
161    def test_harness_select(self):
162        self.construct_job(True)
163
164        # record
165        which = "which"
166        harness.select.expect_call(which, self.job).and_return(None)
167
168        # run and test
169        self.job.harness_select(which)
170        self.god.check_playback()
171
172
173    def test_config_set(self):
174        self.construct_job(True)
175
176        # record
177        name = "foo"
178        val = 10
179        self.config.set.expect_call(name, val)
180
181        # run and test
182        self.job.config_set(name, val)
183        self.god.check_playback()
184
185
186    def test_config_get(self):
187        self.construct_job(True)
188
189        # unstub config_get
190        self.god.unstub(self.job, 'config_get')
191        # record
192        name = "foo"
193        val = 10
194        self.config.get.expect_call(name).and_return(val)
195
196        # run and test
197        self.job.config_get(name)
198        self.god.check_playback()
199
200
201    def test_setup_dirs_raise(self):
202        self.construct_job(True)
203
204        # setup
205        results_dir = 'foo'
206        tmp_dir = 'bar'
207
208        # record
209        os.path.exists.expect_call(tmp_dir).and_return(True)
210        os.path.isdir.expect_call(tmp_dir).and_return(False)
211
212        # test
213        self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir)
214        self.god.check_playback()
215
216
217    def test_setup_dirs(self):
218        self.construct_job(True)
219
220        # setup
221        results_dir1 = os.path.join(self.job.resultdir, 'build')
222        results_dir2 = os.path.join(self.job.resultdir, 'build.2')
223        results_dir3 = os.path.join(self.job.resultdir, 'build.3')
224        tmp_dir = 'bar'
225
226        # record
227        os.path.exists.expect_call(tmp_dir).and_return(False)
228        os.mkdir.expect_call(tmp_dir)
229        os.path.isdir.expect_call(tmp_dir).and_return(True)
230        os.path.exists.expect_call(results_dir1).and_return(True)
231        os.path.exists.expect_call(results_dir2).and_return(True)
232        os.path.exists.expect_call(results_dir3).and_return(False)
233        os.path.exists.expect_call(results_dir3).and_return(False)
234        os.mkdir.expect_call(results_dir3)
235
236        # test
237        self.assertEqual(self.job.setup_dirs(None, tmp_dir),
238                         (results_dir3, tmp_dir))
239        self.god.check_playback()
240
241
242    def test_xen(self):
243        self.construct_job(True)
244
245        # setup
246        self.god.stub_function(self.job, "setup_dirs")
247        self.god.stub_class(xen, "xen")
248        results = 'results_dir'
249        tmp = 'tmp'
250        build = 'xen'
251        base_tree = object()
252
253        # record
254        self.job.setup_dirs.expect_call(results,
255                                        tmp).and_return((results, tmp))
256        myxen = xen.xen.expect_new(self.job, base_tree, results, tmp, build,
257                                   False, None)
258
259        # run job and check
260        axen = self.job.xen(base_tree, results, tmp)
261        self.god.check_playback()
262        self.assertEquals(myxen, axen)
263
264
265    def test_kernel_rpm(self):
266        self.construct_job(True)
267
268        # setup
269        self.god.stub_function(self.job, "setup_dirs")
270        self.god.stub_class(kernel, "rpm_kernel")
271        self.god.stub_function(kernel, "preprocess_path")
272        self.god.stub_function(self.job.pkgmgr, "fetch_pkg")
273        results = 'results_dir'
274        tmp = 'tmp'
275        build = 'xen'
276        path = "somepath.rpm"
277        packages_dir = os.path.join("autodir/packages", path)
278
279        # record
280        self.job.setup_dirs.expect_call(results,
281                                        tmp).and_return((results, tmp))
282        kernel.preprocess_path.expect_call(path).and_return(path)
283        self.job.pkgmgr.fetch_pkg.expect_call(path, packages_dir, repo_url='')
284        mykernel = kernel.rpm_kernel.expect_new(self.job, packages_dir,
285                                                results)
286
287        # check
288        akernel = self.job.kernel(path, results, tmp)
289        self.god.check_playback()
290        self.assertEquals(mykernel, akernel)
291
292
293    def test_kernel(self):
294        self.construct_job(True)
295
296        # setup
297        self.god.stub_function(self.job, "setup_dirs")
298        self.god.stub_class(kernel, "kernel")
299        self.god.stub_function(kernel, "preprocess_path")
300        results = 'results_dir'
301        tmp = 'tmp'
302        build = 'linux'
303        path = "somepath.deb"
304
305        # record
306        self.job.setup_dirs.expect_call(results,
307                                        tmp).and_return((results, tmp))
308        kernel.preprocess_path.expect_call(path).and_return(path)
309        mykernel = kernel.kernel.expect_new(self.job, path, results, tmp,
310                                            build, False)
311
312        # check
313        akernel = self.job.kernel(path, results, tmp)
314        self.god.check_playback()
315        self.assertEquals(mykernel, akernel)
316
317
318    def test_run_test_logs_test_error_from_unhandled_error(self):
319        self.construct_job(True)
320
321        # set up stubs
322        self.god.stub_function(self.job.pkgmgr, 'get_package_name')
323        self.god.stub_function(self.job, "_runtest")
324
325        # create an unhandled error object
326        class MyError(error.TestError):
327            pass
328        real_error = MyError("this is the real error message")
329        unhandled_error = error.UnhandledTestError(real_error)
330
331        # set up the recording
332        testname = "error_test"
333        outputdir = os.path.join(self.job.resultdir, testname)
334        self.job.pkgmgr.get_package_name.expect_call(
335            testname, 'test').and_return(("", testname))
336        self.job.get_state.expect_call(
337                self.job._RUN_NUMBER_STATE, default=0).and_return(0)
338        self.job.get_state.expect_call(
339                self.job._KERNEL_IN_TAG_STATE, default=False).and_return(False)
340        os.path.exists.expect_call(outputdir).and_return(False)
341        os.mkdir.expect_call(outputdir)
342        self.job.record.expect_call("START", testname, testname)
343        self.job._increment_group_level.expect_call()
344        self.job._runtest.expect_call(testname, "", (), {}).and_raises(
345            unhandled_error)
346        self.job.record.expect_call("ERROR", testname, testname,
347                                    first_line_comparator(str(real_error)))
348        self.job._decrement_group_level.expect_call()
349        self.job.record.expect_call("END ERROR", testname, testname,
350                                    first_line_comparator(str(real_error)))
351        self.job.harness.run_test_complete.expect_call()
352        autotest_utils.drop_caches.expect_call()
353
354        # run and check
355        self.job.run_test(testname)
356        self.god.check_playback()
357
358
359    def test_run_test_logs_non_test_error_from_unhandled_error(self):
360        self.construct_job(True)
361
362        # set up stubs
363        self.god.stub_function(self.job.pkgmgr, 'get_package_name')
364        self.god.stub_function(self.job, "_runtest")
365
366        # create an unhandled error object
367        class MyError(Exception):
368            pass
369        real_error = MyError("this is the real error message")
370        unhandled_error = error.UnhandledTestError(real_error)
371        reason = first_line_comparator("Unhandled MyError: %s" % real_error)
372
373        # set up the recording
374        testname = "error_test"
375        outputdir = os.path.join(self.job.resultdir, testname)
376        self.job.pkgmgr.get_package_name.expect_call(
377            testname, 'test').and_return(("", testname))
378        self.job.get_state.expect_call(
379                self.job._RUN_NUMBER_STATE, default=0).and_return(0)
380        self.job.get_state.expect_call(
381                self.job._KERNEL_IN_TAG_STATE, default=False).and_return(False)
382        os.path.exists.expect_call(outputdir).and_return(False)
383        os.mkdir.expect_call(outputdir)
384        self.job.record.expect_call("START", testname, testname)
385        self.job._increment_group_level.expect_call()
386        self.job._runtest.expect_call(testname, "", (), {}).and_raises(
387            unhandled_error)
388        self.job.record.expect_call("ERROR", testname, testname, reason)
389        self.job._decrement_group_level.expect_call()
390        self.job.record.expect_call("END ERROR", testname, testname, reason)
391        self.job.harness.run_test_complete.expect_call()
392        autotest_utils.drop_caches.expect_call()
393
394        # run and check
395        self.job.run_test(testname)
396        self.god.check_playback()
397
398
399    def new_container(self):
400        self.construct_job(True)
401
402        # set up stubs
403        self.god.stub_function(autotest_utils, "grep")
404        self.god.stub_function(os, "getpid")
405        self.god.stub_class(cpuset, "cpuset")
406        pid = 100
407        name = 'test%d' % pid
408
409        # record
410        autotest_utils.grep.expect_call('cpuset',
411            '/proc/filesystems').and_return(True)
412        os.getpid.expect_call().and_return(pid)
413
414        container = cpuset.cpuset.expect_new(name, job_size=None, job_pid=pid,
415                                             cpus=None, root=None, disk=None,
416                                             network=None, kswapd_merge=False)
417
418        # run test
419        self.job.new_container()
420
421        self.god.check_playback()
422        return container
423
424
425    def test_new_container(self):
426        container = self.new_container()
427        self.assertEquals(self.job.container, container)
428
429
430    def test_release_container(self):
431        self.new_container()
432
433        # record
434        self.job.container.release.expect_call()
435
436        # run
437        self.job.release_container()
438        self.god.check_playback()
439        self.assertEquals(self.job.container, None)
440
441
442    def test_record(self):
443        self.construct_job(True)
444
445        # steup
446        self.job.group_level = 1
447        status = ''
448        status_code = "PASS"
449        subdir = "subdir"
450        operation = "super_fun"
451        mytime = "1234"
452        msg_tag = ""
453        if "." in self.job.log_filename:
454            msg_tag = self.job.log_filename.split(".", 1)[1]
455        epoch_time = int(mytime)
456        local_time = time.localtime(epoch_time)
457        optional_fields = {}
458        optional_fields["timestamp"] = str(epoch_time)
459        optional_fields["localtime"] = time.strftime("%b %d %H:%M:%S",
460                                                     local_time)
461        fields = [status_code, subdir, operation]
462        fields += ["%s=%s" % x for x in optional_fields.iteritems()]
463        fields.append(status)
464        msg = '\t'.join(str(x) for x in fields)
465        msg = '\t' * self.job.group_level + msg
466
467        self.god.stub_function(log, "is_valid_status")
468        self.god.stub_function(time, "time")
469        self.god.stub_function(self.job, "open")
470
471
472        # record
473        log.is_valid_status.expect_call(status_code).and_return(True)
474        time.time.expect_call().and_return(mytime)
475        self.job.harness.test_status_detail.expect_call(status_code, subdir,
476                                                        operation, '', msg_tag)
477        self.job.harness.test_status.expect_call(msg, msg_tag)
478        myfile = self.god.create_mock_class(file, "file")
479        status_file = os.path.join(self.job.resultdir, self.job.log_filename)
480        self.job.open.expect_call(status_file, "a").and_return(myfile)
481        myfile.write.expect_call(msg + "\n")
482
483        dir = os.path.join(self.job.resultdir, subdir)
484        status_file = os.path.join(dir, self.job.DEFAULT_LOG_FILENAME)
485        self.job.open.expect_call(status_file, "a").and_return(myfile)
486        myfile.write.expect_call(msg + "\n")
487
488
489        # run test
490        self.god.unstub(self.job, "record")
491        self.job.record(status_code, subdir, operation)
492        self.god.check_playback()
493
494
495    def test_end_boot(self):
496        self.construct_job(True)
497
498        # set up the job class
499        self.job.group_level = 2
500
501        self.job._decrement_group_level.expect_call()
502        self.job.record.expect_call("END GOOD", "sub", "reboot",
503                                    optional_fields={"kernel": "2.6.15-smp",
504                                                     "patch0": "patchname"})
505
506        # run test
507        self.job.end_reboot("sub", "2.6.15-smp", ["patchname"])
508        self.god.check_playback()
509
510
511    def test_end_boot_and_verify_success(self):
512        self.construct_job(True)
513
514        # set up the job class
515        self.job.group_level = 2
516
517        self.god.stub_function(autotest_utils, "running_os_ident")
518        autotest_utils.running_os_ident.expect_call().and_return("2.6.15-smp")
519
520        self.god.stub_function(utils, "read_one_line")
521        utils.read_one_line.expect_call("/proc/cmdline").and_return(
522            "blah more-blah root=lala IDENT=81234567 blah-again")
523
524        self.job.record.expect_call("GOOD", "sub", "reboot.verify",
525                                    "2.6.15-smp")
526        self.job._decrement_group_level.expect_call()
527        self.job.record.expect_call("END GOOD", "sub", "reboot",
528                                    optional_fields={"kernel": "2.6.15-smp"})
529
530        # run test
531        self.job.end_reboot_and_verify(81234567, "2.6.15-smp", "sub")
532        self.god.check_playback()
533
534
535    def test_end_boot_and_verify_failure(self):
536        self.construct_job(True)
537
538        # set up the job class
539        self.job.group_level = 2
540
541        self.god.stub_function(autotest_utils, "running_os_ident")
542        autotest_utils.running_os_ident.expect_call().and_return("2.6.15-smp")
543
544        self.god.stub_function(utils, "read_one_line")
545        utils.read_one_line.expect_call("/proc/cmdline").and_return(
546            "blah more-blah root=lala IDENT=81234567 blah-again")
547
548        self.job.record.expect_call("ABORT", "sub", "reboot.verify",
549                                    "boot failure")
550        self.job._decrement_group_level.expect_call()
551        self.job.record.expect_call("END ABORT", "sub", "reboot",
552                                    optional_fields={"kernel": "2.6.15-smp"})
553
554        # run test
555        self.assertRaises(error.JobError, self.job.end_reboot_and_verify,
556                          91234567, "2.6.16-smp", "sub")
557        self.god.check_playback()
558
559
560if __name__ == "__main__":
561    unittest.main()
562