写的aiossdb库,需要对协程进行测试,但是pytest会默认将定义的函数就当成普通函数来测试,当然可以在一个测试函数里面定义要测试的部分,但是比较麻烦吧,每次都得写loop之类。这篇文章介绍了一种使用pytest对协程进行测试的优雅方法。

翻译自:Testing (asyncio) coroutines with pytest

pytest是一个非常棒的Python测试库,并且也是我最喜欢的Python库之一。它使得写测试很简单,并且它在测试失败的时候的报告能力也非常棒。

但是,它不能在测试协程上给你太大帮助,所以一个简单的测试协程的程序可能会是这样:

1
2
3
4
5
6
7
8
9
10
11
import asyncio
def test_coro():
loop = asyncio.get_event_loop()
@asyncio.coroutine
def do_test():
yield from asyncio.sleep(0.1, loop=loop)
assert 0 # onoes!
loop.run_until_complete(do_test())

这种方法有些许问题,并且很冗余,我们真正感兴趣的代码只有包含yield fromassert的那几行。

如果每个测试用例都有自己的事件循环实例,并且在测试通过或者失败的时候自动正确关闭那会更好。

我们也不能简单的在测试用例里yield,因为pytest会认为我们的测试用例会产生不在这里的新的测试用例。所以我们必须创建一个独立的协程来包含我们真实的测试,并且放到事件循环里去运行它。

如果我们可以像下面这么做,我们的测试将会看起来更加整洁,并且会表现更好

1
2
3
4
@asyncio.coroutine
def test_coro(loop):
yield from asyncio.sleep(0.1, loop=loop)
assert 0

幸亏pytest的灵活的插件系统,使得可以实现我们期望的操作。但是,大多数需要的hooks都没有非常好的形成文档,这使得找出以需要实现的hooks以及如何实现相当难。

我们创建一个本地每个目录的插件,因为这样比创建一个真正的外部插件要容易一些。pytest会查看每一个测试目录去寻找一个conftest.py文件,然后对这个目录下的所有测试应用这里定义的fixtures和hooks。

让我们开始从写一个会给每一个测试用例创建新的事件循环实例,并且会在测试完成后关闭的fixture开始吧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import asyncio
import pytest
# yield_fixture 已经弃用,统一使用fixture
@pytest.fixture
def loop():
# Set-up
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
yield loop
# Clean-up
loop.close()
# tests/test_coros.py
def test_coro(loop):
@asyncio.coroutine
def do_test():
yield from asyncio.sleep(0.1, loop=loop)
assert 0 # onoes!
loop.run_until_complete(do_test())

在每一个测试之前,pytest执行loop fixture直到第一个yield语句,yielded获得的值将会传递给我们的测试用例作为loop参数。当测试完成,无论成功与否,pytest会结束loop fixture的运行,因此会合适的关闭loop。同样的你可以写一个fixture来创建一个socket,并且在每个测试完成之后关闭。你的socket fixture可以依赖loop fixture,就和我们的测试做的一样。很不错,不是吗?

但是我们还没有完成,让我们教pytest怎样去执行我们的测试协程吧。因此,我们需要去改变asyncio协程被收集的行为,它们应该像普通测试函数一样收集,而不是测试生成器。还有要改变它们被执行的方式,通过loop.run_until_complete()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# tests/conftest.py
def pytest_pycollect_makeitem(collector, name, obj):
"""收集asyncio协程作为普通函数,而不是生成器
collector 是收集器实例
name 应该是收集到的函数名字
obj 是函数对象"""
if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
# 我们返回测试函数对象的list,依赖于fixtures,一个测试函数可能会导致多个测试项
# 也就是它被 pytest.mark.parametrize()装饰器装饰的时候
# 将其生成为普通函数,也就是说,这些协程会被当成普通函数收集,但是它还是协程
return list(collector._genfunctions(name, obj))
# else:
# 我们返回None,pytest的默认行为会将其应用到obj
def pytest_pyfunc_call(pyfuncitem):
"""如果 pyfuncitem.obj 是一个asyncio的协程函数,通过在时间循环中执行它而不是直接调用"""
testfunction = pyfuncitem.obj
if not asyncio.iscoroutinefunction(testfunction):
# 如果不是协程的话直接返回,pytest会通过正常方式去处理它
return
# 从所有可用的fixtures中抽取需要的参数
funcargs = pyfuncitem.funcargs # 所有fixtures的字典
argnames = pyfuncitem._fixtureinfo.argnames # 这个测试的参数名字
# 也就是从fixtures字典中来找出所需要的fixture对象,fixture对象可能直接返回值yield return
# 也可能是返回一个函数对象,所以需要调用
testargs = {arg: funcargs[arg] for arg in argnames}
# 为这个测试创建生成器对象,测试这时候将不会被执行
coro = testfunction(**testargs)
# 在事件循环中执行这个协程并且开始测试
loop = testargs['loop'] if loop in testargs else asyncio.get_event_loop()
loop.run_until_complete(coro)
return True # 我们执行完测试后的信号

这一个插件应该在pytest 2.4或者更新中运行。

看起来pytest dev正在试着清理并且重构pytest的插件系统。也许,将来的版本将会包含和我们的插件相同的函数。如果pytest支持即时可用的(out-of-the-box)协程测试方法那将会非常棒,因为协程正在在不同的框架中变得非常广泛。

另外:pytest-asyncio正是pytest support for asyncio,可以拿来用!