Python mocking response.content.iter_chunked

40 Views Asked by At

here is my python 3.8 code, Why I can not mock iter_chuncked?

async def _http_get(part_path):
    async with aiofiles.open(part_path, "ab+") as temp_file:
        async with RetryClient(client_session=ClientSession(),
                               retry_options=RandomRetry(statuses=[408, 429, 500, 502, 503, 504])) as session:
            timeout = aiohttp.ClientTimeout(total=3600)
            async with session.request(method="GET", url="http://mock.com",
                                       ssl=ssl.create_default_context(cafile=where()),
                                       timeout=timeout) as response:

                async for chunk in response.content.iter_chunked(10 * 1024 * 1024): # here what I want to mock
                    if chunk:  # filter out keep-alive new chunks
                        await temp_file.write(chunk)

here is my test code

async def mock_iter_content_async(n):
    for chunk in [b'chunk1', b'chunk2', b'chunk3']:
        yield chunk

class TestFileDownload(unittest.TestCase):
    mock_response = MagicMock()
    mock_response.content = mock.Mock()
    mock_response.content.iter_chunked.return_value = mock_iter_content_async(11)

    @patch("aiohttp_retry.RetryClient.request", return_value=mock_response)
    def test_async_http(self, mockClass1):
        part_path = os.path.join(os.getcwd(), "parts.txt")
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(_http_get(part_path))


but it got the error TypeError: 'async for' requires an object with __aiter__ method, got coroutine I found what I mocked didnt work, how to mock iter_chunked?

0

There are 0 best solutions below