1
# Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2
#
3
# Licensed under the Apache License, Version 2.0 (the "License"). You
4
# may not use this file except in compliance with the License. A copy of
5
# the License is located at
6
#
7
# http://aws.amazon.com/apache2.0/
8
#
9
# or in the "license" file accompanying this file. This file is
10
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11
# ANY KIND, either express or implied. See the License for the specific
12
# language governing permissions and limitations under the License.
13 11
import copy
14 11
from collections import deque
15 11
from pprint import pformat
16

17 11
from botocore.validate import validate_parameters
18 11
from botocore.exceptions import ParamValidationError, \
19
    StubResponseError, StubAssertionError, UnStubbedResponseError
20 11
from botocore.awsrequest import AWSResponse
21

22

23 11
class _ANY(object):
24
    """
25
    A helper object that compares equal to everything. Copied from
26
    unittest.mock
27
    """
28

29 11
    def __eq__(self, other):
30 11
        return True
31

32 11
    def __ne__(self, other):
33 11
        return False
34

35 11
    def __repr__(self):
36 11
        return '<ANY>'
37

38 11
ANY = _ANY()
39

40

41 11
class Stubber(object):
42
    """
43
    This class will allow you to stub out requests so you don't have to hit
44
    an endpoint to write tests. Responses are returned first in, first out.
45
    If operations are called out of order, or are called with no remaining
46
    queued responses, an error will be raised.
47

48
    **Example:**
49
    ::
50
        import datetime
51
        import botocore.session
52
        from botocore.stub import Stubber
53

54

55
        s3 = botocore.session.get_session().create_client('s3')
56
        stubber = Stubber(s3)
57

58
        response = {
59
            'IsTruncated': False,
60
            'Name': 'test-bucket',
61
            'MaxKeys': 1000, 'Prefix': '',
62
            'Contents': [{
63
                'Key': 'test.txt',
64
                'ETag': '"abc123"',
65
                'StorageClass': 'STANDARD',
66
                'LastModified': datetime.datetime(2016, 1, 20, 22, 9),
67
                'Owner': {'ID': 'abc123', 'DisplayName': 'myname'},
68
                'Size': 14814
69
            }],
70
            'EncodingType': 'url',
71
            'ResponseMetadata': {
72
                'RequestId': 'abc123',
73
                'HTTPStatusCode': 200,
74
                'HostId': 'abc123'
75
            },
76
            'Marker': ''
77
        }
78

79
        expected_params = {'Bucket': 'test-bucket'}
80

81
        stubber.add_response('list_objects', response, expected_params)
82
        stubber.activate()
83

84
        service_response = s3.list_objects(Bucket='test-bucket')
85
        assert service_response == response
86

87

88
    This class can also be called as a context manager, which will handle
89
    activation / deactivation for you.
90

91
    **Example:**
92
    ::
93
        import datetime
94
        import botocore.session
95
        from botocore.stub import Stubber
96

97

98
        s3 = botocore.session.get_session().create_client('s3')
99

100
        response = {
101
            "Owner": {
102
                "ID": "foo",
103
                "DisplayName": "bar"
104
            },
105
            "Buckets": [{
106
                "CreationDate": datetime.datetime(2016, 1, 20, 22, 9),
107
                "Name": "baz"
108
            }]
109
        }
110

111

112
        with Stubber(s3) as stubber:
113
            stubber.add_response('list_buckets', response, {})
114
            service_response = s3.list_buckets()
115

116
        assert service_response == response
117

118

119
    If you have an input parameter that is a randomly generated value, or you
120
    otherwise don't care about its value, you can use ``stub.ANY`` to ignore
121
    it in validation.
122

123
    **Example:**
124
    ::
125
        import datetime
126
        import botocore.session
127
        from botocore.stub import Stubber, ANY
128

129

130
        s3 = botocore.session.get_session().create_client('s3')
131
        stubber = Stubber(s3)
132

133
        response = {
134
            'IsTruncated': False,
135
            'Name': 'test-bucket',
136
            'MaxKeys': 1000, 'Prefix': '',
137
            'Contents': [{
138
                'Key': 'test.txt',
139
                'ETag': '"abc123"',
140
                'StorageClass': 'STANDARD',
141
                'LastModified': datetime.datetime(2016, 1, 20, 22, 9),
142
                'Owner': {'ID': 'abc123', 'DisplayName': 'myname'},
143
                'Size': 14814
144
            }],
145
            'EncodingType': 'url',
146
            'ResponseMetadata': {
147
                'RequestId': 'abc123',
148
                'HTTPStatusCode': 200,
149
                'HostId': 'abc123'
150
            },
151
            'Marker': ''
152
        }
153

154
        expected_params = {'Bucket': ANY}
155
        stubber.add_response('list_objects', response, expected_params)
156

157
        with stubber:
158
            service_response = s3.list_objects(Bucket='test-bucket')
159

160
        assert service_response == response
161
    """
162 11
    def __init__(self, client):
163
        """
164
        :param client: The client to add your stubs to.
165
        """
166 11
        self.client = client
167 11
        self._event_id = 'boto_stubber'
168 11
        self._expected_params_event_id = 'boto_stubber_expected_params'
169 11
        self._queue = deque()
170

171 11
    def __enter__(self):
172 11
        self.activate()
173 11
        return self
174

175 11
    def __exit__(self, exception_type, exception_value, traceback):
176 11
        self.deactivate()
177

178 11
    def activate(self):
179
        """
180
        Activates the stubber on the client
181
        """
182 11
        self.client.meta.events.register_first(
183
            'before-parameter-build.*.*',
184
            self._assert_expected_params,
185
            unique_id=self._expected_params_event_id)
186 11
        self.client.meta.events.register(
187
            'before-call.*.*',
188
            self._get_response_handler,
189
            unique_id=self._event_id)
190

191 11
    def deactivate(self):
192
        """
193
        Deactivates the stubber on the client
194
        """
195 11
        self.client.meta.events.unregister(
196
            'before-parameter-build.*.*',
197
            self._assert_expected_params,
198
            unique_id=self._expected_params_event_id)
199 11
        self.client.meta.events.unregister(
200
            'before-call.*.*',
201
            self._get_response_handler,
202
            unique_id=self._event_id)
203

204 11
    def add_response(self, method, service_response, expected_params=None):
205
        """
206
        Adds a service response to the response queue. This will be validated
207
        against the service model to ensure correctness. It should be noted,
208
        however, that while missing attributes are often considered correct,
209
        your code may not function properly if you leave them out. Therefore
210
        you should always fill in every value you see in a typical response for
211
        your particular request.
212

213
        :param method: The name of the client method to stub.
214
        :type method: str
215

216
        :param service_response: A dict response stub. Provided parameters will
217
            be validated against the service model.
218
        :type service_response: dict
219

220
        :param expected_params: A dictionary of the expected parameters to
221
            be called for the provided service response. The parameters match
222
            the names of keyword arguments passed to that client call. If
223
            any of the parameters differ a ``StubResponseError`` is thrown.
224
            You can use stub.ANY to indicate a particular parameter to ignore
225
            in validation. stub.ANY is only valid for top level params.
226
        """
227 11
        self._add_response(method, service_response, expected_params)
228

229 11
    def _add_response(self, method, service_response, expected_params):
230 11
        if not hasattr(self.client, method):
231 11
            raise ValueError(
232
                "Client %s does not have method: %s"
233
                % (self.client.meta.service_model.service_name, method))
234

235
        # Create a successful http response
236 11
        http_response = AWSResponse(None, 200, {}, None)
237

238 11
        operation_name = self.client.meta.method_to_api_mapping.get(method)
239 11
        self._validate_response(operation_name, service_response)
240

241
        # Add the service_response to the queue for returning responses
242 11
        response = {
243
            'operation_name': operation_name,
244
            'response': (http_response, service_response),
245
            'expected_params': expected_params
246
        }
247 11
        self._queue.append(response)
248

249 11
    def add_client_error(self, method, service_error_code='',
250
                         service_message='', http_status_code=400,
251
                         service_error_meta=None, expected_params=None,
252
                         response_meta=None):
253
        """
254
        Adds a ``ClientError`` to the response queue.
255

256
        :param method: The name of the service method to return the error on.
257
        :type method: str
258

259
        :param service_error_code: The service error code to return,
260
                                   e.g. ``NoSuchBucket``
261
        :type service_error_code: str
262

263
        :param service_message: The service message to return, e.g.
264
                        'The specified bucket does not exist.'
265
        :type service_message: str
266

267
        :param http_status_code: The HTTP status code to return, e.g. 404, etc
268
        :type http_status_code: int
269

270
        :param service_error_meta: Additional keys to be added to the
271
            service Error
272
        :type service_error_meta: dict
273

274
        :param expected_params: A dictionary of the expected parameters to
275
            be called for the provided service response. The parameters match
276
            the names of keyword arguments passed to that client call. If
277
            any of the parameters differ a ``StubResponseError`` is thrown.
278
            You can use stub.ANY to indicate a particular parameter to ignore
279
            in validation.
280

281
        :param response_meta: Additional keys to be added to the
282
            response's ResponseMetadata
283
        :type response_meta: dict
284

285
        """
