Best Python code snippet using autotest_python
metahost_scheduler.py
Source:metahost_scheduler.py
2Host Scheduler classes.3"""4class HostSchedulingUtility(object):5 """Interface to host availability information from the scheduler."""6 def hosts_in_label(self, label_id):7 """Return potentially usable hosts with the given label."""8 raise NotImplementedError9 def remove_host_from_label(self, host_id, label_id):10 """Remove this host from the internal list of usable hosts in the label.11 This is provided as an optimization -- when code gets a host from a12 label and concludes it's unusable, it can call this to avoid getting13 that host again in the future (within this tick). This function should14 not affect correctness.15 """16 raise NotImplementedError17 def pop_host(self, host_id):18 """Remove and return a host from the internal list of available hosts.19 """20 raise NotImplementedError21 def ineligible_hosts_for_entry(self, queue_entry):22 """Get the list of hosts ineligible to run the given queue entry."""23 raise NotImplementedError24 def is_host_usable(self, host_id):25 """Determine if the host is currently usable at all."""26 raise NotImplementedError27 def is_host_eligible_for_job(self, host_id, queue_entry):28 """Determine if the host is eligible specifically for this queue entry.29 :param queue_entry: a HostQueueEntry DBObject30 """31 raise NotImplementedError32class MetahostScheduler(object):33 def can_schedule_metahost(self, queue_entry):34 """Return true if this object can schedule the given queue entry.35 At most one MetahostScheduler should return true for any given entry.36 :param queue_entry: a HostQueueEntry DBObject37 """38 raise NotImplementedError39 def schedule_metahost(self, queue_entry, scheduling_utility):40 """Schedule the given queue entry, if possible.41 This method should make necessary database changes culminating in42 assigning a host to the given queue entry in the database. It may43 take no action if no host can be assigned currently.44 :param queue_entry: a HostQueueEntry DBObject45 :param scheduling_utility: a HostSchedulingUtility object46 """47 raise NotImplementedError48 def recovery_on_startup(self):49 """Perform any necessary recovery upon scheduler startup."""50 pass51 def tick(self):52 """Called once per scheduler cycle; any actions are allowed."""53 pass54class LabelMetahostScheduler(MetahostScheduler):55 def can_schedule_metahost(self, queue_entry):56 return bool(queue_entry.meta_host)57 def schedule_metahost(self, queue_entry, scheduling_utility):58 label_id = queue_entry.meta_host59 hosts_in_label = scheduling_utility.hosts_in_label(label_id)60 ineligible_host_ids = scheduling_utility.ineligible_hosts_for_entry(61 queue_entry)62 for host_id in hosts_in_label:63 if not scheduling_utility.is_host_usable(host_id):64 scheduling_utility.remove_host_from_label(host_id, label_id)65 continue66 if host_id in ineligible_host_ids:67 continue68 if not scheduling_utility.is_host_eligible_for_job(host_id,69 queue_entry):70 continue71 # Remove the host from our cached internal state before returning72 scheduling_utility.remove_host_from_label(host_id, label_id)73 host = scheduling_utility.pop_host(host_id)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!