Testing a forever running task in asyncio

Question

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.

duFF · Accepted Answer · 2017-08-03 12:19:37Z

4

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

answered Aug 3, 2017 at 12:19

duFF

6221 gold badge5 silver badges15 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Sardathrion - against SE abuse · Accepted Answer · 2017-08-03 13:16:34Z

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

Collectives™ on Stack Overflow

Testing a forever running task in asyncio

2 Answers 2

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related