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 Github

copy

Full Screen

...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()...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful