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

Question

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

11

我正在更新我的一个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)

and

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

我已经尝试为FPL类添加了一个_close()函数来关闭会话，然后从测试中调用它，但这仍然无效，并且仍显示存在未关闭的客户端会话。

这种方法可行吗？是我做错了什么，还是最好使用像asynctest或pytest-aiohttp这样的工具呢？

编辑：aiohttp的文档中也有一个示例展示如何使用标准库的unittest测试应用程序。不幸的是，我无法使其正常工作，因为AioHTTPTestCase中提供的loop自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'

- Amos

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- SColvin · Accepted Answer

使用 aiohttp-pytest 与 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请求。我没有一个简单的示例，但可以在此处查看完整的工作示例。