How to use _fill_server_control_namespace method in autotest

Best Python code snippet using autotest_python

server_job.py

Source:server_job.py

...758        if log_exists:759            return pickle.load(open(self._uncollected_log_file))760        else:761            return []762    def _fill_server_control_namespace(self, namespace, protect=True):763        """764        Prepare a namespace to be used when executing server control files.765        This sets up the control file API by importing modules and making them766        available under the appropriate names within namespace.767        For use by _execute_code().768        Args:769          namespace: The namespace dictionary to fill in.770          protect: Boolean.  If True (the default) any operation that would771              clobber an existing entry in namespace will cause an error.772        Raises:773          error.AutoservError: When a name would be clobbered by import.774        """775        def _import_names(module_name, names=()):776            """777            Import a module and assign named attributes into namespace.778            Args:779                module_name: The string module name.780                names: A limiting list of names to import from module_name.  If781                    empty (the default), all names are imported from the module782                    similar to a "from foo.bar import *" statement.783            Raises:784                error.AutoservError: When a name being imported would clobber785                    a name already in namespace.786            """787            module = __import__(module_name, {}, {}, names)788            # No names supplied?  Import * from the lowest level module.789            # (Ugh, why do I have to implement this part myself?)790            if not names:791                for submodule_name in module_name.split('.')[1:]:792                    module = getattr(module, submodule_name)793                if hasattr(module, '__all__'):794                    names = getattr(module, '__all__')795                else:796                    names = dir(module)797            # Install each name into namespace, checking to make sure it798            # doesn't override anything that already exists.799            for name in names:800                # Check for conflicts to help prevent future problems.801                if name in namespace and protect:802                    if namespace[name] is not getattr(module, name):803                        raise error.AutoservError('importing name '804                                                  '%s from %s %r would override %r' %805                                                  (name, module_name, getattr(module, name),806                                                   namespace[name]))807                    else:808                        # Encourage cleanliness and the use of __all__ for a809                        # more concrete API with less surprises on '*' imports.810                        warnings.warn('%s (%r) being imported from %s for use '811                                      'in server control files is not the '812                                      'first occurrence of that import.' %813                                      (name, namespace[name], module_name))814                namespace[name] = getattr(module, name)815        # This is the equivalent of prepending a bunch of import statements to816        # the front of the control script.817        namespace.update(os=os, sys=sys, logging=logging)818        _import_names('autotest.server',819                      ('hosts', 'autotest_remote', 'standalone_profiler',820                       'source_kernel', 'rpm_kernel', 'deb_kernel', 'git_kernel'))821        _import_names('autotest.server.subcommand',822                      ('parallel', 'parallel_simple', 'subcommand'))823        _import_names('autotest.server.utils',824                      ('run', 'get_tmp_dir', 'sh_escape', 'parse_machine'))825        _import_names('autotest.client.shared.error')826        _import_names('autotest.client.shared.barrier', ('barrier',))827        # Inject ourself as the job object into other classes within the API.828        # (Yuck, this injection is a gross thing be part of a public API. -gps)829        #830        # XXX Base & SiteAutotest do not appear to use .job.  Who does?831        namespace['autotest_remote'].Autotest.job = self832        # server.hosts.base_classes.Host uses .job.833        namespace['hosts'].Host.job = self834        namespace['hosts'].factory.ssh_user = self._ssh_user835        namespace['hosts'].factory.ssh_port = self._ssh_port836        namespace['hosts'].factory.ssh_pass = self._ssh_pass837    def _execute_code(self, code_file, namespace, protect=True):838        """839        Execute code using a copy of namespace as a server control script.840        Unless protect_namespace is explicitly set to False, the dict will not841        be modified.842        Args:843          code_file: The filename of the control file to execute.844          namespace: A dict containing names to make available during execution.845          protect: Boolean.  If True (the default) a copy of the namespace dict846              is used during execution to prevent the code from modifying its847              contents outside of this function.  If False the raw dict is848              passed in and modifications will be allowed.849        """850        if protect:851            namespace = namespace.copy()852        self._fill_server_control_namespace(namespace, protect=protect)853        # TODO: Simplify and get rid of the special cases for only 1 machine.854        if len(self.machines) > 1:855            machines_text = '\n'.join(self.machines) + '\n'856            # Only rewrite the file if it does not match our machine list.857            try:858                machines_f = open(MACHINES_FILENAME, 'r')859                existing_machines_text = machines_f.read()860                machines_f.close()861            except EnvironmentError:862                existing_machines_text = None863            if machines_text != existing_machines_text:864                utils.open_write_close(MACHINES_FILENAME, machines_text)865        execfile(code_file, namespace, namespace)866    def _parse_status(self, new_line):...

server_job_unittest.py

Source:server_job_unittest.py

...106        # run method107        self.job.init_parser(results)108        # check109        self.god.check_playback()110    def test_fill_server_control_namespace(self):111        class MockAutotest(object):112            job = None113        class MockHosts(object):114            job = None115        # Verify that the job attributes are injected in the expected place.116        self.god.stub_with(autotest, 'Autotest', MockAutotest)117        self.god.stub_with(hosts, 'Host', MockHosts)118        self.job._fill_server_control_namespace({})119        self.assertEqual(hosts.Host.job, self.job)120        self.assertEqual(autotest.Autotest.job, self.job)121        test_ns = {}122        self.job._fill_server_control_namespace(test_ns)123        # Verify that a few of the expected module exports were loaded.124        self.assertEqual(test_ns['sys'], sys)125        self.assert_('git' in test_ns)126        self.assert_('parallel_simple' in test_ns)127        self.assert_('sh_escape' in test_ns)128        self.assert_('barrier' in test_ns)129        self.assert_('format_error' in test_ns)130        self.assert_('AutoservRebootError' in test_ns)131        # This should not exist, client.common_lib.errors does not export it.132        self.assert_('format_exception' not in test_ns)133        # Replacing something that exists with something else is an error.134        orig_test_ns = {'hosts': 'not the autotest_lib.server.hosts module'}135        test_ns = orig_test_ns.copy()136        self.assertRaises(error.AutoservError,137                          self.job._fill_server_control_namespace, test_ns)138        # Replacing something that exists with something else is an error.139        test_ns = orig_test_ns.copy()140        self.assertRaises(error.AutoservError,141                          self.job._fill_server_control_namespace, test_ns)142        # Replacing something without protection should succeed.143        test_ns = orig_test_ns.copy()144        self.job._fill_server_control_namespace(test_ns, protect=False)145        self.assertEqual(test_ns['hosts'], hosts)146        # Replacing something with itself should issue a warning.147        test_ns = {'hosts': hosts}148        self.god.stub_function(warnings, 'showwarning')149        warnings.showwarning.expect_call(150                mock.is_instance_comparator(UserWarning), UserWarning,151                mock.is_string_comparator(), mock.is_instance_comparator(int))152        self.job._fill_server_control_namespace(test_ns)153        self.god.check_playback()154        self.assertEqual(test_ns['hosts'], hosts)155    def test_parallel_simple_with_one_machine(self):156        self.job.machines = ["hostname"]157        # setup158        func = self.god.create_mock_function("wrapper")159        # record160        func.expect_call("hostname")161        # run and check162        self.job.parallel_simple(func, self.job.machines)163        self.god.check_playback()164    def test_run_test(self):165        # setup166        self.god.stub_function(self.job.pkgmgr, 'get_package_name')...

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.