py.test миксеры и сопрограммы asyncio

Я создаю несколько тестов для кода python3, используя py.test. Код обращается к базе данных Postgresql с помощью aiopg (интерфейс на основе Asyncio для postgres).

Мои основные ожидания:

  • Каждый тестовый пример должен иметь доступ к новому циклу событий asyncio.

  • Тест, который выполняется слишком долго, остановится с исключением тайм-аута.

  • Каждый тестовый пример должен иметь доступ к соединению с базой данных.

  • Я не хочу повторяться при написании тестовых случаев.

Используя фикстуры py.test, я могу довольно близко подойти к тому, что хочу, но мне все равно приходится немного повторяться в каждом асинхронном тестовом примере.

Вот как выглядит мой код:

@pytest.fixture(scope='function')
def tloop(request):
    # This fixture is responsible for getting a new event loop
    # for every test, and close it when the test ends.
    ...

def run_timeout(cor,loop,timeout=ASYNC_TEST_TIMEOUT):
    """
    Run a given coroutine with timeout.
    """
    task_with_timeout = asyncio.wait_for(cor,timeout)
    try:
        loop.run_until_complete(task_with_timeout)
    except futures.TimeoutError:
        # Timeout:
        raise ExceptAsyncTestTimeout()


@pytest.fixture(scope='module')
def clean_test_db(request):
    # Empty the test database.
    ...

@pytest.fixture(scope='function')
def udb(request,clean_test_db,tloop):
    # Obtain a connection to the database using aiopg
    # (That's why we need tloop here).
    ...


# An example for a test:
def test_insert_user(tloop,udb):
    @asyncio.coroutine
    def insert_user():
        # Do user insertion here ...
        yield from udb.insert_new_user(...
        ...

    run_timeout(insert_user(),tloop)

Я могу жить с решением, которое у меня есть до сих пор, но определение внутренней сопрограммы и добавление строки run_timeout для каждого асинхронного теста, который я пишу, может стать громоздким.

Я хочу, чтобы мои тесты выглядели примерно так:

@some_magic_decorator
def test_insert_user(udb):
    # Do user insertion here ...
    yield from udb.insert_new_user(...
    ...

Я попытался создать такой декоратор каким-то элегантным способом, но потерпел неудачу. В более общем случае, если мой тест выглядит так:

@some_magic_decorator
def my_test(arg1,arg2,...,arg_n):
    ...

Тогда созданная функция (после применения декоратора) должна быть:

def my_test_wrapper(tloop,arg1,arg2,...,arg_n):
    run_timeout(my_test(),tloop)

Обратите внимание, что в некоторых моих тестах используются другие фикстуры (например, помимо udb), и эти фикстуры должны отображаться как аргументы создаваемой функции, иначе py.test не будет их вызывать.

Я пытался использовать как wrapt, так и decorator для создания такого волшебного декоратора, однако кажется, что оба этих модуля помогают мне создать функцию с сигнатурой, идентичной my_test, что не является хорошим решением в этом случае.

Вероятно, это можно решить с помощью eval или аналогичного хака, но мне было интересно, есть ли что-то элегантное, чего мне здесь не хватает.


postgresql python-asyncio pytest python-decorators fixtures
person real    schedule 02.02.2015    source источник

Ответы (2)


arrow_upward
5
arrow_downward

Я сейчас пытаюсь решить аналогичную проблему. Вот что я придумал на данный момент. Кажется, он работает, но нуждается в некоторой очистке:

# tests/test_foo.py
import asyncio

@asyncio.coroutine
def test_coro(loop):
    yield from asyncio.sleep(0.1)
    assert 0

# tests/conftest.py
import asyncio


@pytest.yield_fixture
def loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()


def pytest_pycollect_makeitem(collector, name, obj):
    """Collect asyncio coroutines as normal functions, not as generators."""
    if asyncio.iscoroutinefunction(obj):
        return list(collector._genfunctions(name, obj))


def pytest_pyfunc_call(pyfuncitem):
    """If ``pyfuncitem.obj`` is an asyncio coroutinefunction, execute it via
    the event loop instead of calling it directly."""
    testfunction = pyfuncitem.obj

    if not asyncio.iscoroutinefunction(testfunction):
        return

    # Copied from _pytest/python.py:pytest_pyfunc_call()
    funcargs = pyfuncitem.funcargs
    testargs = {}
    for arg in pyfuncitem._fixtureinfo.argnames:
        testargs[arg] = funcargs[arg]
    coro = testfunction(**testargs)  # Will no execute the test yet!

    # Run the coro in the event loop
    loop = testargs.get('loop', asyncio.get_event_loop())
    loop.run_until_complete(coro)

    return True  # TODO: What to return here?

Поэтому я в основном позволяю pytest собирать сопрограммы asyncio, как обычные функции. Я также перехватываю выполнение текста для функций. Если тестируемая функция является сопрограммой, я запускаю ее в цикле обработки событий. Он работает с фикстурой или без нее, создавая новый экземпляр цикла событий для каждого теста.

Редактировать. По словам Ронни Пфанншмидта, что-то подобное будет добавлено в pytest после выпуска 2.7. :-)

person Stefan Scherfke    schedule 17.02.2015
comment
Я не знал о pytest_pycollect_makeitem и pytest_pyfunc_call. Очень круто! Откуда ты взял эти вещи? Если вы планируете использовать это, убедитесь, что у вас есть разные циклы для каждого теста, и добавьте мою функцию run_timeout в свой код, чтобы ваши тесты не зависали. - person real; 18.02.2015
comment
Из документации pytest (pytest.org/latest/plugins.html#pytest-hook -reference), от копания в коде и догадок. ИМХО, это действительно не так хорошо документировано. Возможно, я выпущу его как официальный pytest-плагин. - person Stefan Scherfke; 18.02.2015

arrow_upward
-1
arrow_downward

Каждый тестовый пример должен иметь доступ к новому циклу событий asyncio.

Набор тестов asyncio использует unittest.TestCase. Он использует метод setUp() для создания нового цикла событий. addCleanup(loop.close) автоматически закрывает цикл обработки событий даже при ошибке.

Извините, я не знаю, как написать это с помощью py.test, если вы не хотите использовать TestCase. Но если я правильно помню, py.test поддерживает unittest.TestCase.

Тест, который выполняется слишком долго, остановится с исключением тайм-аута.

Вы можете использовать loop.call_later() с функцией, которая вызывает BaseException в качестве сторожевого таймера.

person haypo    schedule 04.02.2015
comment
Спасибо за ответ. Наличие нового цикла событий и исключений тайм-аута — это то, что я уже решил. См. функции tloop и run_timeout в моем коде. Моя проблема заключается в создании волшебного декоратора. Я хочу, чтобы каждый тестовый пример был сопрограммой, которая будет запускаться автоматически. - person real; 05.02.2015