KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use poll_all_jobs method in autotest

Best Python code snippet using autotest_python

frontend.py

Source:frontend.py

...297            return298        tko = TKO()299        while True:300            time.sleep(60 * poll_interval)301            result = self.poll_all_jobs(tko, jobs, email_from, email_to)302            if result is not None:303                return result304    def result_notify(self, job, email_from, email_to):305        """306        Notify about the result of a job. Will always print, if email data307        is provided, will send email for it as well.308            job: job object to notify about309            email_from: send notification email upon completion from here310            email_from: send notification email upon completion to here311        """312        if job.result is True:313            subject = 'Testing PASSED: '314        else:315            subject = 'Testing FAILED: '316        subject += '%s : %s\n' % (job.name, job.id)317        text = []318        for platform in job.results_platform_map:319            for status in job.results_platform_map[platform]:320                if status == 'Total':321                    continue322                for host in job.results_platform_map[platform][status]:323                    text.append('%20s %10s %10s' % (platform, status, host))324                    if status == 'Failed':325                        for test_status in job.test_status[host].fail:326                            text.append('(%s, %s) : %s' %327                                        (host, test_status.test_name,328                                         test_status.reason))329                        text.append('')330        base_url = 'http://' + self.server331        params = ('columns=test',332                  'rows=machine_group',333                  "condition=tag~'%s-%%25'" % job.id,334                  'title=Report')335        query_string = '&'.join(params)336        url = '%s/tko/compose_query.cgi?%s' % (base_url, query_string)337        text.append(url + '\n')338        url = '%s/afe/#tab_id=view_job&object_id=%s' % (base_url, job.id)339        text.append(url + '\n')340        smtp_info = {}341        smtp_info['server'] = settings.get_value('SERVER', 'smtp_server',342                                                 default='localhost')343        smtp_info['port'] = settings.get_value('SERVER', 'smtp_port',344                                               default='')345        smtp_info['user'] = settings.get_value('SERVER', 'smtp_user',346                                               default='')347        smtp_info['password'] = settings.get_value('SERVER', 'smtp_password',348                                                   default='')349        body = '\n'.join(text)350        print '---------------------------------------------------'351        print 'Subject: ', subject352        print body353        print '---------------------------------------------------'354        if email_from and email_to:355            print 'Sending email ...'356            mail.send(from_address=email_from,357                      to_addresses=email_to,358                      subject=subject,359                      body=body,360                      smtp_info=smtp_info)361        print362    def print_job_result(self, job):363        """364        Print the result of a single job.365            job: a job object366        """367        if job.result is None:368            print 'PENDING',369        elif job.result is True:370            print 'PASSED',371        elif job.result is False:372            print 'FAILED',373        elif job.result == "Abort":374            print 'ABORT',375        print ' %s : %s' % (job.id, job.name)376    def poll_all_jobs(self, tko, jobs, email_from=None, email_to=None):377        """378        Poll all jobs in a list.379            jobs: list of job objects to poll380            email_from: send notification email upon completion from here381            email_from: send notification email upon completion to here382        Returns:383            a) All complete successfully (return True)384            b) One or more has failed (return False)385            c) Cannot tell yet (return None)386        """387        results = []388        for job in jobs:389            if getattr(job, 'result', None) is None:390                job.result = self.poll_job_results(tko, job)...

batch_run.py

Source:batch_run.py

...108    job_names = [f"{save_filename}.{line_num+1}" for line_num in range(len(lines))]109    for line_num in range(len(lines)):110        if os.path.exists(f"./job_results/{job_names[line_num]}"):111            print(f"WARNING: job results already exists for line {line_num+1}, skipping evaluation: delete if you wish to rerun")112    def poll_all_jobs():113        not_finished_procs = []114        for p_line_num, p_machine, p_gpu, proc in proc_list:115            if args.dry_run or proc.poll() is not None:116                message = "finished" if args.dry_run or proc.returncode == 0 else "failed"117                job_name = job_names[p_line_num]118                machine_infos[p_machine] = remove_from_machine_state(machine_infos[p_machine], p_gpu)119                print(f"{message}: {job_name}; {lines[p_line_num].strip()}",flush=True)120            else:121                not_finished_procs.append((p_line_num, p_machine, p_gpu, proc))122        proc_list[:] = not_finished_procs123    proc_list = []124    machine_config = args125    os.makedirs("./job_results/",exist_ok=True)126    for line_num in range(len(job_names)):127        job_name = job_names[line_num]128        if os.path.exists(f"./job_results/{job_name}"):129            print("skipping", lines[line_num].strip(),flush=True)130            continue131        not_polled = True132        while not_polled or is_over_limit(machine_cost(machine_config, machine_infos[best_machine_idx])):133            if not not_polled:134                if not args.dry_run:135                    # ssh commands take at least a second to finish136                    # so its a reasonable slow-query time137                    time.sleep(1)138                # revert machine addition from previous loop iteration139                machine_infos[best_machine_idx] = remove_from_machine_state(machine_infos[best_machine_idx], best_gpu_idx)140                poll_all_jobs()141            not_polled = False142            best_machine_idx = get_best_machine(machine_infos, machine_config)143            best_gpu_idx = get_best_gpu(machine_config, machine_infos[best_machine_idx])144            machine_infos[best_machine_idx] = add_to_machine_state(machine_infos[best_machine_idx], machine_config, best_gpu_idx)145        # start new job146        command = lines[line_num].strip()147        export_prefix = f"export CUDA_VISIBLE_DEVICES={best_gpu_idx} &&" if not args.reserve and not args.no_gpu_required else ""148        machine = machine_configs[best_machine_idx]149        job_name = job_names[line_num]150        if not args.dry_run:151            if args.kabuki_commands:152                proc, new_job_name = make_kabuki_run_command(machine, job_name, export_prefix, command, best_gpu_idx, args)153                job_name = job_names[line_num] = new_job_name154            else:155                proc = make_basic_run_command(machine, job_name, export_prefix, command, best_gpu_idx, args)156        else:157            proc = None158        proc_list.append((159            line_num,160            best_machine_idx,161            best_gpu_idx,162            proc163        ))164        print(f"started: {job_name};  {command}",flush=True)165        if not args.dry_run:166            # give time as to not overload the sshd server167            time.sleep(0.1)168    while proc_list:169        poll_all_jobs()170        if not args.dry_run:171            time.sleep(1)172if __name__ == "__main__":...

OVSPackager_CLItools.py

Source:OVSPackager_CLItools.py

...41# # Delete a specific Package Job42# # The Job must be in either the 'Queue' or 'Error' state to be deleted43# r = client.delete("http://ovspackager.indemand.com/api/jobs/1")44# print r.text45def poll_all_jobs():46	r = client.get("http://ovspackager.indemand.com/api/jobs/")47	# print r.text48	return r.json()['objects']49def poll_all_jobs_month(date):50	"""date is YYYY-MM-DD"""51	#see if I can import date and automatically check from one month back from NOW52	payload = {"start_date": date }53	r = client.get("http://ovspackager.indemand.com/api/jobs/", data=json.dumps(payload))54	# print r.text55	return r.json()['objects']56date_floor = datetime.now() - timedelta(days=90)57date_floor = date.isoformat(date_floor)58all_jobs = poll_all_jobs_month(date_floor)59def poll_all_packaged():...

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.