Best Python code snippet using tempest_python
test_qos.py
Source:test_qos.py  
1#2#    Licensed under the Apache License, Version 2.0 (the "License"); you may3#    not use this file except in compliance with the License. You may obtain4#    a copy of the License at5#6#         http://www.apache.org/licenses/LICENSE-2.07#8#    Unless required by applicable law or agreed to in writing, software9#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT10#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the11#    License for the specific language governing permissions and limitations12#    under the License.13import mock14from heat.common import template_format15from heat.engine.clients.os import neutron16from heat.engine import rsrc_defn17from heat.engine import stack18from heat.engine import template19from heat.tests import common20from heat.tests import utils21qos_policy_template = '''22heat_template_version: 2016-04-0823description: This template to define a neutron qos policy.24resources:25  my_qos_policy:26    type: OS::Neutron::QoSPolicy27    properties:28      description: a policy for test29      shared: true30      tenant_id: d66c74c01d6c41b9846088c1ad9634d031'''32bandwidth_limit_rule_template = '''33heat_template_version: 2016-04-0834description: This template to define a neutron bandwidth limit rule.35resources:36  my_bandwidth_limit_rule:37    type: OS::Neutron::QoSBandwidthLimitRule38    properties:39      policy: 477e8273-60a7-4c41-b683-fdb0bc7cd15140      max_kbps: 100041      max_burst_kbps: 100042      tenant_id: d66c74c01d6c41b9846088c1ad9634d043'''44dscp_marking_rule_template = '''45heat_template_version: 2016-04-0846description: This template to define a neutron DSCP marking rule.47resources:48  my_dscp_marking_rule:49    type: OS::Neutron::QoSDscpMarkingRule50    properties:51      policy: 477e8273-60a7-4c41-b683-fdb0bc7cd15152      dscp_mark: 1653      tenant_id: d66c74c01d6c41b9846088c1ad9634d054'''55class NeutronQoSPolicyTest(common.HeatTestCase):56    def setUp(self):57        super(NeutronQoSPolicyTest, self).setUp()58        self.ctx = utils.dummy_context()59        tpl = template_format.parse(qos_policy_template)60        self.stack = stack.Stack(61            self.ctx,62            'neutron_qos_policy_test',63            template.Template(tpl)64        )65        self.neutronclient = mock.MagicMock()66        self.patchobject(neutron.NeutronClientPlugin, 'has_extension',67                         return_value=True)68        self.my_qos_policy = self.stack['my_qos_policy']69        self.my_qos_policy.client = mock.MagicMock(70            return_value=self.neutronclient)71        self.patchobject(self.my_qos_policy, 'physical_resource_name',72                         return_value='test_policy')73    def test_qos_policy_handle_create(self):74        policy = {75            'policy': {76                'description': 'a policy for test',77                'id': '9c1eb3fe-7bba-479d-bd43-1d497e53c384',78                'rules': [],79                'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0',80                'shared': True81            }82        }83        create_props = {'name': 'test_policy',84                        'description': 'a policy for test',85                        'shared': True,86                        'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'}87        self.neutronclient.create_qos_policy.return_value = policy88        self.my_qos_policy.handle_create()89        self.assertEqual('9c1eb3fe-7bba-479d-bd43-1d497e53c384',90                         self.my_qos_policy.resource_id)91        self.neutronclient.create_qos_policy.assert_called_once_with(92            {'policy': create_props}93        )94    def test_qos_policy_handle_delete(self):95        policy_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'96        self.my_qos_policy.resource_id = policy_id97        self.neutronclient.delete_qos_policy.return_value = None98        self.assertIsNone(self.my_qos_policy.handle_delete())99        self.neutronclient.delete_qos_policy.assert_called_once_with(100            self.my_qos_policy.resource_id)101    def test_qos_policy_handle_delete_not_found(self):102        policy_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'103        self.my_qos_policy.resource_id = policy_id104        not_found = self.neutronclient.NotFound105        self.neutronclient.delete_qos_policy.side_effect = not_found106        self.assertIsNone(self.my_qos_policy.handle_delete())107        self.neutronclient.delete_qos_policy.assert_called_once_with(108            self.my_qos_policy.resource_id)109    def test_qos_policy_handle_delete_resource_id_is_none(self):110        self.my_qos_policy.resource_id = None111        self.assertIsNone(self.my_qos_policy.handle_delete())112        self.assertEqual(0, self.neutronclient.delete_qos_policy.call_count)113    def test_qos_policy_handle_update(self):114        policy_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'115        self.my_qos_policy.resource_id = policy_id116        props = {117            'name': 'test_policy',118            'description': 'test',119            'shared': False120        }121        prop_dict = props.copy()122        update_snippet = rsrc_defn.ResourceDefinition(123            self.my_qos_policy.name,124            self.my_qos_policy.type(),125            props)126        # with name127        self.my_qos_policy.handle_update(json_snippet=update_snippet,128                                         tmpl_diff={},129                                         prop_diff=props)130        # without name131        props['name'] = None132        self.my_qos_policy.handle_update(json_snippet=update_snippet,133                                         tmpl_diff={},134                                         prop_diff=props)135        self.assertEqual(2, self.neutronclient.update_qos_policy.call_count)136        self.neutronclient.update_qos_policy.assert_called_with(137            policy_id, {'policy': prop_dict})138    def test_qos_policy_get_attr(self):139        self.my_qos_policy.resource_id = 'test policy'140        policy = {141            'policy': {142                'name': 'test_policy',143                'description': 'a policy for test',144                'id': '9c1eb3fe-7bba-479d-bd43-1d497e53c384',145                'rules': [],146                'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0',147                'shared': True148            }149        }150        self.neutronclient.show_qos_policy.return_value = policy151        self.assertEqual([], self.my_qos_policy.FnGetAtt('rules'))152        self.assertEqual(policy['policy'],153                         self.my_qos_policy.FnGetAtt('show'))154        self.neutronclient.show_qos_policy.assert_has_calls(155            [mock.call(self.my_qos_policy.resource_id)] * 2)156class NeutronQoSBandwidthLimitRuleTest(common.HeatTestCase):157    def setUp(self):158        super(NeutronQoSBandwidthLimitRuleTest, self).setUp()159        self.ctx = utils.dummy_context()160        tpl = template_format.parse(bandwidth_limit_rule_template)161        self.stack = stack.Stack(162            self.ctx,163            'neutron_bandwidth_limit_rule_test',164            template.Template(tpl)165        )166        self.neutronclient = mock.MagicMock()167        self.patchobject(neutron.NeutronClientPlugin, 'has_extension',168                         return_value=True)169        self.bandwidth_limit_rule = self.stack['my_bandwidth_limit_rule']170        self.bandwidth_limit_rule.client = mock.MagicMock(171            return_value=self.neutronclient)172        self.find_mock = self.patchobject(173            neutron.neutronV20,174            'find_resourceid_by_name_or_id')175        self.policy_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'176        self.find_mock.return_value = self.policy_id177    def test_rule_handle_create(self):178        rule = {179            'bandwidth_limit_rule': {180                'id': 'cf0eab12-ef8b-4a62-98d0-70576583c17a',181                'max_kbps': 1000,182                'max_burst_kbps': 1000,183                'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'184            }185        }186        create_props = {'max_kbps': 1000,187                        'max_burst_kbps': 1000,188                        'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'}189        self.neutronclient.create_bandwidth_limit_rule.return_value = rule190        self.bandwidth_limit_rule.handle_create()191        self.assertEqual('cf0eab12-ef8b-4a62-98d0-70576583c17a',192                         self.bandwidth_limit_rule.resource_id)193        self.neutronclient.create_bandwidth_limit_rule.assert_called_once_with(194            self.policy_id,195            {'bandwidth_limit_rule': create_props})196    def test_rule_handle_delete(self):197        rule_id = 'cf0eab12-ef8b-4a62-98d0-70576583c17a'198        self.bandwidth_limit_rule.resource_id = rule_id199        self.neutronclient.delete_bandwidth_limit_rule.return_value = None200        self.assertIsNone(self.bandwidth_limit_rule.handle_delete())201        self.neutronclient.delete_bandwidth_limit_rule.assert_called_once_with(202            rule_id, self.policy_id)203    def test_rule_handle_delete_not_found(self):204        rule_id = 'cf0eab12-ef8b-4a62-98d0-70576583c17a'205        self.bandwidth_limit_rule.resource_id = rule_id206        not_found = self.neutronclient.NotFound207        self.neutronclient.delete_bandwidth_limit_rule.side_effect = not_found208        self.assertIsNone(self.bandwidth_limit_rule.handle_delete())209        self.neutronclient.delete_bandwidth_limit_rule.assert_called_once_with(210            rule_id, self.policy_id)211    def test_rule_handle_delete_resource_id_is_none(self):212        self.bandwidth_limit_rule.resource_id = None213        self.assertIsNone(self.bandwidth_limit_rule.handle_delete())214        self.assertEqual(0,215                         self.neutronclient.bandwidth_limit_rule.call_count)216    def test_rule_handle_update(self):217        rule_id = 'cf0eab12-ef8b-4a62-98d0-70576583c17a'218        self.bandwidth_limit_rule.resource_id = rule_id219        prop_diff = {220            'max_kbps': 500,221            'max_burst_kbps': 400222        }223        self.bandwidth_limit_rule.handle_update(224            json_snippet={},225            tmpl_diff={},226            prop_diff=prop_diff)227        self.neutronclient.update_bandwidth_limit_rule.assert_called_once_with(228            rule_id,229            self.policy_id,230            {'bandwidth_limit_rule': prop_diff})231    def test_rule_get_attr(self):232        self.bandwidth_limit_rule.resource_id = 'test rule'233        rule = {234            'bandwidth_limit_rule': {235                'id': 'cf0eab12-ef8b-4a62-98d0-70576583c17a',236                'max_kbps': 1000,237                'max_burst_kbps': 1000,238                'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'239            }240        }241        self.neutronclient.show_bandwidth_limit_rule.return_value = rule242        self.assertEqual(rule['bandwidth_limit_rule'],243                         self.bandwidth_limit_rule.FnGetAtt('show'))244        self.neutronclient.show_bandwidth_limit_rule.assert_called_once_with(245            self.bandwidth_limit_rule.resource_id, self.policy_id)246class NeutronQoSDscpMarkingRuleTest(common.HeatTestCase):247    def setUp(self):248        super(NeutronQoSDscpMarkingRuleTest, self).setUp()249        self.ctx = utils.dummy_context()250        tpl = template_format.parse(dscp_marking_rule_template)251        self.stack = stack.Stack(252            self.ctx,253            'neutron_dscp_marking_rule_test',254            template.Template(tpl)255        )256        self.neutronclient = mock.MagicMock()257        self.patchobject(neutron.NeutronClientPlugin, 'has_extension',258                         return_value=True)259        self.dscp_marking_rule = self.stack['my_dscp_marking_rule']260        self.dscp_marking_rule.client = mock.MagicMock(261            return_value=self.neutronclient)262        self.find_mock = self.patchobject(263            neutron.neutronV20,264            'find_resourceid_by_name_or_id')265        self.policy_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'266        self.find_mock.return_value = self.policy_id267    def test_rule_handle_create(self):268        rule = {269            'dscp_marking_rule': {270                'id': 'cf0eab12-ef8b-4a62-98d0-70576583c17a',271                'dscp_mark': 16,272                'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'273            }274        }275        create_props = {'dscp_mark': 16,276                        'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'}277        self.neutronclient.create_dscp_marking_rule.return_value = rule278        self.dscp_marking_rule.handle_create()279        self.assertEqual('cf0eab12-ef8b-4a62-98d0-70576583c17a',280                         self.dscp_marking_rule.resource_id)281        self.neutronclient.create_dscp_marking_rule.assert_called_once_with(282            self.policy_id,283            {'dscp_marking_rule': create_props})284    def test_rule_handle_delete(self):285        rule_id = 'cf0eab12-ef8b-4a62-98d0-70576583c17a'286        self.dscp_marking_rule.resource_id = rule_id287        self.neutronclient.delete_dscp_marking_rule.return_value = None288        self.assertIsNone(self.dscp_marking_rule.handle_delete())289        self.neutronclient.delete_dscp_marking_rule.assert_called_once_with(290            rule_id, self.policy_id)291    def test_rule_handle_delete_not_found(self):292        rule_id = 'cf0eab12-ef8b-4a62-98d0-70576583c17a'293        self.dscp_marking_rule.resource_id = rule_id294        not_found = self.neutronclient.NotFound295        self.neutronclient.delete_dscp_marking_rule.side_effect = not_found296        self.assertIsNone(self.dscp_marking_rule.handle_delete())297        self.neutronclient.delete_dscp_marking_rule.assert_called_once_with(298            rule_id, self.policy_id)299    def test_rule_handle_delete_resource_id_is_none(self):300        self.dscp_marking_rule.resource_id = None301        self.assertIsNone(self.dscp_marking_rule.handle_delete())302        self.assertEqual(0,303                         self.neutronclient.dscp_marking_rule.call_count)304    def test_rule_handle_update(self):305        rule_id = 'cf0eab12-ef8b-4a62-98d0-70576583c17a'306        self.dscp_marking_rule.resource_id = rule_id307        prop_diff = {308            'dscp_mark': 8309        }310        self.dscp_marking_rule.handle_update(311            json_snippet={},312            tmpl_diff={},313            prop_diff=prop_diff)314        self.neutronclient.update_dscp_marking_rule.assert_called_once_with(315            rule_id,316            self.policy_id,317            {'dscp_marking_rule': prop_diff})318    def test_rule_get_attr(self):319        self.dscp_marking_rule.resource_id = 'test rule'320        rule = {321            'dscp_marking_rule': {322                'id': 'cf0eab12-ef8b-4a62-98d0-70576583c17a',323                'dscp_mark': 8,324                'tenant_id': 'd66c74c01d6c41b9846088c1ad9634d0'325            }326        }327        self.neutronclient.show_dscp_marking_rule.return_value = rule328        self.assertEqual(rule['dscp_marking_rule'],329                         self.dscp_marking_rule.FnGetAtt('show'))330        self.neutronclient.show_dscp_marking_rule.assert_called_once_with(...qospolicy.py
Source:qospolicy.py  
...58        return self.post(config.NEUTRON_QOS_POLICIES, payload)59    def update_qos_policy(self, qos_policyid, payload):60        logging.info('update qos_policy: ' + qos_policyid)61        return self.put(config.NEUTRON_QOS_POLICIES + '/' + qos_policyid, payload)62    def delete_qos_policy(self, qos_policyid):63        logging.info('delete qos_policy: ' + qos_policyid)64        return self.delete(config.NEUTRON_QOS_POLICIES + '/' + qos_policyid)65    @staticmethod66    def perform_tests(servername, username, count=0):67        logging.info('perform qos_policy tests, server: %s, user: %s' % (servername, username))68        tester = QosPolicy(servername, username)69        utils.assert_status(tester.get_qos_policys(), 200)70        qos_policy_one = tester.create_qos_policy(change_id(QOS_POLICY_ONE, count))71        utils.assert_status(qos_policy_one, 201)72        if qos_policy_one.status_code == 201:73            qos_policy_one_id = json.loads(qos_policy_one.text)['policy']['id']74            utils.assert_status(tester.get_qos_policy(qos_policy_one_id), 200)75        utils.assert_status(tester.update_qos_policy(change_id(QOS_POLICY_UPDATE['policy'], count)['policy']['id'],76                                                     change_id(QOS_POLICY_UPDATE['policy'], count)), 201)77        utils.assert_status(tester.delete_qos_policy(change_id(QOS_POLICY_ONE, count)['policy']['id']), 204)78        utils.assert_status(tester.get_qos_policy(change_id(QOS_POLICY_ONE, count)['policy']['id']), 404)79if __name__ == '__main__':...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!!
