How to use running_os_full_version method in autotest

Best Python code snippet using autotest_python
job_unittest.py
Source:job_unittest.py
1#!/usr/bin/python2import logging, os, shutil, sys, time, StringIO3import common4from autotest_lib.client.bin import job, boottool, config, sysinfo, harness5from autotest_lib.client.bin import test, xen, kernel, utils6from autotest_lib.client.common_lib import packages, error, log7from autotest_lib.client.common_lib import logging_manager, logging_config8from autotest_lib.client.common_lib import base_job_unittest9from autotest_lib.client.common_lib.test_utils import mock, unittest10class job_test_case(unittest.TestCase):11    """Generic job TestCase class that defines a standard job setUp and12    tearDown, with some standard stubs."""13    job_class = job.base_client_job14    def setUp(self):15        self.god = mock.mock_god(ut=self)16        self.god.stub_with(job.base_client_job, '_get_environ_autodir',17                           classmethod(lambda cls: '/adir'))18        self.job = self.job_class.__new__(self.job_class)19        self.job._job_directory = base_job_unittest.stub_job_directory20    def tearDown(self):21        self.god.unstub_all()22class test_find_base_directories(23        base_job_unittest.test_find_base_directories.generic_tests,24        job_test_case):25    def test_autodir_equals_clientdir(self):26        autodir, clientdir, _ = self.job._find_base_directories()27        self.assertEqual(autodir, '/adir')28        self.assertEqual(clientdir, '/adir')29    def test_serverdir_is_none(self):30        _, _, serverdir = self.job._find_base_directories()31        self.assertEqual(serverdir, None)32class abstract_test_init(base_job_unittest.test_init.generic_tests):33    """Generic client job mixin used when defining variations on the34    job.__init__ generic tests."""35    OPTIONAL_ATTRIBUTES = (36        base_job_unittest.test_init.generic_tests.OPTIONAL_ATTRIBUTES37        - set(['control', 'bootloader', 'harness']))38class test_init_minimal_options(abstract_test_init, job_test_case):39    def call_init(self):40        # TODO(jadmanski): refactor more of the __init__ code to not need to41        # stub out countless random APIs42        self.god.stub_function_to_return(job.os, 'mkdir', None)43        self.god.stub_function_to_return(job.os.path, 'exists', True)44        self.god.stub_function_to_return(self.job, '_load_state', None)45        self.god.stub_function_to_return(self.job, 'record', None)46        self.god.stub_function_to_return(job.shutil, 'copyfile', None)47        self.god.stub_function_to_return(job.logging_manager,48                                         'configure_logging', None)49        class manager:50            def start_logging(self):51                return None52        self.god.stub_function_to_return(job.logging_manager,53                                         'get_logging_manager', manager())54        class stub_sysinfo:55            def log_per_reboot_data(self):56                return None57        self.god.stub_function_to_return(job.sysinfo, 'sysinfo',58                                         stub_sysinfo())59        class stub_harness:60            run_start = lambda self: None61        self.god.stub_function_to_return(job.harness, 'select', stub_harness())62        self.god.stub_function_to_return(job.boottool, 'boottool', object())63        class options:64            tag = ''65            verbose = False66            cont = False67            harness = 'stub'68            harness_args = None69            hostname = None70            user = None71            log = False72            args = ''73            output_dir = ''74            tap_report = None75        self.god.stub_function_to_return(job.utils, 'drop_caches', None)76        self.job._job_state = base_job_unittest.stub_job_state77        self.job.__init__('/control', options)78class dummy(object):79    """A simple placeholder for attributes"""80    pass81class first_line_comparator(mock.argument_comparator):82    def __init__(self, first_line):83        self.first_line = first_line84    def is_satisfied_by(self, parameter):85        return self.first_line == parameter.splitlines()[0]86class test_base_job(unittest.TestCase):87    def setUp(self):88        # make god89        self.god = mock.mock_god(ut=self)90        # need to set some environ variables91        self.autodir = "autodir"92        os.environ['AUTODIR'] = self.autodir93        # set up some variables94        self.control = "control"95        self.jobtag = "jobtag"96        # get rid of stdout and logging97        sys.stdout = StringIO.StringIO()98        logging_manager.configure_logging(logging_config.TestingConfig())99        logging.disable(logging.CRITICAL)100        def dummy_configure_logging(*args, **kwargs):101            pass102        self.god.stub_with(logging_manager, 'configure_logging',103                           dummy_configure_logging)104        real_get_logging_manager = logging_manager.get_logging_manager105        def get_logging_manager_no_fds(manage_stdout_and_stderr=False,106                                       redirect_fds=False):107            return real_get_logging_manager(manage_stdout_and_stderr, False)108        self.god.stub_with(logging_manager, 'get_logging_manager',109                           get_logging_manager_no_fds)110        # stub out some stuff111        self.god.stub_function(os.path, 'exists')112        self.god.stub_function(os.path, 'isdir')113        self.god.stub_function(os, 'makedirs')114        self.god.stub_function(os, 'mkdir')115        self.god.stub_function(os, 'remove')116        self.god.stub_function(shutil, 'rmtree')117        self.god.stub_function(shutil, 'copyfile')118        self.god.stub_function(job, 'open')119        self.god.stub_function(utils, 'system')120        self.god.stub_function(utils, 'drop_caches')121        self.god.stub_function(harness, 'select')122        self.god.stub_function(sysinfo, 'log_per_reboot_data')123        self.god.stub_class(config, 'config')124        self.god.stub_class(job.local_host, 'LocalHost')125        self.god.stub_class(boottool, 'boottool')126        self.god.stub_class(sysinfo, 'sysinfo')127        self.god.stub_class_method(job.base_client_job,128                                   '_cleanup_debugdir_files')129        self.god.stub_class_method(job.base_client_job, '_cleanup_results_dir')130        self.god.stub_with(job.base_job.job_directory, '_ensure_valid',131                           lambda *_: None)132    def tearDown(self):133        sys.stdout = sys.__stdout__134        self.god.unstub_all()135    def _setup_pre_record_init(self, cont):136        self.god.stub_function(self.job, '_load_state')137        resultdir = os.path.join(self.autodir, 'results', self.jobtag)138        tmpdir = os.path.join(self.autodir, 'tmp')139        if not cont:140            job.base_client_job._cleanup_debugdir_files.expect_call()141            job.base_client_job._cleanup_results_dir.expect_call()142        self.job._load_state.expect_call()143        my_harness = self.god.create_mock_class(harness.harness,144                                                'my_harness')145        harness.select.expect_call(None,146                                   self.job,147                                   None).and_return(my_harness)148        return resultdir, my_harness149    def _setup_post_record_init(self, cont, resultdir, my_harness):150        # now some specific stubs151        self.god.stub_function(self.job, 'config_get')152        self.god.stub_function(self.job, 'config_set')153        self.god.stub_function(self.job, 'record')154        # other setup155        results = os.path.join(self.autodir, 'results')156        download = os.path.join(self.autodir, 'tests', 'download')157        pkgdir = os.path.join(self.autodir, 'packages')158        utils.drop_caches.expect_call()159        job_sysinfo = sysinfo.sysinfo.expect_new(resultdir)160        if not cont:161            os.path.exists.expect_call(download).and_return(False)162            os.mkdir.expect_call(download)163            shutil.copyfile.expect_call(mock.is_string_comparator(),164                                 os.path.join(resultdir, 'control'))165        self.config = config.config.expect_new(self.job)166        self.job.config_get.expect_call(167                'boottool.executable').and_return(None)168        bootloader = boottool.boottool.expect_new(None)169        job.local_host.LocalHost.expect_new(hostname='localhost',170                                            bootloader=bootloader)171        job_sysinfo.log_per_reboot_data.expect_call()172        if not cont:173            self.job.record.expect_call('START', None, None)174        my_harness.run_start.expect_call()175        self.god.stub_function(utils, 'read_one_line')176        utils.read_one_line.expect_call('/proc/cmdline').and_return(177            'blah more-blah root=lala IDENT=81234567 blah-again console=tty1')178        self.job.config_set.expect_call('boot.default_args',179                                        'more-blah console=tty1')180    def construct_job(self, cont):181        # will construct class instance using __new__182        self.job = job.base_client_job.__new__(job.base_client_job)183        # record184        resultdir, my_harness = self._setup_pre_record_init(cont)185        self._setup_post_record_init(cont, resultdir, my_harness)186        # finish constructor187        options = dummy()188        options.tag = self.jobtag189        options.cont = cont190        options.harness = None191        options.harness_args = None192        options.log = False193        options.verbose = False194        options.hostname = 'localhost'195        options.user = 'my_user'196        options.args = ''197        options.output_dir = ''198        options.tap_report = None199        self.job.__init__(self.control, options,200                          extra_copy_cmdline=['more-blah'])201        # check202        self.god.check_playback()203    def get_partition_mock(self, devname):204        """205        Create a mock of a partition object and return it.206        """207        class mock(object):208            device = devname209            get_mountpoint = self.god.create_mock_function('get_mountpoint')210        return mock211    def test_constructor_first_run(self):212        self.construct_job(False)213    def test_constructor_continuation(self):214        self.construct_job(True)215    def test_constructor_post_record_failure(self):216        """217        Test post record initialization failure.218        """219        self.job = job.base_client_job.__new__(job.base_client_job)220        options = dummy()221        options.tag = self.jobtag222        options.cont = False223        options.harness = None224        options.harness_args = None225        options.log = False226        options.verbose = False227        options.hostname = 'localhost'228        options.user = 'my_user'229        options.args = ''230        options.output_dir = ''231        options.tap_report = None232        error = Exception('fail')233        self.god.stub_function(self.job, '_post_record_init')234        self.god.stub_function(self.job, 'record')235        self._setup_pre_record_init(False)236        self.job._post_record_init.expect_call(237                self.control, options, True, ['more-blah']).and_raises(error)238        self.job.record.expect_call(239                'ABORT', None, None,'client.bin.job.__init__ failed: %s' %240                str(error))241        self.assertRaises(242                Exception, self.job.__init__, self.control, options,243                drop_caches=True, extra_copy_cmdline=['more-blah'])244        # check245        self.god.check_playback()246    def test_relative_path(self):247        self.construct_job(True)248        dummy = "asdf"249        ret = self.job.relative_path(os.path.join(self.job.resultdir, dummy))250        self.assertEquals(ret, dummy)251    def test_control_functions(self):252        self.construct_job(True)253        control_file = "blah"254        self.job.control_set(control_file)255        self.assertEquals(self.job.control_get(), os.path.abspath(control_file))256    def test_harness_select(self):257        self.construct_job(True)258        # record259        which = "which"260        harness_args = ''261        harness.select.expect_call(which, self.job, 262                                   harness_args).and_return(None)263        # run and test264        self.job.harness_select(which, harness_args)265        self.god.check_playback()266    def test_config_set(self):267        self.construct_job(True)268        # unstub config_set269        self.god.unstub(self.job, 'config_set')270        # record271        name = "foo"272        val = 10273        self.config.set.expect_call(name, val)274        # run and test275        self.job.config_set(name, val)276        self.god.check_playback()277    def test_config_get(self):278        self.construct_job(True)279        # unstub config_get280        self.god.unstub(self.job, 'config_get')281        # record282        name = "foo"283        val = 10284        self.config.get.expect_call(name).and_return(val)285        # run and test286        self.job.config_get(name)287        self.god.check_playback()288    def test_setup_dirs_raise(self):289        self.construct_job(True)290        # setup291        results_dir = 'foo'292        tmp_dir = 'bar'293        # record294        os.path.exists.expect_call(tmp_dir).and_return(True)295        os.path.isdir.expect_call(tmp_dir).and_return(False)296        # test297        self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir)298        self.god.check_playback()299    def test_setup_dirs(self):300        self.construct_job(True)301        # setup302        results_dir1 = os.path.join(self.job.resultdir, 'build')303        results_dir2 = os.path.join(self.job.resultdir, 'build.2')304        results_dir3 = os.path.join(self.job.resultdir, 'build.3')305        tmp_dir = 'bar'306        # record307        os.path.exists.expect_call(tmp_dir).and_return(False)308        os.mkdir.expect_call(tmp_dir)309        os.path.isdir.expect_call(tmp_dir).and_return(True)310        os.path.exists.expect_call(results_dir1).and_return(True)311        os.path.exists.expect_call(results_dir2).and_return(True)312        os.path.exists.expect_call(results_dir3).and_return(False)313        os.path.exists.expect_call(results_dir3).and_return(False)314        os.mkdir.expect_call(results_dir3)315        # test316        self.assertEqual(self.job.setup_dirs(None, tmp_dir),317                         (results_dir3, tmp_dir))318        self.god.check_playback()319    def test_xen(self):320        self.construct_job(True)321        # setup322        self.god.stub_function(self.job, "setup_dirs")323        self.god.stub_class(xen, "xen")324        results = 'results_dir'325        tmp = 'tmp'326        build = 'xen'327        base_tree = object()328        # record329        self.job.setup_dirs.expect_call(results,330                                        tmp).and_return((results, tmp))331        myxen = xen.xen.expect_new(self.job, base_tree, results, tmp, build,332                                   False, None)333        # run job and check334        axen = self.job.xen(base_tree, results, tmp)335        self.god.check_playback()336        self.assertEquals(myxen, axen)337    def test_kernel_rpm(self):338        self.construct_job(True)339        # setup340        self.god.stub_function(self.job, "setup_dirs")341        self.god.stub_class(kernel, "rpm_kernel")342        self.god.stub_function(kernel, "preprocess_path")343        self.god.stub_function(self.job.pkgmgr, "fetch_pkg")344        self.god.stub_function(utils, "get_os_vendor")345        results = 'results_dir'346        tmp = 'tmp'347        build = 'xen'348        path = "somepath.rpm"349        packages_dir = os.path.join("autodir/packages", path)350        # record351        self.job.setup_dirs.expect_call(results,352                                        tmp).and_return((results, tmp))353        kernel.preprocess_path.expect_call(path).and_return(path)354        os.path.exists.expect_call(path).and_return(False)355        self.job.pkgmgr.fetch_pkg.expect_call(path, packages_dir, repo_url='')356        utils.get_os_vendor.expect_call()357        mykernel = kernel.rpm_kernel.expect_new(self.job, [packages_dir],358                                                results)359        # check360        akernel = self.job.kernel(path, results, tmp)361        self.god.check_playback()362        self.assertEquals(mykernel, akernel)363    def test_kernel(self):364        self.construct_job(True)365        # setup366        self.god.stub_function(self.job, "setup_dirs")367        self.god.stub_class(kernel, "kernel")368        self.god.stub_function(kernel, "preprocess_path")369        results = 'results_dir'370        tmp = 'tmp'371        build = 'linux'372        path = "somepath.deb"373        # record374        self.job.setup_dirs.expect_call(results,375                                        tmp).and_return((results, tmp))376        kernel.preprocess_path.expect_call(path).and_return(path)377        mykernel = kernel.kernel.expect_new(self.job, path, results, tmp,378                                            build, False)379        # check380        akernel = self.job.kernel(path, results, tmp)381        self.god.check_playback()382        self.assertEquals(mykernel, akernel)383    def test_run_test_logs_test_error_from_unhandled_error(self):384        self.construct_job(True)385        # set up stubs386        self.god.stub_function(self.job.pkgmgr, 'get_package_name')387        self.god.stub_function(self.job, "_runtest")388        # create an unhandled error object389        class MyError(error.TestError):390            pass391        real_error = MyError("this is the real error message")392        unhandled_error = error.UnhandledTestError(real_error)393        # set up the recording394        testname = "error_test"395        outputdir = os.path.join(self.job.resultdir, testname)396        self.job.pkgmgr.get_package_name.expect_call(397            testname, 'test').and_return(("", testname))398        os.path.exists.expect_call(outputdir).and_return(False)399        self.job.record.expect_call("START", testname, testname,400                                    optional_fields=None)401        self.job._runtest.expect_call(testname, "", None, (), {}).and_raises(402            unhandled_error)403        self.job.record.expect_call("ERROR", testname, testname,404                                    first_line_comparator(str(real_error)))405        self.job.record.expect_call("END ERROR", testname, testname)406        self.job.harness.run_test_complete.expect_call()407        utils.drop_caches.expect_call()408        # run and check409        self.job.run_test(testname)410        self.god.check_playback()411    def test_run_test_logs_non_test_error_from_unhandled_error(self):412        self.construct_job(True)413        # set up stubs414        self.god.stub_function(self.job.pkgmgr, 'get_package_name')415        self.god.stub_function(self.job, "_runtest")416        # create an unhandled error object417        class MyError(Exception):418            pass419        real_error = MyError("this is the real error message")420        unhandled_error = error.UnhandledTestError(real_error)421        reason = first_line_comparator("Unhandled MyError: %s" % real_error)422        # set up the recording423        testname = "error_test"424        outputdir = os.path.join(self.job.resultdir, testname)425        self.job.pkgmgr.get_package_name.expect_call(426            testname, 'test').and_return(("", testname))427        os.path.exists.expect_call(outputdir).and_return(False)428        self.job.record.expect_call("START", testname, testname,429                                    optional_fields=None)430        self.job._runtest.expect_call(testname, "", None, (), {}).and_raises(431            unhandled_error)432        self.job.record.expect_call("ERROR", testname, testname, reason)433        self.job.record.expect_call("END ERROR", testname, testname)434        self.job.harness.run_test_complete.expect_call()435        utils.drop_caches.expect_call()436        # run and check437        self.job.run_test(testname)438        self.god.check_playback()439    def test_report_reboot_failure(self):440        self.construct_job(True)441        # record442        self.job.record.expect_call("ABORT", "sub", "reboot.verify",443                                    "boot failure")444        self.job.record.expect_call("END ABORT", "sub", "reboot",445                                    optional_fields={"kernel": "2.6.15-smp"})446        # playback447        self.job._record_reboot_failure("sub", "reboot.verify", "boot failure",448                                        running_id="2.6.15-smp")449        self.god.check_playback()450    def _setup_check_post_reboot(self, mount_info, cpu_count):451        # setup452        self.god.stub_function(job.partition_lib, "get_partition_list")453        self.god.stub_function(utils, "count_cpus")454        part_list = [self.get_partition_mock("/dev/hda1"),455                     self.get_partition_mock("/dev/hdb1")]456        mount_list = ["/mnt/hda1", "/mnt/hdb1"]457        # record458        job.partition_lib.get_partition_list.expect_call(459                self.job, exclude_swap=False).and_return(part_list)460        for i in xrange(len(part_list)):461            part_list[i].get_mountpoint.expect_call().and_return(mount_list[i])462        if cpu_count is not None:463            utils.count_cpus.expect_call().and_return(cpu_count)464        self.job._state.set('client', 'mount_info', mount_info)465        self.job._state.set('client', 'cpu_count', 8)466    def test_check_post_reboot_success(self):467        self.construct_job(True)468        mount_info = set([("/dev/hda1", "/mnt/hda1"),469                          ("/dev/hdb1", "/mnt/hdb1")])470        self._setup_check_post_reboot(mount_info, 8)471        # playback472        self.job._check_post_reboot("sub")473        self.god.check_playback()474    def test_check_post_reboot_mounts_failure(self):475        self.construct_job(True)476        mount_info = set([("/dev/hda1", "/mnt/hda1")])477        self._setup_check_post_reboot(mount_info, None)478        self.god.stub_function(self.job, "_record_reboot_failure")479        self.job._record_reboot_failure.expect_call("sub",480                "reboot.verify_config", "mounted partitions are different after"481                " reboot (old entries: set([]), new entries: set([('/dev/hdb1',"482                " '/mnt/hdb1')]))", running_id=None)483        # playback484        self.assertRaises(error.JobError, self.job._check_post_reboot, "sub")485        self.god.check_playback()486    def test_check_post_reboot_cpu_failure(self):487        self.construct_job(True)488        mount_info = set([("/dev/hda1", "/mnt/hda1"),489                          ("/dev/hdb1", "/mnt/hdb1")])490        self._setup_check_post_reboot(mount_info, 4)491        self.god.stub_function(self.job, "_record_reboot_failure")492        self.job._record_reboot_failure.expect_call(493            'sub', 'reboot.verify_config',494            'Number of CPUs changed after reboot (old count: 8, new count: 4)',495            running_id=None)496        # playback497        self.assertRaises(error.JobError, self.job._check_post_reboot, "sub")498        self.god.check_playback()499    def test_end_boot(self):500        self.construct_job(True)501        self.god.stub_function(self.job, "_check_post_reboot")502        # set up the job class503        self.job._record_prefix = '\t\t'504        self.job._check_post_reboot.expect_call("sub", running_id=None)505        self.job.record.expect_call("END GOOD", "sub", "reboot",506                                    optional_fields={"kernel": "2.6.15-smp",507                                                     "patch0": "patchname"})508        # run test509        self.job.end_reboot("sub", "2.6.15-smp", ["patchname"])510        self.god.check_playback()511    def test_end_boot_and_verify_success(self):512        self.construct_job(True)513        self.god.stub_function(self.job, "_check_post_reboot")514        # set up the job class515        self.job._record_prefix = '\t\t'516        self.god.stub_function(utils, "running_os_ident")517        utils.running_os_ident.expect_call().and_return("2.6.15-smp")518        utils.read_one_line.expect_call("/proc/cmdline").and_return(519            "blah more-blah root=lala IDENT=81234567 blah-again")520        self.god.stub_function(utils, "running_os_full_version")521        running_id = "2.6.15-smp"522        utils.running_os_full_version.expect_call().and_return(running_id)523        self.job.record.expect_call("GOOD", "sub", "reboot.verify",524                                    running_id)525        self.job._check_post_reboot.expect_call("sub", running_id=running_id)526        self.job.record.expect_call("END GOOD", "sub", "reboot",527                                    optional_fields={"kernel": running_id})528        # run test529        self.job.end_reboot_and_verify(81234567, "2.6.15-smp", "sub")530        self.god.check_playback()531    def test_end_boot_and_verify_failure(self):532        self.construct_job(True)533        self.god.stub_function(self.job, "_record_reboot_failure")534        # set up the job class535        self.job._record_prefix = '\t\t'536        self.god.stub_function(utils, "running_os_ident")537        utils.running_os_ident.expect_call().and_return("2.6.15-smp")538        utils.read_one_line.expect_call("/proc/cmdline").and_return(539            "blah more-blah root=lala IDENT=81234567 blah-again")540        self.job._record_reboot_failure.expect_call("sub", "reboot.verify",541                "boot failure", running_id="2.6.15-smp")542        # run test543        self.assertRaises(error.JobError, self.job.end_reboot_and_verify,544                          91234567, "2.6.16-smp", "sub")545        self.god.check_playback()546    def test_parse_args(self):547        test_set = {"a='foo bar baz' b='moo apt'":548                    ["a='foo bar baz'", "b='moo apt'"],549                    "a='foo bar baz' only=gah":550                    ["a='foo bar baz'", "only=gah"],551                    "a='b c d' no=argh":552                    ["a='b c d'", "no=argh"]}553        for t in test_set:554            parsed_args = job.base_client_job._parse_args(t)555            expected_args = test_set[t]556            self.assertEqual(parsed_args, expected_args)557    def test_run_test_timeout_parameter_is_propagated(self):558        self.construct_job(True)559        # set up stubs560        self.god.stub_function(self.job.pkgmgr, 'get_package_name')561        self.god.stub_function(self.job, "_runtest")562        # create an unhandled error object563        #class MyError(error.TestError):564        #    pass565        #real_error = MyError("this is the real error message")566        #unhandled_error = error.UnhandledTestError(real_error)567        # set up the recording568        testname = "test"569        outputdir = os.path.join(self.job.resultdir, testname)570        self.job.pkgmgr.get_package_name.expect_call(571            testname, 'test').and_return(("", testname))572        os.path.exists.expect_call(outputdir).and_return(False)573        timeout = 60574        optional_fields = {}575        optional_fields['timeout'] = timeout576        self.job.record.expect_call("START", testname, testname,577                                    optional_fields=optional_fields)578        self.job._runtest.expect_call(testname, "", timeout, (), {})579        self.job.record.expect_call("GOOD", testname, testname,580                                    "completed successfully")581        self.job.record.expect_call("END GOOD", testname, testname)582        self.job.harness.run_test_complete.expect_call()583        utils.drop_caches.expect_call()584        # run and check585        self.job.run_test(testname, timeout=timeout)586        self.god.check_playback()587if __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.