Блог

Главная - Вопросы по программированию - проблемы pytest с привязкой к сеансу и asyncio

проблемы pytest с привязкой к сеансу и asyncio

#python #pytest #python-asyncio #semaphore #pytest-xdist

#python #pytest #python-asyncio #семафор #pytest-xdist

Вопрос:

У меня есть несколько тестовых файлов, каждый из которых имеет асинхронный параметр, который выглядит следующим образом:

 
@pytest.fixture(scope="module")
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="module")
async def some_fixture():
    return await make_fixture()

Я использую xdist для распараллеливания. Кроме того, у меня есть этот декоратор:

 @toolz.curry
def throttle(limit, f):
    semaphore = asyncio.Semaphore(limit)

    @functools.wraps(f)
    async def wrapped(*args, **kwargs):
        async with semaphore:
            return await f(*args, **kwargs)

    return wrapped

и у меня есть функция, использующая это:

 @throttle(10)
def f():
    ...

Теперь f вызывается из нескольких тестовых файлов, и я получаю исключение, сообщающее мне, что я не могу использовать семафор из разных циклов событий.

Я попытался перейти к привязке цикла событий на уровне сеанса:

 

@pytest.fixture(scope="session", autouse=True)
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

Но это только дало мне:

Несоответствие области действия: вы пытались получить доступ к прибору с привязкой к ‘функции’ ‘event_loop’ с помощью объекта запроса с привязкой к модулю, задействованного фабриками

Возможно ли вообще, чтобы xdist async fixture semaphore работали вместе?

Ответ №1:

В конечном итоге он заработал, используя следующее conftest.py :

 import asyncio

import pytest


@pytest.fixture(scope="session")
def event_loop():
    return asyncio.get_event_loop()

Комментарии:

1. Это сработало для меня, но, похоже, ошибка с pytest-asyncio

2. Это не ошибка, параметр event_loop pytest-asyncio по умолчанию имеет значение «функция», поэтому вы получаете новый event_loop для каждой тестовой функции. Следовательно, другое асинхронное устройство не может работать в течение всего сеанса, поскольку это будет охватывать несколько разных циклов событий. Вам нужно переопределить привязку event_loop с нужной вам областью, как показано в ответе Uri. Смотрите «Приспособления» в pypi.org/project/pytest-asyncio