KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _check_no_acl_for_metahost method in autotest

Best Python code snippet using autotest_python

host_scheduler.py

Source:host_scheduler.py

...174        :return: True if atomic group settings are okay, False otherwise.175        """176        return (self._get_host_atomic_group_id(host_labels, queue_entry) ==177                queue_entry.atomic_group_id)178    def _check_no_acl_for_metahost(self, host_id, queue_entry):179        """180        If the given HostQueueEntry is for a metahost, determine if the given181        host id has any user-specific ACL set. We do not want to schedule a182        metahost job on a reserved host.183        :param host_id: host id to examine.184        :param queue_entry: The HostQueueEntry being considered for the host.185        :return: True if entry is not a metahost or the host only has the "Everyone" ACL, False otherwise.186        """187        host_acls = self._host_acls.get(host_id, set())188        return (queue_entry.meta_host is None or host_acls == self.everyone_acl)189    def _get_host_atomic_group_id(self, host_labels, queue_entry=None):190        """191        Return the atomic group label id for a host with the given set of192        labels if any, or None otherwise.  Raises an exception if more than193        one atomic group are found in the set of labels.194        :param host_labels: A list of label ids that the host has.195        :param queue_entry: The HostQueueEntry we're testing.  Only used for196                extra info in a potential logged error message.197        :return: The id of the atomic group found on a label in host_labels198                or None if no atomic group label is found.199        """200        atomic_labels = [self._labels[label_id] for label_id in host_labels201                         if self._labels[label_id].atomic_group_id is not None]202        atomic_ids = set(label.atomic_group_id for label in atomic_labels)203        if not atomic_ids:204            return None205        if len(atomic_ids) > 1:206            logging.error('More than one Atomic Group on HQE "%s" via: %r',207                          queue_entry, atomic_labels)208        return atomic_ids.pop()209    def _get_atomic_group_labels(self, atomic_group_id):210        """211        Lookup the label ids that an atomic_group is associated with.212        :param atomic_group_id - The id of the AtomicGroup to look up.213        :return: A generator yielding Label ids for this atomic group.214        """215        return (a_id for a_id, label in self._labels.iteritems() if216                label.atomic_group_id == atomic_group_id and not217                label.invalid)218    def _get_eligible_host_ids_in_group(self, group_hosts, queue_entry):219        """220        :param group_hosts - A sequence of Host ids to test for usability221                and eligibility against the Job associated with queue_entry.222        :param queue_entry - The HostQueueEntry that these hosts are being223                tested for eligibility against.224        :return: A subset of group_hosts Host ids that are eligible for the225                supplied queue_entry.226        """227        return set(host_id for host_id in group_hosts if228                   self.is_host_usable(host_id) and229                   self.is_host_eligible_for_job(host_id, queue_entry))230    def is_host_eligible_for_job(self, host_id, queue_entry):231        if self._is_host_invalid(host_id):232            # if an invalid host is scheduled for a job, it's a one-time host233            # and it therefore bypasses eligibility checks. note this can only234            # happen for non-metahosts, because invalid hosts have their label235            # relationships cleared.236            return True237        job_dependencies = self._job_dependencies.get(queue_entry.job_id, set())238        host_labels = self._host_labels.get(host_id, set())239        return (self._is_acl_accessible(host_id, queue_entry) and240                self._check_job_dependencies(job_dependencies, host_labels) and241                self._check_only_if_needed_labels(242                    job_dependencies, host_labels, queue_entry) and243                self._check_atomic_group_labels(host_labels, queue_entry) and244                self._check_no_acl_for_metahost(host_id, queue_entry))245    def _is_host_invalid(self, host_id):246        host_object = self._hosts_available.get(host_id, None)247        return host_object and host_object.invalid248    def _schedule_non_metahost(self, queue_entry):249        if not self.is_host_eligible_for_job(queue_entry.host_id, queue_entry):250            return None251        return self._hosts_available.pop(queue_entry.host_id, None)252    def is_host_usable(self, host_id):253        if host_id not in self._hosts_available:254            # host was already used during this scheduling cycle255            return False256        if self._hosts_available[host_id].invalid:257            # Invalid hosts cannot be used for metahosts.  They're included in258            # the original query because they can be used by non-metahosts....

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.