430 lines
15 KiB
Python
430 lines
15 KiB
Python
# Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"). You
|
|
# may not use this file except in compliance with the License. A copy of
|
|
# the License is located at
|
|
#
|
|
# http://aws.amazon.com/apache2.0/
|
|
#
|
|
# or in the "license" file accompanying this file. This file is
|
|
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
|
|
# ANY KIND, either express or implied. See the License for the specific
|
|
# language governing permissions and limitations under the License.
|
|
import copy
|
|
from collections import deque
|
|
from pprint import pformat
|
|
|
|
from botocore.awsrequest import AWSResponse
|
|
from botocore.exceptions import (
|
|
ParamValidationError,
|
|
StubAssertionError,
|
|
StubResponseError,
|
|
UnStubbedResponseError,
|
|
)
|
|
from botocore.validate import validate_parameters
|
|
|
|
|
|
class _ANY:
|
|
"""
|
|
A helper object that compares equal to everything. Copied from
|
|
unittest.mock
|
|
"""
|
|
|
|
def __eq__(self, other):
|
|
return True
|
|
|
|
def __ne__(self, other):
|
|
return False
|
|
|
|
def __repr__(self):
|
|
return '<ANY>'
|
|
|
|
|
|
ANY = _ANY()
|
|
|
|
|
|
class Stubber:
|
|
"""
|
|
This class will allow you to stub out requests so you don't have to hit
|
|
an endpoint to write tests. Responses are returned first in, first out.
|
|
If operations are called out of order, or are called with no remaining
|
|
queued responses, an error will be raised.
|
|
|
|
**Example:**
|
|
::
|
|
import datetime
|
|
import botocore.session
|
|
from botocore.stub import Stubber
|
|
|
|
|
|
s3 = botocore.session.get_session().create_client('s3')
|
|
stubber = Stubber(s3)
|
|
|
|
response = {
|
|
'IsTruncated': False,
|
|
'Name': 'test-bucket',
|
|
'MaxKeys': 1000, 'Prefix': '',
|
|
'Contents': [{
|
|
'Key': 'test.txt',
|
|
'ETag': '"abc123"',
|
|
'StorageClass': 'STANDARD',
|
|
'LastModified': datetime.datetime(2016, 1, 20, 22, 9),
|
|
'Owner': {'ID': 'abc123', 'DisplayName': 'myname'},
|
|
'Size': 14814
|
|
}],
|
|
'EncodingType': 'url',
|
|
'ResponseMetadata': {
|
|
'RequestId': 'abc123',
|
|
'HTTPStatusCode': 200,
|
|
'HostId': 'abc123'
|
|
},
|
|
'Marker': ''
|
|
}
|
|
|
|
expected_params = {'Bucket': 'test-bucket'}
|
|
|
|
stubber.add_response('list_objects', response, expected_params)
|
|
stubber.activate()
|
|
|
|
service_response = s3.list_objects(Bucket='test-bucket')
|
|
assert service_response == response
|
|
|
|
|
|
This class can also be called as a context manager, which will handle
|
|
activation / deactivation for you.
|
|
|
|
**Example:**
|
|
::
|
|
import datetime
|
|
import botocore.session
|
|
from botocore.stub import Stubber
|
|
|
|
|
|
s3 = botocore.session.get_session().create_client('s3')
|
|
|
|
response = {
|
|
"Owner": {
|
|
"ID": "foo",
|
|
"DisplayName": "bar"
|
|
},
|
|
"Buckets": [{
|
|
"CreationDate": datetime.datetime(2016, 1, 20, 22, 9),
|
|
"Name": "baz"
|
|
}]
|
|
}
|
|
|
|
|
|
with Stubber(s3) as stubber:
|
|
stubber.add_response('list_buckets', response, {})
|
|
service_response = s3.list_buckets()
|
|
|
|
assert service_response == response
|
|
|
|
|
|
If you have an input parameter that is a randomly generated value, or you
|
|
otherwise don't care about its value, you can use ``stub.ANY`` to ignore
|
|
it in validation.
|
|
|
|
**Example:**
|
|
::
|
|
import datetime
|
|
import botocore.session
|
|
from botocore.stub import Stubber, ANY
|
|
|
|
|
|
s3 = botocore.session.get_session().create_client('s3')
|
|
stubber = Stubber(s3)
|
|
|
|
response = {
|
|
'IsTruncated': False,
|
|
'Name': 'test-bucket',
|
|
'MaxKeys': 1000, 'Prefix': '',
|
|
'Contents': [{
|
|
'Key': 'test.txt',
|
|
'ETag': '"abc123"',
|
|
'StorageClass': 'STANDARD',
|
|
'LastModified': datetime.datetime(2016, 1, 20, 22, 9),
|
|
'Owner': {'ID': 'abc123', 'DisplayName': 'myname'},
|
|
'Size': 14814
|
|
}],
|
|
'EncodingType': 'url',
|
|
'ResponseMetadata': {
|
|
'RequestId': 'abc123',
|
|
'HTTPStatusCode': 200,
|
|
'HostId': 'abc123'
|
|
},
|
|
'Marker': ''
|
|
}
|
|
|
|
expected_params = {'Bucket': ANY}
|
|
stubber.add_response('list_objects', response, expected_params)
|
|
|
|
with stubber:
|
|
service_response = s3.list_objects(Bucket='test-bucket')
|
|
|
|
assert service_response == response
|
|
"""
|
|
|
|
def __init__(self, client):
|
|
"""
|
|
:param client: The client to add your stubs to.
|
|
"""
|
|
self.client = client
|
|
self._event_id = 'boto_stubber'
|
|
self._expected_params_event_id = 'boto_stubber_expected_params'
|
|
self._queue = deque()
|
|
|
|
def __enter__(self):
|
|
self.activate()
|
|
return self
|
|
|
|
def __exit__(self, exception_type, exception_value, traceback):
|
|
self.deactivate()
|
|
|
|
def activate(self):
|
|
"""
|
|
Activates the stubber on the client
|
|
"""
|
|
self.client.meta.events.register_first(
|
|
'before-parameter-build.*.*',
|
|
self._assert_expected_params,
|
|
unique_id=self._expected_params_event_id,
|
|
)
|
|
self.client.meta.events.register(
|
|
'before-call.*.*',
|
|
self._get_response_handler,
|
|
unique_id=self._event_id,
|
|
)
|
|
|
|
def deactivate(self):
|
|
"""
|
|
Deactivates the stubber on the client
|
|
"""
|
|
self.client.meta.events.unregister(
|
|
'before-parameter-build.*.*',
|
|
self._assert_expected_params,
|
|
unique_id=self._expected_params_event_id,
|
|
)
|
|
self.client.meta.events.unregister(
|
|
'before-call.*.*',
|
|
self._get_response_handler,
|
|
unique_id=self._event_id,
|
|
)
|
|
|
|
def add_response(self, method, service_response, expected_params=None):
|
|
"""
|
|
Adds a service response to the response queue. This will be validated
|
|
against the service model to ensure correctness. It should be noted,
|
|
however, that while missing attributes are often considered correct,
|
|
your code may not function properly if you leave them out. Therefore
|
|
you should always fill in every value you see in a typical response for
|
|
your particular request.
|
|
|
|
:param method: The name of the client method to stub.
|
|
:type method: str
|
|
|
|
:param service_response: A dict response stub. Provided parameters will
|
|
be validated against the service model.
|
|
:type service_response: dict
|
|
|
|
:param expected_params: A dictionary of the expected parameters to
|
|
be called for the provided service response. The parameters match
|
|
the names of keyword arguments passed to that client call. If
|
|
any of the parameters differ a ``StubResponseError`` is thrown.
|
|
You can use stub.ANY to indicate a particular parameter to ignore
|
|
in validation. stub.ANY is only valid for top level params.
|
|
"""
|
|
self._add_response(method, service_response, expected_params)
|
|
|
|
def _add_response(self, method, service_response, expected_params):
|
|
if not hasattr(self.client, method):
|
|
raise ValueError(
|
|
"Client %s does not have method: %s"
|
|
% (self.client.meta.service_model.service_name, method)
|
|
)
|
|
|
|
# Create a successful http response
|
|
http_response = AWSResponse(None, 200, {}, None)
|
|
|
|
operation_name = self.client.meta.method_to_api_mapping.get(method)
|
|
self._validate_operation_response(operation_name, service_response)
|
|
|
|
# Add the service_response to the queue for returning responses
|
|
response = {
|
|
'operation_name': operation_name,
|
|
'response': (http_response, service_response),
|
|
'expected_params': expected_params,
|
|
}
|
|
self._queue.append(response)
|
|
|
|
def add_client_error(
|
|
self,
|
|
method,
|
|
service_error_code='',
|
|
service_message='',
|
|
http_status_code=400,
|
|
service_error_meta=None,
|
|
expected_params=None,
|
|
response_meta=None,
|
|
modeled_fields=None,
|
|
):
|
|
"""
|
|
Adds a ``ClientError`` to the response queue.
|
|
|
|
:param method: The name of the service method to return the error on.
|
|
:type method: str
|
|
|
|
:param service_error_code: The service error code to return,
|
|
e.g. ``NoSuchBucket``
|
|
:type service_error_code: str
|
|
|
|
:param service_message: The service message to return, e.g.
|
|
'The specified bucket does not exist.'
|
|
:type service_message: str
|
|
|
|
:param http_status_code: The HTTP status code to return, e.g. 404, etc
|
|
:type http_status_code: int
|
|
|
|
:param service_error_meta: Additional keys to be added to the
|
|
service Error
|
|
:type service_error_meta: dict
|
|
|
|
:param expected_params: A dictionary of the expected parameters to
|
|
be called for the provided service response. The parameters match
|
|
the names of keyword arguments passed to that client call. If
|
|
any of the parameters differ a ``StubResponseError`` is thrown.
|
|
You can use stub.ANY to indicate a particular parameter to ignore
|
|
in validation.
|
|
|
|
:param response_meta: Additional keys to be added to the
|
|
response's ResponseMetadata
|
|
:type response_meta: dict
|
|
|
|
:param modeled_fields: Additional keys to be added to the response
|
|
based on fields that are modeled for the particular error code.
|
|
These keys will be validated against the particular error shape
|
|
designated by the error code.
|
|
:type modeled_fields: dict
|
|
|
|
"""
|
|
http_response = AWSResponse(None, http_status_code, {}, None)
|
|
|
|
# We don't look to the model to build this because the caller would
|
|
# need to know the details of what the HTTP body would need to
|
|
# look like.
|
|
parsed_response = {
|
|
'ResponseMetadata': {'HTTPStatusCode': http_status_code},
|
|
'Error': {'Message': service_message, 'Code': service_error_code},
|
|
}
|
|
|
|
if service_error_meta is not None:
|
|
parsed_response['Error'].update(service_error_meta)
|
|
|
|
if response_meta is not None:
|
|
parsed_response['ResponseMetadata'].update(response_meta)
|
|
|
|
if modeled_fields is not None:
|
|
service_model = self.client.meta.service_model
|
|
shape = service_model.shape_for_error_code(service_error_code)
|
|
self._validate_response(shape, modeled_fields)
|
|
parsed_response.update(modeled_fields)
|
|
|
|
operation_name = self.client.meta.method_to_api_mapping.get(method)
|
|
# Note that we do not allow for expected_params while
|
|
# adding errors into the queue yet.
|
|
response = {
|
|
'operation_name': operation_name,
|
|
'response': (http_response, parsed_response),
|
|
'expected_params': expected_params,
|
|
}
|
|
self._queue.append(response)
|
|
|
|
def assert_no_pending_responses(self):
|
|
"""
|
|
Asserts that all expected calls were made.
|
|
"""
|
|
remaining = len(self._queue)
|
|
if remaining != 0:
|
|
raise AssertionError(f"{remaining} responses remaining in queue.")
|
|
|
|
def _assert_expected_call_order(self, model, params):
|
|
if not self._queue:
|
|
raise UnStubbedResponseError(
|
|
operation_name=model.name,
|
|
reason=(
|
|
'Unexpected API Call: A call was made but no additional '
|
|
'calls expected. Either the API Call was not stubbed or '
|
|
'it was called multiple times.'
|
|
),
|
|
)
|
|
|
|
name = self._queue[0]['operation_name']
|
|
if name != model.name:
|
|
raise StubResponseError(
|
|
operation_name=model.name,
|
|
reason=f'Operation mismatch: found response for {name}.',
|
|
)
|
|
|
|
def _get_response_handler(self, model, params, context, **kwargs):
|
|
self._assert_expected_call_order(model, params)
|
|
# Pop off the entire response once everything has been validated
|
|
return self._queue.popleft()['response']
|
|
|
|
def _assert_expected_params(self, model, params, context, **kwargs):
|
|
if self._should_not_stub(context):
|
|
return
|
|
self._assert_expected_call_order(model, params)
|
|
expected_params = self._queue[0]['expected_params']
|
|
if expected_params is None:
|
|
return
|
|
|
|
# Validate the parameters are equal
|
|
for param, value in expected_params.items():
|
|
if param not in params or expected_params[param] != params[param]:
|
|
raise StubAssertionError(
|
|
operation_name=model.name,
|
|
reason='Expected parameters:\n%s,\nbut received:\n%s'
|
|
% (pformat(expected_params), pformat(params)),
|
|
)
|
|
|
|
# Ensure there are no extra params hanging around
|
|
if sorted(expected_params.keys()) != sorted(params.keys()):
|
|
raise StubAssertionError(
|
|
operation_name=model.name,
|
|
reason='Expected parameters:\n%s,\nbut received:\n%s'
|
|
% (pformat(expected_params), pformat(params)),
|
|
)
|
|
|
|
def _should_not_stub(self, context):
|
|
# Do not include presign requests when processing stubbed client calls
|
|
# as a presign request will never have an HTTP request sent over the
|
|
# wire for it and therefore not receive a response back.
|
|
if context and context.get('is_presign_request'):
|
|
return True
|
|
|
|
def _validate_operation_response(self, operation_name, service_response):
|
|
service_model = self.client.meta.service_model
|
|
operation_model = service_model.operation_model(operation_name)
|
|
output_shape = operation_model.output_shape
|
|
|
|
# Remove ResponseMetadata so that the validator doesn't attempt to
|
|
# perform validation on it.
|
|
response = service_response
|
|
if 'ResponseMetadata' in response:
|
|
response = copy.copy(service_response)
|
|
del response['ResponseMetadata']
|
|
|
|
self._validate_response(output_shape, response)
|
|
|
|
def _validate_response(self, shape, response):
|
|
if shape is not None:
|
|
validate_parameters(response, shape)
|
|
elif response:
|
|
# If the output shape is None, that means the response should be
|
|
# empty apart from ResponseMetadata
|
|
raise ParamValidationError(
|
|
report=(
|
|
"Service response should only contain ResponseMetadata."
|
|
)
|
|
)
|