We read every piece of feedback, and take your input very seriously.
To see all available qualifiers, see our documentation.
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
See also:
""" Patch aiobotocore to work with moto """ from typing import Callable from unittest.mock import MagicMock import aiobotocore.endpoint import aiobotocore.awsrequest import aiohttp import aiohttp.client_reqrep import aiohttp.typedefs import botocore.awsrequest import botocore.model import pytest class MockAWSResponse(aiobotocore.awsrequest.AioAWSResponse): def __init__(self, response: botocore.awsrequest.AWSResponse): self._moto_response = response self.status_code = response.status_code self.raw = MockHttpClientResponse(response) # adapt async methods to use moto's response async def _content_prop(self) -> bytes: return self._moto_response.content async def _text_prop(self) -> str: return self._moto_response.text class MockHttpClientResponse(aiohttp.client_reqrep.ClientResponse): def __init__(self, response: botocore.awsrequest.AWSResponse): async def read(self, n: int = -1) -> bytes: # streaming/range requests. used by s3fs return response.content self.content = MagicMock(aiohttp.StreamReader) self.content.read = read self.response = response @property def raw_headers(self) -> aiohttp.typedefs.RawHeaders: # Return the headers encoded the way that aiobotocore expects them return { k.encode("utf-8"): str(v).encode("utf-8") for k, v in self.response.headers.items() }.items() # TODO: clear caches for function scope @pytest.fixture(scope="session") def patch_aiobotocore(): def factory(original: Callable) -> Callable: def patched_convert_to_response_dict( http_response: botocore.awsrequest.AWSResponse, operation_model: botocore.model.OperationModel, ): return original(MockAWSResponse(http_response), operation_model) return patched_convert_to_response_dict aiobotocore.endpoint.convert_to_response_dict = factory( aiobotocore.endpoint.convert_to_response_dict )
The text was updated successfully, but these errors were encountered:
No branches or pull requests
See also:
The text was updated successfully, but these errors were encountered: