py.test混合装置和异步协程

Question

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

我的主要期望：

• 每个测试用例都应该可以访问新的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 python模块来创建这种魔术装饰器，但是似乎这两个模块都可以帮助我创建一个签名与my_test相同的函数，在这种情况下，这不是一个好的解决方案。

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

Answer 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像正常功能一样收集异步协程。我还拦截函数的文本执行。如果要测试的功能是协程，则在事件循环中执行它。无论有没有夹具，它都可以在每个测试中创建新的事件循环实例。

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