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!!
