How to use is_absolute_limit method in tempest

Best Python code snippet using tempest_python

rest_client.py

Source:rest_client.py Github

copy

Full Screen

...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...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful