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

How to use _should_run_cleanup method in autotest

Best Python code snippet using autotest_python

scheduler_models.py

Source:scheduler_models.py

...933        execution_subdir = queue_entry_from_group.execution_subdir934        return list(HostQueueEntry.fetch(935            where='job_id=%s AND execution_subdir=%s',936            params=(self.id, execution_subdir)))937    def _should_run_cleanup(self, queue_entry):938        if self.reboot_before == model_attributes.RebootBefore.ALWAYS:939            return True940        elif self.reboot_before == model_attributes.RebootBefore.IF_DIRTY:941            return queue_entry.host.dirty942        return False943    def _should_run_verify(self, queue_entry):944        do_not_verify = (queue_entry.host.protection ==945                         host_protections.Protection.DO_NOT_VERIFY)946        if do_not_verify:947            return False948        # If RebootBefore is set to NEVER, then we won't run reset because949        # we can't cleanup, so we need to weaken a Reset into a Verify.950        weaker_reset = (self.run_reset and951                self.reboot_before == model_attributes.RebootBefore.NEVER)952        return self.run_verify or weaker_reset953    def _should_run_reset(self, queue_entry):954        can_verify = (queue_entry.host.protection !=955                         host_protections.Protection.DO_NOT_VERIFY)956        can_reboot = self.reboot_before != model_attributes.RebootBefore.NEVER957        return (can_reboot and can_verify and (self.run_reset or958                (self._should_run_cleanup(queue_entry) and959                 self._should_run_verify(queue_entry))))960    def _should_run_provision(self, queue_entry):961        """962        Determine if the queue_entry needs to have a provision task run before963        it to provision queue_entry.host.964        @param queue_entry: The host queue entry in question.965        @returns: True if we should schedule a provision task, False otherwise.966        """967        # If we get to this point, it means that the scheduler has already968        # vetted that all the unprovisionable labels match, so we can just969        # find all labels on the job that aren't on the host to get the list970        # of what we need to provision.  (See the scheduling logic in971        # host_scheduler.py:is_host_eligable_for_job() where we discard all972        # actionable labels when assigning jobs to hosts.)973        job_labels = {x.name for x in queue_entry.get_labels()}974        # Skip provision if `skip_provision` is listed in the job labels.975        if provision.SKIP_PROVISION in job_labels:976            return False977        _, host_labels = queue_entry.host.platform_and_labels()978        # If there are any labels on the job that are not on the host and they979        # are labels that provisioning knows how to change, then that means980        # there is provisioning work to do.  If there's no provisioning work to981        # do, then obviously we have no reason to schedule a provision task!982        diff = job_labels - set(host_labels)983        if any([provision.Provision.acts_on(x) for x in diff]):984            return True985        return False986    def _queue_special_task(self, queue_entry, task):987        """988        Create a special task and associate it with a host queue entry.989        @param queue_entry: The queue entry this special task should be990                            associated with.991        @param task: One of the members of the enum models.SpecialTask.Task.992        @returns: None993        """994        models.SpecialTask.objects.create(995                host=models.Host.objects.get(id=queue_entry.host_id),996                queue_entry=queue_entry, task=task)997    def schedule_pre_job_tasks(self, queue_entry):998        """999        Queue all of the special tasks that need to be run before a host1000        queue entry may run.1001        If no special taskes need to be scheduled, then |on_pending| will be1002        called directly.1003        @returns None1004        """1005        task_queued = False1006        hqe_model = models.HostQueueEntry.objects.get(id=queue_entry.id)1007        if self._should_run_provision(queue_entry):1008            self._queue_special_task(hqe_model,1009                                     models.SpecialTask.Task.PROVISION)1010            task_queued = True1011        elif self._should_run_reset(queue_entry):1012            self._queue_special_task(hqe_model, models.SpecialTask.Task.RESET)1013            task_queued = True1014        else:1015            if self._should_run_cleanup(queue_entry):1016                self._queue_special_task(hqe_model,1017                                         models.SpecialTask.Task.CLEANUP)1018                task_queued = True1019            if self._should_run_verify(queue_entry):1020                self._queue_special_task(hqe_model,1021                                         models.SpecialTask.Task.VERIFY)1022                task_queued = True1023        if not task_queued:1024            queue_entry.on_pending()1025    def _assign_new_group(self, queue_entries):1026        if len(queue_entries) == 1:1027            group_subdir_name = queue_entries[0].host.hostname1028        else:1029            group_subdir_name = self._next_group_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.