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

How to use is_absolute_limit method in tempest

Best Python code snippet using tempest_python

rest_client.py

Source:rest_client.py

...241        resp, resp_body = self._request(method, url,242                                        headers=headers, body=body)243        while (resp.status == 413 and244               'retry-after' in resp and245                not self.is_absolute_limit(246                    resp, self._parse_resp(resp_body)) and247                retry < MAX_RECURSION_DEPTH):248            retry += 1249            delay = int(resp['retry-after'])250            time.sleep(delay)251            resp, resp_body = self._request(method, url,252                                            headers=headers, body=body)253        self._error_checker(method, url, headers, body,254                            resp, resp_body)255        return resp, resp_body256    def _error_checker(self, method, url,257                       headers, body, resp, resp_body):258        # NOTE(mtreinish): Check for httplib response from glance_http. The259        # object can't be used here because importing httplib breaks httplib2.260        # If another object from a class not imported were passed here as261        # resp this could possibly fail262        if str(type(resp)) == "<type 'instance'>":263            ctype = resp.getheader('content-type')264        else:265            try:266                ctype = resp['content-type']267            # NOTE(mtreinish): Keystone delete user responses doesn't have a268            # content-type header. (They don't have a body) So just pretend it269            # is set.270            except KeyError:271                ctype = 'application/json'272        # It is not an error response273        if resp.status < 400:274            return275        JSON_ENC = ['application/json; charset=UTF-8', 'application/json',276                    'application/json; charset=utf-8']277        # NOTE(mtreinish): This is for compatibility with Glance and swift278        # APIs. These are the return content types that Glance api v1279        # (and occasionally swift) are using.280        TXT_ENC = ['text/plain; charset=UTF-8', 'text/html; charset=UTF-8',281                   'text/plain; charset=utf-8']282        XML_ENC = ['application/xml', 'application/xml; charset=UTF-8']283        if ctype in JSON_ENC or ctype in XML_ENC:284            parse_resp = True285        elif ctype in TXT_ENC:286            parse_resp = False287        else:288            raise exceptions.RestClientException(str(resp.status))289        if resp.status == 401 or resp.status == 403:290            raise exceptions.Unauthorized()291        if resp.status == 404:292            raise exceptions.NotFound(resp_body)293        if resp.status == 400:294            if parse_resp:295                resp_body = self._parse_resp(resp_body)296            raise exceptions.BadRequest(resp_body)297        if resp.status == 409:298            if parse_resp:299                resp_body = self._parse_resp(resp_body)300            raise exceptions.Duplicate(resp_body)301        if resp.status == 413:302            if parse_resp:303                resp_body = self._parse_resp(resp_body)304            if self.is_absolute_limit(resp, resp_body):305                raise exceptions.OverLimit(resp_body)306            else:307                raise exceptions.RateLimitExceeded(resp_body)308        if resp.status == 422:309            if parse_resp:310                resp_body = self._parse_resp(resp_body)311            raise exceptions.UnprocessableEntity(resp_body)312        if resp.status in (500, 501):313            message = resp_body314            if parse_resp:315                resp_body = self._parse_resp(resp_body)316                #I'm seeing both computeFault and cloudServersFault come back.317                #Will file a bug to fix, but leave as is for now.318                if 'cloudServersFault' in resp_body:319                    message = resp_body['cloudServersFault']['message']320                elif 'computeFault' in resp_body:321                    message = resp_body['computeFault']['message']322                elif 'error' in resp_body:  # Keystone errors323                    message = resp_body['error']['message']324                    raise exceptions.IdentityError(message)325                elif 'message' in resp_body:326                    message = resp_body['message']327            raise exceptions.ComputeFault(message)328        if resp.status >= 400:329            if parse_resp:330                resp_body = self._parse_resp(resp_body)331            raise exceptions.RestClientException(str(resp.status))332    def is_absolute_limit(self, resp, resp_body):333        if (not isinstance(resp_body, collections.Mapping) or334            'retry-after' not in resp):335            return True336        over_limit = resp_body.get('overLimit', None)337        if not over_limit:338            return True339        return 'exceed' in over_limit.get('message', 'blabla')340    def wait_for_resource_deletion(self, id):341        """Waits for a resource to be deleted."""342        start_time = int(time.time())343        while True:344            if self.is_resource_deleted(id):345                return346            if int(time.time()) - start_time >= self.build_timeout:347                raise exceptions.TimeoutException348            time.sleep(self.build_interval)349    def is_resource_deleted(self, id):350        """351        Subclasses override with specific deletion detection.352        """353        message = ('"%s" does not implement is_resource_deleted'354                   % self.__class__.__name__)355        raise NotImplementedError(message)356class RestClientXML(RestClient):357    TYPE = "xml"358    def _parse_resp(self, body):359        return xml_to_json(etree.fromstring(body))360    def is_absolute_limit(self, resp, resp_body):361        if (not isinstance(resp_body, collections.Mapping) or362            'retry-after' not in resp):363            return True...

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.