我正在使用pytest和pytest-asyncio来运行异步测试。奇怪的是,这些测试只有在只有一个时才有效。例如:
@pytest.mark.asyncio
async def test_foo():
client = await get_db_connection()
user = await client.DO_QUERY
response = await FunctionUnderTest(
db=client, ...args
)
assert response.id is not None
@pytest.mark.asyncio
async def test_error_foo():
client = await get_db_connection()
with pytest.raises(MissingRequiredError):
await FunctionUnderTest(
db=client, ...args
)
如果我注释掉其中任何一个测试,剩下的一个将通过,但同时运行这两个测试会得到:
RuntimeError: Task <Task pending name='Task-5' coro=<test_error_foo() running at /tests/test_function_under_test.py:44> cb=[_run_until_complete_cb() at /usr/lib/python3.10/asyncio/base_events.py:184]> got Future <Future pending> attached to a different loop
我本来希望pytest-asyncio创建一个事件循环并按顺序运行所有测试,但这似乎不起作用。
通过使用pytest-asyncio-cooperative
,您希望获得更好的里程,该 是围绕所有测试使用单个事件循环而设计的。 另一方面,pytest-asyncio
每个测试使用事件循环。
- 安装:
pip install pytest-asyncio-cooperative
- 将
pytest.mark.asyncio
替换为pytest.mark.asyncio_cooperative
@pytest.mark.asyncio_cooperative
async def test_foo():
client = await get_db_connection()
user = await client.DO_QUERY
response = await FunctionUnderTest(
db=client, ...args
)
assert response.id is not None
@pytest.mark.asyncio_cooperative
async def test_error_foo():
client = await get_db_connection()
with pytest.raises(MissingRequiredError):
await FunctionUnderTest(
db=client, ...args
)
- 运行测试时
pytest-asyncio
关闭并按顺序运行测试打开(即 --max-asyncio-tasks 1):
pytest -p no:asyncio --max-asyncio-tasks 1 test_function_under_test.py
诚然,如果你想继续使用pytest-asyncio
你可以重写测试以使用pytest-asyncio
提供的事件循环,然后在测试中显式使用它。我认为您必须将其传递到引用事件循环的代码中。
免责声明:我是此pytest.mark.asyncio_cooperative
的维护者