合并 py.test 和 trio/curio
Combining py.test and trio/curio
我会结合 pytest 和 trio(或 curio,如果这样更容易的话),即将我的测试用例编写为协程函数。通过在 conftest.py
:
中声明自定义测试运行器,这相对容易实现
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
'''If item is a coroutine function, run it under trio'''
if not inspect.iscoroutinefunction(pyfuncitem.obj):
return
kernel = trio.Kernel()
funcargs = pyfuncitem.funcargs
testargs = {arg: funcargs[arg]
for arg in pyfuncitem._fixtureinfo.argnames}
try:
kernel.run(functools.partial(pyfuncitem.obj, **testargs))
finally:
kernel.run(shutdown=True)
return True
这让我可以像这样编写测试用例:
async def test_something():
server = MockServer()
server_task = await trio.run(server.serve)
try:
# test the server
finally:
server.please_terminate()
try:
with trio.fail_after(30):
server_task.join()
except TooSlowError:
server_task.cancel()
但这是很多样板文件。在非异步代码中,我会将其分解为固定装置:
@pytest.yield_fixture()
def mock_server():
server = MockServer()
thread = threading.Thread(server.serve)
thread.start()
try:
yield server
finally:
server.please_terminate()
thread.join()
server.server_close()
def test_something(mock_server):
# do the test..
有没有办法在 trio 中做同样的事情,即实现异步装置?理想情况下,我会写:
async def test_something(mock_server):
# do the test..
编辑:下面的答案现在大多不相关——而是使用 pytest-trio 并遵循 instructions in its manual.
您的示例 pytest_pyfunc_call
代码不起作用,因为它混合了 trio 和 curio :-)。对于 trio,有一个装饰器 trio.testing.trio_test
可用于标记单个测试(就像您使用的是经典单元测试或其他东西),因此编写 pytest 插件函数的最简单方法是将其应用于每个异步测试:
from trio.testing import trio_test
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
if inspect.iscoroutine(pyfuncitem.obj):
# Apply the @trio_test decorator
pyfuncitem.obj = trio_test(pyfuncitem.obj)
如果你好奇的话,这基本上等同于:
import trio
from functools import wraps, partial
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
if inspect.iscoroutine(pyfuncitem.obj):
fn = pyfuncitem.obj
@wraps(fn)
def wrapper(**kwargs):
trio.run(partial(fn, **kwargs))
pyfuncitem.obj = wrapper
无论如何,这并不能解决你的固定装置问题——因为你需要更多的参与。
我会结合 pytest 和 trio(或 curio,如果这样更容易的话),即将我的测试用例编写为协程函数。通过在 conftest.py
:
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
'''If item is a coroutine function, run it under trio'''
if not inspect.iscoroutinefunction(pyfuncitem.obj):
return
kernel = trio.Kernel()
funcargs = pyfuncitem.funcargs
testargs = {arg: funcargs[arg]
for arg in pyfuncitem._fixtureinfo.argnames}
try:
kernel.run(functools.partial(pyfuncitem.obj, **testargs))
finally:
kernel.run(shutdown=True)
return True
这让我可以像这样编写测试用例:
async def test_something():
server = MockServer()
server_task = await trio.run(server.serve)
try:
# test the server
finally:
server.please_terminate()
try:
with trio.fail_after(30):
server_task.join()
except TooSlowError:
server_task.cancel()
但这是很多样板文件。在非异步代码中,我会将其分解为固定装置:
@pytest.yield_fixture()
def mock_server():
server = MockServer()
thread = threading.Thread(server.serve)
thread.start()
try:
yield server
finally:
server.please_terminate()
thread.join()
server.server_close()
def test_something(mock_server):
# do the test..
有没有办法在 trio 中做同样的事情,即实现异步装置?理想情况下,我会写:
async def test_something(mock_server):
# do the test..
编辑:下面的答案现在大多不相关——而是使用 pytest-trio 并遵循 instructions in its manual.
您的示例 pytest_pyfunc_call
代码不起作用,因为它混合了 trio 和 curio :-)。对于 trio,有一个装饰器 trio.testing.trio_test
可用于标记单个测试(就像您使用的是经典单元测试或其他东西),因此编写 pytest 插件函数的最简单方法是将其应用于每个异步测试:
from trio.testing import trio_test
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
if inspect.iscoroutine(pyfuncitem.obj):
# Apply the @trio_test decorator
pyfuncitem.obj = trio_test(pyfuncitem.obj)
如果你好奇的话,这基本上等同于:
import trio
from functools import wraps, partial
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
if inspect.iscoroutine(pyfuncitem.obj):
fn = pyfuncitem.obj
@wraps(fn)
def wrapper(**kwargs):
trio.run(partial(fn, **kwargs))
pyfuncitem.obj = wrapper
无论如何,这并不能解决你的固定装置问题——因为你需要更多的参与。