Я обновляю один из моих пакетов 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
и нашел пример , показывающий, как тестировать приложения с помощью unittest стандартной библиотеки. К сожалению, я не могу заставить его работать, так как 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'