Best Python code snippet using tempest_python
test_qos.py
Source:test_qos.py  
...435        self.assertEqual(200, retrieved_rule['max_kbps'])436        self.assertEqual(1337, retrieved_rule['max_burst_kbps'])437        self.assertEqual('ingress', retrieved_rule['direction'])438        # Test 'list rules'439        rules = self.qos_bw_limit_rule_client.list_limit_bandwidth_rules(440            policy['id'])441        rules = rules['bandwidth_limit_rules']442        rules_ids = [r['id'] for r in rules]443        self.assertIn(rule['id'], rules_ids)444        # Test 'show policy'445        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])446        policy_rules = retrieved_policy['policy']['rules']447        self.assertEqual(1, len(policy_rules))448        self.assertEqual(rule['id'], policy_rules[0]['id'])449        self.assertEqual(qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,450                         policy_rules[0]['type'])451    @decorators.idempotent_id('8a59b00b-ab01-4787-92f8-93a5cdf5e378')452    def test_rule_create_fail_for_the_same_type(self):453        policy = self.create_qos_policy(name=self.policy_name,454                                        description='test policy',455                                        shared=False)456        self._create_qos_bw_limit_rule(457            policy['id'], {'max_kbps': 200, 'max_burst_kbps': 1337})458        self.assertRaises(459            exceptions.Conflict,460            self._create_qos_bw_limit_rule,461            policy['id'],462            {'max_kbps': 201, 'max_burst_kbps': 1338})463    @decorators.idempotent_id('149a6988-2568-47d2-931e-2dbc858943b3')464    def test_rule_update(self):465        policy = self.create_qos_policy(name=self.policy_name,466                                        description='test policy',467                                        shared=False)468        rule = self._create_qos_bw_limit_rule(469            policy['id'], {'max_kbps': 1, 'max_burst_kbps': 1})470        if self.opposite_direction:471            self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(472                policy['id'], rule['id'],473                **{'max_kbps': 200, 'max_burst_kbps': 1337,474                   'direction': self.opposite_direction})475        else:476            self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(477                policy['id'], rule['id'],478                **{'max_kbps': 200, 'max_burst_kbps': 1337})479        retrieved_policy = self.qos_bw_limit_rule_client.\480            show_limit_bandwidth_rule(policy['id'], rule['id'])481        retrieved_policy = retrieved_policy['bandwidth_limit_rule']482        self.assertEqual(200, retrieved_policy['max_kbps'])483        self.assertEqual(1337, retrieved_policy['max_burst_kbps'])484        if self.opposite_direction:485            self.assertEqual(self.opposite_direction,486                             retrieved_policy['direction'])487    @decorators.idempotent_id('67ee6efd-7b33-4a68-927d-275b4f8ba958')488    def test_rule_delete(self):489        policy = self.create_qos_policy(name=self.policy_name,490                                        description='test policy',491                                        shared=False)492        rule = self._create_qos_bw_limit_rule(493            policy['id'], {'max_kbps': 200, 'max_burst_kbps': 1337})494        retrieved_policy = \495            self.qos_bw_limit_rule_client.show_limit_bandwidth_rule(496                policy['id'], rule['id'])497        retrieved_policy = retrieved_policy['bandwidth_limit_rule']498        self.assertEqual(rule['id'], retrieved_policy['id'])499        self.qos_bw_limit_rule_client.delete_limit_bandwidth_rule(500            policy['id'], rule['id'])501        self.assertRaises(502            exceptions.NotFound,503            self.qos_bw_limit_rule_client.show_limit_bandwidth_rule,504            policy['id'], rule['id'])505    @decorators.idempotent_id('f211222c-5808-46cb-a961-983bbab6b852')506    def test_rule_create_rule_nonexistent_policy(self):507        self.assertRaises(508            exceptions.NotFound,509            self._create_qos_bw_limit_rule,510            'policy', {'max_kbps': 200, 'max_burst_kbps': 1337})511    @decorators.idempotent_id('a4a2e7ad-786f-4927-a85a-e545a93bd274')512    def test_rule_create_forbidden_for_regular_tenants(self):513        self.assertRaises(514            exceptions.Forbidden,515            self.qos_bw_limit_rule_client_primary.create_limit_bandwidth_rule,516            'policy', **{'max_kbps': 1, 'max_burst_kbps': 2})517    @decorators.idempotent_id('1bfc55d9-6fd8-4293-ab3a-b1d69bf7cd2e')518    def test_rule_update_forbidden_for_regular_tenants_own_policy(self):519        policy = self.create_qos_policy(name=self.policy_name,520                                        description='test policy',521                                        shared=False,522                                        project_id=self.client.tenant_id)523        rule = self._create_qos_bw_limit_rule(524            policy['id'],525            {'max_kbps': 1, 'max_burst_kbps': 1})526        self.assertRaises(527            exceptions.Forbidden,528            self.qos_bw_limit_rule_client_primary.update_limit_bandwidth_rule,529            policy['id'], rule['id'], **{'max_kbps': 2, 'max_burst_kbps': 4})530    @decorators.idempotent_id('9a607936-4b6f-4c2f-ad21-bd5b3d4fc91f')531    def test_rule_update_forbidden_for_regular_tenants_foreign_policy(self):532        policy = self.create_qos_policy(533            name=self.policy_name,534            description='test policy',535            shared=False,536            project_id=self.admin_client.tenant_id)537        rule = self._create_qos_bw_limit_rule(538            policy['id'], {'max_kbps': 1, 'max_burst_kbps': 1})539        self.assertRaises(540            exceptions.NotFound,541            self.qos_bw_limit_rule_client_primary.update_limit_bandwidth_rule,542            policy['id'], rule['id'], **{'max_kbps': 2, 'max_burst_kbps': 4})543    @decorators.idempotent_id('ce0bd0c2-54d9-4e29-85f1-cfb36ac3ebe2')544    def test_get_rules_by_policy(self):545        policy1 = self.create_qos_policy(546            name='test-policy1',547            description='test policy1',548            shared=False)549        rule1 = self._create_qos_bw_limit_rule(550            policy1['id'], {'max_kbps': 200, 'max_burst_kbps': 1337})551        policy2 = self.create_qos_policy(552            name='test-policy2',553            description='test policy2',554            shared=False)555        rule2 = self._create_qos_bw_limit_rule(556            policy2['id'], {'max_kbps': 5000, 'max_burst_kbps': 2523})557        # Test 'list rules'558        rules = self.qos_bw_limit_rule_client.list_limit_bandwidth_rules(559            policy1['id'])560        rules = rules['bandwidth_limit_rules']561        rules_ids = [r['id'] for r in rules]562        self.assertIn(rule1['id'], rules_ids)563        self.assertNotIn(rule2['id'], rules_ids)564    @testtools.skipUnless(565        CONF.neutron_plugin_options.create_shared_resources,566        """Creation of shared resources should be allowed,567        setting the create_shared_resources option as 'True' is needed""")568    @decorators.idempotent_id('d911707e-fa2c-11e9-9553-5076af30bbf5')569    def test_attach_and_detach_a_policy_by_a_tenant(self):570        # As an admin create an non shared QoS policy,add a rule571        # and associate it with a network572        self.network = self.create_network()573        policy = self.create_qos_policy(name=self.policy_name,574                                        description='test policy for attach',575                                        shared=False)576        self._create_qos_bw_limit_rule(577            policy['id'], {'max_kbps': 1024, 'max_burst_kbps': 1024})578        self.admin_client.update_network(579            self.network['id'], qos_policy_id=policy['id'])580        # As a tenant, try to detach the policy from the network581        # The operation should be forbidden582        self.assertRaises(583            exceptions.Forbidden,584            self.client.update_network,585            self.network['id'], qos_policy_id=None)586        # As an admin, make the policy shared587        self.admin_client.update_qos_policy(policy['id'], shared=True)588        # As a tenant, try to detach the policy from the network589        # The operation should be allowed590        self.client.update_network(self.network['id'],591                                   qos_policy_id=None)592        retrieved_network = self.admin_client.show_network(self.network['id'])593        self.assertIsNone(retrieved_network['network']['qos_policy_id'])594        # As a tenant, try to delete the policy from the network595        # should be forbidden596        self.assertRaises(597            exceptions.Forbidden,598            self.client.delete_qos_policy,599            policy['id'])600class QosBandwidthLimitRuleWithDirectionTestJSON(601        QosBandwidthLimitRuleTestJSON):602    required_extensions = (603        QosBandwidthLimitRuleTestJSON.required_extensions +604        ['qos-bw-limit-direction']605    )606    scenarios = [607        ('ingress', {'direction': 'ingress'}),608        ('egress', {'direction': 'egress'}),609    ]610    @classmethod611    @base.require_qos_rule_type(qos_consts.RULE_TYPE_BANDWIDTH_LIMIT)612    def resource_setup(cls):613        super(QosBandwidthLimitRuleWithDirectionTestJSON, cls).resource_setup()614    def setUp(self):615        super(QosBandwidthLimitRuleWithDirectionTestJSON, self).setUp()616        self.policy_name = data_utils.rand_name(name='test', prefix='policy')617    @decorators.idempotent_id('c8cbe502-0f7e-11ea-8d71-362b9e155667')618    def test_create_policy_with_multiple_rules(self):619        # Create a policy with multiple rules620        policy = self.create_qos_policy(name=self.policy_name,621                                        description='test policy1',622                                        shared=False)623        rule1 = self._create_qos_bw_limit_rule(624            policy['id'], {'max_kbps': 1024, 'max_burst_kbps': 1024,625                           'direction': n_constants.EGRESS_DIRECTION})626        rule2 = self._create_qos_bw_limit_rule(627            policy['id'], {'max_kbps': 1024, 'max_burst_kbps': 1024,628                           'direction': n_constants.INGRESS_DIRECTION})629        # Check that the rules were added to the policy630        rules = self.qos_bw_limit_rule_client.list_limit_bandwidth_rules(631            policy['id'])['bandwidth_limit_rules']632        rules_ids = [rule['id'] for rule in rules]633        self.assertIn(rule1['id'], rules_ids)634        self.assertIn(rule2['id'], rules_ids)635        # Check that the rules creation fails for the same rule types636        self.assertRaises(637            exceptions.Conflict,638            self._create_qos_bw_limit_rule,639            policy['id'],640            {'max_kbps': 1025, 'max_burst_kbps': 1025,641             'direction': n_constants.EGRESS_DIRECTION})642        self.assertRaises(643            exceptions.Conflict,644            self._create_qos_bw_limit_rule,...test_qos_limit_bandwidth_rules_client.py
Source:test_qos_limit_bandwidth_rules_client.py  
1# Copyright 2021 Red Hat.2#3# Licensed under the Apache License, Version 2.0 (the "License"); you may4# not use this file except in compliance with the License. You may obtain5# a copy of the License at6#7#      http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12# License for the specific language governing permissions and limitations13# under the License.14import copy15from tempest.lib import decorators16from tempest.lib.services.network import qos_limit_bandwidth_rules_client17from tempest.tests.lib import fake_auth_provider18from tempest.tests.lib.services import base19from oslo_log import log as logging20LOG = logging.getLogger('tempest')21class TestQosLimitBandwidthRulesClient(base.BaseServiceTest):22    FAKE_QOS_POLICY_ID = "f1011b08-1297-11e9-a1e7-c7e6825a2616"23    FAKE_MAX_BW_RULE_ID = "e758c89e-1297-11e9-a6cf-cf46a71e6699"24    FAKE_MAX_BW_RULE_REQUEST = {25        'qos_policy_id': FAKE_QOS_POLICY_ID,26        'max_kbps': 1000,27        'max_burst_kbps': 0,28        'direction': 'ingress'29    }30    FAKE_MAX_BW_RULE_RESPONSE = {31        'bandwidth_limit_rule': {32            'id': FAKE_MAX_BW_RULE_ID,33            'max_kbps': 10000,34            'max_burst_kbps': 0,35            'direction': 'ingress'36        }37    }38    FAKE_MAX_BW_RULES = {39        'bandwidth_limit_rules': [40            FAKE_MAX_BW_RULE_RESPONSE['bandwidth_limit_rule']41        ]42    }43    def setUp(self):44        super(TestQosLimitBandwidthRulesClient, self).setUp()45        fake_auth = fake_auth_provider.FakeAuthProvider()46        self.qos_limit_bw_client = qos_limit_bandwidth_rules_client.\47            QosLimitBandwidthRulesClient(fake_auth, "network", "regionOne")48    @decorators.idempotent_id('cde981fa-e93b-11eb-aacb-74e5f9e2a801')49    def test_create_limit_bandwidth_rules(self, bytes_body=False):50        self.check_service_client_function(51            self.qos_limit_bw_client.create_limit_bandwidth_rule,52            "tempest.lib.common.rest_client.RestClient.post",53            self.FAKE_MAX_BW_RULE_RESPONSE,54            bytes_body,55            201,56            **self.FAKE_MAX_BW_RULE_REQUEST57        )58    @decorators.idempotent_id('86e6803a-e974-11eb-aacb-74e5f9e2a801')59    def test_update_limit_bandwidth_rules(self, bytes_body=False):60        update_kwargs = {61            "max_kbps": "20000"62        }63        resp_body = {64            "bandwidth_limit_rule": copy.deepcopy(65                self.FAKE_MAX_BW_RULE_RESPONSE['bandwidth_limit_rule']66            )67        }68        resp_body["bandwidth_limit_rule"].update(update_kwargs)69        self.check_service_client_function(70            self.qos_limit_bw_client.update_limit_bandwidth_rule,71            "tempest.lib.common.rest_client.RestClient.put",72            resp_body,73            bytes_body,74            200,75            qos_policy_id=self.FAKE_QOS_POLICY_ID,76            rule_id=self.FAKE_MAX_BW_RULE_ID,77            **update_kwargs)78    @decorators.idempotent_id('be60ae6e-e979-11eb-aacb-74e5f9e2a801')79    def test_show_limit_bandwidth_rules(self, bytes_body=False):80        self.check_service_client_function(81            self.qos_limit_bw_client.show_limit_bandwidth_rule,82            "tempest.lib.common.rest_client.RestClient.get",83            self.FAKE_MAX_BW_RULE_RESPONSE,84            bytes_body,85            200,86            qos_policy_id=self.FAKE_QOS_POLICY_ID,87            rule_id=self.FAKE_MAX_BW_RULE_ID88        )89    @decorators.idempotent_id('0a7c0964-e97b-11eb-aacb-74e5f9e2a801')90    def test_delete_limit_bandwidth_rule(self):91        self.check_service_client_function(92            self.qos_limit_bw_client.delete_limit_bandwidth_rule,93            "tempest.lib.common.rest_client.RestClient.delete",94            {},95            status=204,96            qos_policy_id=self.FAKE_QOS_POLICY_ID,97            rule_id=self.FAKE_MAX_BW_RULE_ID)98    @decorators.idempotent_id('08df88ae-e97d-11eb-aacb-74e5f9e2a801')99    def test_list_minimum_bandwidth_rules(self, bytes_body=False):100        self.check_service_client_function(101            self.qos_limit_bw_client.list_limit_bandwidth_rules,102            "tempest.lib.common.rest_client.RestClient.get",103            self.FAKE_MAX_BW_RULES,104            bytes_body,105            200,106            qos_policy_id=self.FAKE_QOS_POLICY_ID...qos_limit_bandwidth_rules_client.py
Source:qos_limit_bandwidth_rules_client.py  
...50        """51        uri = '/qos/policies/{}/bandwidth_limit_rules/{}'.format(52            qos_policy_id, rule_id)53        return self.delete_resource(uri)54    def list_limit_bandwidth_rules(self, qos_policy_id, **filters):55        """Lists all limit bandwidth rules for a QoS policy.56        For full list of available parameters, please refer to the official57        API reference:58        https://docs.openstack.org/api-ref/network/v2/index.html#list-bandwidth-limit-rules-for-qos-policy59        """60        uri = '/qos/policies/{}/bandwidth_limit_rules'.format(qos_policy_id)...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!!
