codekingpro/portable-devtools

1# Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13import copy14from collections import deque15from pprint import pformat16 17from botocore.awsrequest import AWSResponse18from botocore.exceptions import (19    ParamValidationError,20    StubAssertionError,21    StubResponseError,22    UnStubbedResponseError,23)24from botocore.validate import validate_parameters25 26 27class _ANY:28    """29    A helper object that compares equal to everything. Copied from30    unittest.mock31    """32 33    def __eq__(self, other):34        return True35 36    def __ne__(self, other):37        return False38 39    def __repr__(self):40        return '<ANY>'41 42 43ANY = _ANY()44 45 46class Stubber:47    """48    This class will allow you to stub out requests so you don't have to hit49    an endpoint to write tests. Responses are returned first in, first out.50    If operations are called out of order, or are called with no remaining51    queued responses, an error will be raised.52 53    **Example:**54    ::55        import datetime56        import botocore.session57        from botocore.stub import Stubber58 59 60        s3 = botocore.session.get_session().create_client('s3')61        stubber = Stubber(s3)62 63        response = {64            'IsTruncated': False,65            'Name': 'test-bucket',66            'MaxKeys': 1000, 'Prefix': '',67            'Contents': [{68                'Key': 'test.txt',69                'ETag': '"abc123"',70                'StorageClass': 'STANDARD',71                'LastModified': datetime.datetime(2016, 1, 20, 22, 9),72                'Owner': {'ID': 'abc123', 'DisplayName': 'myname'},73                'Size': 1481474            }],75            'EncodingType': 'url',76            'ResponseMetadata': {77                'RequestId': 'abc123',78                'HTTPStatusCode': 200,79                'HostId': 'abc123'80            },81            'Marker': ''82        }83 84        expected_params = {'Bucket': 'test-bucket'}85 86        stubber.add_response('list_objects', response, expected_params)87        stubber.activate()88 89        service_response = s3.list_objects(Bucket='test-bucket')90        assert service_response == response91 92 93    This class can also be called as a context manager, which will handle94    activation / deactivation for you.95 96    **Example:**97    ::98        import datetime99        import botocore.session100        from botocore.stub import Stubber101 102 103        s3 = botocore.session.get_session().create_client('s3')104 105        response = {106            "Owner": {107                "ID": "foo",108                "DisplayName": "bar"109            },110            "Buckets": [{111                "CreationDate": datetime.datetime(2016, 1, 20, 22, 9),112                "Name": "baz"113            }]114        }115 116 117        with Stubber(s3) as stubber:118            stubber.add_response('list_buckets', response, {})119            service_response = s3.list_buckets()120 121        assert service_response == response122 123 124    If you have an input parameter that is a randomly generated value, or you125    otherwise don't care about its value, you can use ``stub.ANY`` to ignore126    it in validation.127 128    **Example:**129    ::130        import datetime131        import botocore.session132        from botocore.stub import Stubber, ANY133 134 135        s3 = botocore.session.get_session().create_client('s3')136        stubber = Stubber(s3)137 138        response = {139            'IsTruncated': False,140            'Name': 'test-bucket',141            'MaxKeys': 1000, 'Prefix': '',142            'Contents': [{143                'Key': 'test.txt',144                'ETag': '"abc123"',145                'StorageClass': 'STANDARD',146                'LastModified': datetime.datetime(2016, 1, 20, 22, 9),147                'Owner': {'ID': 'abc123', 'DisplayName': 'myname'},148                'Size': 14814149            }],150            'EncodingType': 'url',151            'ResponseMetadata': {152                'RequestId': 'abc123',153                'HTTPStatusCode': 200,154                'HostId': 'abc123'155            },156            'Marker': ''157        }158 159        expected_params = {'Bucket': ANY}160        stubber.add_response('list_objects', response, expected_params)161 162        with stubber:163            service_response = s3.list_objects(Bucket='test-bucket')164 165        assert service_response == response166    """167 168    def __init__(self, client):169        """170        :param client: The client to add your stubs to.171        """172        self.client = client173        self._event_id = 'boto_stubber'174        self._expected_params_event_id = 'boto_stubber_expected_params'175        self._queue = deque()176 177    def __enter__(self):178        self.activate()179        return self180 181    def __exit__(self, exception_type, exception_value, traceback):182        self.deactivate()183 184    def activate(self):185        """186        Activates the stubber on the client187        """188        self.client.meta.events.register_first(189            'before-parameter-build.*.*',190            self._assert_expected_params,191            unique_id=self._expected_params_event_id,192        )193        self.client.meta.events.register(194            'before-call.*.*',195            self._get_response_handler,196            unique_id=self._event_id,197        )198 199    def deactivate(self):200        """201        Deactivates the stubber on the client202        """203        self.client.meta.events.unregister(204            'before-parameter-build.*.*',205            self._assert_expected_params,206            unique_id=self._expected_params_event_id,207        )208        self.client.meta.events.unregister(209            'before-call.*.*',210            self._get_response_handler,211            unique_id=self._event_id,212        )213 214    def add_response(self, method, service_response, expected_params=None):215        """216        Adds a service response to the response queue. This will be validated217        against the service model to ensure correctness. It should be noted,218        however, that while missing attributes are often considered correct,219        your code may not function properly if you leave them out. Therefore220        you should always fill in every value you see in a typical response for221        your particular request.222 223        :param method: The name of the client method to stub.224        :type method: str225 226        :param service_response: A dict response stub. Provided parameters will227            be validated against the service model.228        :type service_response: dict229 230        :param expected_params: A dictionary of the expected parameters to231            be called for the provided service response. The parameters match232            the names of keyword arguments passed to that client call. If233            any of the parameters differ a ``StubResponseError`` is thrown.234            You can use stub.ANY to indicate a particular parameter to ignore235            in validation. stub.ANY is only valid for top level params.236        """237        self._add_response(method, service_response, expected_params)238 239    def _add_response(self, method, service_response, expected_params):240        if not hasattr(self.client, method):241            raise ValueError(242                "Client %s does not have method: %s"243                % (self.client.meta.service_model.service_name, method)244            )245 246        # Create a successful http response247        http_response = AWSResponse(None, 200, {}, None)248 249        operation_name = self.client.meta.method_to_api_mapping.get(method)250        self._validate_operation_response(operation_name, service_response)251 252        # Add the service_response to the queue for returning responses253        response = {254            'operation_name': operation_name,255            'response': (http_response, service_response),256            'expected_params': expected_params,257        }258        self._queue.append(response)259 260    def add_client_error(261        self,262        method,263        service_error_code='',264        service_message='',265        http_status_code=400,266        service_error_meta=None,267        expected_params=None,268        response_meta=None,269        modeled_fields=None,270    ):271        """272        Adds a ``ClientError`` to the response queue.273 274        :param method: The name of the service method to return the error on.275        :type method: str276 277        :param service_error_code: The service error code to return,278                                   e.g. ``NoSuchBucket``279        :type service_error_code: str280 281        :param service_message: The service message to return, e.g.282                        'The specified bucket does not exist.'283        :type service_message: str284 285        :param http_status_code: The HTTP status code to return, e.g. 404, etc286        :type http_status_code: int287 288        :param service_error_meta: Additional keys to be added to the289            service Error290        :type service_error_meta: dict291 292        :param expected_params: A dictionary of the expected parameters to293            be called for the provided service response. The parameters match294            the names of keyword arguments passed to that client call. If295            any of the parameters differ a ``StubResponseError`` is thrown.296            You can use stub.ANY to indicate a particular parameter to ignore297            in validation.298 299        :param response_meta: Additional keys to be added to the300            response's ResponseMetadata301        :type response_meta: dict302 303        :param modeled_fields: Additional keys to be added to the response304            based on fields that are modeled for the particular error code.305            These keys will be validated against the particular error shape306            designated by the error code.307        :type modeled_fields: dict308 309        """310        http_response = AWSResponse(None, http_status_code, {}, None)311 312        # We don't look to the model to build this because the caller would313        # need to know the details of what the HTTP body would need to314        # look like.315        parsed_response = {316            'ResponseMetadata': {'HTTPStatusCode': http_status_code},317            'Error': {'Message': service_message, 'Code': service_error_code},318        }319 320        if service_error_meta is not None:321            parsed_response['Error'].update(service_error_meta)322 323        if response_meta is not None:324            parsed_response['ResponseMetadata'].update(response_meta)325 326        if modeled_fields is not None:327            service_model = self.client.meta.service_model328            shape = service_model.shape_for_error_code(service_error_code)329            self._validate_response(shape, modeled_fields)330            parsed_response.update(modeled_fields)331 332        operation_name = self.client.meta.method_to_api_mapping.get(method)333        # Note that we do not allow for expected_params while334        # adding errors into the queue yet.335        response = {336            'operation_name': operation_name,337            'response': (http_response, parsed_response),338            'expected_params': expected_params,339        }340        self._queue.append(response)341 342    def assert_no_pending_responses(self):343        """344        Asserts that all expected calls were made.345        """346        remaining = len(self._queue)347        if remaining != 0:348            raise AssertionError(f"{remaining} responses remaining in queue.")349 350    def _assert_expected_call_order(self, model, params):351        if not self._queue:352            raise UnStubbedResponseError(353                operation_name=model.name,354                reason=(355                    'Unexpected API Call: A call was made but no additional '356                    'calls expected. Either the API Call was not stubbed or '357                    'it was called multiple times.'358                ),359            )360 361        name = self._queue[0]['operation_name']362        if name != model.name:363            raise StubResponseError(364                operation_name=model.name,365                reason=f'Operation mismatch: found response for {name}.',366            )367 368    def _get_response_handler(self, model, params, context, **kwargs):369        self._assert_expected_call_order(model, params)370        # Pop off the entire response once everything has been validated371        return self._queue.popleft()['response']372 373    def _assert_expected_params(self, model, params, context, **kwargs):374        if self._should_not_stub(context):375            return376        self._assert_expected_call_order(model, params)377        expected_params = self._queue[0]['expected_params']378        if expected_params is None:379            return380 381        # Validate the parameters are equal382        for param, value in expected_params.items():383            if param not in params or expected_params[param] != params[param]:384                raise StubAssertionError(385                    operation_name=model.name,386                    reason='Expected parameters:\n%s,\nbut received:\n%s'387                    % (pformat(expected_params), pformat(params)),388                )389 390        # Ensure there are no extra params hanging around391        if sorted(expected_params.keys()) != sorted(params.keys()):392            raise StubAssertionError(393                operation_name=model.name,394                reason='Expected parameters:\n%s,\nbut received:\n%s'395                % (pformat(expected_params), pformat(params)),396            )397 398    def _should_not_stub(self, context):399        # Do not include presign requests when processing stubbed client calls400        # as a presign request will never have an HTTP request sent over the401        # wire for it and therefore not receive a response back.402        if context and context.get('is_presign_request'):403            return True404 405    def _validate_operation_response(self, operation_name, service_response):406        service_model = self.client.meta.service_model407        operation_model = service_model.operation_model(operation_name)408        output_shape = operation_model.output_shape409 410        # Remove ResponseMetadata so that the validator doesn't attempt to411        # perform validation on it.412        response = service_response413        if 'ResponseMetadata' in response:414            response = copy.copy(service_response)415            del response['ResponseMetadata']416 417        self._validate_response(output_shape, response)418 419    def _validate_response(self, shape, response):420        if shape is not None:421            validate_parameters(response, shape)422        elif response:423            # If the output shape is None, that means the response should be424            # empty apart from ResponseMetadata425            raise ParamValidationError(426                report=(427                    "Service response should only contain ResponseMetadata."428                )429            )430