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
144
145
146
147
148
|
from contextlib import contextmanager
import mock
from tests import BaseSessionTest, ClientHTTPStubber
from botocore.history import BaseHistoryHandler
from botocore.history import get_global_history_recorder
class RecordingHandler(BaseHistoryHandler):
def __init__(self):
self.recorded_calls = []
def emit(self, event_type, payload, source):
self.recorded_calls.append((event_type, payload, source))
class TestRecordStatementsInjections(BaseSessionTest):
def setUp(self):
super(TestRecordStatementsInjections, self).setUp()
self.client = self.session.create_client('s3', 'us-west-2')
self.http_stubber = ClientHTTPStubber(self.client)
self.s3_response_body = (
'<ListAllMyBucketsResult '
' xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
' <Owner>'
' <ID>d41d8cd98f00b204e9800998ecf8427e</ID>'
' <DisplayName>foo</DisplayName>'
' </Owner>'
' <Buckets>'
' <Bucket>'
' <Name>bar</Name>'
' <CreationDate>1912-06-23T22:57:02.000Z</CreationDate>'
' </Bucket>'
' </Buckets>'
'</ListAllMyBucketsResult>'
).encode('utf-8')
self.recording_handler = RecordingHandler()
history_recorder = get_global_history_recorder()
history_recorder.enable()
history_recorder.add_handler(self.recording_handler)
def _get_all_events_of_type(self, event_type):
recorded_calls = self.recording_handler.recorded_calls
matching = [call for call in recorded_calls
if call[0] == event_type]
return matching
def test_does_record_api_call(self):
self.http_stubber.add_response(body=self.s3_response_body)
with self.http_stubber:
self.client.list_buckets()
api_call_events = self._get_all_events_of_type('API_CALL')
self.assertEqual(len(api_call_events), 1)
event = api_call_events[0]
event_type, payload, source = event
self.assertEqual(payload, {
'operation': u'ListBuckets',
'params': {},
'service': 's3'
})
self.assertEqual(source, 'BOTOCORE')
def test_does_record_http_request(self):
self.http_stubber.add_response(body=self.s3_response_body)
with self.http_stubber:
self.client.list_buckets()
http_request_events = self._get_all_events_of_type('HTTP_REQUEST')
self.assertEqual(len(http_request_events), 1)
event = http_request_events[0]
event_type, payload, source = event
method = payload['method']
self.assertEqual(method, u'GET')
# The header values vary too much per request to verify them here.
# Instead just check the presense of each expected header.
headers = payload['headers']
for expected_header in ['Authorization', 'User-Agent', 'X-Amz-Date',
'X-Amz-Content-SHA256']:
self.assertIn(expected_header, headers)
body = payload['body']
self.assertIsNone(body)
streaming = payload['streaming']
self.assertEquals(streaming, False)
url = payload['url']
self.assertEquals(url, 'https://s3.us-west-2.amazonaws.com/')
self.assertEqual(source, 'BOTOCORE')
def test_does_record_http_response(self):
self.http_stubber.add_response(body=self.s3_response_body)
with self.http_stubber:
self.client.list_buckets()
http_response_events = self._get_all_events_of_type('HTTP_RESPONSE')
self.assertEqual(len(http_response_events), 1)
event = http_response_events[0]
event_type, payload, source = event
self.assertEqual(payload, {
'status_code': 200,
'headers': {},
'streaming': False,
'body': self.s3_response_body,
'context': {'operation_name': 'ListBuckets'}
}
)
self.assertEqual(source, 'BOTOCORE')
def test_does_record_parsed_response(self):
self.http_stubber.add_response(body=self.s3_response_body)
with self.http_stubber:
self.client.list_buckets()
parsed_response_events = self._get_all_events_of_type(
'PARSED_RESPONSE')
self.assertEqual(len(parsed_response_events), 1)
event = parsed_response_events[0]
event_type, payload, source = event
# Given that the request contains headers with a user agent string
# a date and a signature we need to disassemble the call and manually
# assert the interesting bits since mock can only assert if the args
# all match exactly.
owner = payload['Owner']
self.assertEqual(owner, {
'DisplayName': 'foo',
'ID': 'd41d8cd98f00b204e9800998ecf8427e'
})
buckets = payload['Buckets']
self.assertEqual(len(buckets), 1)
bucket = buckets[0]
self.assertEqual(bucket['Name'], 'bar')
metadata = payload['ResponseMetadata']
self.assertEqual(metadata, {
'HTTPHeaders': {},
'HTTPStatusCode': 200,
'RetryAttempts': 0
})
|