286 11
        http_response = AWSResponse(None, http_status_code, {}, None)
287

288
        # We don't look to the model to build this because the caller would
289
        # need to know the details of what the HTTP body would need to
290
        # look like.
291 11
        parsed_response = {
292
            'ResponseMetadata': {'HTTPStatusCode': http_status_code},
293
            'Error': {
294
                'Message': service_message,
295
                'Code': service_error_code
296
            }
297
        }
298

299 11
        if service_error_meta is not None:
300 11
            parsed_response['Error'].update(service_error_meta)
301

302 11
        if response_meta is not None:
303 11
            parsed_response['ResponseMetadata'].update(response_meta)
304

305 11
        operation_name = self.client.meta.method_to_api_mapping.get(method)
306
        # Note that we do not allow for expected_params while
307
        # adding errors into the queue yet.
308 11
        response = {
309
            'operation_name': operation_name,
310
            'response': (http_response, parsed_response),
311
            'expected_params': expected_params,
312
        }
313 11
        self._queue.append(response)
314

315 11
    def assert_no_pending_responses(self):
316
        """
317
        Asserts that all expected calls were made.
318
        """
319 11
        remaining = len(self._queue)
320 11
        if remaining != 0:
321 11
            raise AssertionError(
322
                "%d responses remaining in queue." % remaining)
323

324 11
    def _assert_expected_call_order(self, model, params):
325 11
        if not self._queue:
326 11
            raise UnStubbedResponseError(
327
                operation_name=model.name,
328
                reason=(
329
                        'Unexpected API Call: A call was made but no additional calls expected. '
330
                        'Either the API Call was not stubbed or it was called multiple times.'
331
                        )
332
            )
333

334 11
        name = self._queue[0]['operation_name']
335 11
        if name != model.name:
336 0
            raise StubResponseError(
337
                operation_name=model.name,
338
                reason='Operation mismatch: found response for %s.' % name)
339

340 11
    def _get_response_handler(self, model, params, context, **kwargs):
341 11
        self._assert_expected_call_order(model, params)
342
        # Pop off the entire response once everything has been validated
343 11
        return self._queue.popleft()['response']
344

345 11
    def _assert_expected_params(self, model, params, context, **kwargs):
346 11
        if self._should_not_stub(context):
347 11
            return
348 11
        self._assert_expected_call_order(model, params)
349 11
        expected_params = self._queue[0]['expected_params']
350 11
        if expected_params is None:
351 11
            return
352

353
        # Validate the parameters are equal
354 11
        for param, value in expected_params.items():
355 11
            if param not in params or expected_params[param] != params[param]:
356 11
                raise StubAssertionError(
357
                    operation_name=model.name,
358
                    reason='Expected parameters:\n%s,\nbut received:\n%s' % (
359
                        pformat(expected_params), pformat(params)))
360

361
        # Ensure there are no extra params hanging around
362 11
        if sorted(expected_params.keys()) != sorted(params.keys()):
363 0
            raise StubAssertionError(
364
                operation_name=model.name,
365
                reason='Expected parameters:\n%s,\nbut received:\n%s' % (
366
                    pformat(expected_params), pformat(params)))
367

368 11
    def _should_not_stub(self, context):
369
        # Do not include presign requests when processing stubbed client calls
370
        # as a presign request will never have an HTTP request sent over the
371
        # wire for it and therefore not receive a response back.
372 11
        if context and context.get('is_presign_request'):
373 11
            return True
374

375 11
    def _validate_response(self, operation_name, service_response):
376 11
        service_model = self.client.meta.service_model
377 11
        operation_model = service_model.operation_model(operation_name)
378 11
        output_shape = operation_model.output_shape
379

380
        # Remove ResponseMetadata so that the validator doesn't attempt to
381
        # perform validation on it.
382 11
        response = service_response
383 11
        if 'ResponseMetadata' in response:
384 11
            response = copy.copy(service_response)
385 11
            del response['ResponseMetadata']
386

387 11
        if output_shape is not None:
388 11
            validate_parameters(response, output_shape)
389 11
        elif response:
390
            # If the output shape is None, that means the response should be
391
            # empty apart from ResponseMetadata
392 11
            raise ParamValidationError(
393
                report=(
394
                    "Service response should only contain ResponseMetadata."))

Read our documentation on viewing source code .

Loading