KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use show_association_qos method in tempest

Best Python code snippet using tempest_python

qos_client.py

Source:qos_client.py

...41                body = self.show_qos(qos_id)42                if not any(key in body['specs'] for key in args):43                    return44            elif operation == 'disassociate':45                body = self.show_association_qos(qos_id)46                if not any(args in body[i]['id'] for i in range(0, len(body))):47                    return48            elif operation == 'disassociate-all':49                body = self.show_association_qos(qos_id)50                if not body:51                    return52            else:53                msg = (" operation value is either not defined or incorrect.")54                raise exceptions.UnprocessableEntity(msg)55            if int(time.time()) - start_time >= self.build_timeout:56                raise exceptions.TimeoutException57            time.sleep(self.build_interval)58    def create_qos(self, name, consumer, **kwargs):59        """Create a QoS Specification.60        name : name of the QoS specifications61        consumer : conumer of Qos ( front-end / back-end / both )62        """63        post_body = {'name': name, 'consumer': consumer}64        post_body.update(kwargs)65        post_body = json.dumps({'qos_specs': post_body})66        resp, body = self.post('qos-specs', post_body)67        self.expected_success(200, resp.status)68        body = json.loads(body)69        return service_client.ResponseBody(resp, body['qos_specs'])70    def delete_qos(self, qos_id, force=False):71        """Delete the specified QoS specification."""72        resp, body = self.delete(73            "qos-specs/%s?force=%s" % (str(qos_id), force))74        self.expected_success(202, resp.status)75        return service_client.ResponseBody(resp, body)76    def list_qos(self):77        """List all the QoS specifications created."""78        url = 'qos-specs'79        resp, body = self.get(url)80        body = json.loads(body)81        self.expected_success(200, resp.status)82        return service_client.ResponseBodyList(resp, body['qos_specs'])83    def show_qos(self, qos_id):84        """Get the specified QoS specification."""85        url = "qos-specs/%s" % str(qos_id)86        resp, body = self.get(url)87        body = json.loads(body)88        self.expected_success(200, resp.status)89        return service_client.ResponseBody(resp, body['qos_specs'])90    def set_qos_key(self, qos_id, **kwargs):91        """Set the specified keys/values of QoS specification.92        kwargs : it is the dictionary of the key=value pairs to set93        """94        put_body = json.dumps({"qos_specs": kwargs})95        resp, body = self.put('qos-specs/%s' % qos_id, put_body)96        body = json.loads(body)97        self.expected_success(200, resp.status)98        return service_client.ResponseBody(resp, body['qos_specs'])99    def unset_qos_key(self, qos_id, keys):100        """Unset the specified keys of QoS specification.101        keys : it is the array of the keys to unset102        """103        put_body = json.dumps({'keys': keys})104        resp, body = self.put('qos-specs/%s/delete_keys' % qos_id, put_body)105        self.expected_success(202, resp.status)106        return service_client.ResponseBody(resp, body)107    def associate_qos(self, qos_id, vol_type_id):108        """Associate the specified QoS with specified volume-type."""109        url = "qos-specs/%s/associate" % str(qos_id)110        url += "?vol_type_id=%s" % vol_type_id111        resp, body = self.get(url)112        self.expected_success(202, resp.status)113        return service_client.ResponseBody(resp, body)114    def show_association_qos(self, qos_id):115        """Get the association of the specified QoS specification."""116        url = "qos-specs/%s/associations" % str(qos_id)117        resp, body = self.get(url)118        body = json.loads(body)119        self.expected_success(200, resp.status)120        return service_client.ResponseBodyList(resp, body['qos_associations'])121    def disassociate_qos(self, qos_id, vol_type_id):122        """Disassociate the specified QoS with specified volume-type."""123        url = "qos-specs/%s/disassociate" % str(qos_id)124        url += "?vol_type_id=%s" % vol_type_id125        resp, body = self.get(url)126        self.expected_success(202, resp.status)127        return service_client.ResponseBody(resp, body)128    def disassociate_all_qos(self, qos_id):...

