我正在更新我的一个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'
pytest
可以完美地运行标准的 unittest 风格测试。 - SColvinfpl
对象。转换所有内容甚至没有花费那么长时间,看起来比以前更清晰。 - Amos