Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _get_hosts method in autotest

Best Python code snippet using autotest_python

notifier.py

Source:notifier.py

...81        gnp = GrowlNotificationPacket(notification='attachment',82                                      title='Attachment added',83                                      description=attachment.title)84        gs = GrowlSender(self.env)85        gs.notify(self._get_hosts('attachment'), gnp)86    def attachment_deleted(self, attachment):87        """Called when an attachment is deleted."""88        if 'attachment' not in self.sources:89            return90        gnp = GrowlNotificationPacket(notification='ticket',91                                      title='Attachment deleted',92                                      description=attachment.title)93        gs = GrowlSender(self.env)94        gs.notify(self._get_hosts('attachment'), gnp)95    # ITicketChangeListener Interface96    def ticket_created(self, ticket):97        """Called when a ticket is created."""98        if 'ticket' not in self.sources:99            return100        gnp = GrowlNotificationPacket(notification='ticket',101                                      title='Ticket #%d created' % ticket.id,102                                      description=self._ticket_repr(ticket))103        gs = GrowlSender(self.env)104        gs.notify(self._get_hosts('ticket'), gnp)105    def ticket_changed(self, ticket, comment, author, old_values):106        """Called when a ticket is modified."""107        if 'ticket' not in self.sources:108            return109        gnp = GrowlNotificationPacket(notification='ticket',110                                      title='Ticket #%d updated' % ticket.id,111                                      description=self._ticket_repr(ticket))112        gs = GrowlSender(self.env)113        gs.notify(self._get_hosts('ticket'), gnp)114    def ticket_deleted(self, ticket):115        """Called when a ticket is deleted."""116        if 'ticket' not in self.sources:117            return118        gnp = GrowlNotificationPacket(notification='ticket',119                                      title='Ticket #%d deleted' % ticket.id,120                                      description=self._ticket_repr(ticket))121        gs = GrowlSender(self.env)122        gs.notify(self._get_hosts('ticket'), gnp)123    # IWikiChangeListener Interface124    def wiki_page_added(self, page):125        """Called whenever a new Wiki page is added."""126        if 'wiki' not in self.sources:127            return128        gnp = GrowlNotificationPacket(notification='wiki',129                                      title='Page created',130                                      description=page.name)131        gs = GrowlSender(self.env)132        gs.notify(self._get_hosts('wiki'), gnp)133    def wiki_page_changed(self, page, version, t, comment, author, ipnr):134        """Called when a page has been modified."""135        if 'wiki' not in self.sources:136            return137        gnp = GrowlNotificationPacket(notification='wiki',138                                      title='Page updated',139                                      description=self._wiki_repr(page,140                                                                  comment))141        gs = GrowlSender(self.env)142        gs.notify(self._get_hosts('wiki'), gnp)143    def wiki_page_deleted(self, page):144        """Called when a page has been deleted."""145        if 'wiki' not in self.sources:146            return147        gnp = GrowlNotificationPacket(notification='wiki',148                                      title='Page deleted',149                                      description=self._wiki_repr(page))150        gs = GrowlSender(self.env)151        gs.notify(self._get_hosts('wiki'), gnp)152    def wiki_page_version_deleted(self, page):153        """Called when a version of a page has been deleted."""154        if 'wiki' not in self.sources:155            return156        gnp = GrowlNotificationPacket(notification='wiki',157                                      title='Page suppressed',158                                      description=self._wiki_repr(page))159        gs = GrowlSender(self.env)160        gs.notify(self._get_hosts('wiki'), gnp)161    # IBuildListener Interface162    def build_started(build):163        """Called when a build slave has accepted a build initiation."""164        if 'bitten' not in self.sources:165            return166        gnp = GrowlNotificationPacket(notification='bitten',167                                      title='Build started',168                                      description=self._bitten_repr(build),169                                      priority=-2)170        gs = GrowlSender(self.env)171        gs.notify(self._get_hosts('bitten'), gnp)172    173    def build_aborted(build):174        """Called when a build slave cancels a build or disconnects."""175        if 'bitten' not in self.sources:176            return177        gnp = GrowlNotificationPacket(notification='bitten',178                                      title='Build aborted',179                                      description=self._bitten_repr(build))180        gs = GrowlSender(self.env)181        gs.notify(self._get_hosts('bitten'), gnp)182    183    def build_completed(build):184        """Called when a build slave has completed a build, regardless of the185        outcome."""186        if 'bitten' not in self.sources:187            return188        failure = self.status == Build.FAILURE189        status = 'Build %s' % failure and 'failed' or 'successful'190        gnp = GrowlNotificationPacket(notification='bitten',191                                      title=status,192                                      description=self._bitten_repr(build),193                                      sticky=failure,194                                      priority=failure and 2 or 0)195        gs = GrowlSender(self.env)196        gs.notify(self._get_hosts('bitten'), gnp)197    # API198    199    def get_available_sources(self):200        return self.sources201    def is_userprefs_enabled(self):202        return self.userprefs_enabled203        204    def register_notifications(self, hosts):205        """Asks Growl clients to register Trac application. A bit suboptimal, 206        but that's the only way to register without an explicit user 207        registration""" 208        grp = GrowlRegistrationPacket()209        for n in self.avail_sources:210            grp.addNotification(n, n in self.sources)211        gs = GrowlSender(self.env)212        gs.notify(hosts, grp)213    def validate_host(self, admin, host):214        if host == '<broadcast>':215            if not admin:216                raise PermissionError('Broadcast: GROWL_ADMIN')217            return True218        # TODO: implement host validation219        try:220            r = gethostbyname(host)221            self.log.info("Address of %s: %s" % (host, r))222        except gaierror:223            raise TracError("Host '%s' is invalid" % host)224        return True225    # Implementation226    227    def __init__(self):228        self.sources = [s.strip().lower() for s in self.avail_sources]229        self.hosts = filter(None, [h.strip() for h in self.dest_hosts])230        # register project-defined hosts231        self.register_notifications(self.hosts)232        233    def _ticket_repr(self, ticket):234        """String representation of a Trac ticket"""235        rep = '%s (%s)' % (ticket['summary'], ticket['status'])236        return rep237        238    def _wiki_repr(self, page, comment=None):239        """String representation of a Trac wiki page"""240        rep = page.name241        if comment:242            rep += '\n%s' % comment243        return rep244        245    def _bitten_repr(self, build):246        """String representation of a Bitten build"""247        rep = '%s (%d) r%d' % (build.config, build.id, build.rev)248        if build.platform:249            rep += ' %s' % build.platform 250        if build.slave:251            rep += ' %s' % build.slave252        return rep253    254    def _get_hosts(self, source):255        # get user-specific hosts256        hosts = self._get_user_hosts(source)257        # add hosts defined in the project config, removing duplicates258        hosts.extend([h for h in self.hosts if h not in hosts])259        return hosts260    def _get_user_hosts(self, source):261        db = self.env.get_db_cnx()262        cursor = db.cursor()263        cursor.execute("SELECT DISTINCT H.value " \264            "FROM session_attribute S, session_attribute H " \265            "WHERE (S.name=%s AND S.value='1') " \266            "AND H.name='growl.host' AND S.sid=H.sid", 267            ("growl.source.%s" % source,))268        hosts = []...

