py.test 混合夹具和异步协程

Posted 2023-02-23

【中文标题】py.test 混合夹具和异步协程

【英文标题】：py.test mixing fixtures and asyncio coroutines

【发布时间】：2015-04-01 17:41:55

【问题描述】：

我正在使用 py.test 为 python3 代码构建一些测试。该代码使用 aiopg（基于 Asyncio 的 postgres 接口）访问 Postgresql 数据库。

我的主要期望：

每个测试用例都应该可以访问一个新的异步事件循环。

运行时间过长的测试会因超时异常而停止。

每个测试用例都应该能够访问数据库连接。

我不想在编写测试用例时重复自己。

使用 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 python 模块来创建这样一个神奇的装饰器，但是这两个模块似乎都帮助我创建了一个签名与 my_test 相同的函数，这不是一个好的解决方案这种情况。

这可能可以使用 eval 或类似的 hack 解决，但我想知道这里是否缺少一些优雅的东西。

【参考方案1】：

我目前正在尝试解决类似的问题。到目前为止，这是我想出的。它似乎有效，但需要一些清理：

# 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 协程。我还截取函数的文本执行。如果要测试的函数是协程，我在事件循环中执行。它可以在有或没有固定装置的情况下为每个测试创建一个新的事件循环实例。

编辑：根据 Ronny Pfannschmidt 的说法，在 2.7 版本之后，类似这样的内容将被添加到 pytest。 :-)

【讨论】：

我不知道 pytest_pycollect_makeitem 和 pytest_pyfunc_call。真的很酷！你从哪里弄来的那些东西？如果您打算使用它，请确保为每个测试使用不同的循环，并将我的 run_timeout 函数添加到您的代码中，以使您的测试不会卡住。

来自 pytest 文档 (pytest.org/latest/plugins.html#pytest-hook-reference)，来自挖掘代码和猜测。恕我直言，它真的没有那么好记录。也许我会把它作为一个官方的 pytest-plugin 发布。

【参考方案2】：

每个测试用例都应该可以访问一个新的异步事件循环。

asyncio 的测试套件使用 unittest.TestCase。它使用 setUp() 方法来创建一个新的事件循环。 addCleanup(loop.close) 会自动关闭事件循环，即使发生错误。

抱歉，如果您不想使用 TestCase，我不知道如何使用 py.test 编写此内容。但如果我没记错的话，py.test 支持 unittest.TestCase。

运行时间过长的测试会因超时异常而停止。

您可以将 loop.call_later() 与引发 BaseException 作为看门狗的函数一起使用。

【讨论】：

感谢您的回复。拥有一个新的事件循环和超时异常是我已经解决的问题。请参阅我的代码中的函数 tloop 和 run_timeout。我的问题是创建魔法​​装饰器。我希望每个测试用例都是一个自动运行的协程。