Best Python code snippet using autotest_python
scheduler_models.py
Source:scheduler_models.py  
...404        new_row = [getattr(template, field) for field in cls._fields]405        clone = cls(row=new_row, new_record=True)406        clone.id = None407        return clone408    def _view_job_url(self):409        return "%s#tab_id=view_job&object_id=%s" % (_base_url, self.job.id)410    def get_labels(self):411        """412        Get all labels associated with this host queue entry (either via the413        meta_host or as a job dependency label).  The labels yielded are not414        guaranteed to be unique.415        @yields Label instances associated with this host_queue_entry.416        """417        if self.meta_host:418            yield Label(id=self.meta_host, always_query=False)419        labels = Label.fetch(420            joins="JOIN afe_jobs_dependency_labels AS deps "421            "ON (afe_labels.id = deps.label_id)",422            where="deps.job_id = %d" % self.job.id)423        for label in labels:424            yield label425    def set_host(self, host):426        if host:427            logging.info('Assigning host %s to entry %s', host.hostname, self)428            self.queue_log_record('Assigning host ' + host.hostname)429            self.update_field('host_id', host.id)430            self.block_host(host.id)431        else:432            logging.info('Releasing host from %s', self)433            self.queue_log_record('Releasing host')434            self.unblock_host(self.host.id)435            self.update_field('host_id', None)436        self.host = host437    def queue_log_record(self, log_line):438        now = str(datetime.datetime.now())439        _drone_manager.write_lines_to_file(self.queue_log_path,440                                           [now + ' ' + log_line])441    def block_host(self, host_id):442        logging.info("creating block %s/%s", self.job.id, host_id)443        row = [0, self.job.id, host_id]444        block = IneligibleHostQueue(row=row, new_record=True)445        block.save()446    def unblock_host(self, host_id):447        logging.info("removing block %s/%s", self.job.id, host_id)448        blocks = IneligibleHostQueue.fetch(449            'job_id=%d and host_id=%d' % (self.job.id, host_id))450        for block in blocks:451            block.delete()452    def set_execution_subdir(self, subdir=None):453        if subdir is None:454            assert self.host455            subdir = self.host.hostname456        self.update_field('execution_subdir', subdir)457    def _get_hostname(self):458        if self.host:459            return self.host.hostname460        return 'no host'461    def __str__(self):462        flags = []463        if self.active:464            flags.append('active')465        if self.complete:466            flags.append('complete')467        if self.deleted:468            flags.append('deleted')469        if self.aborted:470            flags.append('aborted')471        flags_str = ','.join(flags)472        if flags_str:473            flags_str = ' [%s]' % flags_str474        return "%s/%d (%d) %s%s" % (self._get_hostname(), self.job.id, self.id,475                                    self.status, flags_str)476    def set_status(self, status):477        logging.info("%s -> %s", self, status)478        self.update_field('status', status)479        active = (status in models.HostQueueEntry.ACTIVE_STATUSES)480        complete = (status in models.HostQueueEntry.COMPLETE_STATUSES)481        assert not (active and complete)482        self.update_field('active', active)483        self.update_field('complete', complete)484        should_email_status = (status.lower() in _notify_email_statuses or485                               'all' in _notify_email_statuses)486        should_email_admin_status = (status.lower() in487                                     _notify_admin_email_statuses or488                                     'all' in _notify_admin_email_statuses)489        if complete:490            self._on_complete(status)491            self._email_on_job_complete(should_email_admin_status)492        if should_email_status:493            self._email_on_status(status)494        if should_email_admin_status:495            self._email_admin_on_status(status)496    def _on_complete(self, status):497        if status is not models.HostQueueEntry.Status.ABORTED:498            self.job.stop_if_necessary()499        if not self.execution_subdir:500            return501        # unregister any possible pidfiles associated with this queue entry502        for pidfile_name in drone_manager.ALL_PIDFILE_NAMES:503            pidfile_id = _drone_manager.get_pidfile_id_from(504                self.execution_path(), pidfile_name=pidfile_name)505            _drone_manager.unregister_pidfile(pidfile_id)506    def _get_status_email_contents(self, status, summary=None, hostname=None):507        """508        Gather info for the status notification e-mails.509        If needed, we could start using the Django templating engine to create510        the subject and the e-mail body, but that doesn't seem necessary right511        now.512        :param status: Job status text. Mandatory.513        :param summary: Job summary text. Optional.514        :param hostname: A hostname for the job. Optional.515        :return: Tuple (subject, body) for the notification e-mail.516        """517        job_stats = Job(id=self.job.id).get_execution_details()518        subject = 'Autotest #%s' % self.job.id519        if hostname is not None:520            subject += ' | %s on %s' % (self.job.name, hostname)521        else:522            subject += ' | %s' % self.job.name523        status = status.split()[-1]524        if status == "Completed":525            subject += ' | success: %.2f%%' % job_stats['success_rate']526        else:527            subject += ' | %s' % status528        body = ""529        if int(job_stats['total_executed']) > 0:530            body += ("run: %s | pass: %s | skip: %s | fail: %s" %531                     (job_stats['total_executed'], job_stats['total_passed'],532                      job_stats['total_skipped'], job_stats['total_failed']))533            e_time = job_stats['execution_time']534            if e_time not in ['(could not determine)', '(none)']:535                body += " | runtime: %s" % e_time536            if status == "Completed":537                body += " | success: %.2f%%" % job_stats['success_rate']538            else:539                body += "\n[Warning] Job status: %s" % status540            body += "\n%s\n\n" % self._view_job_url()541        body += job_stats['fail_detail']542        body += job_stats['warn_detail']543        body += job_stats['skip_detail']544        body += job_stats['pass_detail']545        keyval_list = job_stats['keyval_dict_list']546        if keyval_list:547            for kv in keyval_list:548                k, v = kv.items()[0]549                body += "%s:\n" % k550                for part in v.split():551                    body += "  %s\n" % part552                body += "\n"553        if hostname is not None:554            body += "Job was run on host %s\n" % hostname555        body += ("For more details, check the full report on the web:\n%s\n" %556                 self._view_job_url())557        return subject, body558    def _email_on_status(self, status):559        hostname = self._get_hostname()560        subject, body = self._get_status_email_contents(status, None, hostname)561        mail.manager.send(self.job.email_list, subject, body)562    def _email_admin_on_status(self, status):563        hostname = self._get_hostname()564        subject, body = self._get_status_email_contents(status, None, hostname)565        mail.manager.send(_grid_admin_email, subject, body)566    def _email_on_job_complete(self, email_admin=False):567        if not self.job.is_finished():568            return569        summary = []570        hosts_queue = HostQueueEntry.fetch('job_id = %s' % self.job.id)...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
