How to use _filter_by_tenant_id method in tempest

Best Python code snippet using tempest_python
cleanup_service.py
Source:cleanup_service.py
...64    def __init__(self, kwargs):65        self.client = None66        for key, value in kwargs.items():67            setattr(self, key, value)68    def _filter_by_tenant_id(self, item_list):69        if (item_list is None70                or len(item_list) == 071                or not hasattr(self, 'tenant_id')72                or self.tenant_id is None73                or 'tenant_id' not in item_list[0]):74            return item_list75        _filtered_list = []76        for item in item_list:77            if item['tenant_id'] == self.tenant_id:78                _filtered_list.append(item)79        return _filtered_list80    def list(self):81        pass82    def delete(self):83        pass84    def dry_run(self):85        pass86    def save_state(self):87        pass88    def run(self):89        if self.is_dry_run:90            self.dry_run()91        elif self.is_save_state:92            self.save_state()93        else:94            self.delete()95class SnapshotService(BaseService):96    def __init__(self, manager, **kwargs):97        super(SnapshotService, self).__init__(kwargs)98        self.client = manager.snapshots_client99    def list(self):100        client = self.client101        __, snaps = client.list_snapshots()102        LOG.debug("List count, %s Snapshots" % len(snaps))103        return snaps104    def delete(self):105        snaps = self.list()106        client = self.client107        for snap in snaps:108            try:109                client.delete_snapshot(snap['id'])110            except Exception as e:111                LOG.exception("Delete Snapshot exception: %s" % e)112                pass113    def dry_run(self):114        snaps = self.list()115        self.data['snapshots'] = snaps116class ServerService(BaseService):117    def __init__(self, manager, **kwargs):118        super(ServerService, self).__init__(kwargs)119        self.client = manager.servers_client120    def list(self):121        client = self.client122        _, servers_body = client.list_servers()123        servers = servers_body['servers']124        LOG.debug("List count, %s Servers" % len(servers))125        return servers126    def delete(self):127        client = self.client128        servers = self.list()129        for server in servers:130            try:131                client.delete_server(server['id'])132            except Exception as e:133                LOG.exception("Delete Server exception: %s" % e)134                pass135    def dry_run(self):136        servers = self.list()137        self.data['servers'] = servers138class ServerGroupService(ServerService):139    def list(self):140        client = self.client141        _, sgs = client.list_server_groups()142        LOG.debug("List count, %s Server Groups" % len(sgs))143        return sgs144    def delete(self):145        client = self.client146        sgs = self.list()147        for sg in sgs:148            try:149                client.delete_server_group(sg['id'])150            except Exception as e:151                LOG.exception("Delete Server Group exception: %s" % e)152                pass153    def dry_run(self):154        sgs = self.list()155        self.data['server_groups'] = sgs156class StackService(BaseService):157    def __init__(self, manager, **kwargs):158        super(StackService, self).__init__(kwargs)159        self.client = manager.orchestration_client160    def list(self):161        client = self.client162        _, stacks = client.list_stacks()163        LOG.debug("List count, %s Stacks" % len(stacks))164        return stacks165    def delete(self):166        client = self.client167        stacks = self.list()168        for stack in stacks:169            try:170                client.delete_stack(stack['id'])171            except Exception as e:172                LOG.exception("Delete Stack exception: %s " % e)173                pass174    def dry_run(self):175        stacks = self.list()176        self.data['stacks'] = stacks177class KeyPairService(BaseService):178    def __init__(self, manager, **kwargs):179        super(KeyPairService, self).__init__(kwargs)180        self.client = manager.keypairs_client181    def list(self):182        client = self.client183        _, keypairs = client.list_keypairs()184        LOG.debug("List count, %s Keypairs" % len(keypairs))185        return keypairs186    def delete(self):187        client = self.client188        keypairs = self.list()189        for k in keypairs:190            try:191                name = k['keypair']['name']192                client.delete_keypair(name)193            except Exception as e:194                LOG.exception("Delete Keypairs exception: %s" % e)195                pass196    def dry_run(self):197        keypairs = self.list()198        self.data['keypairs'] = keypairs199class SecurityGroupService(BaseService):200    def __init__(self, manager, **kwargs):201        super(SecurityGroupService, self).__init__(kwargs)202        self.client = manager.security_groups_client203    def list(self):204        client = self.client205        _, secgrps = client.list_security_groups()206        secgrp_del = [grp for grp in secgrps if grp['name'] != 'default']207        LOG.debug("List count, %s Security Groups" % len(secgrp_del))208        return secgrp_del209    def delete(self):210        client = self.client211        secgrp_del = self.list()212        for g in secgrp_del:213            try:214                client.delete_security_group(g['id'])215            except Exception as e:216                LOG.exception("Delete Security Groups exception: %s" % e)217    def dry_run(self):218        secgrp_del = self.list()219        self.data['security_groups'] = secgrp_del220class FloatingIpService(BaseService):221    def __init__(self, manager, **kwargs):222        super(FloatingIpService, self).__init__(kwargs)223        self.client = manager.floating_ips_client224    def list(self):225        client = self.client226        _, floating_ips = client.list_floating_ips()227        LOG.debug("List count, %s Floating IPs" % len(floating_ips))228        return floating_ips229    def delete(self):230        client = self.client231        floating_ips = self.list()232        for f in floating_ips:233            try:234                client.delete_floating_ip(f['id'])235            except Exception as e:236                LOG.exception("Delete Floating IPs exception: %s" % e)237                pass238    def dry_run(self):239        floating_ips = self.list()240        self.data['floating_ips'] = floating_ips241class VolumeService(BaseService):242    def __init__(self, manager, **kwargs):243        super(VolumeService, self).__init__(kwargs)244        self.client = manager.volumes_client245    def list(self):246        client = self.client247        _, vols = client.list_volumes()248        LOG.debug("List count, %s Volumes" % len(vols))249        return vols250    def delete(self):251        client = self.client252        vols = self.list()253        for v in vols:254            try:255                client.delete_volume(v['id'])256            except Exception as e:257                LOG.exception("Delete Volume exception: %s" % e)258                pass259    def dry_run(self):260        vols = self.list()261        self.data['volumes'] = vols262# Begin network service classes263class NetworkService(BaseService):264    def __init__(self, manager, **kwargs):265        super(NetworkService, self).__init__(kwargs)266        self.client = manager.network_client267    def list(self):268        client = self.client269        _, networks = client.list_networks()270        networks = self._filter_by_tenant_id(networks['networks'])271        # filter out networks declared in tempest.conf272        if self.is_preserve:273            networks = [network for network in networks274                        if (network['name'] != CONF_PRIV_NETWORK_NAME275                            and network['id'] != CONF_PUB_NETWORK)]276        LOG.debug("List count, %s Networks" % networks)277        return networks278    def delete(self):279        client = self.client280        networks = self.list()281        for n in networks:282            try:283                client.delete_network(n['id'])284            except Exception as e:285                LOG.exception("Delete Network exception: %s" % e)286                pass287    def dry_run(self):288        networks = self.list()289        self.data['networks'] = networks290class NetworkIpSecPolicyService(NetworkService):291    def list(self):292        client = self.client293        _, ipsecpols = client.list_ipsecpolicies()294        ipsecpols = ipsecpols['ipsecpolicies']295        ipsecpols = self._filter_by_tenant_id(ipsecpols)296        LOG.debug("List count, %s IP Security Policies" % len(ipsecpols))297        return ipsecpols298    def delete(self):299        client = self.client300        ipsecpols = self.list()301        for ipsecpol in ipsecpols:302            try:303                client.delete_ipsecpolicy(ipsecpol['id'])304            except Exception as e:305                LOG.exception("Delete IP Securty Policy exception: %s" % e)306                pass307    def dry_run(self):308        ipsecpols = self.list()309        self.data['ip_security_policies'] = ipsecpols310class NetworkFwPolicyService(NetworkService):311    def list(self):312        client = self.client313        _, fwpols = client.list_firewall_policies()314        fwpols = fwpols['firewall_policies']315        fwpols = self._filter_by_tenant_id(fwpols)316        LOG.debug("List count, %s Firewall Policies" % len(fwpols))317        return fwpols318    def delete(self):319        client = self.client320        fwpols = self.list()321        for fwpol in fwpols:322            try:323                client.delete_firewall_policy(fwpol['id'])324            except Exception as e:325                LOG.exception("Delete Firewall Policy exception: %s" % e)326                pass327    def dry_run(self):328        fwpols = self.list()329        self.data['firewall_policies'] = fwpols330class NetworkFwRulesService(NetworkService):331    def list(self):332        client = self.client333        _, fwrules = client.list_firewall_rules()334        fwrules = fwrules['firewall_rules']335        fwrules = self._filter_by_tenant_id(fwrules)336        LOG.debug("List count, %s Firewall Rules" % len(fwrules))337        return fwrules338    def delete(self):339        client = self.client340        fwrules = self.list()341        for fwrule in fwrules:342            try:343                client.delete_firewall_rule(fwrule['id'])344            except Exception as e:345                LOG.exception("Delete Firewall Rule exception: %s" % e)346                pass347    def dry_run(self):348        fwrules = self.list()349        self.data['firewall_rules'] = fwrules350class NetworkIkePolicyService(NetworkService):351    def list(self):352        client = self.client353        _, ikepols = client.list_ikepolicies()354        ikepols = ikepols['ikepolicies']355        ikepols = self._filter_by_tenant_id(ikepols)356        LOG.debug("List count, %s IKE Policies" % len(ikepols))357        return ikepols358    def delete(self):359        client = self.client360        ikepols = self.list()361        for ikepol in ikepols:362            try:363                client.delete_firewall_rule(ikepol['id'])364            except Exception as e:365                LOG.exception("Delete IKE Policy exception: %s" % e)366                pass367    def dry_run(self):368        ikepols = self.list()369        self.data['ike_policies'] = ikepols370class NetworkVpnServiceService(NetworkService):371    def list(self):372        client = self.client373        _, vpnsrvs = client.list_vpnservices()374        vpnsrvs = vpnsrvs['vpnservices']375        vpnsrvs = self._filter_by_tenant_id(vpnsrvs)376        LOG.debug("List count, %s VPN Services" % len(vpnsrvs))377        return vpnsrvs378    def delete(self):379        client = self.client380        vpnsrvs = self.list()381        for vpnsrv in vpnsrvs:382            try:383                client.delete_vpnservice(vpnsrv['id'])384            except Exception as e:385                LOG.exception("Delete VPN Service exception: %s" % e)386                pass387    def dry_run(self):388        vpnsrvs = self.list()389        self.data['vpn_services'] = vpnsrvs390class NetworkFloatingIpService(NetworkService):391    def list(self):392        client = self.client393        _, flips = client.list_floatingips()394        flips = flips['floatingips']395        flips = self._filter_by_tenant_id(flips)396        LOG.debug("List count, %s Network Floating IPs" % len(flips))397        return flips398    def delete(self):399        client = self.client400        flips = self.list()401        for flip in flips:402            try:403                client.delete_floatingip(flip['id'])404            except Exception as e:405                LOG.exception("Delete Network Floating IP exception: %s" % e)406                pass407    def dry_run(self):408        flips = self.list()409        self.data['floating_ips'] = flips410class NetworkRouterService(NetworkService):411    def list(self):412        client = self.client413        _, routers = client.list_routers()414        routers = routers['routers']415        routers = self._filter_by_tenant_id(routers)416        if self.is_preserve:417            routers = [router for router in routers418                       if router['id'] != CONF_PUB_ROUTER]419        LOG.debug("List count, %s Routers" % len(routers))420        return routers421    def delete(self):422        client = self.client423        routers = self.list()424        for router in routers:425            try:426                rid = router['id']427                _, ports = client.list_router_interfaces(rid)428                ports = ports['ports']429                for port in ports:430                    subid = port['fixed_ips'][0]['subnet_id']431                    client.remove_router_interface_with_subnet_id(rid, subid)432                    client.delete_router(rid)433            except Exception as e:434                LOG.exception("Delete Router exception: %s" % e)435                pass436    def dry_run(self):437        routers = self.list()438        self.data['routers'] = routers439class NetworkHealthMonitorService(NetworkService):440    def list(self):441        client = self.client442        _, hms = client.list_health_monitors()443        hms = hms['health_monitors']444        hms = self._filter_by_tenant_id(hms)445        LOG.debug("List count, %s Health Monitors" % len(hms))446        return hms447    def delete(self):448        client = self.client449        hms = self.list()450        for hm in hms:451            try:452                client.delete_health_monitor(hm['id'])453            except Exception as e:454                LOG.exception("Delete Health Monitor exception: %s" % e)455                pass456    def dry_run(self):457        hms = self.list()458        self.data['health_monitors'] = hms459class NetworkMemberService(NetworkService):460    def list(self):461        client = self.client462        _, members = client.list_members()463        members = members['members']464        members = self._filter_by_tenant_id(members)465        LOG.debug("List count, %s Members" % len(members))466        return members467    def delete(self):468        client = self.client469        members = self.list()470        for member in members:471            try:472                client.delete_member(member['id'])473            except Exception as e:474                LOG.exception("Delete Member exception: %s" % e)475                pass476    def dry_run(self):477        members = self.list()478        self.data['members'] = members479class NetworkVipService(NetworkService):480    def list(self):481        client = self.client482        _, vips = client.list_vips()483        vips = vips['vips']484        vips = self._filter_by_tenant_id(vips)485        LOG.debug("List count, %s VIPs" % len(vips))486        return vips487    def delete(self):488        client = self.client489        vips = self.list()490        for vip in vips:491            try:492                client.delete_vip(vip['id'])493            except Exception as e:494                LOG.exception("Delete VIP exception: %s" % e)495                pass496    def dry_run(self):497        vips = self.list()498        self.data['vips'] = vips499class NetworkPoolService(NetworkService):500    def list(self):501        client = self.client502        _, pools = client.list_pools()503        pools = pools['pools']504        pools = self._filter_by_tenant_id(pools)505        LOG.debug("List count, %s Pools" % len(pools))506        return pools507    def delete(self):508        client = self.client509        pools = self.list()510        for pool in pools:511            try:512                client.delete_pool(pool['id'])513            except Exception as e:514                LOG.exception("Delete Pool exception: %s" % e)515                pass516    def dry_run(self):517        pools = self.list()518        self.data['pools'] = pools519class NetworMeteringLabelRuleService(NetworkService):520    def list(self):521        client = self.client522        _, rules = client.list_metering_label_rules()523        rules = rules['metering_label_rules']524        rules = self._filter_by_tenant_id(rules)525        LOG.debug("List count, %s Metering Label Rules" % len(rules))526        return rules527    def delete(self):528        client = self.client529        rules = self.list()530        for rule in rules:531            try:532                client.delete_metering_label_rule(rule['id'])533            except Exception as e:534                LOG.exception("Delete Metering Label Rule exception: %s" % e)535                pass536    def dry_run(self):537        rules = self.list()538        self.data['rules'] = rules539class NetworMeteringLabelService(NetworkService):540    def list(self):541        client = self.client542        _, labels = client.list_metering_labels()543        labels = labels['metering_labels']544        labels = self._filter_by_tenant_id(labels)545        LOG.debug("List count, %s Metering Labels" % len(labels))546        return labels547    def delete(self):548        client = self.client549        labels = self.list()550        for label in labels:551            try:552                client.delete_metering_label(label['id'])553            except Exception as e:554                LOG.exception("Delete Metering Label exception: %s" % e)555                pass556    def dry_run(self):557        labels = self.list()558        self.data['labels'] = labels559class NetworkPortService(NetworkService):560    def list(self):561        client = self.client562        _, ports = client.list_ports()563        ports = ports['ports']564        ports = self._filter_by_tenant_id(ports)565        LOG.debug("List count, %s Ports" % len(ports))566        return ports567    def delete(self):568        client = self.client569        ports = self.list()570        for port in ports:571            try:572                client.delete_port(port['id'])573            except Exception as e:574                LOG.exception("Delete Port exception: %s" % e)575                pass576    def dry_run(self):577        ports = self.list()578        self.data['ports'] = ports579class NetworkSubnetService(NetworkService):580    def list(self):581        client = self.client582        _, subnets = client.list_subnets()583        subnets = subnets['subnets']584        subnets = self._filter_by_tenant_id(subnets)585        LOG.debug("List count, %s Subnets" % len(subnets))586        return subnets587    def delete(self):588        client = self.client589        subnets = self.list()590        for subnet in subnets:591            try:592                client.delete_subnet(subnet['id'])593            except Exception as e:594                LOG.exception("Delete Subnet exception: %s" % e)595                pass596    def dry_run(self):597        subnets = self.list()598        self.data['subnets'] = subnets...
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.