Reputation: 5048
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.
Upvotes: 1
Views: 1743
Reputation: 5048
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)
Upvotes: 1
Reputation: 125
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.
Upvotes: 1