首页 > 解决方案 > 为异步 aiohttp 和 aio 文件编写单元测试

问题描述

我是 asyncio 的新手。我正在使用 aiohttp 和 aiofiles 来下载图像。如何为这两个编写单元测试。

class ImageDownloader:
    def __init__(self, folder_path: str):
        self.folder_path = folder_path

    async def async_download_image(self, image_name: str, image_url: str):
        logging.info("%s downloading is started", image_name)
        async with aiohttp.ClientSession() as session:
            async with session.get(image_url) as resp:
                if resp.status == 200:
                    logging.info(" %s downloading is finished", image_name)
                    image_saving_path = os.path.join(self.folder_path, image_name)
                    logging.info(" %s saving to directory is started", image_name)
                    file = await aiofiles.open(image_saving_path, mode='wb')
                    await file.write(await resp.read())
                    await file.close()
                    logging.info(" %s saving to directory is finished", image_name)
                else:
                    logging.exception(IMAGE_DOWNLOADER_EXCEPTION + image_name)
                    raise ImageDownloaderError(IMAGE_DOWNLOADER_EXCEPTION + image_name)

标签: python-asyncioaiohttppython-aiofiles

解决方案


从 Python 3.8 开始,unittest.IsolatedAsyncioTestCase您可以方便地测试任何 asyncio 代码的编写单元测试:

class MyFixture(unittest.IsolatedAsyncioTestCase):
    async def test_1(self):
        result = await production_code()

        self.assertEqual(result, 42)

关于aiohttp官方建议(请参阅文档“Faking request object”中的警告)运行本地服务器来测试您的客户端。老实说,我不知道为什么,因为它不符合模拟昂贵依赖项的标准规则。无论如何,要这样做,您必须重新设计您的函数,以便它接受会话对象作为参数。这样,您可以在模拟解析器的帮助下将请求重定向到本地测试服务器。

async def production_code(client_session):
    aync with client_session.get(...) as response:
        ...

...
    async def test_2(self):
        with create_mocked_session() as mock:
            await production_code(mock)
        ...

aiohttp通过模拟会话对象本身并产生准备手工测试响应,可能更容易完全绕过整个库。

我不知道,aiofiles但同样的模式也适用于文件输入/输出。传入一个模拟的 file_like。它最好将所有内容都保存在内存中。


推荐阅读