翻译自 Advanced asyncio testing

其实就是这个库的使用:pytest-asyncio

参考使用例子:aiossdb

上篇文章中,我展示了如何使用pytest的fixture系统和插件结构来帮助你写出更简洁和更好的测试。fixture允许你为每一个测试用例创建一个更清晰的事件循环实例。插件系统允许你将异步的协程作为测试用例。但是还有更方便的pytest-asyncio可以用来写测试用例

1
2
3
4
5
6
7
8
9
10
11
import asyncio
import time
import pytest
@pytest.mark.asyncio
async def test_coro(event_loop):
before = time.monotonic()
await asyncio.sleep(0.1, loop=event_loop)
after = time.monotonic()
assert after - before >= 0.1

event_loop是内置的fixture,因此可以直接使用

因此使用pytest-asyncio很好的提高了你的测试,并且还有更好用的

其包含以下特性:

  • 创建和注入asyncio的事件循环的fixture
  • 注入未使用的tcp端口的fixture
  • 可以将测试用例标记为协程的markers
  • 非常容易的测试非默认事件循环
  • 支持定义异步fixture和异步生成器fixture

Fixtures

event_loop

创建并且注入一个默认异步事件循环的新实例,默认情况下,这个循环将在测试结束后关闭,也就是说默认的fixture范围是function级别的

只用event_loopfixture是不会使得你的测试函数变成协程的,你必须和event_loop进行交互,使用event_loop.run_until_complete方法。

或者使用pytest.mark.asyncio将测试函数变成协程,查看pytest.mark.asyncio是如何将测试函数当成协程来处理的。

只是简单的使用这个event_loop不会将其变成全局的loop,必须使用pytest.mark.asyncio才可以,这样其他的fixture以及pytest来测试的事件循环就使用的是同一个全局loop,在一个测试中可能有多个fixture,这些fixture可能也使用loop,所以这样可以确保其fixtures都是使用的一个loop.

1
2
3
4
def test_http_client(event_loop):
url = 'http://httpbin.org/get'
resp = event_loop.run_until_complete(http_client(url))
assert b'HTTP/1.1 200 OK' in resp

这个fixture可以在标准的pytest测试文件中被覆盖,比如直接在测试文件里,或者是conftest.py,定义一个自定义的事件循环。这样自定义的event_loop在测试中就会生效。

1
2
3
4
5
@pytest.fixture()
def event_loop():
loop = MyCustomLoop()
yield loop
loop.close()

如果使用了pytest.mark.asyncio,pytest的hook就会确保产生的loop会被作为默认的全局loop。依赖于event_loop的fixture就可以在运行的时候期望它们的loop被合适的修改为自定义的loop。

event_loop_process_pool

event_loop_process_pool几乎和event_loop是一直的,除了创建的事件循环会将concurrent.futures.ProcessPoolExecutor作为默认的处理器。

unused_tcp_port

查找并且产生一个没有用过的tcp端口,对于绑定一个临时的测试服务器是比较有用的。

unused_tcp_port_factory

在每次调用时候都会产生一个没有使用的tcp端口,在一个测试中要使用多个端口的情况比较有用

1
2
3
def a_test(unused_tcp_port_factory):
port1, port2 = unused_tcp_port_factory(), unused_tcp_port_factory()
...

Async fixtures

异步的fixture就和普通的pytest fixture一样定义,除了它们应该是协程或者异步的生成器

1
2
3
4
5
6
7
8
9
@pytest.fixture
async def async_gen_fixture():
await asyncio.sleep(0.1)
# 同样,生成一个值,作为调用fixture结果
yield 'a value'
@pytest.fixture(scope='module')
async def async_fixture():
return await asyncio.sleep(0.1)

所有的范围都支持,如果你需要使用一个不是函数级别的范围的话,你需要重新定义event_loop来获取更大的范围,异步的fixture需要事件循环,因此必须定义和event_loop一个一样的或者更窄的范围

如果你想在Python 3.5中使用,yield语句必须用await yield_()来替换,并且协程函数必须使用@async_generator来替换

1
2
3
4
5
6
7
from async_generator import yield_, async_generator
@pytest.fixture
@async_generator
async def async_gen_fixture():
await asyncio.sleep(0.1)
await yield_('a value')

Markers

pytest.mark.asyncio

使用这个marker来标记你的测试协程,然后pytest就会使用event_loop提供的事件循环来运行你的异步测试函数。

可以将event_loop覆盖来使用新的事件循环(参考上面)

为了让你的测试代码更加简洁,pytestmark特性可以用来将整个的模块或者类来使用这个标记。只有测试协程会被影响,默认的协程应该以test_为前缀,因此,fixtures是可以安全定义的。

1
2
3
4
5
6
7
8
9
import asyncio
import pytest
# All test coroutines will be treated as marked.
pytestmark = pytest.mark.asyncio
async def test_example(event_loop):
"""No marker!"""
await asyncio.sleep(0, loop=event_loop)

pytest.mark.asyncio_process_pool

asyncio_process_poolasyncio marker是基本一致的,除了事件循环会使用concurrent.futures.ProcessPoolExecutor作为默认的处理器。