availability_zones.py

Source:availability_zones.py

...36        Raises:37            AssertionError: if not all hosts are active38            TimeoutExpired: if there is no updates for hosts39        """40        def _get_hosts():41            zone = waiting.wait(42                lambda: self._client.find(zoneName=zone_name),43                timeout_seconds=config.NOVA_AVAILABILITY_TIMEOUT,44                expected_exceptions=nova_exceptions.ClientException)45            for hosts_dict in zone.hosts.values():46                for host in hosts_dict.values():47                    host['updated_at'] = parser.parse(host['updated_at'])48                    yield host49        last_updated = max([x['updated_at'] for x in _get_hosts()])50        def _predicate():51            return all([x['updated_at'] > last_updated for x in _get_hosts()])52        waiting.wait(53            _predicate,54            timeout_seconds=config.NOVA_AVAILABILITY_TIMEOUT)55        active_hosts = [x for x in _get_hosts() if x['active']]56        assert_that(active_hosts, only_contains(has_entries(available=True)))57    @steps_checker.step58    def get_zones(self, check=True, **kwargs):59        """Step to find all zones matching **kwargs.60        Args:61            check (bool): flag whether to check step or not62            **kwargs: like: {'zoneName': 'nova'},63                            {'zoneState': {u'available': True}}64        Returns:65            list: nova zones66        """67        zones = self._client.findall(**kwargs)68        if check:69            assert_that(zones, is_not(empty()))...

config.py

Source:config.py

...3    data = {}4    with open(filepath) as f:5        data = json.loads(f.read())6    return data7def _get_hosts(config: dict) -> dict:8    hosts = config.get('hosts')9    if not hosts:10        raise Exception('load config fialed')11    data = {}12    for hconfig in hosts:13        if not isinstance(hconfig, dict):14            raise Exception('load config fialed')15        name = hconfig.get('name')16        _type = hconfig.get('type')17        uri = hconfig.get('uri')18        if not name or not _type or not uri:19            raise Exception('load config fialed')20        data[name] = hconfig21    return data22def load(filepath: str, need_check: bool = True) -> dict:23    data = _load_config(filepath)...

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.