Что ж, после долгих копаний, испытаний и, конечно же, изучения асинцио, я сам все заработал.Спасибо за предложения.
Проблема заключалась в том, что event_loop из asyncio не работал;как упоминал @hoefling, сам pytest не поддерживает сопрограммы, но pytest-asyncio добавляет такую полезную функцию в ваши тесты.Это очень хорошо объяснено здесь: https://medium.com/ideas-at-igenius/testing-asyncio-python-code-with-pytest-a2f3628f82bc
Итак, без pytest-asyncio ваш асинхронный код, который необходимо протестировать, будет выглядеть следующим образом:
def test_this_is_an_async_test():
loop = asyncio.get_event_loop()
result = loop.run_until_complete(my_async_function(param1, param2, param3)
assert result == 'expected'
Мы используем loop.run_until_complete() поскольку в противном случае цикл никогда не будет работать, так как по умолчанию asyncio работает (а pytest ничего не делает, чтобы заставить его работать по-другому).
С pytest-asyncio ваш тест работает схорошо известные части async / await:
async def test_this_is_an_async_test(event_loop):
result = await my_async_function(param1, param2, param3)
assert result == 'expected'
В этом случае pytest-asyncio оборачивает вызов run_until_complete (), обобщая его, поэтому цикл обработки событий будет доступен для вашего асинхронного кода.it.
Обратите внимание: параметр event_loop во втором случае здесь даже не нужен, pytest-asyncio предоставляет один доступный для вашего теста.
С другой стороны, когда вы тестируете свойПриложение Tornado, вам обычно нужно настроить http-сервер и запустить его во время ваших тестов, прослушивания в известном порту и т. Д., Поэтому обычным способом является написание фикстур для получения http server, base_url (обычно http://localhost:, с неиспользуемым портом и т. д. и т. д.).
pytest-tornado очень полезен, поскольку предлагает вам несколько таких приспособлений: http_server, http_client, unused_port, base_url и т. Д.
Кроме того, он получаетФункция pytest mark gen_test (), которая преобразует любой стандартный тест для использования сопрограмм через yield и даже утверждает, что он будет работать с заданным временем ожидания, например:
@pytest.mark.gen_test(timeout=3)
def test_fetch_my_data(http_client, base_url):
result = yield http_client.fetch('/'.join([base_url, 'result']))
assert len(result) == 1000
Но этот способ не поддерживаетasync / await, и фактически только ioloop Tornado будет доступен через фикстуру io_loop (хотя ioloop Tornado по умолчанию использует asyncio снизу от Tornado 5.0), поэтому вам нужно объединить как pytest.mark.gen_test, так и pytest.mark.asyncio, но в правильном порядке! (что мне не удалось).
Как только я понял, в чем может быть проблема, это был следующий подход:
@pytest.mark.gen_test(timeout=2)
@pytest.mark.asyncio
async def test_score_returns_204_empty(http_client, base_url):
score_url = '/'.join([base_url, URL_PREFIX, 'score'])
token = create_token('test', scopes=['score:get'])
headers = {
'Authorization': f'Bearer {token}',
'Accept': 'application/json',
}
response = await http_client.fetch(score_url, headers=headers, raise_error=False)
assert response.code == 204
Но это совершенно неправильно, если вы понимаете, как работают оболочки Python для декоратора.С помощью приведенного выше кода сопрограмма pytest-asyncio затем оборачивается в pytest-tornado yield gen.coroutine, который не запустит цикл обработки событий ... поэтому мои тесты по-прежнему не выполнялись с той же проблемой.Любой запрос к базе данных возвращал Future в ожидании запуска цикла событий.
Мой обновленный код, как только я допустил глупую ошибку:
@pytest.mark.asyncio
@pytest.mark.gen_test(timeout=2)
async def test_score_returns_204_empty(http_client, base_url):
score_url = '/'.join([base_url, URL_PREFIX, 'score'])
token = create_token('test', scopes=['score:get'])
headers = {
'Authorization': f'Bearer {token}',
'Accept': 'application/json',
}
response = await http_client.fetch(score_url, headers=headers, raise_error=False)
assert response.code == 204
В этом случаеgen.coroutine обернут в сопрограмму pytest-asyncio, а event_loop запускает сопрограммы, как и ожидалось!
Но была еще небольшая проблема, от которой мне тоже потребовалось немного времени;Приспособление event_loop pytest-asyncio создает для каждого теста новый цикл обработки событий, в то время как pytest-tornado тоже создает новый IOloop.И тесты все еще терпели неудачу, но на этот раз с другой ошибкой.
Файл conftest.py теперь выглядит следующим образом;пожалуйста, обратите внимание, что я повторно объявил светильник event_loop, чтобы использовать event_loop из самого прибора pytest-tornado io_loop (помните, что pytest-tornado создает новый io_loop для каждой тестовой функции):
@pytest.fixture(scope='function')
def event_loop(io_loop):
loop = io_loop.current().asyncio_loop
yield loop
loop.stop()
@pytest.fixture(scope='function')
async def db():
dbe = await setup_db()
yield dbe
@pytest.fixture
def app(db):
"""
Returns a valid testing Tornado Application instance.
:return:
"""
app = make_app(db)
settings.JWT_SECRET = 'its_secret_one'
yield app
Теперь все моиТесты работают, я снова счастливый человек и очень горжусь тем, что теперь я лучше понимаю асинцио.Круто!