diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/botocore/stub.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/botocore/stub.py | 433 |
1 files changed, 433 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/botocore/stub.py b/.venv/lib/python3.12/site-packages/botocore/stub.py new file mode 100644 index 00000000..018fc087 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/botocore/stub.py @@ -0,0 +1,433 @@ +# 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( + f"Client {self.client.meta.service_model.service_name} " + f"does not have method: {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=( + f'Expected parameters:\n{pformat(expected_params)},\n' + f'but received:\n{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=( + f'Expected parameters:\n{pformat(expected_params)},\n' + f'but received:\n{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." + ) + ) |