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

问题描述：

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

这里是一个玩具例如模仿真正的代码：

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

答

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

由于ook方法，它将始终是一个阻塞操作。我猜想，因为您使用asyncio，您希望ook在主线程上是非阻塞的？

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

文档/对事件循环样本here

答

基于关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 ===========================

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

相关推荐