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には実際にはイベントループが組み込まれています。this comment for an example.を参照してください。別のスレッドでタスクを実行し、そのタスクを制御する方法を示します。

ドキュメント/イベントループのためのサンプルですhere

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

は、