How to start an asyncio background task in unit test setUpModule?

user_412896 0
Viewed 79

In an E2E test I need to start an asyncio background task (a simulator that connects to the server under test) which will be used by all the tests in the module. The setUpModule would be the natural place to start it, but how can I start an asyncio background task from there?

To my understanding, the background task would need to be in the same asyncio event loop that runs the individual tests. Is this possible? I'm hoping to avoid threads / subprocesses.

I'm currently using unittest framework but can change that if another test framework supports this.

python-unittest python-asyncio python
2 Answers

One solution is using pytest with the pytest-asyncio addon and performing this action using an auto-used module-level fixture:

@pytest_asyncio.fixture(autouse=True, scope="module")
async def setup_foo():
    foo_task = asyncio.create_task(run_foo())
    yield
    foo_task.cancel()
    with suppress(asyncio.CancelledError):
        await foo_task

Note that if you run this as-is, you will get the error ScopeMismatch: You tried to access the function scoped fixture event_loop with a module scoped request object, involved factories. To fix this, you need to configure all tests within the module to run using the same event loop:

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

Putting this all together, we have the following full example that can be run with pytest test_example.py:

import pytest
import pytest_asyncio

import asyncio
from contextlib import suppress


# Function to be executed asynchronounsly in background during module tests
async def run_foo():
    while True:
        await asyncio.sleep(10)


# Function run once per module to execute run_foo in background
@pytest_asyncio.fixture(autouse=True, scope="module")
async def setup_foo():
    foo_task = asyncio.create_task(run_foo())
    yield  # Runs the rest of your module tests
    foo_task.cancel()
    with suppress(asyncio.CancelledError):
        await foo_task


# Configures the event loop to be created only once per module
@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


# Actual test case run within the same module-level event loop
@pytest.mark.asyncio
async def test_app():
    await asyncio.sleep(3)

Note that the module-level setup code can all be extracted into a conftest.py file in the same directory as your test_foo.py files.

Read more about Pytest fixtures here.

user_136768820

Apparently it cannot currently be done on module / class level using the unittest framework. Here is an example of starting the background task in asyncSetUp (which is run before each test). The setup is extracted in a mixin so it can be easily applied to different tests.

The background tasks are automatically canceled by the framework, so it doesn't need to be done manually.

(I'll accept the pytest answer, as it answers the module-level question.)

import asyncio
import itertools
import unittest

async def bg_task():
    for n in itertools.count():
        print(n)
        await asyncio.sleep(0.25)

class BgTaskMixin:
    async def asyncSetUp(self):
        await super().asyncSetUp()
        asyncio.create_task(bg_task())

class FooTest(
    BgTaskMixin,
    unittest.IsolatedAsyncioTestCase,
):
    async def test_bg(self):
        print('TEST START')
        await asyncio.sleep(2)
user_4128960
Related