Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use construct_job method in autotest

Best Python code snippet using autotest_python
job_unittest.py
Source:job_unittest.py
...57        self.god.stub_class_method(job.base_job, '_cleanup_results_dir')58    def tearDown(self):59        sys.stdout = sys.__stdout__60        self.god.unstub_all()61    def construct_job(self, cont):62        # will construct class instance using __new__63        self.job = job.base_job.__new__(job.base_job)64        # now some specific stubs65        self.god.stub_function(self.job, '_load_state')66        self.god.stub_function(self.job, '_init_group_level')67        self.god.stub_function(self.job, 'config_get')68        self.god.stub_function(self.job, 'config_set')69        self.god.stub_function(self.job, 'record')70        self.god.stub_function(self.job, '_increment_group_level')71        self.god.stub_function(self.job, '_decrement_group_level')72        self.god.stub_function(self.job, 'get_state')73        self.god.stub_function(self.job, 'set_state')74        # other setup75        tmpdir = os.path.join(self.autodir, 'tmp')76        results = os.path.join(self.autodir, 'results')77        download = os.path.join(self.autodir, 'tests', 'download')78        resultdir = os.path.join(self.autodir, 'results', self.jobtag)79        pkgdir = os.path.join(self.autodir, 'packages')80        # record81        os.path.exists.expect_call(resultdir).and_return(False)82        os.makedirs.expect_call(resultdir)83        if not cont:84            job.base_job._cleanup_results_dir.expect_call()85        utils.drop_caches.expect_call()86        self.job._load_state.expect_call()87        self.job.get_state.expect_call("__run_test_cleanup",88                                       default=True).and_return(True)89        job_sysinfo = sysinfo.sysinfo.expect_new(resultdir)90        self.job.get_state.expect_call("__sysinfo",91                                       None).and_return(None)92        self.job.get_state.expect_call("__last_boot_tag",93                                       default=None).and_return(None)94        self.job.get_state.expect_call("__job_tag",95                                       default=None).and_return('1337-gps')96        if not cont:97            os.path.exists.expect_call(tmpdir).and_return(False)98            os.mkdir.expect_call(tmpdir)99            os.path.exists.expect_call(pkgdir).and_return(False)100            os.mkdir.expect_call(pkgdir)101            os.path.exists.expect_call(results).and_return(False)102            os.mkdir.expect_call(results)103            os.path.exists.expect_call(download).and_return(False)104            os.mkdir.expect_call(download)105            os.makedirs.expect_call(os.path.join(resultdir, 'analysis'))106            shutil.copyfile.expect_call(mock.is_string_comparator(),107                                 os.path.join(resultdir, 'control'))108        self.job._init_group_level.expect_call()109        self.config = config.config.expect_new(self.job)110        my_harness = self.god.create_mock_class(harness.harness,111                                                'my_harness')112        harness.select.expect_call(None,113                                   self.job).and_return(my_harness)114        self.job.config_get.expect_call(115                'boottool.executable').and_return(None)116        bootloader = boottool.boottool.expect_new(None)117        job_sysinfo.log_per_reboot_data.expect_call()118        if not cont:119            self.job.record.expect_call('START', None, None)120            self.job._increment_group_level.expect_call()121        my_harness.run_start.expect_call()122        self.job.get_state.expect_call('__monitor_disk',123                                       default=0.0).and_return(0.0)124        self.god.stub_function(utils, 'read_one_line')125        utils.read_one_line.expect_call('/proc/cmdline').and_return(126            'blah more-blah root=lala IDENT=81234567 blah-again console=tty1')127        self.job.config_set.expect_call('boot.default_args',128                                        'more-blah console=tty1')129        # finish constructor130        options = Dummy()131        options.tag = self.jobtag132        options.cont = cont133        options.harness = None134        options.log = False135        options.verbose = False136        self.job.__init__(self.control, options,137                          extra_copy_cmdline=['more-blah'])138        # check139        self.god.check_playback()140    def get_partition_mock(self, devname):141        """142        Create a mock of a partition object and return it.143        """144        class mock(object):145            device = devname146            get_mountpoint = self.god.create_mock_function('get_mountpoint')147        return mock148    def test_constructor(self):149        self.construct_job(False)150    def test_monitor_disk_usage(self):151        self.construct_job(True)152        # record153        max_rate = 10.0154        self.job.set_state.expect_call('__monitor_disk', max_rate)155        # test156        self.job.monitor_disk_usage(max_rate)157        self.god.check_playback()158    def test_relative_path(self):159        self.construct_job(True)160        dummy = "asdf"161        ret = self.job.relative_path(os.path.join(self.job.resultdir, dummy))162        self.assertEquals(ret, dummy)163    def test_control_functions(self):164        self.construct_job(True)165        control_file = "blah"166        self.job.control_set(control_file)167        self.assertEquals(self.job.control_get(), os.path.abspath(control_file))168    def test_harness_select(self):169        self.construct_job(True)170        # record171        which = "which"172        harness.select.expect_call(which, self.job).and_return(None)173        # run and test174        self.job.harness_select(which)175        self.god.check_playback()176    def test_config_set(self):177        self.construct_job(True)178        # unstub config_set179        self.god.unstub(self.job, 'config_set')180        # record181        name = "foo"182        val = 10183        self.config.set.expect_call(name, val)184        # run and test185        self.job.config_set(name, val)186        self.god.check_playback()187    def test_config_get(self):188        self.construct_job(True)189        # unstub config_get190        self.god.unstub(self.job, 'config_get')191        # record192        name = "foo"193        val = 10194        self.config.get.expect_call(name).and_return(val)195        # run and test196        self.job.config_get(name)197        self.god.check_playback()198    def test_setup_dirs_raise(self):199        self.construct_job(True)200        # setup201        results_dir = 'foo'202        tmp_dir = 'bar'203        # record204        os.path.exists.expect_call(tmp_dir).and_return(True)205        os.path.isdir.expect_call(tmp_dir).and_return(False)206        # test207        self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir)208        self.god.check_playback()209    def test_setup_dirs(self):210        self.construct_job(True)211        # setup212        results_dir1 = os.path.join(self.job.resultdir, 'build')213        results_dir2 = os.path.join(self.job.resultdir, 'build.2')214        results_dir3 = os.path.join(self.job.resultdir, 'build.3')215        tmp_dir = 'bar'216        # record217        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        # test226        self.assertEqual(self.job.setup_dirs(None, tmp_dir),227                         (results_dir3, tmp_dir))228        self.god.check_playback()229    def test_xen(self):230        self.construct_job(True)231        # setup232        self.god.stub_function(self.job, "setup_dirs")233        self.god.stub_class(xen, "xen")234        results = 'results_dir'235        tmp = 'tmp'236        build = 'xen'237        base_tree = object()238        # record239        self.job.setup_dirs.expect_call(results,240                                        tmp).and_return((results, tmp))241        myxen = xen.xen.expect_new(self.job, base_tree, results, tmp, build,242                                   False, None)243        # run job and check244        axen = self.job.xen(base_tree, results, tmp)245        self.god.check_playback()246        self.assertEquals(myxen, axen)247    def test_kernel_rpm(self):248        self.construct_job(True)249        # setup250        self.god.stub_function(self.job, "setup_dirs")251        self.god.stub_class(kernel, "rpm_kernel")252        self.god.stub_function(kernel, "preprocess_path")253        self.god.stub_function(self.job.pkgmgr, "fetch_pkg")254        self.god.stub_function(utils, "get_os_vendor")255        results = 'results_dir'256        tmp = 'tmp'257        build = 'xen'258        path = "somepath.rpm"259        packages_dir = os.path.join("autodir/packages", path)260        # record261        self.job.setup_dirs.expect_call(results,262                                        tmp).and_return((results, tmp))263        kernel.preprocess_path.expect_call(path).and_return(path)264        os.path.exists.expect_call(path).and_return(False)265        self.job.pkgmgr.fetch_pkg.expect_call(path, packages_dir, repo_url='')266        utils.get_os_vendor.expect_call()267        mykernel = kernel.rpm_kernel.expect_new(self.job, [packages_dir],268                                                results)269        # check270        akernel = self.job.kernel(path, results, tmp)271        self.god.check_playback()272        self.assertEquals(mykernel, akernel)273    def test_kernel(self):274        self.construct_job(True)275        # setup276        self.god.stub_function(self.job, "setup_dirs")277        self.god.stub_class(kernel, "kernel")278        self.god.stub_function(kernel, "preprocess_path")279        results = 'results_dir'280        tmp = 'tmp'281        build = 'linux'282        path = "somepath.deb"283        # record284        self.job.setup_dirs.expect_call(results,285                                        tmp).and_return((results, tmp))286        kernel.preprocess_path.expect_call(path).and_return(path)287        mykernel = kernel.kernel.expect_new(self.job, path, results, tmp,288                                            build, False)289        # check290        akernel = self.job.kernel(path, results, tmp)291        self.god.check_playback()292        self.assertEquals(mykernel, akernel)293    def test_run_test_logs_test_error_from_unhandled_error(self):294        self.construct_job(True)295        # set up stubs296        self.god.stub_function(self.job.pkgmgr, 'get_package_name')297        self.god.stub_function(self.job, "_runtest")298        # create an unhandled error object299        class MyError(error.TestError):300            pass301        real_error = MyError("this is the real error message")302        unhandled_error = error.UnhandledTestError(real_error)303        # set up the recording304        testname = "error_test"305        outputdir = os.path.join(self.job.resultdir, testname)306        self.job.pkgmgr.get_package_name.expect_call(307            testname, 'test').and_return(("", testname))308        self.job.get_state.expect_call(309                self.job._RUN_NUMBER_STATE, default=0).and_return(0)310        self.job.get_state.expect_call(311                self.job._KERNEL_IN_TAG_STATE, default=False).and_return(False)312        os.path.exists.expect_call(outputdir).and_return(False)313        os.mkdir.expect_call(outputdir)314        self.job.record.expect_call("START", testname, testname)315        self.job._increment_group_level.expect_call()316        self.job._runtest.expect_call(testname, "", (), {}).and_raises(317            unhandled_error)318        self.job.record.expect_call("ERROR", testname, testname,319                                    first_line_comparator(str(real_error)))320        self.job._decrement_group_level.expect_call()321        self.job.record.expect_call("END ERROR", testname, testname)322        self.job.harness.run_test_complete.expect_call()323        utils.drop_caches.expect_call()324        # run and check325        self.job.run_test(testname)326        self.god.check_playback()327    def test_run_test_logs_non_test_error_from_unhandled_error(self):328        self.construct_job(True)329        # set up stubs330        self.god.stub_function(self.job.pkgmgr, 'get_package_name')331        self.god.stub_function(self.job, "_runtest")332        # create an unhandled error object333        class MyError(Exception):334            pass335        real_error = MyError("this is the real error message")336        unhandled_error = error.UnhandledTestError(real_error)337        reason = first_line_comparator("Unhandled MyError: %s" % real_error)338        # set up the recording339        testname = "error_test"340        outputdir = os.path.join(self.job.resultdir, testname)341        self.job.pkgmgr.get_package_name.expect_call(342            testname, 'test').and_return(("", testname))343        self.job.get_state.expect_call(344                self.job._RUN_NUMBER_STATE, default=0).and_return(0)345        self.job.get_state.expect_call(346                self.job._KERNEL_IN_TAG_STATE, default=False).and_return(False)347        os.path.exists.expect_call(outputdir).and_return(False)348        os.mkdir.expect_call(outputdir)349        self.job.record.expect_call("START", testname, testname)350        self.job._increment_group_level.expect_call()351        self.job._runtest.expect_call(testname, "", (), {}).and_raises(352            unhandled_error)353        self.job.record.expect_call("ERROR", testname, testname, reason)354        self.job._decrement_group_level.expect_call()355        self.job.record.expect_call("END ERROR", testname, testname)356        self.job.harness.run_test_complete.expect_call()357        utils.drop_caches.expect_call()358        # run and check359        self.job.run_test(testname)360        self.god.check_playback()361    def test_record(self):362        self.construct_job(True)363        # steup364        self.job.group_level = 1365        status = ''366        status_code = "PASS"367        subdir = "subdir"368        operation = "super_fun"369        mytime = "1234"370        msg_tag = ""371        if "." in self.job.log_filename:372            msg_tag = self.job.log_filename.split(".", 1)[1]373        epoch_time = int(mytime)374        local_time = time.localtime(epoch_time)375        optional_fields = {}376        optional_fields["timestamp"] = str(epoch_time)377        optional_fields["localtime"] = time.strftime("%b %d %H:%M:%S",378                                                     local_time)379        fields = [status_code, subdir, operation]380        fields += ["%s=%s" % x for x in optional_fields.iteritems()]381        fields.append(status)382        msg = '\t'.join(str(x) for x in fields)383        msg = '\t' * self.job.group_level + msg384        self.god.stub_function(log, "is_valid_status")385        self.god.stub_function(time, "time")386        self.god.stub_function(self.job, "open")387        # record388        log.is_valid_status.expect_call(status_code).and_return(True)389        time.time.expect_call().and_return(mytime)390        self.job.harness.test_status_detail.expect_call(status_code, subdir,391                                                        operation, '', msg_tag)392        self.job.harness.test_status.expect_call(msg, msg_tag)393        myfile = self.god.create_mock_class(file, "file")394        status_file = os.path.join(self.job.resultdir, self.job.log_filename)395        self.job.open.expect_call(status_file, "a").and_return(myfile)396        myfile.write.expect_call(msg + "\n")397        dir = os.path.join(self.job.resultdir, subdir)398        status_file = os.path.join(dir, self.job.DEFAULT_LOG_FILENAME)399        self.job.open.expect_call(status_file, "a").and_return(myfile)400        myfile.write.expect_call(msg + "\n")401        # run test402        self.god.unstub(self.job, "record")403        self.job.record(status_code, subdir, operation)404        self.god.check_playback()405    def test_report_reboot_failure(self):406        self.construct_job(True)407        # record408        self.job.record.expect_call("ABORT", "sub", "reboot.verify",409                                    "boot failure")410        self.job._decrement_group_level.expect_call()411        self.job.record.expect_call("END ABORT", "sub", "reboot",412                                    optional_fields={"kernel": "2.6.15-smp"})413        # playback414        self.job._record_reboot_failure("sub", "reboot.verify", "boot failure",415                                        running_id="2.6.15-smp")416        self.god.check_playback()417    def _setup_check_post_reboot(self, mount_info):418        # setup419        self.god.stub_function(job.partition_lib, "get_partition_list")420        part_list = [self.get_partition_mock("/dev/hda1"),421                     self.get_partition_mock("/dev/hdb1")]422        mount_list = ["/mnt/hda1", "/mnt/hdb1"]423        # record424        job.partition_lib.get_partition_list.expect_call(self.job).and_return(425                part_list)426        for i in xrange(len(part_list)):427            part_list[i].get_mountpoint.expect_call().and_return(mount_list[i])428        self.job.get_state.expect_call("__mount_info").and_return(mount_info)429    def test_check_post_reboot_success(self):430        self.construct_job(True)431        mount_info = set([("/dev/hda1", "/mnt/hda1"),432                          ("/dev/hdb1", "/mnt/hdb1")])433        self._setup_check_post_reboot(mount_info)434        # playback435        self.job._check_post_reboot("sub")436        self.god.check_playback()437    def test_check_post_reboot_mounts_failure(self):438        self.construct_job(True)439        mount_info = set([("/dev/hda1", "/mnt/hda1")])440        self._setup_check_post_reboot(mount_info)441        self.god.stub_function(self.job, "_record_reboot_failure")442        self.job._record_reboot_failure.expect_call("sub",443                "reboot.verify_config", "mounted partitions are different after"444                " reboot (old entries: set([]), new entries: set([('/dev/hdb1',"445                " '/mnt/hdb1')]))", running_id=None)446        # playback447        self.assertRaises(error.JobError, self.job._check_post_reboot, "sub")448        self.god.check_playback()449    def test_end_boot(self):450        self.construct_job(True)451        self.god.stub_function(self.job, "_check_post_reboot")452        # set up the job class453        self.job.group_level = 2454        self.job._check_post_reboot.expect_call("sub", running_id=None)455        self.job._decrement_group_level.expect_call()456        self.job.record.expect_call("END GOOD", "sub", "reboot",457                                    optional_fields={"kernel": "2.6.15-smp",458                                                     "patch0": "patchname"})459        # run test460        self.job.end_reboot("sub", "2.6.15-smp", ["patchname"])461        self.god.check_playback()462    def test_end_boot_and_verify_success(self):463        self.construct_job(True)464        self.god.stub_function(self.job, "_check_post_reboot")465        # set up the job class466        self.job.group_level = 2467        self.god.stub_function(utils, "running_os_ident")468        utils.running_os_ident.expect_call().and_return("2.6.15-smp")469        utils.read_one_line.expect_call("/proc/cmdline").and_return(470            "blah more-blah root=lala IDENT=81234567 blah-again")471        self.god.stub_function(utils, "running_os_full_version")472        running_id = "2.6.15-smp"473        utils.running_os_full_version.expect_call().and_return(running_id)474        self.job.record.expect_call("GOOD", "sub", "reboot.verify",475                                    running_id)476        self.job._check_post_reboot.expect_call("sub", running_id=running_id)477        self.job._decrement_group_level.expect_call()478        self.job.record.expect_call("END GOOD", "sub", "reboot",479                                    optional_fields={"kernel": running_id})480        # run test481        self.job.end_reboot_and_verify(81234567, "2.6.15-smp", "sub")482        self.god.check_playback()483    def test_end_boot_and_verify_failure(self):484        self.construct_job(True)485        self.god.stub_function(self.job, "_record_reboot_failure")486        # set up the job class487        self.job.group_level = 2488        self.god.stub_function(utils, "running_os_ident")489        utils.running_os_ident.expect_call().and_return("2.6.15-smp")490        utils.read_one_line.expect_call("/proc/cmdline").and_return(491            "blah more-blah root=lala IDENT=81234567 blah-again")492        self.job._record_reboot_failure.expect_call("sub", "reboot.verify",493                "boot failure", running_id="2.6.15-smp")494        # run test495        self.assertRaises(error.JobError, self.job.end_reboot_and_verify,496                          91234567, "2.6.16-smp", "sub")497        self.god.check_playback()498    def test_default_tag(self):499        self.construct_job(cont=False)500        self.job.set_state.expect_call("__job_tag", None)501        self.job.default_tag(None)502        self.assertEqual(None, self.job.tag)503        self.job.set_state.expect_call("__job_tag", '1337-gps')504        self.job.default_tag('1337-gps')505        self.assertEqual('1337-gps', self.job.tag)506        self.construct_job(cont=True)507        self.job.default_tag(None)508        self.god.check_playback()509if __name__ == "__main__":...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.