CODEWITHSUNDEEP
pythonpython-3.xpytestpython-asyncio
Sardathrion - against SE abuse
Sardathrion - against SE abuse

Reputation: 18246

Testing a forever running task in asyncio

I need to call a task every second (say) to poll some sensor data on a piece of hardware. In the unit test, all I want to do, is to check that the right methods are called and that errors (such as sensor has blown up or vanished) do get caught.

Here is a toy example that mimics the real code:

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

As expected, running this will block forever.

How can I cancel the ook task so that the unit test does not block?

A work around would be to split the loop into another function and define it as no testable. I want to avoid doing that. Note also that messing with func (to call loop.close() or some such) does not work either as it is there just so the toy example test can assert something.

Upvotes: 2

Views: 2699

Answers (2)

Sardathrion - against SE abuse
Sardathrion - against SE abuse

Reputation: 18246

Based off duFF's answer, here is the fixed toy code:

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()

When run:

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

Upvotes: 3

duFF
duFF

Reputation: 622

As it stands, the way you've designed the ook method is the cause of your issue.

Due to the ook method, it will always be a blocking operation. I assume, since you're using asyncio you want ook to be non-blocking on the main thread?

If thats the case, asyncio actually comes with an event loop built in, see this comment for an example., which will run an task on another thread and gives you ways to control that task.

Documentation/samples for the event loop are here

Upvotes: 4

Related Questions