How to use _get_prioritized_special_tasks method in autotest

Best Python code snippet using autotest_python

monitor_db.py

Source:monitor_db.py

...459            message = '\n'.join(str(hqe) for hqe in unrecovered_hqes)460            raise host_scheduler.SchedulerError(461                '%d unrecovered verifying host queue entries:\n%s' %462                (len(unrecovered_hqes), message))463    def _get_prioritized_special_tasks(self):464        """465        Returns all queued SpecialTasks prioritized for repair first, then466        cleanup, then verify.467        """468        queued_tasks = models.SpecialTask.objects.filter(is_active=False,469                                                         is_complete=False,470                                                         host__locked=False)471        # exclude hosts with active queue entries unless the SpecialTask is for472        # that queue entry473        queued_tasks = models.SpecialTask.objects.add_join(474            queued_tasks, 'afe_host_queue_entries', 'host_id',475            join_condition='afe_host_queue_entries.active',476            join_from_key='host_id', force_left_join=True)477        queued_tasks = queued_tasks.extra(478            where=['(afe_host_queue_entries.id IS NULL OR '479                   'afe_host_queue_entries.id = '480                   'afe_special_tasks.queue_entry_id)'])481        # reorder tasks by priority482        task_priority_order = [models.SpecialTask.Task.REPAIR,483                               models.SpecialTask.Task.CLEANUP,484                               models.SpecialTask.Task.VERIFY]485        def task_priority_key(task):486            return task_priority_order.index(task.task)487        return sorted(queued_tasks, key=task_priority_key)488    def _schedule_special_tasks(self):489        """490        Execute queued SpecialTasks that are ready to run on idle hosts.491        """492        for task in self._get_prioritized_special_tasks():493            if self.host_has_agent(task.host):494                continue495            self.add_agent_task(self._get_agent_task_for_special_task(task))496    def _reverify_remaining_hosts(self):497        # recover active hosts that have not yet been recovered, although this498        # should never happen499        message = ('Recovering active host %s - this probably indicates a '500                   'scheduler bug')501        self._reverify_hosts_where(502            "status IN ('Repairing', 'Verifying', 'Cleaning')",503            print_message=message)504    def _reverify_hosts_where(self, where,505                              print_message='Reverifying host %s'):506        full_where = 'locked = 0 AND invalid = 0 AND ' + where...

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.