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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
from botocore.awsrequest import AWSResponse
from botocore.retries import special, standard
from tests import mock, unittest
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))
|