base_qos_client.py

Source:base_qos_client.py

...42                body = self.show_qos(qos_id)['qos_specs']43                if not any(key in body['specs'] for key in args):44                    return45            elif operation == 'disassociate':46                body = self.show_association_qos(qos_id)['qos_associations']47                if not any(args in body[i]['id'] for i in range(0, len(body))):48                    return49            elif operation == 'disassociate-all':50                body = self.show_association_qos(qos_id)['qos_associations']51                if not body:52                    return53            else:54                msg = (" operation value is either not defined or incorrect.")55                raise lib_exc.UnprocessableEntity(msg)56            if int(time.time()) - start_time >= self.build_timeout:57                raise exceptions.TimeoutException58            time.sleep(self.build_interval)59    def create_qos(self, **kwargs):60        """Create a QoS Specification.61        Available params: see http://developer.openstack.org/62                              api-ref-blockstorage-v2.html#createQoSSpec63        """64        post_body = json.dumps({'qos_specs': kwargs})65        resp, body = self.post('qos-specs', post_body)66        self.expected_success(200, resp.status)67        body = json.loads(body)68        return service_client.ResponseBody(resp, body)69    def delete_qos(self, qos_id, force=False):70        """Delete the specified QoS specification."""71        resp, body = self.delete(72            "qos-specs/%s?force=%s" % (str(qos_id), force))73        self.expected_success(202, resp.status)74        return service_client.ResponseBody(resp, body)75    def list_qos(self):76        """List all the QoS specifications created."""77        url = 'qos-specs'78        resp, body = self.get(url)79        body = json.loads(body)80        self.expected_success(200, resp.status)81        return service_client.ResponseBody(resp, body)82    def show_qos(self, qos_id):83        """Get the specified QoS specification."""84        url = "qos-specs/%s" % str(qos_id)85        resp, body = self.get(url)86        body = json.loads(body)87        self.expected_success(200, resp.status)88        return service_client.ResponseBody(resp, body)89    def set_qos_key(self, qos_id, **kwargs):90        """Set the specified keys/values of QoS specification.91        Available params: see http://developer.openstack.org/92                              api-ref-blockstorage-v2.html#setQoSKey93        """94        put_body = json.dumps({"qos_specs": kwargs})95        resp, body = self.put('qos-specs/%s' % qos_id, put_body)96        body = json.loads(body)97        self.expected_success(200, resp.status)98        return service_client.ResponseBody(resp, body)99    def unset_qos_key(self, qos_id, keys):100        """Unset the specified keys of QoS specification.101        :param keys: keys to delete from the QoS specification.102        TODO(jordanP): Add a link once LP #1524877 is fixed.103        """104        put_body = json.dumps({'keys': keys})105        resp, body = self.put('qos-specs/%s/delete_keys' % qos_id, put_body)106        self.expected_success(202, resp.status)107        return service_client.ResponseBody(resp, body)108    def associate_qos(self, qos_id, vol_type_id):109        """Associate the specified QoS with specified volume-type."""110        url = "qos-specs/%s/associate" % str(qos_id)111        url += "?vol_type_id=%s" % vol_type_id112        resp, body = self.get(url)113        self.expected_success(202, resp.status)114        return service_client.ResponseBody(resp, body)115    def show_association_qos(self, qos_id):116        """Get the association of the specified QoS specification."""117        url = "qos-specs/%s/associations" % str(qos_id)118        resp, body = self.get(url)119        body = json.loads(body)120        self.expected_success(200, resp.status)121        return service_client.ResponseBody(resp, body)122    def disassociate_qos(self, qos_id, vol_type_id):123        """Disassociate the specified QoS with specified volume-type."""124        url = "qos-specs/%s/disassociate" % str(qos_id)125        url += "?vol_type_id=%s" % vol_type_id126        resp, body = self.get(url)127        self.expected_success(202, resp.status)128        return service_client.ResponseBody(resp, body)129    def disassociate_all_qos(self, qos_id):...

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.