在 asyncio 中测试一个永远运行的任务答案

【问题标题】：Testing a forever running task in asyncio在 asyncio 中测试一个永远运行的任务
【发布时间】：2018-01-11 00:43:20
【问题描述】：

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

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

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() 或类似的东西）也不起作用，因为它在那里只是玩具示例测试可以断言一些东西。

【问题讨论】：

标签： python python-3.x pytest python-asyncio

【解决方案1】：

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

由于ook 方法，它总是一个阻塞操作。我假设，既然您使用的是asyncio，那么您希望ook 在主线程上是非阻塞的？

如果是这样，asyncio 实际上带有一个内置的事件循环，请参阅this comment for an example.，它将在另一个线程上运行一个任务并为您提供控制该任务的方法。

事件循环的文档/示例是here

【讨论】：

【解决方案2】：

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

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 ===========================

【讨论】：