运行并发测试

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/[=12= 的同一个调用下同时运行多个测试]. （这在某些方面也很好——它可以防止状态在测试之间泄漏，并且至少使用 trio 可以让你为不同的测试配置不同的 trio，例如设置一个测试使用 autojump clock 而另一个测试不使用.)

绝对最简单的选择是在单独的线程中使用 pytest-xdist 到运行测试。您仍然可以在每个测试内部使用异步——所有这些异步库都支持运行在不同的线程中使用不同的循环。

如果你真的需要使用异步并发，那么我认为你必须编写一个单独的 pytest 测试函数，然后在该函数中进行你自己的调度和并发。如果你这样做，那么从 pytest 的角度来看只有一个测试，所以要获得好的 per-test 输出并不容易。我想你可以尝试使用 pytest-subtests?

Answer 2

按照 Nathaniel 的建议使用 pytest-subtests 似乎是一个可行的解决方案。这是使用 trio 解决它的方法，它运行对名称以 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)
最后，nursey 循环应该被替换为：

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

Answer 3

现在有https://github.com/willemt/pytest-asyncio-cooperative。

今天 2020-03-25，限制非常严格 — 您必须确保您的测试不共享 anything（好吧，从技术上讲，不共享可变状态）而且您不能使用 mock.patch（技术上不要模拟其他测试可能使用的任何东西）。

您可以关注 https://github.com/pytest-dev/pytest-asyncio/issues/69 上的讨论，我认为这很难，但可以想出一种方法来标记每个 fixture 以允许或禁止并发使用，并安排测试以保留这些限制。

运行并发测试

Run tests concurrently

pytest

python-asyncio

curio

python-trio

pytest-asyncio

运行 并发测试

Run tests concurrently

pytest

python-asyncio

curio

python-trio

pytest-asyncio

运行并发测试