KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _assert_job_scheduled_on method in autotest

Best Python code snippet using autotest_python
atomic_group_unittests.py
Source:atomic_group_unittests.py
...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)142        self._run_scheduler()143        self._assert_job_scheduled_on(job_b.id, 8)144        self._assert_job_scheduled_on(job_b.id, 9)145        self._check_for_extra_schedulings()146    def test_atomic_group_skips_ineligible_hosts(self):147        # Test hosts marked ineligible for this job are not eligible.148        # How would this ever happen anyways?149        job = self._create_job(metahosts=[self.label4.id], atomic_group=1)150        models.IneligibleHostQueue.objects.create(job=job, host_id=5)151        models.IneligibleHostQueue.objects.create(job=job, host_id=6)152        models.IneligibleHostQueue.objects.create(job=job, host_id=7)153        self._run_scheduler()154        # No scheduling should occur as all desired hosts were ineligible.155        self._check_for_extra_schedulings()156    def test_atomic_group_scheduling_fail(self):157        # If synch_count is > the atomic group number of machines, the job158        # should be aborted immediately.159        model_job = self._create_job(synchronous=True, atomic_group=1)160        model_job.synch_count = 4161        model_job.save()162        job = scheduler_models.Job(id=model_job.id)163        self._run_scheduler()164        self._check_for_extra_schedulings()165        queue_entries = job.get_host_queue_entries()166        self.assertEqual(1, len(queue_entries))167        self.assertEqual(queue_entries[0].status,168                         models.HostQueueEntry.Status.ABORTED)169    def test_atomic_group_no_labels_no_scheduling(self):170        # Never schedule on atomic groups marked invalid.171        job = self._create_job(metahosts=[self.label5.id], synchronous=True,172                               atomic_group=1)173        # Deleting an atomic group via the frontend marks it invalid and174        # removes all label references to the group.  The job now references175        # an invalid atomic group with no labels associated with it.176        self.label5.atomic_group.invalid = True177        self.label5.atomic_group.save()178        self.label5.atomic_group = None179        self.label5.save()180        self._run_scheduler()181        self._check_for_extra_schedulings()182    def test_schedule_directly_on_atomic_group_host_fail(self):183        # Scheduling a job directly on hosts in an atomic group must184        # fail to avoid users inadvertently holding up the use of an185        # entire atomic group by using the machines individually.186        job = self._create_job(hosts=[5])187        self._run_scheduler()188        self._check_for_extra_schedulings()189    def test_schedule_directly_on_atomic_group_host(self):190        # Scheduling a job directly on one host in an atomic group will191        # work when the atomic group is listed on the HQE in addition192        # to the host (assuming the sync count is 1).193        job = self._create_job(hosts=[5], atomic_group=1)194        self._run_scheduler()195        self._assert_job_scheduled_on(job.id, 5)196        self._check_for_extra_schedulings()197    def test_schedule_directly_on_atomic_group_hosts_sync2(self):198        job = self._create_job(hosts=[5,8], atomic_group=1, synchronous=True)199        self._run_scheduler()200        self._assert_job_scheduled_on(job.id, 5)201        self._assert_job_scheduled_on(job.id, 8)202        self._check_for_extra_schedulings()203    def test_schedule_directly_on_atomic_group_hosts_wrong_group(self):204        job = self._create_job(hosts=[5,8], atomic_group=2, synchronous=True)205        self._run_scheduler()206        self._check_for_extra_schedulings()207    # TODO(gps): These should probably live in their own TestCase class208    # specific to testing HostScheduler methods directly.  It was convenient209    # to put it here for now to share existing test environment setup code.210    def test_HostScheduler_check_atomic_group_labels(self):211        normal_job = self._create_job(metahosts=[0])212        atomic_job = self._create_job(atomic_group=1)213        # Indirectly initialize the internal state of the host scheduler.214        self._dispatcher._refresh_pending_queue_entries()215        atomic_hqe = scheduler_models.HostQueueEntry.fetch(where='job_id=%d' %216                                                     atomic_job.id)[0]217        normal_hqe = scheduler_models.HostQueueEntry.fetch(where='job_id=%d' %218                                                     normal_job.id)[0]219        host_scheduler = self._dispatcher._host_scheduler220        self.assertTrue(host_scheduler._check_atomic_group_labels(221                [self.label4.id], atomic_hqe))222        self.assertFalse(host_scheduler._check_atomic_group_labels(223                [self.label4.id], normal_hqe))224        self.assertFalse(host_scheduler._check_atomic_group_labels(225                [self.label5.id, self.label6.id, self.label7.id], normal_hqe))226        self.assertTrue(host_scheduler._check_atomic_group_labels(227                [self.label4.id, self.label6.id], atomic_hqe))228        self.assertTrue(host_scheduler._check_atomic_group_labels(229                        [self.label4.id, self.label5.id],230                        atomic_hqe))231class OnlyIfNeededTest(monitor_db_unittest.DispatcherSchedulingTest):232    def _setup_test_only_if_needed_labels(self):233        # apply only_if_needed label3 to host1234        models.Host.smart_get('host1').labels.add(self.label3)235        return self._create_job_simple([1], use_metahost=True)236    def test_only_if_needed_labels_avoids_host(self):237        job = self._setup_test_only_if_needed_labels()238        # if the job doesn't depend on label3, there should be no scheduling239        self._run_scheduler()240        self._check_for_extra_schedulings()241    def test_only_if_needed_labels_schedules(self):242        job = self._setup_test_only_if_needed_labels()243        job.dependency_labels.add(self.label3)244        self._run_scheduler()245        self._assert_job_scheduled_on(1, 1)246        self._check_for_extra_schedulings()247    def test_only_if_needed_labels_via_metahost(self):248        job = self._setup_test_only_if_needed_labels()249        job.dependency_labels.add(self.label3)250        # should also work if the metahost is the only_if_needed label251        self._do_query('DELETE FROM afe_jobs_dependency_labels')252        self._create_job(metahosts=[3])253        self._run_scheduler()254        self._assert_job_scheduled_on(2, 1)255        self._check_for_extra_schedulings()256    def test_metahosts_obey_blocks(self):257        """258        Metahosts can't get scheduled on hosts already scheduled for259        that job.260        """261        self._create_job(metahosts=[1], hosts=[1])262        # make the nonmetahost entry complete, so the metahost can try263        # to get scheduled264        self._update_hqe(set='complete = 1', where='host_id=1')265        self._run_scheduler()...
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.