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

8 

9import logging 

10from binascii import crc32 

11 

12from botocore.retries.base import BaseRetryableChecker 

13 

14logger = logging.getLogger(__name__) 

15 

16 

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

18# available. 

19class RetryIDPCommunicationError(BaseRetryableChecker): 

20 _SERVICE_NAME = 'sts' 

21 

22 def is_retryable(self, context): 

23 service_name = context.operation_model.service_model.service_name 

24 if service_name != self._SERVICE_NAME: 

25 return False 

26 error_code = context.get_error_code() 

27 return error_code == 'IDPCommunicationError' 

28 

29 

30class RetryDDBChecksumError(BaseRetryableChecker): 

31 _CHECKSUM_HEADER = 'x-amz-crc32' 

32 _SERVICE_NAME = 'dynamodb' 

33 

34 def is_retryable(self, context): 

35 service_name = context.operation_model.service_model.service_name 

36 if service_name != self._SERVICE_NAME: 

37 return False 

38 if context.http_response is None: 

39 return False 

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

41 if checksum is None: 

42 return False 

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

44 if actual_crc32 != int(checksum): 

45 logger.debug( 

46 "DynamoDB crc32 checksum does not match, " 

47 "expected: %s, actual: %s", 

48 checksum, 

49 actual_crc32, 

50 ) 

51 return True