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

How to use _check_hqe_statuses method in autotest

Best Python code snippet using autotest_python

monitor_db_functional_test.py

Source:monitor_db_functional_test.py

...715        job = self._create_job(hosts=[1,2,3])716        job.synch_count = 3717        job.save()718        queue_entries = list(job.hostqueueentry_set.all())719        def _check_hqe_statuses(*statuses):720            for queue_entry, status in zip(queue_entries, statuses):721                self._check_statuses(queue_entry, status)722        self.mock_drone_manager.process_capacity = 2723        self._run_dispatcher() # verify runs on 1 and 2724        queue_entries = list(job.hostqueueentry_set.all())725        _check_hqe_statuses(HqeStatus.QUEUED,726                            HqeStatus.VERIFYING, HqeStatus.VERIFYING)727        self.assertEquals(len(self.mock_drone_manager.running_pidfile_ids()), 2)728        self.mock_drone_manager.finish_specific_process(729                'hosts/host3/1-verify', drone_manager.AUTOSERV_PID_FILE)730        self.mock_drone_manager.finish_process(_PidfileType.VERIFY)731        self._run_dispatcher() # verify runs on 3732        _check_hqe_statuses(HqeStatus.VERIFYING, HqeStatus.PENDING,733                            HqeStatus.PENDING)734        self.mock_drone_manager.finish_process(_PidfileType.VERIFY)735        self._run_dispatcher() # job won't run due to throttling736        _check_hqe_statuses(HqeStatus.STARTING, HqeStatus.STARTING,737                            HqeStatus.STARTING)738        self._assert_nothing_is_running()739        self.mock_drone_manager.process_capacity = 3740        self._run_dispatcher() # now job runs741        _check_hqe_statuses(HqeStatus.RUNNING, HqeStatus.RUNNING,742                            HqeStatus.RUNNING)743        self.mock_drone_manager.process_capacity = 2744        self.mock_drone_manager.finish_process(_PidfileType.JOB,745                                               exit_status=271)746        self._run_dispatcher() # gathering won't run due to throttling747        _check_hqe_statuses(HqeStatus.GATHERING, HqeStatus.GATHERING,748                            HqeStatus.GATHERING)749        self._assert_nothing_is_running()750        self.mock_drone_manager.process_capacity = 3751        self._run_dispatcher() # now gathering runs752        self.mock_drone_manager.process_capacity = 0753        self.mock_drone_manager.finish_process(_PidfileType.GATHER)754        self._run_dispatcher() # parsing runs despite throttling755        _check_hqe_statuses(HqeStatus.PARSING, HqeStatus.PARSING,756                            HqeStatus.PARSING)757    def test_abort_starting_while_throttling(self):758        self._initialize_test()759        job = self._create_job(hosts=[1,2], synchronous=True)760        queue_entry = job.hostqueueentry_set.all()[0]761        job.run_verify = False762        job.run_reset = False763        job.reboot_after = model_attributes.RebootAfter.NEVER764        job.save()765        self.mock_drone_manager.process_capacity = 0766        self._run_dispatcher() # go to starting, but don't start job767        self._check_statuses(queue_entry, HqeStatus.STARTING,768                             HostStatus.PENDING)769        job.hostqueueentry_set.update(aborted=True)...

monitor_db_functional_unittest.py

Source:monitor_db_functional_unittest.py

...690        job = self._create_job(hosts=[1, 2, 3])691        job.synch_count = 3692        job.save()693        queue_entries = list(job.hostqueueentry_set.all())694        def _check_hqe_statuses(*statuses):695            for queue_entry, status in zip(queue_entries, statuses):696                self._check_statuses(queue_entry, status)697        self.mock_drone_manager.process_capacity = 2698        self._run_dispatcher()  # verify runs on 1 and 2699        _check_hqe_statuses(HqeStatus.VERIFYING, HqeStatus.VERIFYING,700                            HqeStatus.VERIFYING)701        self.assertEquals(len(self.mock_drone_manager.running_pidfile_ids()), 2)702        self.mock_drone_manager.finish_specific_process(703            'hosts/host1/1-verify', drone_manager.AUTOSERV_PID_FILE)704        self.mock_drone_manager.finish_process(_PidfileType.VERIFY)705        self._run_dispatcher()  # verify runs on 3706        _check_hqe_statuses(HqeStatus.PENDING, HqeStatus.PENDING,707                            HqeStatus.VERIFYING)708        self.mock_drone_manager.finish_process(_PidfileType.VERIFY)709        self._run_dispatcher()  # job won't run due to throttling710        _check_hqe_statuses(HqeStatus.STARTING, HqeStatus.STARTING,711                            HqeStatus.STARTING)712        self._assert_nothing_is_running()713        self.mock_drone_manager.process_capacity = 3714        self._run_dispatcher()  # now job runs715        _check_hqe_statuses(HqeStatus.RUNNING, HqeStatus.RUNNING,716                            HqeStatus.RUNNING)717        self.mock_drone_manager.process_capacity = 2718        self.mock_drone_manager.finish_process(_PidfileType.JOB,719                                               exit_status=271)720        self._run_dispatcher()  # gathering won't run due to throttling721        _check_hqe_statuses(HqeStatus.GATHERING, HqeStatus.GATHERING,722                            HqeStatus.GATHERING)723        self._assert_nothing_is_running()724        self.mock_drone_manager.process_capacity = 3725        self._run_dispatcher()  # now gathering runs726        self.mock_drone_manager.process_capacity = 0727        self.mock_drone_manager.finish_process(_PidfileType.GATHER)728        self._run_dispatcher()  # parsing runs despite throttling729        _check_hqe_statuses(HqeStatus.PARSING, HqeStatus.PARSING,730                            HqeStatus.PARSING)731    def test_abort_starting_while_throttling(self):732        self._initialize_test()733        job = self._create_job(hosts=[1, 2], synchronous=True)734        queue_entry = job.hostqueueentry_set.all()[0]735        job.run_verify = False736        job.reboot_after = model_attributes.RebootAfter.NEVER737        job.save()738        self.mock_drone_manager.process_capacity = 0739        self._run_dispatcher()  # go to starting, but don't start job740        self._check_statuses(queue_entry, HqeStatus.STARTING,741                             HostStatus.PENDING)742        job.hostqueueentry_set.update(aborted=True)743        self._run_dispatcher()...

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.