Запускать тесты одновременно

Question

Я хотел бы запустить несколько тестов одновременно, используя asyncio (/curio/trio) и pytest, но я не смог найти никакой информации об этом. Нужно ли мне самому их планировать? И если да, есть ли способ получить хороший результат, разделяющий (под) тестовые случаи?

Вот небольшой пример игрушки, для которого я пробую это:

import pytest
import time
import asyncio

pytestmark = pytest.mark.asyncio
expected_duration = 1
accepted_error = 0.1

async def test_sleep():
    start = time.time()
    time.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

async def test_async_sleep():
    start = time.time()
    await asyncio.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

Answer 1

К сожалению, из-за того, что pytest работает внутри, вы не можете запускать несколько тестов одновременно с одним и тем же вызовом trio.run/asyncio.run/curio.run. (Это также хорошо в некоторых отношениях - это предотвращает утечку состояния между тестами, и, по крайней мере, с трио позволяет вам настроить трио по-разному для разных тестов, например, настроить один тест на использование часов автоперехода, а другой тест - нет.)

Определенно, самый простой вариант - использовать pytest-xdist для запуска тестов в отдельных потоках. Вы по-прежнему можете использовать async внутри каждого теста - все эти библиотеки async поддерживают выполнение разных циклов в разных потоках.

Если вам действительно нужно использовать асинхронный параллелизм, я думаю, вам придется написать одну тестовую функцию pytest, а затем внутри этой функции выполнить свое собственное планирование и параллелизм. Если вы сделаете это таким образом, то с точки зрения pytest будет только один тест, поэтому будет непросто получить хороший результат для каждого теста. Думаю, вы могли бы попробовать использовать pytest-subtests?

Answer 2

Сейчас есть https://github.com/willemt/pytest-asyncio-cooperative.

сегодня 2020-03-25, ограничения довольно велики - вы должны убедиться, что ваши тесты не разделяют anything (ну, технически, не делитесь изменяемым состоянием), и вы не можете использовать mock.patch (технически не имитируйте ничего, что может использовать другой тест).

Вы можете следить за обсуждением на https://github.com/pytest-dev/pytest-asyncio/issues/69, я считаю, что это сложно, но возможно придумать способ пометить каждый прибор, чтобы разрешить или запретить одновременное использование, и запланировать тесты, чтобы сохранить эти ограничения.

Answer 3

Использование pytest-subtests, предложенное Натаниэлем, кажется жизнеспособным решением. Вот как это можно решить с помощью трио: он запускает подтесты для каждой функции, имя которой начинается сio_.

import pytest
import sys
import trio
import inspect
import re
import time


pytestmark = pytest.mark.trio
io_test_pattern = re.compile("io_.*")


async def tests(subtests):

    def find_io_tests(subtests, ignored_names):
        functions = inspect.getmembers(sys.modules[__name__], inspect.isfunction)
        for (f_name, function) in functions:
            if f_name in ignored_names:
                continue
            if re.search(io_test_pattern, f_name):
                yield (run, subtests, f_name, function)

    async def run(subtests, test_name, test_function):
        with subtests.test(msg=test_name):
            await test_function()

    self_name = inspect.currentframe().f_code.co_name
    async with trio.open_nursery() as nursery:
        for io_test in find_io_tests(subtests, {self_name}):
            nursery.start_soon(*io_test)


accepted_error = 0.1

async def io_test_1():
    await assert_sleep_duration_ok(1)

async def io_test_2():
    await assert_sleep_duration_ok(2)

async def io_test_3():
    await assert_sleep_duration_ok(3)

async def io_test_4():
    await assert_sleep_duration_ok(4)

async def assert_sleep_duration_ok(duration):
    start = time.time()
    await trio.sleep(duration)
    actual_duration = time.time() - start
    assert abs(actual_duration - duration) < accepted_error

Бег python -m pytest -v выходы:

============================ test session starts =============================
platform darwin -- Python 3.7.0, pytest-4.6.2, py-1.8.0, pluggy-0.12.0
plugins: asyncio-0.10.0, trio-0.5.2, subtests-0.2.1
collected 1 item

tests/stripe_test.py::tests PASSED                                     [100%]
tests/stripe_test.py::tests PASSED                                     [100%]
tests/stripe_test.py::tests PASSED                                     [100%]
tests/stripe_test.py::tests PASSED                                     [100%]
tests/stripe_test.py::tests PASSED                                     [100%]

========================== 1 passed in 4.07 seconds ==========================

Это не идеально, поскольку процентное соотношение зависит только от количества тестов, а не от количества подтестов (т.е. io_* отмеченные функции здесь), но это кажется хорошим началом.

Также обратите внимание, что time.time() используется, поэтому имеет смысл как для trio, так и для asyncio, но в реальном случае использования trio.current_time() следует использовать вместо этого.

Те же тесты могут быть выполнены с использованием asyncio, вам в основном придется заменить три вещи:

• pytestmark = pytest.mark.trio â † ' pytestmark = pytest.mark.asyncio
• yield (run, subtests, f_name, function) â † ' yield run(subtests, f_name, function)
• И, наконец, цикл медсестры следует заменить на что-то вроде:

await asyncio.gather(*find_io_tests(subtests, {self_name}))