Best Python code snippet using autotest_python
rpc_utils.py
Source:rpc_utils.py
...268 fp.close()269 except:270 pass271 return text272def _get_metahost_counts(metahost_objects):273 metahost_counts = {}274 for metahost in metahost_objects:275 metahost_counts.setdefault(metahost, 0)276 metahost_counts[metahost] += 1277 return metahost_counts278def get_job_info(job, preserve_metahosts=False, queue_entry_filter_data=None):279 hosts = []280 one_time_hosts = []281 meta_hosts = []282 atomic_group = None283 queue_entries = job.hostqueueentry_set.all()284 if queue_entry_filter_data:285 queue_entries = models.HostQueueEntry.query_objects(286 queue_entry_filter_data, initial_query=queue_entries)287 for queue_entry in queue_entries:288 if (queue_entry.host and (preserve_metahosts or289 not queue_entry.meta_host)):290 if queue_entry.deleted:291 continue292 if queue_entry.host.invalid:293 one_time_hosts.append(queue_entry.host)294 else:295 hosts.append(queue_entry.host)296 else:297 meta_hosts.append(queue_entry.meta_host)298 if atomic_group is None:299 if queue_entry.atomic_group is not None:300 atomic_group = queue_entry.atomic_group301 else:302 assert atomic_group.name == queue_entry.atomic_group.name, (303 'DB inconsistency. HostQueueEntries with multiple atomic'304 ' groups on job %s: %s != %s' % (305 id, atomic_group.name, queue_entry.atomic_group.name))306 meta_host_counts = _get_metahost_counts(meta_hosts)307 info = dict(dependencies=[label.name for label308 in job.dependency_labels.all()],309 hosts=hosts,310 meta_hosts=meta_hosts,311 meta_host_counts=meta_host_counts,312 one_time_hosts=one_time_hosts,313 atomic_group=atomic_group)314 return info315def create_new_job(owner, options, host_objects, metahost_objects,316 atomic_group=None):317 labels_by_name = dict((label.name, label)318 for label in models.Label.objects.all())319 all_host_objects = host_objects + metahost_objects320 metahost_counts = _get_metahost_counts(metahost_objects)321 dependencies = options.get('dependencies', [])322 synch_count = options.get('synch_count')323 # check that each metahost request has enough hosts under the label324 for label, requested_count in metahost_counts.iteritems():325 available_count = label.host_set.count()326 if requested_count > available_count:327 error = ("You have requested %d %s's, but there are only %d."328 % (requested_count, label.name, available_count))329 raise model_logic.ValidationError({'meta_hosts' : error})330 if atomic_group:331 check_atomic_group_create_job(332 synch_count, host_objects, metahost_objects,333 dependencies, atomic_group, labels_by_name)334 else:...
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!!