Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/botocore/retries/special.py: 39%

1"""Special cased retries. 

2 

3These are additional retry cases we still have to handle from the legacy 

4retry handler. They don't make sense as part of the standard mode retry 

5module. Ideally we should be able to remove this module. 

6 

7""" 

8import logging 

9from binascii import crc32 

10 

11from botocore.retries.base import BaseRetryableChecker 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16# TODO: This is an ideal candidate for the retryable trait once that's 

17# available. 

18class RetryIDPCommunicationError(BaseRetryableChecker): 

19 _SERVICE_NAME = 'sts' 

20 

21 def is_retryable(self, context): 

22 service_name = context.operation_model.service_model.service_name 

23 if service_name != self._SERVICE_NAME: 

24 return False 

25 error_code = context.get_error_code() 

26 return error_code == 'IDPCommunicationError' 

27 

28 

29class RetryDDBChecksumError(BaseRetryableChecker): 

30 _CHECKSUM_HEADER = 'x-amz-crc32' 

31 _SERVICE_NAME = 'dynamodb' 

32 

33 def is_retryable(self, context): 

34 service_name = context.operation_model.service_model.service_name 

35 if service_name != self._SERVICE_NAME: 

36 return False 

37 if context.http_response is None: 

38 return False 

39 checksum = context.http_response.headers.get(self._CHECKSUM_HEADER) 

40 if checksum is None: 

41 return False 

42 actual_crc32 = crc32(context.http_response.content) & 0xFFFFFFFF 

43 if actual_crc32 != int(checksum): 

44 logger.debug( 

45 "DynamoDB crc32 checksum does not match, " 

46 "expected: %s, actual: %s", 

47 checksum, 

48 actual_crc32, 

49 ) 

50 return True