我不明白装饰器
@pytest.mark.asyncio
可以用于什么目的。
我尝试在安装了
pytest
和 pytest-asyncio
插件的情况下运行以下代码片段,但失败了,所以我得出的结论是 pytest 在没有装饰器的情况下收集测试协程。为什么会这样存在?
async def test_div():
return 1 / 0
当你的测试被标记为
@pytest.mark.asyncio
时,它们就变成了协程,与正文中的关键字await
一起
pytest
将使用 event_loop
夹具提供的事件循环将其作为异步任务执行:
此代码带有装饰器
@pytest.mark.asyncio
async def test_example(event_loop):
do_stuff()
await asyncio.sleep(0.1, loop=event_loop)
等于写这个:
def test_example():
loop = asyncio.new_event_loop()
try:
do_stuff()
asyncio.set_event_loop(loop)
loop.run_until_complete(asyncio.sleep(0.1, loop=loop))
finally:
loop.close()
从
pytest-asyncio>=0.17
开始,如果您将 asyncio_mode = auto
添加到您的配置(pyproject.toml
、setup.cfg
或 pytest.ini
),则不需要标记,即自动为异步测试启用此行为。
参见 https://pytest-asyncio.readthedocs.io/en/latest/reference/configuration.html
Sławomir Lenart的答案仍然是正确的,但请注意,从
pytest-asyncio>=0.17
开始,如果将asyncio_mode = auto
添加到pyproject.toml
或pytest.ini
,则不需要标记,即自动为异步测试启用此行为.
参见 https://pytest-asyncio.readthedocs.io/en/latest/reference/configuration.html
我尝试使用 pytest 运行以下代码片段并 安装了 pytest-asyncio 插件,但它“失败”
实际上测试不会运行。即使您安装了
pytest
和 pytest-asyncio
,它也会被 skipped 并且您会在输出中看到如下警告:
================================== warnings summary ====================================
tests.py::test_div
/Users/.../ PytestUnhandledCoroutineWarning: async def functions are not natively supported and have been skipped.
You need to install a suitable plugin for your async framework, for example:
- anyio
- pytest-asyncio
- pytest-tornasync
- pytest-trio
- pytest-twisted
warnings.warn(PytestUnhandledCoroutineWarning(msg.format(nodeid)))
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
============================= 1 skipped, 1 warning in 0.02s =============================
正如文档所说:
带有此标记的协程或异步生成器被视为测试 通过 pytest 函数。
所以你必须标记你的异步函数。
@pytest.mark.asyncio
装饰器还允许您指定scope
,以便某些测试可以共享相同的事件循环,否则每个测试将在其自己的事件循环中运行:
import asyncio
import pytest
@pytest.mark.asyncio(scope="class")
class TestClassScopedLoop:
loop: asyncio.AbstractEventLoop
async def test_remember_loop(self):
TestClassScopedLoop.loop = asyncio.get_running_loop()
async def test_this_runs_in_same_loop(self):
assert asyncio.get_running_loop() is TestClassScopedLoop.loop