Использование @pytest.fixture(scope="module") с @pytest.mark.asyncio
Я думаю, что приведенный ниже пример является действительно распространенным вариантом использования:
- создать соединение с базой данных один раз,
- передать это соединение, чтобы проверить, какие данные вставки
- передать соединение тесту, который проверяет данные.
Изменение объема @pytest.fixture(scope="module")
причины ScopeMismatch: You tried to access the 'function' scoped fixture 'event_loop' with a 'module' scoped request object, involved factories
,
Так же test_insert
а также test_find
сопрограмме не нужен аргумент event_loop, потому что цикл доступен уже при передаче соединения.
Есть идеи, как исправить эти две проблемы?
import pytest
@pytest.fixture(scope="function") # <-- want this to be scope="module"; run once!
@pytest.mark.asyncio
async def connection(event_loop):
""" Expensive function; want to do in the module scope. Only this function needs `event_loop`!
"""
conn await = make_connection(event_loop)
return conn
@pytest.mark.dependency()
@pytest.mark.asyncio
async def test_insert(connection, event_loop): # <-- does not need event_loop arg
""" Test insert into database.
NB does not need event_loop argument; just the connection.
"""
_id = 0
success = await connection.insert(_id, "data")
assert success == True
@pytest.mark.dependency(depends=['test_insert'])
@pytest.mark.asyncio
async def test_find(connection, event_loop): # <-- does not need event_loop arg
""" Test database find.
NB does not need event_loop argument; just the connection.
"""
_id = 0
data = await connection.find(_id)
assert data == "data"
1 ответ
Решение состоит в том, чтобы переопределить фиксатор event_loop с областью действия модуля. Включите это в тестовый файл.
@pytest.yield_fixture(scope="module")
def event_loop():
loop = asyncio.get_event_loop()
yield loop
loop.close()
Аналогичная проблема ScopeMismatch была поднята в github для pytest-asyncio (ссылка). Решение (ниже) работает для меня:
@pytest.yield_fixture(scope='class')
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()