Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _make_parallel_wrapper method in autotest

Best Python code snippet using autotest_python

server_job.py

Source:server_job.py

...327        """328        Return True if external logging should be used.329        """330        return False331    def _make_parallel_wrapper(self, function, machines, log):332        """Wrap function as appropriate for calling by parallel_simple."""333        is_forking = not (len(machines) == 1 and self.machines == machines)334        if self._parse_job and is_forking and log:335            def wrapper(machine):336                self._parse_job += "/" + machine337                self._using_parser = True338                self.machines = [machine]339                self.push_execution_context(machine)340                os.chdir(self.resultdir)341                utils.write_keyval(self.resultdir, {"hostname": machine})342                self.init_parser()343                result = function(machine)344                self.cleanup_parser()345                return result346        elif len(machines) > 1 and log:347            def wrapper(machine):348                self.push_execution_context(machine)349                os.chdir(self.resultdir)350                machine_data = {'hostname': machine,351                                'status_version': str(self._STATUS_VERSION)}352                utils.write_keyval(self.resultdir, machine_data)353                result = function(machine)354                return result355        else:356            wrapper = function357        return wrapper358    def parallel_simple(self, function, machines, log=True, timeout=None,359                        return_results=False):360        """361        Run 'function' using parallel_simple, with an extra wrapper to handle362        the necessary setup for continuous parsing, if possible. If continuous363        parsing is already properly initialized then this should just work.364        :param function: A callable to run in parallel given each machine.365        :param machines: A list of machine names to be passed one per subcommand366                invocation of function.367        :param log: If True, output will be written to output in a subdirectory368                named after each machine.369        :param timeout: Seconds after which the function call should timeout.370        :param return_results: If True instead of an AutoServError being raised371                on any error a list of the results|exceptions from the function372                called on each arg is returned.  [default: False]373        :raise error.AutotestError: If any of the functions failed.374        """375        wrapper = self._make_parallel_wrapper(function, machines, log)376        return subcommand.parallel_simple(wrapper, machines,377                                          log=log, timeout=timeout,378                                          return_results=return_results)379    def parallel_on_machines(self, function, machines, timeout=None):380        """381        :param function: Called in parallel with one machine as its argument.382        :param machines: A list of machines to call function(machine) on.383        :param timeout: Seconds after which the function call should timeout.384        :return: A list of machines on which function(machine) returned385                without raising an exception.386        """387        results = self.parallel_simple(function, machines, timeout=timeout,388                                       return_results=True)389        success_machines = []...

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.