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

Issue

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.

Solution

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.

Answered By - user13676882