def fetch_task()

in lisa/runners/lisa_runner.py [0:0]

• 36 lines of code
• 19 McCabe index (conditional complexity)

    def fetch_task(self) -> Optional[Task[None]]:
        self._prepare_environments(
            platform=self.platform,
        )

        # sort environments by status
        available_environments = self._sort_environments(self.environments)
        available_results = [x for x in self.test_results if x.can_run]
        self._sort_test_results(available_results)

        # check deleteable environments
        delete_task = self._delete_unused_environments()
        if delete_task:
            return delete_task

        if available_results and available_environments:
            for priority in range(6):
                can_run_results = self._get_results_by_priority(
                    available_results, priority
                )
                if not can_run_results:
                    continue

                # it means there are test cases and environment, so it needs to
                # schedule task.
                for environment in available_environments:
                    if environment.is_in_use:
                        # skip in used environments
                        continue

                    environment_results = self._get_runnable_test_results(
                        test_results=can_run_results, environment=environment
                    )

                    if not environment_results:
                        continue

                    task = self._associate_environment_test_results(
                        environment=environment, test_results=environment_results
                    )
                    # there is more checking conditions. If some conditions doesn't
                    # meet, the task is None. If so, not return, and try next
                    # conditions or skip this test case.
                    if task:
                        return task
                if not any(x.is_in_use for x in available_environments):
                    # no environment in used, and not fit. those results cannot be run.
                    self._skip_test_results(can_run_results)
        elif available_results:
            # no available environments, so mark all test results skipped.
            self._skip_test_results(available_results)
            self.status = ActionStatus.SUCCESS
        return None