1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
from tests import unittest
import mock
from nose.tools import assert_equal, assert_is_instance
from botocore.compat import six
from botocore.awsrequest import AWSResponse
from botocore.retries import standard, special
def create_fake_op_model(service_name):
model = mock.Mock()
model.service_model.service_name = service_name
return model
class TestRetryIDPCommunicationError(unittest.TestCase):
def setUp(self):
self.checker = special.RetryIDPCommunicationError()
def test_only_retries_error_for_sts(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('s3'),
parsed_response={
'Error': {'Code': 'IDPCommunicationError',
'Message': 'message'}},
http_response=AWSResponse(
status_code=400, raw=None, headers={},
url='https://foo'),
caught_exception=None, )
self.assertFalse(self.checker.is_retryable(context))
def test_can_retry_idp_communication_error(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('sts'),
parsed_response={
'Error': {'Code': 'IDPCommunicationError',
'Message': 'message'}},
http_response=AWSResponse(
status_code=400, raw=None, headers={},
url='https://foo'),
caught_exception=None, )
self.assertTrue(self.checker.is_retryable(context))
def test_not_idp_communication_error(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('sts'),
parsed_response={
'Error': {'Code': 'NotIDPCommunicationError',
'Message': 'message'}},
http_response=AWSResponse(
status_code=400, raw=None, headers={},
url='https://foo'),
caught_exception=None, )
self.assertFalse(self.checker.is_retryable(context))
class TestRetryDDBChecksumError(unittest.TestCase):
def setUp(self):
self.checker = special.RetryDDBChecksumError()
def raw_stream(self, contents):
raw = mock.Mock()
raw.stream.return_value = [contents]
return raw
def test_checksum_not_in_header(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('dynamodb'),
parsed_response={
'Anything': ["foo"],
},
http_response=AWSResponse(
status_code=200, raw=self.raw_stream(b'foo'),
headers={},
url='https://foo'),
caught_exception=None,
)
self.assertFalse(self.checker.is_retryable(context))
def test_checksum_matches(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('dynamodb'),
parsed_response={
'Anything': ["foo"],
},
http_response=AWSResponse(
status_code=200, raw=self.raw_stream(b'foo'),
headers={'x-amz-crc32': '2356372769'},
url='https://foo'),
caught_exception=None
)
self.assertFalse(self.checker.is_retryable(context))
def test_checksum_not_matches(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('dynamodb'),
parsed_response={
'Anything': ["foo"],
},
http_response=AWSResponse(
status_code=200, raw=self.raw_stream(b'foo'),
headers={'x-amz-crc32': '2356372768'},
url='https://foo'),
caught_exception=None
)
self.assertTrue(self.checker.is_retryable(context))
def test_checksum_check_only_for_dynamodb(self):
context = standard.RetryContext(
attempt_number=1, operation_model=create_fake_op_model('s3'),
parsed_response={
'Anything': ["foo"],
},
http_response=AWSResponse(
status_code=200, raw=self.raw_stream(b'foo'),
headers={'x-amz-crc32': '2356372768'},
url='https://foo'),
caught_exception=None
)
self.assertFalse(self.checker.is_retryable(context))
|
