AioStubber that returns AioAWSResponse()

CHANGES.rst

@@ -1,6 +1,10 @@
 Changes
 -------
 
+2.7.1 (2023-10-19)
+^^^^^^^^^^^^^^^^^^
+* added AioStubber that return AioAWSResponse()
+
 2.7.0 (2023-10-17)
 ^^^^^^^^^^^^^^^^^^
 * add support for Python 3.12

aiobotocore/__init__.py

@@ -1 +1 @@
-__version__ = '2.7.0'
+__version__ = '2.7.1'

aiobotocore/stub.py

@@ -0,0 +1,108 @@
+from botocore.stub import Stubber
+
+from .awsrequest import AioAWSResponse
+
+
+class AioStubber(Stubber):
+    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)
+            )  # pragma: no cover
+
+        # Create a successful http response
+        http_response = AioAWSResponse(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 = AioAWSResponse(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)

tests/test_patches.py

@@ -2,7 +2,7 @@
 
 import botocore
 import pytest
-from botocore import retryhandler
+from botocore import retryhandler, stub
 from botocore.args import ClientArgsCreator
 from botocore.awsrequest import AWSResponse
 from botocore.client import BaseClient, ClientCreator, Config
@@ -650,6 +650,7 @@
     retryhandler.CRC32Checker._check_response: {
         '3ee7afd0bb1a3bf53934d77e44f619962c52b0c9'
     },
+    stub.Stubber: {'bccf23c3733cc656b909f5130cba80dbc9540b05'},
 }
 
 

tests/test_stubber.py

@@ -0,0 +1,74 @@
+import pytest
+
+from aiobotocore.awsrequest import AioAWSResponse
+from aiobotocore.session import AioSession
+from aiobotocore.stub import AioStubber
+
+from .mock_server import AIOServer
+
+
+@pytest.mark.moto
+@pytest.mark.asyncio
+async def test_add_response():
+    session = AioSession()
+
+    async with AIOServer() as server, session.create_client(
+        's3',
+        endpoint_url=server.endpoint_url,
+        aws_secret_access_key='xxx',
+        aws_access_key_id='xxx',
+    ) as s3_client:
+        stubber = AioStubber(s3_client)
+        operation_name = 'put_object'
+        service_response = dict(
+            ETag="6805f2cfc46c0f04559748bb039d69ae",
+            VersionId="psM2sYY4.o1501dSx8wMvnkOzSBB.V4a",
+        )
+        expected_params = dict()
+        stubber.add_response(operation_name, service_response, expected_params)
+
+        assert len(stubber._queue) == 1
+        assert stubber._queue[0][
+            'operation_name'
+        ] == s3_client.meta.method_to_api_mapping.get(operation_name)
+        assert isinstance(stubber._queue[0]['response'][0], AioAWSResponse)
+        assert stubber._queue[0]['response'][1] == service_response
+        assert stubber._queue[0]['expected_params'] == expected_params
+
+
+@pytest.mark.moto
+@pytest.mark.asyncio
+async def test_add_client_error():
+    session = AioSession()
+
+    async with AIOServer() as server, session.create_client(
+        's3',
+        endpoint_url=server.endpoint_url,
+        aws_secret_access_key='xxx',
+        aws_access_key_id='xxx',
+    ) as s3_client:
+        stubber = AioStubber(s3_client)
+        operation_name = 'put_object'
+        service_error_code = 'InvalidObjectState'
+        service_message = 'Object is in invalid state'
+        http_status_code = 400
+        service_error_meta = {"AdditionalInfo": "value"}
+        response_meta = {"AdditionalResponseInfo": "value"}
+        modeled_fields = {'StorageClass': 'foo', 'AccessTier': 'bar'}
+
+        stubber.add_client_error(
+            operation_name,
+            service_error_code,
+            service_message,
+            http_status_code,
+            service_error_meta,
+            response_meta=response_meta,
+            modeled_fields=modeled_fields,
+        )
+
+        assert len(stubber._queue) == 1
+        assert stubber._queue[0][
+            'operation_name'
+        ] == s3_client.meta.method_to_api_mapping.get(operation_name)
+        assert isinstance(stubber._queue[0]['response'][0], AioAWSResponse)
+        assert stubber._queue[0]['response'][1]