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

How to use test_job_abort_in_verify method in autotest

Best Python code snippet using autotest_python

monitor_db_functional_test.py

Source:monitor_db_functional_test.py

...487                                               exit_status=256)488        self._run_dispatcher()489        self._check_host_status(host, HostStatus.READY) # ignore failure490        self._assert_nothing_is_running()491    def test_job_abort_in_verify(self):492        self._initialize_test()493        job = self._create_job(hosts=[1])494        queue_entries = list(job.hostqueueentry_set.all())495        self._run_dispatcher() # launches verify496        self._check_statuses(queue_entries[0], HqeStatus.VERIFYING)497        job.hostqueueentry_set.update(aborted=True)498        self._run_dispatcher() # kills verify, launches cleanup499        self.assert_(self.mock_drone_manager.was_last_process_killed(500                _PidfileType.VERIFY, set([signal.SIGKILL])))501        self.mock_drone_manager.finish_process(_PidfileType.CLEANUP)502        self._run_dispatcher()503    def test_job_abort(self):504        self._initialize_test()505        job = self._create_job(hosts=[1])506        job.run_reset = False507        job.save()508        queue_entries = list(job.hostqueueentry_set.all())509        self._run_dispatcher() # launches job510        self._check_statuses(queue_entries[0], HqeStatus.RUNNING)511        job.hostqueueentry_set.update(aborted=True)512        self._run_dispatcher() # kills job, launches gathering513        self._check_statuses(queue_entries[0], HqeStatus.GATHERING)514        self.mock_drone_manager.finish_process(_PidfileType.GATHER)515        self._run_dispatcher() # launches parsing + cleanup516        queue_entry = job.hostqueueentry_set.all()[0]517        self._finish_parsing()518        # The abort will cause gathering to launch a cleanup.519        self.mock_drone_manager.finish_process(_PidfileType.CLEANUP)520        self._run_dispatcher()521        self.mock_drone_manager.finish_process(_PidfileType.RESET)522        self._run_dispatcher()523    def test_job_abort_queued_synchronous(self):524        self._initialize_test()525        job = self._create_job(hosts=[1,2])526        job.synch_count = 2527        job.save()528        job.hostqueueentry_set.update(aborted=True)529        self._run_dispatcher()530        for host_queue_entry in job.hostqueueentry_set.all():531            self.assertEqual(host_queue_entry.status,532                             HqeStatus.ABORTED)533    def test_no_pidfile_leaking(self):534        self._initialize_test()535        self.test_simple_job()536        self.mock_drone_manager.refresh()537        self.assertEquals(self.mock_drone_manager._pidfiles, {})538        self.test_job_abort_in_verify()539        self.mock_drone_manager.refresh()540        self.assertEquals(self.mock_drone_manager._pidfiles, {})541        self.test_job_abort()542        self.mock_drone_manager.refresh()543        self.assertEquals(self.mock_drone_manager._pidfiles, {})544    def _make_job_and_queue_entry(self):545        job = self._create_job(hosts=[1])546        queue_entry = job.hostqueueentry_set.all()[0]547        return job, queue_entry548    def test_recover_running_no_process(self):549        # recovery should re-execute a Running HQE if no process is found550        _, queue_entry = self._make_job_and_queue_entry()551        queue_entry.status = HqeStatus.RUNNING552        queue_entry.execution_subdir = '1-myuser/host1'...

monitor_db_functional_unittest.py

Source:monitor_db_functional_unittest.py

...467                                               exit_status=256)468        self._run_dispatcher()469        self._check_host_status(host, HostStatus.READY)  # ignore failure470        self._assert_nothing_is_running()471    def test_job_abort_in_verify(self):472        self._initialize_test()473        job = self._create_job(hosts=[1])474        self._run_dispatcher()  # launches verify475        job.hostqueueentry_set.update(aborted=True)476        self._run_dispatcher()  # kills verify, launches cleanup477        self.assert_(self.mock_drone_manager.was_last_process_killed(478            _PidfileType.VERIFY))479        self.mock_drone_manager.finish_process(_PidfileType.CLEANUP)480        self._run_dispatcher()481    def test_job_abort(self):482        self._initialize_test()483        job = self._create_job(hosts=[1])484        job.run_verify = False485        job.save()486        self._run_dispatcher()  # launches job487        job.hostqueueentry_set.update(aborted=True)488        self._run_dispatcher()  # kills job, launches gathering489        self.assert_(self.mock_drone_manager.was_last_process_killed(490            _PidfileType.JOB))491        self.mock_drone_manager.finish_process(_PidfileType.GATHER)492        self._run_dispatcher()  # launches parsing + cleanup493        queue_entry = job.hostqueueentry_set.all()[0]494        self._finish_parsing_and_cleanup(queue_entry)495    def test_job_abort_queued_synchronous(self):496        self._initialize_test()497        job = self._create_job(hosts=[1, 2])498        job.synch_count = 2499        job.save()500        job.hostqueueentry_set.update(aborted=True)501        self._run_dispatcher()502        for host_queue_entry in job.hostqueueentry_set.all():503            self.assertEqual(host_queue_entry.status,504                             HqeStatus.ABORTED)505    def test_no_pidfile_leaking(self):506        self._initialize_test()507        self.test_simple_job()508        self.assertEquals(self.mock_drone_manager._pidfiles, {})509        self.test_job_abort_in_verify()510        self.assertEquals(self.mock_drone_manager._pidfiles, {})511        self.test_job_abort()512        self.assertEquals(self.mock_drone_manager._pidfiles, {})513    def _make_job_and_queue_entry(self):514        job = self._create_job(hosts=[1])515        queue_entry = job.hostqueueentry_set.all()[0]516        return job, queue_entry517    def test_recover_running_no_process(self):518        # recovery should re-execute a Running HQE if no process is found519        _, queue_entry = self._make_job_and_queue_entry()520        queue_entry.status = HqeStatus.RUNNING521        queue_entry.execution_subdir = '1-myuser/host1'522        queue_entry.save()523        queue_entry.host.status = HostStatus.RUNNING...

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.