How to use update_qos_policy method in tempest

Best Python code snippet using tempest_python

na_elementsw_qos_policy.py

Source:na_elementsw_qos_policy.py

...156            self.sfe.create_qos_policy(name=name, qos=qos)157        except (solidfire.common.ApiServerError, solidfire.common.ApiConnectionError) as exc:158            self.module.fail_json(msg="Error creating qos policy: %s: %s" %159                                  (name, to_native(exc)), exception=traceback.format_exc())160    def update_qos_policy(self, qos_policy_id, modify, name=None):161        """162        Update the QOS Policy if the policy already exists163        """164        options = dict(165            qos_policy_id=qos_policy_id166        )167        if name is not None:168            options['name'] = name169        if 'qos' in modify:170            options['qos'] = modify['qos']171        try:172            self.sfe.modify_qos_policy(**options)173        except (solidfire.common.ApiServerError, solidfire.common.ApiConnectionError) as exc:174            self.module.fail_json(msg="Error updating qos policy: %s: %s" %175                                  (self.parameters['from_name'] if name is not None else self.parameters['name'], to_native(exc)),176                                  exception=traceback.format_exc())177    def delete_qos_policy(self, qos_policy_id):178        """179        Delete the QOS Policy180        """181        try:182            self.sfe.delete_qos_policy(qos_policy_id=qos_policy_id)183        except (solidfire.common.ApiServerError, solidfire.common.ApiConnectionError) as exc:184            self.module.fail_json(msg="Error deleting qos policy: %s: %s" %185                                  (self.parameters['name'], to_native(exc)), exception=traceback.format_exc())186    def apply(self):187        """188        Process the create/delete/rename/modify actions for qos policy on the Element Software Cluster189        """190        modify = dict()191        current = self.get_qos_policy(self.parameters['name'])192        qos_policy_id = None if current is None else current['qos_policy_id']193        cd_action = self.na_helper.get_cd_action(current, self.parameters)194        modify = self.na_helper.get_modified_attributes(current, self.parameters)195        if cd_action == 'create' and self.parameters.get('from_name') is not None:196            from_qos_policy = self.get_qos_policy(self.parameters['from_name'])197            if from_qos_policy is None:198                self.module.fail_json(msg="Error renaming qos policy, no existing policy with name/id: %s" % self.parameters['from_name'])199            cd_action = 'rename'200            qos_policy_id = from_qos_policy['qos_policy_id']201            self.na_helper.changed = True202            modify = self.na_helper.get_modified_attributes(from_qos_policy, self.parameters)203        if cd_action == 'create' and 'qos' not in self.parameters:204            self.module.fail_json(msg="Error creating qos policy: %s, 'qos:' option is required" % self.parameters['name'])205        self.debug['modify'] = modify206        if not self.module.check_mode:207            if cd_action == 'create':208                self.create_qos_policy(self.parameters['name'], self.parameters['qos'])209            elif cd_action == 'delete':210                self.delete_qos_policy(qos_policy_id)211            elif cd_action == 'rename':212                self.update_qos_policy(qos_policy_id, modify, name=self.parameters['name'])213            elif modify:214                self.update_qos_policy(qos_policy_id, modify)215        results = dict(changed=self.na_helper.changed)216        if self.parameters['debug']:217            results['debug'] = self.debug218        self.module.exit_json(**results)219def main():220    """221    Main function222    """223    na_elementsw_qos_policy = ElementSWQosPolicy()224    na_elementsw_qos_policy.apply()225if __name__ == '__main__':...

qospolicy.py

Source:qospolicy.py

...55        return self.get(config.NEUTRON_QOS_POLICIES)56    def create_qos_policy(self, payload):57        logging.info('create qos_policy: ' + payload['policy']['id'])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__':...

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.