How to use get_partition_mock method in autotest

Best Python code snippet using autotest_python

job_unittest.py

Source:job_unittest.py

...193        options.output_dir = ''194        self.job.__init__(self.control, options)195        # check196        self.god.check_playback()197    def get_partition_mock(self, devname):198        """199        Create a mock of a partition object and return it.200        """201        class mock(object):202            device = devname203            get_mountpoint = self.god.create_mock_function('get_mountpoint')204        return mock205    def test_constructor_first_run(self):206        self.construct_job(False)207    def test_constructor_continuation(self):208        self.construct_job(True)209    def test_constructor_post_record_failure(self):210        """211        Test post record initialization failure.212        """213        self.job = job.base_client_job.__new__(job.base_client_job)214        options = dummy()215        options.tag = self.jobtag216        options.cont = False217        options.harness = None218        options.harness_args = None219        options.log = False220        options.verbose = False221        options.hostname = 'localhost'222        options.user = 'my_user'223        options.args = ''224        options.output_dir = ''225        error = Exception('fail')226        self.god.stub_function(self.job, '_post_record_init')227        self.god.stub_function(self.job, 'record')228        self._setup_pre_record_init(False)229        self.job._post_record_init.expect_call(230                self.control, options, True).and_raises(error)231        self.job.record.expect_call(232                'ABORT', None, None,'client.bin.job.__init__ failed: %s' %233                str(error))234        self.assertRaises(235                Exception, self.job.__init__, self.control, options,236                drop_caches=True)237        # check238        self.god.check_playback()239    def test_control_functions(self):240        self.construct_job(True)241        control_file = "blah"242        self.job.control_set(control_file)243        self.assertEquals(self.job.control_get(), os.path.abspath(control_file))244    def test_harness_select(self):245        self.construct_job(True)246        # record247        which = "which"248        harness_args = ''249        harness.select.expect_call(which, self.job, 250                                   harness_args).and_return(None)251        # run and test252        self.job.harness_select(which, harness_args)253        self.god.check_playback()254    def test_setup_dirs_raise(self):255        self.construct_job(True)256        # setup257        results_dir = 'foo'258        tmp_dir = 'bar'259        # record260        os.path.exists.expect_call(tmp_dir).and_return(True)261        os.path.isdir.expect_call(tmp_dir).and_return(False)262        # test263        self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir)264        self.god.check_playback()265    def test_setup_dirs(self):266        self.construct_job(True)267        # setup268        results_dir1 = os.path.join(self.job.resultdir, 'build')269        results_dir2 = os.path.join(self.job.resultdir, 'build.2')270        results_dir3 = os.path.join(self.job.resultdir, 'build.3')271        tmp_dir = 'bar'272        # record273        os.path.exists.expect_call(tmp_dir).and_return(False)274        os.mkdir.expect_call(tmp_dir)275        os.path.isdir.expect_call(tmp_dir).and_return(True)276        os.path.exists.expect_call(results_dir1).and_return(True)277        os.path.exists.expect_call(results_dir2).and_return(True)278        os.path.exists.expect_call(results_dir3).and_return(False)279        os.path.exists.expect_call(results_dir3).and_return(False)280        os.mkdir.expect_call(results_dir3)281        # test282        self.assertEqual(self.job.setup_dirs(None, tmp_dir),283                         (results_dir3, tmp_dir))284        self.god.check_playback()285    def test_run_test_logs_test_error_from_unhandled_error(self):286        self.construct_job(True)287        # set up stubs288        self.god.stub_function(self.job.pkgmgr, 'get_package_name')289        self.god.stub_function(self.job, "_runtest")290        # create an unhandled error object291        class MyError(error.TestError):292            pass293        real_error = MyError("this is the real error message")294        unhandled_error = error.UnhandledTestError(real_error)295        # set up the recording296        testname = "error_test"297        outputdir = os.path.join(self.job.resultdir, testname)298        self.job.pkgmgr.get_package_name.expect_call(299            testname, 'test').and_return(("", testname))300        os.path.exists.expect_call(outputdir).and_return(False)301        self.job.record.expect_call("START", testname, testname,302                                    optional_fields=None)303        self.job._runtest.expect_call(testname, "", None, (), {}).and_raises(304            unhandled_error)305        self.job.record.expect_call("ERROR", testname, testname,306                                    first_line_comparator(str(real_error)))307        self.job.record.expect_call("END ERROR", testname, testname)308        self.job.harness.run_test_complete.expect_call()309        utils.drop_caches.expect_call()310        # run and check311        self.job.run_test(testname)312        self.god.check_playback()313    def test_run_test_logs_non_test_error_from_unhandled_error(self):314        self.construct_job(True)315        # set up stubs316        self.god.stub_function(self.job.pkgmgr, 'get_package_name')317        self.god.stub_function(self.job, "_runtest")318        # create an unhandled error object319        class MyError(Exception):320            pass321        real_error = MyError("this is the real error message")322        unhandled_error = error.UnhandledTestError(real_error)323        reason = first_line_comparator("Unhandled MyError: %s" % real_error)324        # set up the recording325        testname = "error_test"326        outputdir = os.path.join(self.job.resultdir, testname)327        self.job.pkgmgr.get_package_name.expect_call(328            testname, 'test').and_return(("", testname))329        os.path.exists.expect_call(outputdir).and_return(False)330        self.job.record.expect_call("START", testname, testname,331                                    optional_fields=None)332        self.job._runtest.expect_call(testname, "", None, (), {}).and_raises(333            unhandled_error)334        self.job.record.expect_call("ERROR", testname, testname, reason)335        self.job.record.expect_call("END ERROR", testname, testname)336        self.job.harness.run_test_complete.expect_call()337        utils.drop_caches.expect_call()338        # run and check339        self.job.run_test(testname)340        self.god.check_playback()341    def test_report_reboot_failure(self):342        self.construct_job(True)343        # record344        self.job.record.expect_call("ABORT", "sub", "reboot.verify",345                                    "boot failure")346        self.job.record.expect_call("END ABORT", "sub", "reboot",347                                    optional_fields={"kernel": "2.6.15-smp"})348        # playback349        self.job._record_reboot_failure("sub", "reboot.verify", "boot failure",350                                        running_id="2.6.15-smp")351        self.god.check_playback()352    def _setup_check_post_reboot(self, mount_info, cpu_count):353        # setup354        self.god.stub_function(job.partition_lib, "get_partition_list")355        self.god.stub_function(utils, "count_cpus")356        part_list = [self.get_partition_mock("/dev/hda1"),357                     self.get_partition_mock("/dev/hdb1")]358        mount_list = ["/mnt/hda1", "/mnt/hdb1"]359        # record360        job.partition_lib.get_partition_list.expect_call(361                self.job, exclude_swap=False).and_return(part_list)362        for i in xrange(len(part_list)):363            part_list[i].get_mountpoint.expect_call().and_return(mount_list[i])364        if cpu_count is not None:365            utils.count_cpus.expect_call().and_return(cpu_count)366        self.job._state.set('client', 'mount_info', mount_info)367        self.job._state.set('client', 'cpu_count', 8)368    def test_check_post_reboot_success(self):369        self.construct_job(True)370        mount_info = set([("/dev/hda1", "/mnt/hda1"),371                          ("/dev/hdb1", "/mnt/hdb1")])...

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.