KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use show_qos_policy method in tempest

Best Python code snippet using tempest_python
test_qos.py
Source:test_qos.py
...27        policy = self.create_qos_policy(name='test-policy',28                                        description='test policy desc1',29                                        shared=False)30        # Test 'show policy'31        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])32        retrieved_policy = retrieved_policy['policy']33        self.assertEqual('test-policy', retrieved_policy['name'])34        self.assertEqual('test policy desc1', retrieved_policy['description'])35        self.assertFalse(retrieved_policy['shared'])36        # Test 'list policies'37        policies = self.admin_client.list_qos_policies()['policies']38        policies_ids = [p['id'] for p in policies]39        self.assertIn(policy['id'], policies_ids)40    @test.idempotent_id('606a48e2-5403-4052-b40f-4d54b855af76')41    @test.requires_ext(extension="project-id", service="network")42    def test_show_policy_has_project_id(self):43        policy = self.create_qos_policy(name='test-policy', shared=False)44        body = self.admin_client.show_qos_policy(policy['id'])45        show_policy = body['policy']46        self.assertIn('project_id', show_policy)47        self.assertIn('tenant_id', show_policy)48        self.assertEqual(self.admin_client.tenant_id,49                         show_policy['project_id'])50        self.assertEqual(self.admin_client.tenant_id,51                         show_policy['tenant_id'])52    @test.idempotent_id('f8d20e92-f06d-4805-b54f-230f77715815')53    def test_list_policy_filter_by_name(self):54        self.create_qos_policy(name='test', description='test policy',55                               shared=False)56        self.create_qos_policy(name='test2', description='test policy',57                               shared=False)58        policies = (self.admin_client.59                    list_qos_policies(name='test')['policies'])60        self.assertEqual(1, len(policies))61        retrieved_policy = policies[0]62        self.assertEqual('test', retrieved_policy['name'])63    @test.idempotent_id('8e88a54b-f0b2-4b7d-b061-a15d93c2c7d6')64    def test_policy_update(self):65        policy = self.create_qos_policy(name='test-policy',66                                        description='',67                                        shared=False)68        self.admin_client.update_qos_policy(policy['id'],69                                            description='test policy desc2',70                                            shared=True)71        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])72        retrieved_policy = retrieved_policy['policy']73        self.assertEqual('test policy desc2', retrieved_policy['description'])74        self.assertTrue(retrieved_policy['shared'])75        self.assertEqual([], retrieved_policy['rules'])76    @test.idempotent_id('6e880e0f-bbfc-4e54-87c6-680f90e1b618')77    def test_policy_update_forbidden_for_regular_tenants_own_policy(self):78        policy = self.create_qos_policy(name='test-policy',79                                        description='',80                                        shared=False,81                                        tenant_id=self.client.tenant_id)82        self.assertRaises(83            exceptions.Forbidden,84            self.client.update_qos_policy,85            policy['id'], description='test policy')86    @test.idempotent_id('4ecfd7e7-47b6-4702-be38-be9235901a87')87    def test_policy_update_forbidden_for_regular_tenants_foreign_policy(self):88        policy = self.create_qos_policy(name='test-policy',89                                        description='',90                                        shared=False,91                                        tenant_id=self.admin_client.tenant_id)92        self.assertRaises(93            exceptions.NotFound,94            self.client.update_qos_policy,95            policy['id'], description='test policy')96    @test.idempotent_id('ee263db4-009a-4641-83e5-d0e83506ba4c')97    def test_shared_policy_update(self):98        policy = self.create_qos_policy(name='test-policy',99                                        description='',100                                        shared=True)101        self.admin_client.update_qos_policy(policy['id'],102                                            description='test policy desc2')103        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])104        retrieved_policy = retrieved_policy['policy']105        self.assertTrue(retrieved_policy['shared'])106        self.admin_client.update_qos_policy(policy['id'],107                                            shared=False)108        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])109        retrieved_policy = retrieved_policy['policy']110        self.assertFalse(retrieved_policy['shared'])111    @test.idempotent_id('1cb42653-54bd-4a9a-b888-c55e18199201')112    def test_delete_policy(self):113        policy = self.admin_client.create_qos_policy(114            'test-policy', 'desc', True)['policy']115        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])116        retrieved_policy = retrieved_policy['policy']117        self.assertEqual('test-policy', retrieved_policy['name'])118        self.admin_client.delete_qos_policy(policy['id'])119        self.assertRaises(exceptions.NotFound,120                          self.admin_client.show_qos_policy, policy['id'])121    @test.idempotent_id('cf776f77-8d3d-49f2-8572-12d6a1557224')122    def test_list_admin_rule_types(self):123        self._test_list_rule_types(self.admin_client)124    @test.idempotent_id('49c8ea35-83a9-453a-bd23-239cf3b13929')125    def test_list_regular_rule_types(self):126        self._test_list_rule_types(self.client)127    def _test_list_rule_types(self, client):128        # List supported rule types129        # TODO(QoS): since in gate we run both ovs and linuxbridge ml2 drivers,130        # and since Linux Bridge ml2 driver does not have QoS support yet, ml2131        # plugin reports no rule types are supported. Once linuxbridge will132        # receive support for QoS, the list of expected rule types will change.133        #134        # In theory, we could make the test conditional on which ml2 drivers135        # are enabled in gate (or more specifically, on which supported qos136        # rules are claimed by core plugin), but that option doesn't seem to be137        # available through tempest.lib framework138        expected_rule_types = []139        expected_rule_details = ['type']140        rule_types = client.list_qos_rule_types()141        actual_list_rule_types = rule_types['rule_types']142        actual_rule_types = [rule['type'] for rule in actual_list_rule_types]143        # Verify that only required fields present in rule details144        for rule in actual_list_rule_types:145            self.assertEqual(tuple(rule.keys()), tuple(expected_rule_details))146        # Verify if expected rules are present in the actual rules list147        for rule in expected_rule_types:148            self.assertIn(rule, actual_rule_types)149    def _disassociate_network(self, client, network_id):150        updated_network = client.update_network(network_id,151                                                qos_policy_id=None)152        self.assertIsNone(updated_network['network']['qos_policy_id'])153    @test.idempotent_id('65b9ef75-1911-406a-bbdb-ca1d68d528b0')154    def test_policy_association_with_admin_network(self):155        policy = self.create_qos_policy(name='test-policy',156                                        description='test policy',157                                        shared=False)158        network = self.create_shared_network('test network',159                                             qos_policy_id=policy['id'])160        retrieved_network = self.admin_client.show_network(network['id'])161        self.assertEqual(162            policy['id'], retrieved_network['network']['qos_policy_id'])163        self._disassociate_network(self.admin_client, network['id'])164    @test.idempotent_id('1738de5d-0476-4163-9022-5e1b548c208e')165    def test_policy_association_with_tenant_network(self):166        policy = self.create_qos_policy(name='test-policy',167                                        description='test policy',168                                        shared=True)169        network = self.create_network('test network',170                                      qos_policy_id=policy['id'])171        retrieved_network = self.admin_client.show_network(network['id'])172        self.assertEqual(173            policy['id'], retrieved_network['network']['qos_policy_id'])174        self._disassociate_network(self.client, network['id'])175    @test.idempotent_id('9efe63d0-836f-4cc2-b00c-468e63aa614e')176    def test_policy_association_with_network_nonexistent_policy(self):177        self.assertRaises(178            exceptions.NotFound,179            self.create_network,180            'test network',181            qos_policy_id='9efe63d0-836f-4cc2-b00c-468e63aa614e')182    @test.idempotent_id('1aa55a79-324f-47d9-a076-894a8fc2448b')183    def test_policy_association_with_network_non_shared_policy(self):184        policy = self.create_qos_policy(name='test-policy',185                                        description='test policy',186                                        shared=False)187        self.assertRaises(188            exceptions.NotFound,189            self.create_network,190            'test network', qos_policy_id=policy['id'])191    @test.idempotent_id('09a9392c-1359-4cbb-989f-fb768e5834a8')192    def test_policy_update_association_with_admin_network(self):193        policy = self.create_qos_policy(name='test-policy',194                                        description='test policy',195                                        shared=False)196        network = self.create_shared_network('test network')197        retrieved_network = self.admin_client.show_network(network['id'])198        self.assertIsNone(retrieved_network['network']['qos_policy_id'])199        self.admin_client.update_network(network['id'],200                                         qos_policy_id=policy['id'])201        retrieved_network = self.admin_client.show_network(network['id'])202        self.assertEqual(203            policy['id'], retrieved_network['network']['qos_policy_id'])204        self._disassociate_network(self.admin_client, network['id'])205    def _disassociate_port(self, port_id):206        self.client.update_port(port_id, qos_policy_id=None)207        updated_port = self.admin_client.show_port(port_id)208        self.assertIsNone(updated_port['port']['qos_policy_id'])209    @test.idempotent_id('98fcd95e-84cf-4746-860e-44692e674f2e')210    def test_policy_association_with_port_shared_policy(self):211        policy = self.create_qos_policy(name='test-policy',212                                        description='test policy',213                                        shared=True)214        network = self.create_shared_network('test network')215        port = self.create_port(network, qos_policy_id=policy['id'])216        retrieved_port = self.admin_client.show_port(port['id'])217        self.assertEqual(218            policy['id'], retrieved_port['port']['qos_policy_id'])219        self._disassociate_port(port['id'])220    @test.idempotent_id('49e02f5a-e1dd-41d5-9855-cfa37f2d195e')221    def test_policy_association_with_port_nonexistent_policy(self):222        network = self.create_shared_network('test network')223        self.assertRaises(224            exceptions.NotFound,225            self.create_port,226            network,227            qos_policy_id='49e02f5a-e1dd-41d5-9855-cfa37f2d195e')228    @test.idempotent_id('f53d961c-9fe5-4422-8b66-7add972c6031')229    def test_policy_association_with_port_non_shared_policy(self):230        policy = self.create_qos_policy(name='test-policy',231                                        description='test policy',232                                        shared=False)233        network = self.create_shared_network('test network')234        self.assertRaises(235            exceptions.NotFound,236            self.create_port,237            network, qos_policy_id=policy['id'])238    @test.idempotent_id('f8163237-fba9-4db5-9526-bad6d2343c76')239    def test_policy_update_association_with_port_shared_policy(self):240        policy = self.create_qos_policy(name='test-policy',241                                        description='test policy',242                                        shared=True)243        network = self.create_shared_network('test network')244        port = self.create_port(network)245        retrieved_port = self.admin_client.show_port(port['id'])246        self.assertIsNone(retrieved_port['port']['qos_policy_id'])247        self.client.update_port(port['id'], qos_policy_id=policy['id'])248        retrieved_port = self.admin_client.show_port(port['id'])249        self.assertEqual(250            policy['id'], retrieved_port['port']['qos_policy_id'])251        self._disassociate_port(port['id'])252    @test.idempotent_id('18163237-8ba9-4db5-9525-bad6d2343c75')253    def test_delete_not_allowed_if_policy_in_use_by_network(self):254        policy = self.create_qos_policy(name='test-policy',255                                        description='test policy',256                                        shared=True)257        network = self.create_shared_network(258            'test network', qos_policy_id=policy['id'])259        self.assertRaises(260            exceptions.Conflict,261            self.admin_client.delete_qos_policy, policy['id'])262        self._disassociate_network(self.admin_client, network['id'])263        self.admin_client.delete_qos_policy(policy['id'])264    @test.idempotent_id('24153230-84a9-4dd5-9525-bad6d2343c75')265    def test_delete_not_allowed_if_policy_in_use_by_port(self):266        policy = self.create_qos_policy(name='test-policy',267                                        description='test policy',268                                        shared=True)269        network = self.create_shared_network('test network')270        port = self.create_port(network, qos_policy_id=policy['id'])271        self.assertRaises(272            exceptions.Conflict,273            self.admin_client.delete_qos_policy, policy['id'])274        self._disassociate_port(port['id'])275        self.admin_client.delete_qos_policy(policy['id'])276    @test.idempotent_id('a2a5849b-dd06-4b18-9664-0b6828a1fc27')277    def test_qos_policy_delete_with_rules(self):278        policy = self.create_qos_policy(name='test-policy',279                                        description='test policy',280                                        shared=False)281        self.admin_client.create_bandwidth_limit_rule(282            policy['id'], 200, 1337)['bandwidth_limit_rule']283        self.admin_client.delete_qos_policy(policy['id'])284        with testtools.ExpectedException(exceptions.NotFound):285            self.admin_client.show_qos_policy(policy['id'])286    @test.idempotent_id('fb384bde-a973-41c3-a542-6f77a092155f')287    def test_get_policy_that_is_shared(self):288        policy = self.create_qos_policy(289            name='test-policy-shared',290            description='shared policy',291            shared=True,292            tenant_id=self.admin_client.tenant_id)293        obtained_policy = self.client.show_qos_policy(policy['id'])['policy']294        self.assertEqual(obtained_policy, policy)295    @test.idempotent_id('aed8e2a6-22da-421b-89b9-935a2c1a1b50')296    def test_policy_create_forbidden_for_regular_tenants(self):297        self.assertRaises(298            exceptions.Forbidden,299            self.client.create_qos_policy,300            'test-policy', 'test policy', False)301class QosBandwidthLimitRuleTestJSON(base.BaseAdminNetworkTest):302    @classmethod303    @test.requires_ext(extension="qos", service="network")304    @base.require_qos_rule_type(qos_consts.RULE_TYPE_BANDWIDTH_LIMIT)305    def resource_setup(cls):306        super(QosBandwidthLimitRuleTestJSON, cls).resource_setup()307    @test.idempotent_id('8a59b00b-3e9c-4787-92f8-93a5cdf5e378')308    def test_rule_create(self):309        policy = self.create_qos_policy(name='test-policy',310                                        description='test policy',311                                        shared=False)312        rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],313                                                    max_kbps=200,314                                                    max_burst_kbps=1337)315        # Test 'show rule'316        retrieved_rule = self.admin_client.show_bandwidth_limit_rule(317            policy['id'], rule['id'])318        retrieved_rule = retrieved_rule['bandwidth_limit_rule']319        self.assertEqual(rule['id'], retrieved_rule['id'])320        self.assertEqual(200, retrieved_rule['max_kbps'])321        self.assertEqual(1337, retrieved_rule['max_burst_kbps'])322        # Test 'list rules'323        rules = self.admin_client.list_bandwidth_limit_rules(policy['id'])324        rules = rules['bandwidth_limit_rules']325        rules_ids = [r['id'] for r in rules]326        self.assertIn(rule['id'], rules_ids)327        # Test 'show policy'328        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])329        policy_rules = retrieved_policy['policy']['rules']330        self.assertEqual(1, len(policy_rules))331        self.assertEqual(rule['id'], policy_rules[0]['id'])332        self.assertEqual(qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,333                         policy_rules[0]['type'])334    @test.idempotent_id('8a59b00b-ab01-4787-92f8-93a5cdf5e378')335    def test_rule_create_fail_for_the_same_type(self):336        policy = self.create_qos_policy(name='test-policy',337                                        description='test policy',338                                        shared=False)339        self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],340                                             max_kbps=200,341                                             max_burst_kbps=1337)342        self.assertRaises(exceptions.Conflict,343                          self.create_qos_bandwidth_limit_rule,344                          policy_id=policy['id'],345                          max_kbps=201, max_burst_kbps=1338)346    @test.idempotent_id('149a6988-2568-47d2-931e-2dbc858943b3')347    def test_rule_update(self):348        policy = self.create_qos_policy(name='test-policy',349                                        description='test policy',350                                        shared=False)351        rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],352                                                    max_kbps=1,353                                                    max_burst_kbps=1)354        self.admin_client.update_bandwidth_limit_rule(policy['id'],355                                                      rule['id'],356                                                      max_kbps=200,357                                                      max_burst_kbps=1337)358        retrieved_policy = self.admin_client.show_bandwidth_limit_rule(359            policy['id'], rule['id'])360        retrieved_policy = retrieved_policy['bandwidth_limit_rule']361        self.assertEqual(200, retrieved_policy['max_kbps'])362        self.assertEqual(1337, retrieved_policy['max_burst_kbps'])363    @test.idempotent_id('67ee6efd-7b33-4a68-927d-275b4f8ba958')364    def test_rule_delete(self):365        policy = self.create_qos_policy(name='test-policy',366                                        description='test policy',367                                        shared=False)368        rule = self.admin_client.create_bandwidth_limit_rule(369            policy['id'], 200, 1337)['bandwidth_limit_rule']370        retrieved_policy = self.admin_client.show_bandwidth_limit_rule(371            policy['id'], rule['id'])372        retrieved_policy = retrieved_policy['bandwidth_limit_rule']373        self.assertEqual(rule['id'], retrieved_policy['id'])374        self.admin_client.delete_bandwidth_limit_rule(policy['id'], rule['id'])375        self.assertRaises(exceptions.NotFound,376                          self.admin_client.show_bandwidth_limit_rule,377                          policy['id'], rule['id'])378    @test.idempotent_id('f211222c-5808-46cb-a961-983bbab6b852')379    def test_rule_create_rule_nonexistent_policy(self):380        self.assertRaises(381            exceptions.NotFound,382            self.create_qos_bandwidth_limit_rule,383            'policy', 200, 1337)384    @test.idempotent_id('a4a2e7ad-786f-4927-a85a-e545a93bd274')385    def test_rule_create_forbidden_for_regular_tenants(self):386        self.assertRaises(387            exceptions.Forbidden,388            self.client.create_bandwidth_limit_rule,389            'policy', 1, 2)390    @test.idempotent_id('1bfc55d9-6fd8-4293-ab3a-b1d69bf7cd2e')391    def test_rule_update_forbidden_for_regular_tenants_own_policy(self):392        policy = self.create_qos_policy(name='test-policy',393                                        description='test policy',394                                        shared=False,395                                        tenant_id=self.client.tenant_id)396        rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],397                                                    max_kbps=1,398                                                    max_burst_kbps=1)399        self.assertRaises(400            exceptions.NotFound,401            self.client.update_bandwidth_limit_rule,402            policy['id'], rule['id'], max_kbps=2, max_burst_kbps=4)403    @test.idempotent_id('9a607936-4b6f-4c2f-ad21-bd5b3d4fc91f')404    def test_rule_update_forbidden_for_regular_tenants_foreign_policy(self):405        policy = self.create_qos_policy(name='test-policy',406                                        description='test policy',407                                        shared=False,408                                        tenant_id=self.admin_client.tenant_id)409        rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],410                                                    max_kbps=1,411                                                    max_burst_kbps=1)412        self.assertRaises(413            exceptions.NotFound,414            self.client.update_bandwidth_limit_rule,415            policy['id'], rule['id'], max_kbps=2, max_burst_kbps=4)416    @test.idempotent_id('ce0bd0c2-54d9-4e29-85f1-cfb36ac3ebe2')417    def test_get_rules_by_policy(self):418        policy1 = self.create_qos_policy(name='test-policy1',419                                         description='test policy1',420                                         shared=False)421        rule1 = self.create_qos_bandwidth_limit_rule(policy_id=policy1['id'],422                                                     max_kbps=200,423                                                     max_burst_kbps=1337)424        policy2 = self.create_qos_policy(name='test-policy2',425                                         description='test policy2',426                                         shared=False)427        rule2 = self.create_qos_bandwidth_limit_rule(policy_id=policy2['id'],428                                                     max_kbps=5000,429                                                     max_burst_kbps=2523)430        # Test 'list rules'431        rules = self.admin_client.list_bandwidth_limit_rules(policy1['id'])432        rules = rules['bandwidth_limit_rules']433        rules_ids = [r['id'] for r in rules]434        self.assertIn(rule1['id'], rules_ids)435        self.assertNotIn(rule2['id'], rules_ids)436class RbacSharedQosPoliciesTest(base.BaseAdminNetworkTest):437    force_tenant_isolation = True438    credentials = ['primary', 'alt', 'admin']439    @classmethod440    @test.requires_ext(extension="qos", service="network")441    def resource_setup(cls):442        super(RbacSharedQosPoliciesTest, cls).resource_setup()443        cls.client2 = cls.alt_manager.network_client444    def _create_qos_policy(self, tenant_id=None):445        args = {'name': data_utils.rand_name('test-policy'),446                'description': 'test policy',447                'shared': False,448                'tenant_id': tenant_id}449        qos_policy = self.admin_client.create_qos_policy(**args)['policy']450        self.addCleanup(self.admin_client.delete_qos_policy, qos_policy['id'])451        return qos_policy452    def _make_admin_policy_shared_to_tenant_id(self, tenant_id):453        policy = self._create_qos_policy()454        rbac_policy = self.admin_client.create_rbac_policy(455            object_type='qos_policy',456            object_id=policy['id'],457            action='access_as_shared',458            target_tenant=tenant_id,459        )['rbac_policy']460        return {'policy': policy, 'rbac_policy': rbac_policy}461    def _create_network(self, qos_policy_id, client, should_cleanup=True):462        net = client.create_network(463            name=data_utils.rand_name('test-network'),464            qos_policy_id=qos_policy_id)['network']465        if should_cleanup:466            self.addCleanup(client.delete_network, net['id'])467        return net468    @test.idempotent_id('b9dcf582-d3b3-11e5-950a-54ee756c66df')469    def test_policy_sharing_with_wildcard(self):470        qos_pol = self.create_qos_policy(471            name=data_utils.rand_name('test-policy'),472            description='test-shared-policy', shared=False)473        self.assertNotIn(qos_pol, self.client2.list_qos_policies()['policies'])474        # test update shared False -> True475        self.admin_client.update_qos_policy(qos_pol['id'], shared=True)476        qos_pol['shared'] = True477        self.client2.show_qos_policy(qos_pol['id'])478        rbac_pol = {'target_tenant': '*',479                    'tenant_id': self.admin_client.tenant_id,480                    'project_id': self.admin_client.tenant_id,481                    'object_type': 'qos_policy',482                    'object_id': qos_pol['id'],483                    'action': 'access_as_shared'}484        rbac_policies = self.admin_client.list_rbac_policies()['rbac_policies']485        rbac_policies = [r for r in rbac_policies if r.pop('id')]486        self.assertIn(rbac_pol, rbac_policies)487        # update shared True -> False should fail because the policy is bound488        # to a network489        net = self._create_network(qos_pol['id'], self.admin_client, False)490        with testtools.ExpectedException(exceptions.Conflict):491            self.admin_client.update_qos_policy(qos_pol['id'], shared=False)492        # delete the network, and update shared True -> False should pass now493        self.admin_client.delete_network(net['id'])494        self.admin_client.update_qos_policy(qos_pol['id'], shared=False)495        qos_pol['shared'] = False496        self.assertNotIn(qos_pol, self.client2.list_qos_policies()['policies'])497    def _create_net_bound_qos_rbacs(self):498        res = self._make_admin_policy_shared_to_tenant_id(499            self.client.tenant_id)500        qos_policy, rbac_for_client_tenant = res['policy'], res['rbac_policy']501        # add a wildcard rbac rule - now the policy globally shared502        rbac_wildcard = self.admin_client.create_rbac_policy(503            object_type='qos_policy',504            object_id=qos_policy['id'],505            action='access_as_shared',506            target_tenant='*',507        )['rbac_policy']508        # tenant1 now uses qos policy for net509        self._create_network(qos_policy['id'], self.client)510        return rbac_for_client_tenant, rbac_wildcard511    @test.idempotent_id('328b1f70-d424-11e5-a57f-54ee756c66df')512    def test_net_bound_shared_policy_wildcard_and_tenant_id_wild_remove(self):513        client_rbac, wildcard_rbac = self._create_net_bound_qos_rbacs()514        # globally unshare the qos-policy, the specific share should remain515        self.admin_client.delete_rbac_policy(wildcard_rbac['id'])516        self.client.list_rbac_policies(id=client_rbac['id'])517    @test.idempotent_id('1997b00c-0c75-4e43-8ce2-999f9fa555ee')518    def test_net_bound_shared_policy_wildcard_and_tenant_id_wild_remains(self):519        client_rbac, wildcard_rbac = self._create_net_bound_qos_rbacs()520        # remove client_rbac policy the wildcard share should remain521        self.admin_client.delete_rbac_policy(client_rbac['id'])522        self.client.list_rbac_policies(id=wildcard_rbac['id'])523    @test.idempotent_id('2ace9adc-da6e-11e5-aafe-54ee756c66df')524    def test_policy_sharing_with_wildcard_and_tenant_id(self):525        res = self._make_admin_policy_shared_to_tenant_id(526            self.client.tenant_id)527        qos_policy, rbac = res['policy'], res['rbac_policy']528        qos_pol = self.client.show_qos_policy(qos_policy['id'])['policy']529        self.assertTrue(qos_pol['shared'])530        with testtools.ExpectedException(exceptions.NotFound):531            self.client2.show_qos_policy(qos_policy['id'])532        # make the qos-policy globally shared533        self.admin_client.update_qos_policy(qos_policy['id'], shared=True)534        qos_pol = self.client2.show_qos_policy(qos_policy['id'])['policy']535        self.assertTrue(qos_pol['shared'])536        # globally unshare the qos-policy, the specific share should remain537        self.admin_client.update_qos_policy(qos_policy['id'], shared=False)538        self.client.show_qos_policy(qos_policy['id'])539        with testtools.ExpectedException(exceptions.NotFound):540            self.client2.show_qos_policy(qos_policy['id'])541        self.assertIn(rbac,542                      self.admin_client.list_rbac_policies()['rbac_policies'])543    @test.idempotent_id('9f85c76a-a350-11e5-8ae5-54ee756c66df')544    def test_policy_target_update(self):545        res = self._make_admin_policy_shared_to_tenant_id(546            self.client.tenant_id)547        # change to client2548        update_res = self.admin_client.update_rbac_policy(549                res['rbac_policy']['id'], target_tenant=self.client2.tenant_id)550        self.assertEqual(self.client2.tenant_id,551                         update_res['rbac_policy']['target_tenant'])552        # make sure everything else stayed the same553        res['rbac_policy'].pop('target_tenant')554        update_res['rbac_policy'].pop('target_tenant')555        self.assertEqual(res['rbac_policy'], update_res['rbac_policy'])556    @test.idempotent_id('a9b39f46-a350-11e5-97c7-54ee756c66df')557    def test_network_presence_prevents_policy_rbac_policy_deletion(self):558        res = self._make_admin_policy_shared_to_tenant_id(559            self.client2.tenant_id)560        qos_policy_id = res['policy']['id']561        self._create_network(qos_policy_id, self.client2)562        # a network with shared qos-policy should prevent the deletion of an563        # rbac-policy required for it to be shared564        with testtools.ExpectedException(exceptions.Conflict):565            self.admin_client.delete_rbac_policy(res['rbac_policy']['id'])566        # a wildcard policy should allow the specific policy to be deleted567        # since it allows the remaining port568        wild = self.admin_client.create_rbac_policy(569            object_type='qos_policy', object_id=res['policy']['id'],570            action='access_as_shared', target_tenant='*')['rbac_policy']571        self.admin_client.delete_rbac_policy(res['rbac_policy']['id'])572        # now that wildcard is the only remaining, it should be subjected to573        # the same restriction574        with testtools.ExpectedException(exceptions.Conflict):575            self.admin_client.delete_rbac_policy(wild['id'])576        # we can't update the policy to a different tenant577        with testtools.ExpectedException(exceptions.Conflict):578            self.admin_client.update_rbac_policy(579                wild['id'], target_tenant=self.client2.tenant_id)580    @test.idempotent_id('b0fe87e8-a350-11e5-9f08-54ee756c66df')581    def test_regular_client_shares_to_another_regular_client(self):582        # owned by self.admin_client583        policy = self._create_qos_policy()584        with testtools.ExpectedException(exceptions.NotFound):585            self.client.show_qos_policy(policy['id'])586        rbac_policy = self.admin_client.create_rbac_policy(587            object_type='qos_policy', object_id=policy['id'],588            action='access_as_shared',589            target_tenant=self.client.tenant_id)['rbac_policy']590        self.client.show_qos_policy(policy['id'])591        self.assertIn(rbac_policy,592                      self.admin_client.list_rbac_policies()['rbac_policies'])593        # ensure that 'client2' can't see the rbac-policy sharing the594        # qos-policy to it because the rbac-policy belongs to 'client'595        self.assertNotIn(rbac_policy['id'], [p['id'] for p in596                          self.client2.list_rbac_policies()['rbac_policies']])597    @test.idempotent_id('ba88d0ca-a350-11e5-a06f-54ee756c66df')598    def test_filter_fields(self):599        policy = self._create_qos_policy()600        self.admin_client.create_rbac_policy(601            object_type='qos_policy', object_id=policy['id'],602            action='access_as_shared', target_tenant=self.client2.tenant_id)603        field_args = (('id',), ('id', 'action'), ('object_type', 'object_id'),604                      ('tenant_id', 'target_tenant'))605        for fields in field_args:606            res = self.admin_client.list_rbac_policies(fields=fields)607            self.assertEqual(set(fields), set(res['rbac_policies'][0].keys()))608    @test.idempotent_id('c10d993a-a350-11e5-9c7a-54ee756c66df')609    def test_rbac_policy_show(self):610        res = self._make_admin_policy_shared_to_tenant_id(611            self.client.tenant_id)612        p1 = res['rbac_policy']613        p2 = self.admin_client.create_rbac_policy(614            object_type='qos_policy', object_id=res['policy']['id'],615            action='access_as_shared',616            target_tenant='*')['rbac_policy']617        self.assertEqual(618            p1, self.admin_client.show_rbac_policy(p1['id'])['rbac_policy'])619        self.assertEqual(620            p2, self.admin_client.show_rbac_policy(p2['id'])['rbac_policy'])621    @test.idempotent_id('c7496f86-a350-11e5-b380-54ee756c66df')622    def test_filter_rbac_policies(self):623        policy = self._create_qos_policy()624        rbac_pol1 = self.admin_client.create_rbac_policy(625            object_type='qos_policy', object_id=policy['id'],626            action='access_as_shared',627            target_tenant=self.client2.tenant_id)['rbac_policy']628        rbac_pol2 = self.admin_client.create_rbac_policy(629            object_type='qos_policy', object_id=policy['id'],630            action='access_as_shared',631            target_tenant=self.admin_client.tenant_id)['rbac_policy']632        res1 = self.admin_client.list_rbac_policies(id=rbac_pol1['id'])[633            'rbac_policies']634        res2 = self.admin_client.list_rbac_policies(id=rbac_pol2['id'])[635            'rbac_policies']636        self.assertEqual(1, len(res1))637        self.assertEqual(1, len(res2))638        self.assertEqual(rbac_pol1['id'], res1[0]['id'])639        self.assertEqual(rbac_pol2['id'], res2[0]['id'])640    @test.idempotent_id('cd7d755a-a350-11e5-a344-54ee756c66df')641    def test_regular_client_blocked_from_sharing_anothers_policy(self):642        qos_policy = self._make_admin_policy_shared_to_tenant_id(643            self.client.tenant_id)['policy']644        with testtools.ExpectedException(exceptions.BadRequest):645            self.client.create_rbac_policy(646                object_type='qos_policy', object_id=qos_policy['id'],647                action='access_as_shared',648                target_tenant=self.client2.tenant_id)649        # make sure the rbac-policy is invisible to the tenant for which it's650        # being shared651        self.assertFalse(self.client.list_rbac_policies()['rbac_policies'])652class QosDscpMarkingRuleTestJSON(base.BaseAdminNetworkTest):653    VALID_DSCP_MARK1 = 56654    VALID_DSCP_MARK2 = 48655    @classmethod656    @test.requires_ext(extension="qos", service="network")657    @base.require_qos_rule_type(qos_consts.RULE_TYPE_DSCP_MARKING)658    def resource_setup(cls):659        super(QosDscpMarkingRuleTestJSON, cls).resource_setup()660    @test.idempotent_id('f5cbaceb-5829-497c-9c60-ad70969e9a08')661    def test_rule_create(self):662        policy = self.create_qos_policy(name='test-policy',663                                        description='test policy',664                                        shared=False)665        rule = self.admin_client.create_dscp_marking_rule(666            policy['id'], self.VALID_DSCP_MARK1)['dscp_marking_rule']667        # Test 'show rule'668        retrieved_rule = self.admin_client.show_dscp_marking_rule(669            policy['id'], rule['id'])670        retrieved_rule = retrieved_rule['dscp_marking_rule']671        self.assertEqual(rule['id'], retrieved_rule['id'])672        self.assertEqual(self.VALID_DSCP_MARK1, retrieved_rule['dscp_mark'])673        # Test 'list rules'674        rules = self.admin_client.list_dscp_marking_rules(policy['id'])675        rules = rules['dscp_marking_rules']676        rules_ids = [r['id'] for r in rules]677        self.assertIn(rule['id'], rules_ids)678        # Test 'show policy'679        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])680        policy_rules = retrieved_policy['policy']['rules']681        self.assertEqual(1, len(policy_rules))682        self.assertEqual(rule['id'], policy_rules[0]['id'])683        self.assertEqual(qos_consts.RULE_TYPE_DSCP_MARKING,684                         policy_rules[0]['type'])685    @test.idempotent_id('08553ffe-030f-4037-b486-7e0b8fb9385a')686    def test_rule_create_fail_for_the_same_type(self):687        policy = self.create_qos_policy(name='test-policy',688                                        description='test policy',689                                        shared=False)690        self.admin_client.create_dscp_marking_rule(691            policy['id'], self.VALID_DSCP_MARK1)['dscp_marking_rule']692        self.assertRaises(exceptions.Conflict,693                          self.admin_client.create_dscp_marking_rule,694                          policy_id=policy['id'],695                          dscp_mark=self.VALID_DSCP_MARK2)696    @test.idempotent_id('76f632e5-3175-4408-9a32-3625e599c8a2')697    def test_rule_update(self):698        policy = self.create_qos_policy(name='test-policy',699                                        description='test policy',700                                        shared=False)701        rule = self.admin_client.create_dscp_marking_rule(702            policy['id'], self.VALID_DSCP_MARK1)['dscp_marking_rule']703        self.admin_client.update_dscp_marking_rule(704            policy['id'], rule['id'], dscp_mark=self.VALID_DSCP_MARK2)705        retrieved_policy = self.admin_client.show_dscp_marking_rule(706            policy['id'], rule['id'])707        retrieved_policy = retrieved_policy['dscp_marking_rule']708        self.assertEqual(self.VALID_DSCP_MARK2, retrieved_policy['dscp_mark'])709    @test.idempotent_id('74f81904-c35f-48a3-adae-1f5424cb3c18')710    def test_rule_delete(self):711        policy = self.create_qos_policy(name='test-policy',712                                        description='test policy',713                                        shared=False)714        rule = self.admin_client.create_dscp_marking_rule(715            policy['id'], self.VALID_DSCP_MARK1)['dscp_marking_rule']716        retrieved_policy = self.admin_client.show_dscp_marking_rule(717            policy['id'], rule['id'])718        retrieved_policy = retrieved_policy['dscp_marking_rule']719        self.assertEqual(rule['id'], retrieved_policy['id'])720        self.admin_client.delete_dscp_marking_rule(policy['id'], rule['id'])721        self.assertRaises(exceptions.NotFound,722                          self.admin_client.show_dscp_marking_rule,723                          policy['id'], rule['id'])724    @test.idempotent_id('9cb8ef5c-96fc-4978-9ee0-e3b02bab628a')725    def test_rule_create_rule_nonexistent_policy(self):726        self.assertRaises(727            exceptions.NotFound,728            self.admin_client.create_dscp_marking_rule,729            'policy', self.VALID_DSCP_MARK1)730    @test.idempotent_id('bf6002ea-29de-486f-b65d-08aea6d4c4e2')731    def test_rule_create_forbidden_for_regular_tenants(self):732        self.assertRaises(733            exceptions.Forbidden,734            self.client.create_dscp_marking_rule,735            'policy', self.VALID_DSCP_MARK1)736    @test.idempotent_id('33646b08-4f05-4493-a48a-bde768a18533')737    def test_invalid_rule_create(self):738        policy = self.create_qos_policy(name='test-policy',739                                        description='test policy',740                                        shared=False)741        self.assertRaises(742            exceptions.BadRequest,743            self.admin_client.create_dscp_marking_rule,744            policy['id'], 58)745    @test.idempotent_id('c565131d-4c80-4231-b0f3-9ae2be4de129')746    def test_get_rules_by_policy(self):747        policy1 = self.create_qos_policy(name='test-policy1',748                                         description='test policy1',749                                         shared=False)750        rule1 = self.admin_client.create_dscp_marking_rule(751            policy1['id'], self.VALID_DSCP_MARK1)['dscp_marking_rule']752        policy2 = self.create_qos_policy(name='test-policy2',753                                         description='test policy2',754                                         shared=False)755        rule2 = self.admin_client.create_dscp_marking_rule(756            policy2['id'], self.VALID_DSCP_MARK2)['dscp_marking_rule']757        # Test 'list rules'758        rules = self.admin_client.list_dscp_marking_rules(policy1['id'])759        rules = rules['dscp_marking_rules']760        rules_ids = [r['id'] for r in rules]761        self.assertIn(rule1['id'], rules_ids)762        self.assertNotIn(rule2['id'], rules_ids)763class QosMinimumBandwidthRuleTestJSON(base.BaseAdminNetworkTest):764    DIRECTION_EGRESS = "egress"765    DIRECTION_INGRESS = "ingress"766    RULE_NAME = qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH + "_rule"767    RULES_NAME = RULE_NAME + "s"768    @classmethod769    @test.requires_ext(extension="qos", service="network")770    @base.require_qos_rule_type(qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH)771    def resource_setup(cls):772        super(QosMinimumBandwidthRuleTestJSON, cls).resource_setup()773    @test.idempotent_id('aa59b00b-3e9c-4787-92f8-93a5cdf5e378')774    def test_rule_create(self):775        policy = self.create_qos_policy(name='test-policy',776                                        description='test policy',777                                        shared=False)778        rule = self.admin_client.create_minimum_bandwidth_rule(779            policy_id=policy['id'],780            direction=self.DIRECTION_EGRESS,781            min_kbps=1138)[self.RULE_NAME]782        # Test 'show rule'783        retrieved_rule = self.admin_client.show_minimum_bandwidth_rule(784            policy['id'], rule['id'])785        retrieved_rule = retrieved_rule[self.RULE_NAME]786        self.assertEqual(rule['id'], retrieved_rule['id'])787        self.assertEqual(1138, retrieved_rule['min_kbps'])788        self.assertEqual(self.DIRECTION_EGRESS, retrieved_rule['direction'])789        # Test 'list rules'790        rules = self.admin_client.list_minimum_bandwidth_rules(policy['id'])791        rules = rules[self.RULES_NAME]792        rules_ids = [r['id'] for r in rules]793        self.assertIn(rule['id'], rules_ids)794        # Test 'show policy'795        retrieved_policy = self.admin_client.show_qos_policy(policy['id'])796        policy_rules = retrieved_policy['policy']['rules']797        self.assertEqual(1, len(policy_rules))798        self.assertEqual(rule['id'], policy_rules[0]['id'])799        self.assertEqual(qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH,800                         policy_rules[0]['type'])801    @test.idempotent_id('266d9b87-e51c-48bd-9aa7-8269573621be')802    def test_rule_create_fail_for_missing_min_kbps(self):803        policy = self.create_qos_policy(name='test-policy',804                                        description='test policy',805                                        shared=False)806        self.assertRaises(exceptions.BadRequest,807                          self.admin_client.create_minimum_bandwidth_rule,808                          policy_id=policy['id'],809                          direction=self.DIRECTION_EGRESS)...
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.