使用aiohttp和asyncio时编写单元测试

Question

我正在更新我的一个 Python 包，因此它是异步的（使用

aiohttp

而不是

requests

）。我还在更新我的单元测试，以便它们可以使用新的异步版本，但我在这方面遇到了一些麻烦。

这是我的包裹中的片段：

async def fetch(session, url):
    while True:
        try:
            async with session.get(url) as response:
                assert response.status == 200
                return await response.json()
        except Exception as error:
            pass


class FPL():
    def __init__(self, session):
        self.session = session

    async def get_user(self, user_id, return_json=False):
        url = API_URLS["user"].format(user_id)
        user = await fetch(self.session, url)

        if return_json:
            return user
        return User(user, session=self.session)

使用时似乎一切正常：

async def main():
    async with aiohttp.ClientSession() as session:
         fpl = FPL(session)
         user = await fpl.get_user(3808385)
         print(user)

loop = asynio.get_event_loop()
loop.run_until_complete(main())

>>> User 3808385

不幸的是，我的单元测试遇到了一些问题。我想我可以简单地做类似的事情

def _run(coroutine):
    return asyncio.get_event_loop().run_until_complete(coroutine)


class FPLTest(unittest.TestCase):
    def setUp(self):
        session = aiohttp.ClientSession()
        self.fpl = FPL(session)

    def test_user(self):
        user = _run(self.fpl.get_user("3523615"))
        self.assertIsInstance(user, User)

        user = _run(self.fpl.get_user("3523615", True))
        self.assertIsInstance(user, dict)

if __name__ == '__main__':
    unittest.main()

它给出了诸如

之类的错误

DeprecationWarning: The object should be created from async function loop=loop)

和

ResourceWarning: Unclosed client session <aiohttp.client.ClientSession object at 0x7fbe647fd208>

我尝试向

_close()

类添加一个

FPL

函数来关闭会话，然后从测试中调用它，但这也不起作用，并且仍然显示有一个未关闭的客户端会话。

是否可以这样做，我只是做错了什么，还是我最好使用

asynctest

或

pytest-aiohttp

之类的东西来代替？

编辑：我还检查了

aiohttp

的文档，并找到了一个 example 展示了如何使用标准库的单元测试来测试应用程序。不幸的是我无法让它工作，因为

loop

中提供的

AioHTTPTestCase

自 3.5 以来已被弃用并抛出错误：

class FPLTest(AioHTTPTestCase):
    def setUp(self):
        session = aiohttp.ClientSession()
        self.fpl = FPL(session)

    @unittest_run_loop
    async def test_user(self):
        user = await self.fpl.get_user("3523615")
        self.assertIsInstance(user, User)

        user = await self.fpl.get_user("3523615", True)
        self.assertIsInstance(user, dict)

给予

tests/test_fpl.py:20: DeprecationWarning: The object should be created from async function
  session = aiohttp.ClientSession()
  ...
======================================================================
ERROR: test_user (__main__.FPLTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/amos/Documents/fpl/venv/lib/python3.7/site-packages/aiohttp/test_utils.py", line 477, in new_func
    return self.loop.run_until_complete(
AttributeError: 'FPLTest' object has no attribute 'loop'

======================================================================
ERROR: test_user (__main__.FPLTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/amos/Documents/fpl/venv/lib/python3.7/site-packages/aiohttp/test_utils.py", line 451, in tearDown
    self.loop.run_until_complete(self.tearDownAsync())
AttributeError: 'FPLTest' object has no attribute 'loop'

Answer 1

将 pytest 与 aiohttp-pytest 一起使用：

async def test_test_user(loop):
    async with aiohttp.ClientSession() as session:
         fpl = FPL(session)
         user = await fpl.get_user(3808385)
    assert isinstance(user, User)

现代Python开发者的谚语：生命太短，不能不使用pytest。

您可能还想设置一个模拟服务器来在测试期间接收您的http请求，我没有一个简单的示例，但可以在here看到完整的工作示例。

Answer 2

您可以使用标准

pytest-asyncio

库中的

AsyncMock

插件和

unittest

类。确保您安装了软件包：

pip install pytest pytest-asyncio

要允许等待代码，请使用来自插件的

@pytest.mark.asyncio

标记来装饰您的测试函数。接下来，创建一个模拟实例，然后修补使用

aiohttp.ClientSession.get

类返回的

AsyncMock

方法：

from unittest.mock import AsyncMock, patch

import aiohttp
import pytest

# The fetch function to be tested
async def fetch_data(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as response:
        return await response.json()

@pytest.mark.asyncio
async def test_fetch_data_success():
    url = 'http://test.com'
    expected_response = {'key': 'value'}

    mock_response = AsyncMock()
    mock_response.__aenter__.return_value = mock_response
    mock_response.__aexit__.return_value = False
    mock_response.json = AsyncMock(return_value=expected_response)

    with patch('aiohttp.ClientSession.get', return_value=mock_response):
        async with aiohttp.ClientSession() as session:
            response = await fetch_data(session, url)
            assert response == expected_response

你看，我们需要两个异步模拟实例，因为

response.json()

也是可等待的。

由于我们使用的是异步上下文管理器，因此我们需要模拟

__aenter__

和

__aexit__

方法。如果不这样做，您可能会收到如下错误：

session = <aiohttp.client.ClientSession object at 0x0000018C35EF0880>, url = 'http://test.com'

    async def fetch(session: aiohttp.ClientSession, url: str):
>       async with session.get(url) as response:
E       AttributeError: __aenter__

tests\async_fetch_test.py:8: AttributeError