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

How to use _assert_job_scheduled_on_number_of method in autotest

Best Python code snippet using autotest_python

atomic_group_unittests.py

Source:atomic_group_unittests.py

...39        job_b = self._create_job(synchronous=True, metahosts=[self.label5.id],40                         atomic_group=1)41        self._run_scheduler()42        # atomic_group.max_number_of_machines was 2 so we should run on 2.43        self._assert_job_scheduled_on_number_of(job_a.id, (5, 6, 7), 2)44        self._assert_job_scheduled_on(job_b.id, 8)  # label545        self._assert_job_scheduled_on(job_b.id, 9)  # label546        self._check_for_extra_schedulings()47        # The three host label4 atomic group still has one host available.48        # That means a job with a synch_count of 1 asking to be scheduled on49        # the atomic group can still use the final machine.50        #51        # This may seem like a somewhat odd use case.  It allows the use of an52        # atomic group as a set of machines to run smaller jobs within (a set53        # of hosts configured for use in network tests with eachother perhaps?)54        onehost_job = self._create_job(atomic_group=1)55        self._run_scheduler()56        self._assert_job_scheduled_on_number_of(onehost_job.id, (5, 6, 7), 1)57        self._check_for_extra_schedulings()58        # No more atomic groups have hosts available, no more jobs should59        # be scheduled.60        self._create_job(atomic_group=1)61        self._run_scheduler()62        self._check_for_extra_schedulings()63    def test_atomic_group_scheduling_obeys_acls(self):64        # Request scheduling on a specific atomic label but be denied by ACLs.65        self._do_query('DELETE FROM afe_acl_groups_hosts '66                       'WHERE host_id in (8,9)')67        job = self._create_job(metahosts=[self.label5.id], atomic_group=1)68        self._run_scheduler()69        self._check_for_extra_schedulings()70    def test_atomic_group_scheduling_dependency_label_exclude(self):71        # A dependency label that matches no hosts in the atomic group.72        job_a = self._create_job(atomic_group=1)73        job_a.dependency_labels.add(self.label3)74        self._run_scheduler()75        self._check_for_extra_schedulings()76    def test_atomic_group_scheduling_metahost_dependency_label_exclude(self):77        # A metahost and dependency label that excludes too many hosts.78        job_b = self._create_job(synchronous=True, metahosts=[self.label4.id],79                                 atomic_group=1)80        job_b.dependency_labels.add(self.label7)81        self._run_scheduler()82        self._check_for_extra_schedulings()83    def test_atomic_group_scheduling_dependency_label_match(self):84        # A dependency label that exists on enough atomic group hosts in only85        # one of the two atomic group labels.86        job_c = self._create_job(synchronous=True, atomic_group=1)87        job_c.dependency_labels.add(self.label7)88        self._run_scheduler()89        self._assert_job_scheduled_on_number_of(job_c.id, (8, 9), 2)90        self._check_for_extra_schedulings()91    def test_atomic_group_scheduling_no_metahost(self):92        # Force it to schedule on the other group for a reliable test.93        self._do_query('UPDATE afe_hosts SET invalid=1 WHERE id=9')94        # An atomic job without a metahost.95        job = self._create_job(synchronous=True, atomic_group=1)96        self._run_scheduler()97        self._assert_job_scheduled_on_number_of(job.id, (5, 6, 7), 2)98        self._check_for_extra_schedulings()99    def test_atomic_group_scheduling_partial_group(self):100        # Make one host in labels[3] unavailable so that there are only two101        # hosts left in the group.102        self._do_query('UPDATE afe_hosts SET status="Repair Failed" WHERE id=5')103        job = self._create_job(synchronous=True, metahosts=[self.label4.id],104                         atomic_group=1)105        self._run_scheduler()106        # Verify that it was scheduled on the 2 ready hosts in that group.107        self._assert_job_scheduled_on(job.id, 6)108        self._assert_job_scheduled_on(job.id, 7)109        self._check_for_extra_schedulings()110    def test_atomic_group_scheduling_not_enough_available(self):111        # Mark some hosts in each atomic group label as not usable.112        # One host running, another invalid in the first group label.113        self._do_query('UPDATE afe_hosts SET status="Running" WHERE id=5')114        self._do_query('UPDATE afe_hosts SET invalid=1 WHERE id=6')115        # One host invalid in the second group label.116        self._do_query('UPDATE afe_hosts SET invalid=1 WHERE id=9')117        # Nothing to schedule when no group label has enough (2) good hosts..118        self._create_job(atomic_group=1, synchronous=True)119        self._run_scheduler()120        # There are not enough hosts in either atomic group,121        # No more scheduling should occur.122        self._check_for_extra_schedulings()123        # Now create an atomic job that has a synch count of 1.  It should124        # schedule on exactly one of the hosts.125        onehost_job = self._create_job(atomic_group=1)126        self._run_scheduler()127        self._assert_job_scheduled_on_number_of(onehost_job.id, (7, 8), 1)128    def test_atomic_group_scheduling_no_valid_hosts(self):129        self._do_query('UPDATE afe_hosts SET invalid=1 WHERE id in (8,9)')130        self._create_job(synchronous=True, metahosts=[self.label5.id],131                         atomic_group=1)132        self._run_scheduler()133        # no hosts in the selected group and label are valid.  no schedulings.134        self._check_for_extra_schedulings()135    def test_atomic_group_scheduling_metahost_works(self):136        # Test that atomic group scheduling also obeys metahosts.137        self._create_job(metahosts=[0], atomic_group=1)138        self._run_scheduler()139        # There are no atomic group hosts that also have that metahost.140        self._check_for_extra_schedulings()141        job_b = self._create_job(metahosts=[self.label5.id], atomic_group=1)...

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.