在 asyncio 中测试永远的运行任务

Question

我需要每秒（比如说）调用一个任务来轮询硬件上的一些传感器数据。在单元测试中，我想做的就是检查是否调用了正确的方法，是否确实捕获了错误（例如传感器已爆炸或消失）。

这是一个模仿真实代码的玩具示例：

import pytest
import asyncio
import mock


async def ook(func):
    while True:
        await asyncio.sleep(1)
        func()


@pytest.mark.asyncio
async def test_ook():
    func = mock.Mock()
    await ook(func)
    assert func.called is True

正如预期的那样，运行这将永远阻塞。

如何取消 ook 任务以使单元测试不阻塞？

解决方法是将循环拆分为另一个函数并将其定义为不可测试。我想避免这样做。另请注意，弄乱 func（调用 loop.close() 或类似的东西）也不起作用，因为它在那里只是为了让玩具示例测试可以断言一些东西。

Answer 1

就目前而言，您设计 ook 方法的方式是造成问题的原因。

由于ook方法，它总是一个阻塞操作。我假设，因为你正在使用 asyncio 你希望 ook 在主线程上是非阻塞的？

如果是这样，asyncio 实际上内置了一个事件循环，请参阅，它将运行另一个线程上的任务并为您提供控制它的方法任务。

Documentation/samples 对于事件循环是 here

Answer 2

基于duFF's ，这里是固定的玩具代码：

import pytest
import asyncio
import mock


async def ook(func):
    await asyncio.sleep(1)
    func()
    return asyncio.ensure_future(ook(func))


@pytest.mark.asyncio
async def test_ook():
    func = mock.Mock()
    task = await ook(func)
    assert func.called is True
    task.cancel()

当运行:

; py.test tests/ook.py
============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.1.3, py-1.4.34, pluggy-0.4.0           
run-last-failure: rerun last 4 failures first                                   
rootdir: /home/usr/blah, inifile: setup.cfg                             
plugins: xvfb-1.0.0, xdist-1.18.2, colordots-0.1, asyncio-0.6.0                 
collected 1 item s 

ook.py::test_ook PASSED

---------- generated xml file: /home/yann/repos/raiju/unit_tests.xml -----------
============================== 0 tests deselected ==============================
=========================== 1 passed in 0.02 seconds ===========================

在 asyncio 中测试永远的运行任务

Testing a forever running task in asyncio

python

pytest

python-3.x

python-asyncio

在 asyncio 中测试永远的 运行 任务

Testing a forever running task in asyncio

python

pytest

python-3.x

python-asyncio

在 asyncio 中测试永远的运行任务