如何让pytest-bdd与pytest-asyncio兼容实现异步测试？

我要做的是创建一个测试流程：先给设备设置一些值，之后每隔固定时间多次查询这些值，断言查询结果和设置值一致。因为不能等一次响应回来再继续，所以必须异步执行查询和断言，但现在没法把pytest-bdd的then函数和pytest.mark.asyncio结合起来，求解决办法。

两次失败的尝试 #

尝试1：给then步骤加asyncio标记报错 #

直接给then装饰的异步函数加@pytest.mark.asyncio，代码如下：

@pytest.mark.asyncio
@then(parsers.parse("The query result for {setting} is {value:n}"))
@then(parsers.parse("The query result for {setting} is {value:bool}", extra_types={"bool": bool}))
async def test_get_value(destination, setting, value):
    responses = await run_queries(destination, setting)
    for response in responses:
        assert response == value

配套的run_queries函数：

async def run_queries(destination: int, query_name: str, connection_file: str) -> List[int]:
    # 创建包含query_name、connection file和destination的自定义消息对象
    responses = client.execute(message) # 执行查询并收集响应，这部分功能正常
    query_result = []

    for response in responses:
        if response.tid == tid: # 校验是否拿到正确响应
            query_result.append(response.value[0])

    return query_result

结果触发错误：coroutine 'test_get_value' was never awaited

尝试2：给场景加asyncio标记，场景不执行 #

给@scenario装饰的测试函数加@pytest.mark.asyncio，代码：

@scenario('../features/scenario_fixture.feature', 'Hope this works')
@pytest.mark.asyncio
def test_query():
    pass

对应的Feature文件：

Feature: Scenario test
  Scenario: Hope this works
    Then The query result for setting is 100

结果场景完全不运行。

可行的解决方案 #

pytest-bdd本身不直接支持异步then步骤，但可以通过两种方式绕开限制：

方案1：用异步夹具封装异步逻辑 #

把异步查询逻辑放到带@pytest.mark.asyncio的夹具中，pytest会自动处理异步执行，再把结果传给同步的then步骤：

import pytest
from pytest_bdd import then, parsers, scenario

@pytest.fixture(scope="function")
async def query_results(destination, setting):
    return await run_queries(destination, setting)

@then(parsers.parse("The query result for {setting} is {value:n}"))
@then(parsers.parse("The query result for {setting} is {value:bool}", extra_types={"bool": bool}))
def check_query_results(query_results, value):
    for response in query_results:
        assert response == value

@scenario('../features/scenario_fixture.feature', 'Hope this works')
def test_query():
    pass

方案2：在同步步骤中手动运行异步函数 #

如果不想用夹具，直接在同步then步骤里用asyncio.run()执行异步逻辑：

import asyncio
from pytest_bdd import then, parsers, scenario

@then(parsers.parse("The query result for {setting} is {value:n}"))
@then(parsers.parse("The query result for {setting} is {value:bool}", extra_types={"bool": bool}))
def test_get_value(destination, setting, value):
    responses = asyncio.run(run_queries(destination, setting))
    for response in responses:
        assert response == value

@scenario('../features/scenario_fixture.feature', 'Hope this works')
def test_query():
    pass

注意：如果测试环境已存在事件循环（比如用了pytest-asyncio的默认循环），改用asyncio.get_running_loop().run_until_complete()避免循环冲突。

内容的提问来源于stack exchange，提问作者FroMoSouOmou