Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.