TestMu 2025 - The Largest Virtual Software Testing Conference is Back | 50,000+ Attendees

How to use _fetch_runners method in lisa

Best Python code snippet using lisa_python

runner.py

...200        await super().stop()201        # TODO: to be implemented202    async def close(self) -> None:203        await super().close()204    def _fetch_runners(self) -> Iterator[BaseRunner]:205        root_runbook = self._runbook_builder.resolve(self._runbook_builder.variables)206        if root_runbook.combinator:207            combinator_factory = Factory[Combinator](Combinator)208            combinator = combinator_factory.create_by_runbook(root_runbook.combinator)209            del self._runbook_builder.raw_data[constants.COMBINATOR]210            self._log.debug(211                f"found combinator '{combinator.type_name()}', to expand runbook."212            )213            combinator.initialize()214            while True:215                variables = combinator.fetch(self._runbook_builder.variables)216                if variables is None:217                    break218                sub_runbook_builder = self._runbook_builder.derive(variables=variables)219                transformer.run(220                    sub_runbook_builder, phase=constants.TRANSFORMER_PHASE_EXPANDED221                )222                runners = self._generate_runners(223                    sub_runbook_builder.resolve(), variables224                )225                for runner in runners:226                    yield runner227        else:228            # no combinator, use the root runbook229            transformer.run(230                self._runbook_builder, phase=constants.TRANSFORMER_PHASE_EXPANDED231            )232            for runner in self._generate_runners(233                root_runbook, self._runbook_builder.variables234            ):235                yield runner236    def _generate_runners(237        self, runbook: schema.Runbook, variables: Dict[str, VariableEntry]238    ) -> Iterator[BaseRunner]:239        # group filters by runner type240        case_variables = get_case_variables(variables)241        runner_filters: Dict[str, List[schema.BaseTestCaseFilter]] = {}242        for raw_filter in runbook.testcase_raw:243            # by default run all filtered cases unless 'enable' is specified as false244            filter = schema.load_by_type(schema.BaseTestCaseFilter, raw_filter)245            if filter.enabled:246                raw_filters: List[schema.BaseTestCaseFilter] = runner_filters.get(247                    filter.type, []248                )249                if not raw_filters:250                    runner_filters[filter.type] = raw_filters251                raw_filters.append(raw_filter)252            else:253                self._log.debug(f"Skip disabled filter: {raw_filter}.")254        # initialize runners255        factory = Factory[BaseRunner](BaseRunner)256        for runner_name, raw_filters in runner_filters.items():257            self._log.debug(258                f"create runner {runner_name} with {len(raw_filters)} filter(s)."259            )260            # keep filters to current runner's only.261            runbook.testcase = parse_testcase_filters(raw_filters)262            runner = factory.create_by_type_name(263                type_name=runner_name,264                runbook=runbook,265                index=self._runner_count,266                case_variables=case_variables,267            )268            runner.initialize()269            self._runners.append(runner)270            self._runner_count += 1271            yield runner272    def _submit_runner_tasks(273        self,274        runner: BaseRunner,275        task_manager: TaskManager[None],276    ) -> bool:277        has_task: bool = False278        while not runner.is_done and task_manager.has_idle_worker():279            # fetch a task and submit280            task = runner.fetch_task()281            if task:282                if isinstance(task, Task):283                    task_manager.submit_task(task)284                else:285                    raise LisaException(f"Unknown task type: '{type(task)}'")286                has_task = True287            else:288                # current runner may not be done, but it doesn't289                # have task temporarily. The root runner can start290                # tasks from next runner.291                break292        return has_task293    def _start_loop(self) -> None:294        # in case all of runners are disabled295        runner_iterator = self._fetch_runners()296        remaining_runners: List[BaseRunner] = []297        run_message = messages.TestRunMessage(298            status=messages.TestRunStatus.RUNNING,299        )300        notifier.notify(run_message)301        task_manager = TaskManager[None](self._max_concurrency, is_verbose=True)302        # set the global task manager for cancellation check303        set_global_task_manager(task_manager)304        has_more_runner = True305        # run until no idle workers are available and all runner are closed306        while task_manager.wait_worker() or has_more_runner or remaining_runners:307            assert task_manager.has_idle_worker()308            # submit tasks until idle workers are available309            while task_manager.has_idle_worker():...

Blogs

Check out the latest blogs from LambdaTest on this topic:

Desired Capabilities in Selenium Webdriver

Desired Capabilities is a class used to declare a set of basic requirements such as combinations of browsers, operating systems, browser versions, etc. to perform automated cross browser testing of a web application.

13 Best Test Automation Frameworks: The 2021 List

Automation frameworks enable automation testers by simplifying the test development and execution activities. A typical automation framework provides an environment for executing test plans and generating repeatable output. They are specialized tools that assist you in your everyday test automation tasks. Whether it is a test runner, an action recording tool, or a web testing tool, it is there to remove all the hard work from building test scripts and leave you with more time to do quality checks. Test Automation is a proven, cost-effective approach to improving software development. Therefore, choosing the best test automation framework can prove crucial to your test results and QA timeframes.

How To Handle Multiple Windows In Selenium Python

Automating testing is a crucial step in the development pipeline of a software product. In an agile development environment, where there is continuous development, deployment, and maintenance of software products, automation testing ensures that the end software products delivered are error-free.

Continuous delivery and continuous deployment offer testers opportunities for growth

Development practices are constantly changing and as testers, we need to embrace change. One of the changes that we can experience is the move from monthly or quarterly releases to continuous delivery or continuous deployment. This move to continuous delivery or deployment offers testers the chance to learn new skills.

Do you possess the necessary characteristics to adopt an Agile testing mindset?

To understand the agile testing mindset, we first need to determine what makes a team “agile.” To me, an agile team continually focuses on becoming self-organized and cross-functional to be able to complete any challenge they may face during a project.

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.