在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()或其他一些）不起作用，因爲它只是在那裏玩具示例測試可以斷言的東西。

來源

2017-08-03 Sardathrion

就目前而言，您設計ook方法的方式是造成問題的原因。

由於ook方法，它將始終是一個阻塞操作。我猜想，因爲您使用asyncio，您希望ook在主線程上是非阻塞的？

如果是這種情況，asyncio實際上帶有內置的事件循環，請參閱this comment for an example.，它將在另一個線程上運行任務併爲您提供控制該任務的方法。

文檔/對事件循環樣本here

來源

2017-08-03 12:19:37 duFF

基於關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 ===========================

來源

2017-08-03 13:16:34 Sardathrion

在asyncio中測試永久運行的任務

回答

相關問題