KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use retry_config method in localstack
Best Python code snippet using localstack_python
test_horde.py
Source:test_horde.py 
1# pylint: disable=redefined-outer-name2"""3Tests the RequestsStampede horde module.4"""5import logging6import math7import pytest8import requests9import requests.structures10import RequestsStampede.horde11import RequestsStampede.config12@pytest.fixture13def retry_request_default():14 """15 TODO16 """17 return RequestsStampede.horde.RetryRequest()18@pytest.fixture19def retry_request_custom():20 """21 TODO22 """23 return RequestsStampede.horde.RetryRequest(24 retry_config=RequestsStampede.config.AggressiveRetryConfig(25 retry_policy=RequestsStampede.policy.retry.InfiniteRetryPolicy(),26 backoff_policy=RequestsStampede.policy.backoff.RandomBackoffPolicy(),27 ),28 session=requests.Session(),29 )30@pytest.fixture31def retry_session_default():32 """33 TODO34 """35 return RequestsStampede.horde.RetrySession()36@pytest.fixture37def retry_session_custom():38 """39 TODO40 """41 custom_session = requests.Session()42 custom_session.headers.update(43 {"User-Agent": "RonSwanson/1.0", "Authorization": "Bearer ICanDoWhatIWant"}44 )45 return RequestsStampede.horde.RetrySession(46 retry_config=RequestsStampede.config.AggressiveRetryConfig(47 retry_policy=RequestsStampede.policy.retry.InfiniteRetryPolicy(),48 backoff_policy=RequestsStampede.policy.backoff.RandomBackoffPolicy(),49 ),50 session=custom_session,51 )52def test_retry_request_default_parameters(retry_request_default):53 """54 Ensures that RetryRequest default parameters are reasonable.55 """56 assert isinstance(57 retry_request_default, RequestsStampede.horde.AbstractRetryRequest58 )59 assert isinstance(retry_request_default, RequestsStampede.horde.RetryRequest)60 # Validate retry_config61 retry_config = retry_request_default.retry_config62 assert isinstance(retry_config, RequestsStampede.config.AbstractRetryConfig)63 assert isinstance(retry_config, RequestsStampede.config.DefaultRetryConfig)64 assert isinstance(retry_config.retry_enabled, bool)65 assert isinstance(retry_config.backoff_enabled, bool)66 # Ensure retries and backoff delays are enabled67 assert retry_config.retry_enabled68 assert retry_config.backoff_enabled69 # Validate retry_config.retry_policy70 retry_policy = retry_config.retry_policy71 assert isinstance(retry_policy, RequestsStampede.policy.retry.AbstractRetryPolicy)72 assert isinstance(retry_policy, RequestsStampede.policy.retry.FixedRetryPolicy)73 assert isinstance(retry_policy.attempts, int)74 assert retry_policy.attempts == 575 # Validate retry_config.backoff_policy76 backoff_policy = retry_config.backoff_policy77 assert isinstance(78 backoff_policy, RequestsStampede.policy.backoff.AbstractBackoffPolicy79 )80 assert isinstance(81 backoff_policy, RequestsStampede.policy.backoff.FibonacciBackoffPolicy82 )83 assert isinstance(backoff_policy.initial_delay, float)84 assert isinstance(backoff_policy.maximum_delay, float)85 assert backoff_policy.initial_delay == 0.086 assert backoff_policy.maximum_delay == 144.087 # Validate session88 assert retry_request_default.session is None89 # Validate logger90 assert isinstance(retry_request_default.logger, logging.Logger)91def test_retry_request_custom_parameters(retry_request_custom):92 """93 Ensures that a custom retry configuration passed to RetryRequest actually94 modifies the necessary instance parameters.95 """96 assert isinstance(retry_request_custom, RequestsStampede.horde.AbstractRetryRequest)97 assert isinstance(retry_request_custom, RequestsStampede.horde.RetryRequest)98 # Validate retry_config99 retry_config = retry_request_custom.retry_config100 assert isinstance(retry_config, RequestsStampede.config.AbstractRetryConfig)101 assert isinstance(retry_config, RequestsStampede.config.AggressiveRetryConfig)102 assert isinstance(retry_config.retry_enabled, bool)103 assert isinstance(retry_config.backoff_enabled, bool)104 # Ensure retries and backoff delays are enabled105 assert retry_config.retry_enabled106 assert retry_config.backoff_enabled107 # Validate retry_config.retry_policy108 retry_policy = retry_config.retry_policy109 assert isinstance(retry_policy, RequestsStampede.policy.retry.AbstractRetryPolicy)110 assert isinstance(retry_policy, RequestsStampede.policy.retry.InfiniteRetryPolicy)111 assert isinstance(retry_policy.attempts, float)112 assert retry_policy.attempts == math.inf113 # Validate retry_config.backoff_policy114 backoff_policy = retry_config.backoff_policy115 assert isinstance(116 backoff_policy, RequestsStampede.policy.backoff.AbstractBackoffPolicy117 )118 assert isinstance(119 backoff_policy, RequestsStampede.policy.backoff.RandomBackoffPolicy120 )121 assert isinstance(backoff_policy.minimum_delay, float)122 assert isinstance(backoff_policy.maximum_delay, float)123 assert backoff_policy.minimum_delay == 3.0124 assert backoff_policy.maximum_delay == 15.0125 # Validate session126 assert isinstance(retry_request_custom.session, requests.Session)127 # Validate logger128 assert isinstance(retry_request_custom.logger, logging.Logger)129def test_retry_session_default_parameters(retry_session_default):130 """131 Ensures that RetrySession default parameters are reasonable.132 """133 assert isinstance(134 retry_session_default, RequestsStampede.horde.AbstractRetryRequest135 )136 assert isinstance(retry_session_default, RequestsStampede.horde.RetryRequest)137 assert isinstance(retry_session_default, RequestsStampede.horde.RetrySession)138 # Validate retry_config139 retry_config = retry_session_default.retry_config140 assert isinstance(retry_config, RequestsStampede.config.AbstractRetryConfig)141 assert isinstance(retry_config, RequestsStampede.config.DefaultRetryConfig)142 assert isinstance(retry_config.retry_enabled, bool)143 assert isinstance(retry_config.backoff_enabled, bool)144 # Ensure retries and backoff delays are enabled145 assert retry_config.retry_enabled146 assert retry_config.backoff_enabled147 # Validate retry_config.retry_policy148 retry_policy = retry_config.retry_policy149 assert isinstance(retry_policy, RequestsStampede.policy.retry.AbstractRetryPolicy)150 assert isinstance(retry_policy, RequestsStampede.policy.retry.FixedRetryPolicy)151 assert isinstance(retry_policy.attempts, int)152 assert retry_policy.attempts == 5153 # Validate retry_config.backoff_policy154 backoff_policy = retry_config.backoff_policy155 assert isinstance(156 backoff_policy, RequestsStampede.policy.backoff.AbstractBackoffPolicy157 )158 assert isinstance(159 backoff_policy, RequestsStampede.policy.backoff.FibonacciBackoffPolicy160 )161 assert isinstance(backoff_policy.initial_delay, float)162 assert isinstance(backoff_policy.maximum_delay, float)163 assert backoff_policy.initial_delay == 0.0164 assert backoff_policy.maximum_delay == 144.0165 # Validate session166 assert isinstance(retry_session_default.session, requests.Session)167 # Validate logger168 assert isinstance(retry_session_default.logger, logging.Logger)169def test_retry_session_custom_parameters(retry_session_custom):170 """171 TODO172 """173 assert isinstance(retry_session_custom, RequestsStampede.horde.AbstractRetryRequest)174 assert isinstance(retry_session_custom, RequestsStampede.horde.RetryRequest)175 assert isinstance(retry_session_custom, RequestsStampede.horde.RetrySession)176 # Validate retry_config177 retry_config = retry_session_custom.retry_config178 assert isinstance(retry_config, RequestsStampede.config.AbstractRetryConfig)179 assert isinstance(retry_config, RequestsStampede.config.AggressiveRetryConfig)180 assert isinstance(retry_config.retry_enabled, bool)181 assert isinstance(retry_config.backoff_enabled, bool)182 # Ensure retries and backoff delays are enabled183 assert retry_config.retry_enabled184 assert retry_config.backoff_enabled185 # Validate retry_config.retry_policy186 retry_policy = retry_config.retry_policy187 assert isinstance(retry_policy, RequestsStampede.policy.retry.AbstractRetryPolicy)188 assert isinstance(retry_policy, RequestsStampede.policy.retry.InfiniteRetryPolicy)189 assert isinstance(retry_policy.attempts, float)190 assert retry_policy.attempts == math.inf191 # Validate retry_config.backoff_policy192 backoff_policy = retry_config.backoff_policy193 assert isinstance(194 backoff_policy, RequestsStampede.policy.backoff.AbstractBackoffPolicy195 )196 assert isinstance(197 backoff_policy, RequestsStampede.policy.backoff.RandomBackoffPolicy198 )199 assert isinstance(backoff_policy.minimum_delay, float)200 assert isinstance(backoff_policy.maximum_delay, float)201 assert backoff_policy.minimum_delay == 3.0202 assert backoff_policy.maximum_delay == 15.0203 # Validate session204 session = retry_session_custom.session205 assert isinstance(session, requests.Session)206 assert isinstance(session.headers, requests.structures.CaseInsensitiveDict)207 assert "User-Agent" in session.headers208 assert session.headers.get("User-Agent") == "RonSwanson/1.0"209 assert "Authorization" in session.headers210 assert session.headers.get("Authorization") == "Bearer ICanDoWhatIWant"211 # Validate logger...
retry_condition.py
Source:retry_condition.py
...17import aliyunsdkcore.utils.validation as validation18from aliyunsdkcore.acs_exception.exceptions import ClientException, ServerException19import aliyunsdkcore.acs_exception.error_code as error_code20logger = logging.getLogger(__name__)21def _find_data_in_retry_config(key_name, request, retry_config):22 if request.get_product() is None:23 return None24 path = '"{0}"."{1}"."{2}"'.format(request.get_product().lower(),25 request.get_version(),26 key_name)27 return jmespath.search(path, retry_config)28class RetryCondition(object):29 BLANK_STATUS = 030 NO_RETRY = 131 SHOULD_RETRY = 232 SHOULD_RETRY_WITH_CLIENT_TOKEN = 433 SHOULD_RETRY_WITH_THROTTLING_BACKOFF = 834 def should_retry(self, retry_policy_context):35 """Decide whether the previous request should be retried."""36 pass37class NoRetryCondition(RetryCondition):38 def should_retry(self, retry_policy_context):39 return RetryCondition.NO_RETRY40class MaxRetryTimesCondition(RetryCondition):41 def __init__(self, max_retry_times):42 validation.assert_integer_positive(max_retry_times, "max_retry_times")43 self.max_retry_times = max_retry_times44 def should_retry(self, retry_policy_context):45 if retry_policy_context.retries_attempted < self.max_retry_times:46 return RetryCondition.SHOULD_RETRY47 else:48 logger.debug("Reached the maximum number of retry. Attempts:%d",49 retry_policy_context.retries_attempted)50 return RetryCondition.NO_RETRY51class RetryOnExceptionCondition(RetryCondition):52 def __init__(self, retry_config):53 self.retry_config = retry_config54 def should_retry(self, retry_policy_context):55 request = retry_policy_context.original_request56 exception = retry_policy_context.exception57 if isinstance(exception, ClientException):58 if exception.get_error_code() == error_code.SDK_HTTP_ERROR:59 logger.debug("Retryable ClientException occurred. ClientException:%s",60 exception)61 return RetryCondition.SHOULD_RETRY62 if isinstance(exception, ServerException):63 error_code_ = exception.get_error_code()64 normal_errors = _find_data_in_retry_config("RetryableNormalErrors",65 request,66 self.retry_config)67 if isinstance(normal_errors, list) and error_code_ in normal_errors:68 logger.debug("Retryable ServerException occurred. ServerException:%s",69 exception)70 return RetryCondition.SHOULD_RETRY71 throttling_errors = _find_data_in_retry_config("RetryableThrottlingErrors",72 request,73 self.retry_config)74 if isinstance(throttling_errors, list) and error_code_ in throttling_errors:75 logger.debug("Retryable ThrottlingError occurred. ThrottlingError:%s",76 exception)77 return RetryCondition.SHOULD_RETRY | \78 RetryCondition.SHOULD_RETRY_WITH_THROTTLING_BACKOFF79 return RetryCondition.NO_RETRY80class RetryOnHttpStatusCondition(RetryCondition):81 DEFAULT_RETRYABLE_HTTP_STATUS_LIST = [82 500, 502, 503, 50483 ]84 def __init__(self, retryable_http_status_list=None):85 if retryable_http_status_list:86 self.retryable_http_status_list = retryable_http_status_list87 else:88 self.retryable_http_status_list = self.DEFAULT_RETRYABLE_HTTP_STATUS_LIST89 def should_retry(self, retry_policy_context):90 if retry_policy_context.http_status_code in self.retryable_http_status_list:91 logger.debug(92 "Retryable HTTP error occurred. HTTP status code: %s",93 retry_policy_context.http_status_code)94 return RetryCondition.SHOULD_RETRY95 else:96 return RetryCondition.NO_RETRY97class RetryOnApiCondition(RetryCondition):98 def __init__(self, retry_config):99 self.retry_config = retry_config100 def should_retry(self, retry_policy_context):101 request = retry_policy_context.original_request102 retryable_apis = _find_data_in_retry_config("RetryableAPIs", request, self.retry_config)103 if isinstance(retryable_apis, list) and request.get_action_name() in retryable_apis:104 return RetryCondition.SHOULD_RETRY105 else:106 return RetryCondition.NO_RETRY107class RetryOnApiWithClientTokenCondition(RetryCondition):108 def __init__(self, retry_config):109 self.retry_config = retry_config110 def should_retry(self, retry_policy_context):111 request = retry_policy_context.original_request112 retryable_apis = _find_data_in_retry_config(113 "RetryableAPIsWithClientToken", request, self.retry_config)114 if isinstance(retryable_apis, list) and request.get_action_name() in retryable_apis:115 return RetryCondition.SHOULD_RETRY | RetryCondition.SHOULD_RETRY_WITH_THROTTLING_BACKOFF116 else:117 return RetryCondition.NO_RETRY118class AndRetryCondition(RetryCondition):119 def __init__(self, conditions):120 self.conditions = conditions121 def should_retry(self, retry_policy_context):122 retryable = RetryCondition.BLANK_STATUS123 for condition in self.conditions:124 retryable |= condition.should_retry(retry_policy_context)125 return retryable126class OrRetryCondition(RetryCondition):...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack 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!!
Was this article helpful?
