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

How to use call_and_ignore_notfound_exc method in tempest

Best Python code snippet using tempest_python

base.py

Source:base.py

...102        return volume_type103    @classmethod104    def clear_qos_specs(cls):105        for qos_id in cls.qos_specs:106            test_utils.call_and_ignore_notfound_exc(107                cls.admin_volume_qos_client.delete_qos, qos_id)108        for qos_id in cls.qos_specs:109            test_utils.call_and_ignore_notfound_exc(110                cls.admin_volume_qos_client.wait_for_resource_deletion, qos_id)111    @classmethod112    def clear_volume_types(cls):113        for vol_type in cls.volume_types:114            test_utils.call_and_ignore_notfound_exc(115                cls.admin_volume_types_client.delete_volume_type, vol_type)116        for vol_type in cls.volume_types:117            # Resource dictionary uses for is_resource_deleted method,118            # to distinguish between volume-type to encryption-type.119            resource = {'id': vol_type, 'type': 'volume-type'}120            test_utils.call_and_ignore_notfound_exc(121                cls.admin_volume_types_client.wait_for_resource_deletion,...

fwaas_client.py

Source:fwaas_client.py

1# Copyright (c) 2015 Midokura SARL2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15import time16from tempest import config17from tempest import exceptions18from tempest.lib.common.utils import data_utils19from tempest.lib.common.utils import test_utils20from tempest.lib import exceptions as lib_exc21from neutron.plugins.common import constants as p_const22from neutron_fwaas.tests.tempest_plugin.services import client23CONF = config.CONF24class FWaaSClientMixin(object):25    @classmethod26    def resource_setup(cls):27        super(FWaaSClientMixin, cls).resource_setup()28        manager = cls.manager29        cls.firewalls_client = client.FirewallsClient(30            manager.auth_provider,31            CONF.network.catalog_type,32            CONF.network.region or CONF.identity.region,33            endpoint_type=CONF.network.endpoint_type,34            build_interval=CONF.network.build_interval,35            build_timeout=CONF.network.build_timeout,36            **manager.default_params)37        cls.firewall_policies_client = client.FirewallPoliciesClient(38            manager.auth_provider,39            CONF.network.catalog_type,40            CONF.network.region or CONF.identity.region,41            endpoint_type=CONF.network.endpoint_type,42            build_interval=CONF.network.build_interval,43            build_timeout=CONF.network.build_timeout,44            **manager.default_params)45        cls.firewall_rules_client = client.FirewallRulesClient(46            manager.auth_provider,47            CONF.network.catalog_type,48            CONF.network.region or CONF.identity.region,49            endpoint_type=CONF.network.endpoint_type,50            build_interval=CONF.network.build_interval,51            build_timeout=CONF.network.build_timeout,52            **manager.default_params)53    def create_firewall_rule(self, **kwargs):54        body = self.firewall_rules_client.create_firewall_rule(55            name=data_utils.rand_name("fw-rule"),56            **kwargs)57        fw_rule = body['firewall_rule']58        self.addCleanup(test_utils.call_and_ignore_notfound_exc,59                        self.firewall_rules_client.delete_firewall_rule,60                        fw_rule['id'])61        return fw_rule62    def create_firewall_policy(self, **kwargs):63        body = self.firewall_policies_client.create_firewall_policy(64            name=data_utils.rand_name("fw-policy"),65            **kwargs)66        fw_policy = body['firewall_policy']67        self.addCleanup(test_utils.call_and_ignore_notfound_exc,68                        self.firewall_policies_client.delete_firewall_policy,69                        fw_policy['id'])70        return fw_policy71    def create_firewall(self, **kwargs):72        body = self.firewalls_client.create_firewall(73            name=data_utils.rand_name("fw"),74            **kwargs)75        fw = body['firewall']76        self.addCleanup(test_utils.call_and_ignore_notfound_exc,77                        self.delete_firewall_and_wait,78                        fw['id'])79        return fw80    def delete_firewall_and_wait(self, firewall_id):81        self.firewalls_client.delete_firewall(firewall_id)82        self._wait_firewall_while(firewall_id, [p_const.PENDING_DELETE],83                                  not_found_ok=True)84    def _wait_firewall_ready(self, firewall_id):85        self._wait_firewall_while(firewall_id,86                                  [p_const.PENDING_CREATE,87                                   p_const.PENDING_UPDATE])88    def _wait_firewall_while(self, firewall_id, statuses, not_found_ok=False):89        start = int(time.time())90        if not_found_ok:91            expected_exceptions = (lib_exc.NotFound)92        else:93            expected_exceptions = ()94        while True:95            try:96                fw = self.firewalls_client.show_firewall(firewall_id)97            except expected_exceptions:98                break99            status = fw['firewall']['status']100            if status not in statuses:101                break102            if int(time.time()) - start >= self.firewalls_client.build_timeout:103                msg = ("Firewall %(firewall)s failed to reach "104                       "non PENDING status (current %(status)s)") % {105                    "firewall": firewall_id,106                    "status": status,107                }108                raise exceptions.TimeoutException(msg)...

taas_client.py

Source:taas_client.py

1# Copyright (c) 2015 Midokura SARL2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15from tempest import config16from tempest.lib.common.utils import data_utils17from tempest.lib.common.utils import test_utils18from neutron_taas.tests.tempest_plugin.services import client19CONF = config.CONF20class TaaSClientMixin(object):21    @classmethod22    def resource_setup(cls):23        super(TaaSClientMixin, cls).resource_setup()24        os_primary = cls.os_primary25        cls.tap_services_client = client.TapServicesClient(26            os_primary.auth_provider,27            CONF.network.catalog_type,28            CONF.network.region or CONF.identity.region,29            endpoint_type=CONF.network.endpoint_type,30            build_interval=CONF.network.build_interval,31            build_timeout=CONF.network.build_timeout,32            **os_primary.default_params)33        cls.tap_flows_client = client.TapFlowsClient(34            os_primary.auth_provider,35            CONF.network.catalog_type,36            CONF.network.region or CONF.identity.region,37            endpoint_type=CONF.network.endpoint_type,38            build_interval=CONF.network.build_interval,39            build_timeout=CONF.network.build_timeout,40            **os_primary.default_params)41    def create_tap_service(self, **kwargs):42        body = self.tap_services_client.create_tap_service(43            name=data_utils.rand_name("tap_service"),44            **kwargs)45        tap_service = body['tap_service']46        self.addCleanup(test_utils.call_and_ignore_notfound_exc,47                        self.tap_services_client.delete_tap_service,48                        tap_service['id'])49        return tap_service50    def create_tap_flow(self, **kwargs):51        body = self.tap_flows_client.create_tap_flow(52            name=data_utils.rand_name("tap_service"),53            **kwargs)54        tap_flow = body['tap_flow']55        self.addCleanup(test_utils.call_and_ignore_notfound_exc,56                        self.tap_flows_client.delete_tap_flow,57                        tap_flow['id'])58        return tap_flow59    def update_tap_service(self, tap_service_id, **kwargs):60        body = self.tap_services_client.update_tap_service(61            tap_service_id,62            **kwargs)63        tap_service = body['tap_service']64        self.addCleanup(test_utils.call_and_ignore_notfound_exc,65                        self.tap_services_client.delete_tap_service,66                        tap_service['id'])67    def update_tap_flow(self, tap_flow_id, **kwargs):68        body = self.tap_flows_client.update_tap_flow(69            tap_flow_id,70            **kwargs)71        tap_flow = body['tap_flow']72        self.addCleanup(test_utils.call_and_ignore_notfound_exc,73                        self.tap_flows_client.delete_tap_flow,...

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.