Set a shared event loop per test case

Using Django Getting Started
1

Hi all,

Is there a way to use the same event loop for all the tests in a class? We are facing an issue related to using the Strict Redis async client (we don’t use Django Redis client). Currently, we need to create a new Redis client for each test because Django creates a different event loop. This behavior is similar to IsolatedAsyncioTestCase. However, we would prefer to use a single event loop shared by all tests, eliminating the need to create multiple Redis clients.

Has anyone faced this issue?

Kind regards!

2

(This probably belngs in the “Using Django” category?)

Firstly, this is why your tests are running in their own event loops.

Over in django/test/testcases.py::SimpleTestCase._setup_and_call:


        # Convert async test methods.
        if iscoroutinefunction(testMethod):
            setattr(self, self._testMethodName, async_to_sync(testMethod))

async tests are just wrapped in async_to_sync (which spins up its own event loop at the point of decoration)

        # from AsyncToSync.__init__
        self.main_event_loop = None
        try:
            self.main_event_loop = asyncio.get_running_loop()
        except RuntimeError:
            # There's no event loop in this thread.
            pass

One idea: figure out how to get an event loop set up by the time test collection by the test runner is happening. I don’t know what would make sense there, and it’d be a hack.

Another idea: write yourself a decorator for your async tests. Bit silly but would at least be a touch more principled.

import asyncio
import functools

test_loop = asyncio.new_event_loop()

def share_event_loop(test):
    @functools.wraps(test)
    def wrapped(*args, **kwargs):
        coro = test(*args, **kwargs)
        return test_loop.run_until_complete(coro)
    return wrapped

Then in your test case:

class MyTestCase(TestCase):
    @share_event_loop
    async def test_some_async_thing(self):
        ...

This likely doesn’t handle much though.

3

At first glance, I think it could also be an acceptable change in Django to share the same event loop across a whole test process.

1 Like
4

Hi, I faced the same issue.
This is because by default, get_running_loop() returns different objects between test cases.

Using the sample code from @rtpg , I was able to run tests on a common event loop using the following decorator, which can be used on the TestCase class.

So let me introduce my code :slight_smile:

import asyncio
from collection.abc import Callable
import functools
import inspect
from typing import Any, TypeVar

T = TypeVar("T", bound=type[Any])


try:
    default_testing_loop = asyncio.get_running_loop()
except RuntimeError:
    default_testing_loop = asyncio.new_event_loop()


def shared_event_loop(
    target: T | None = None,
    *,
    loop: asyncio.AbstractEventLoop = default_testing_loop,
) -> T | Callable[[T], T]:

    def _decorator(_target: T) -> T:
        if not inspect.isclass(_target):
            raise TypeError("@shared_event_loop can only be applied to classes.")

        for name, method in inspect.getmembers(_target, inspect.isfunction):
            if (name.startswith("test_") or name.endswith("_test")) and inspect.iscoroutinefunction(method):

                @functools.wraps(method)
                def wrapped_method(self, *args: Any, __method=method, **kwargs: Any) -> Any:
                    return loop.run_until_complete(__method(self, *args, **kwargs))

                setattr(_target, name, wrapped_method)

        return _target

    if target is None:
        return _decorator

    return _decorator(target)

Usage:

from django.test import TestCase


@shared_event_loop
class TestClass(TestCase):
    async def test_1(self):
        assert asyncio.get_running_loop() is default_testing_loop

    async def test_2(self):
        ...


# Custom event loop object
my_loop = asyncio.new_event_loop()

@shared_event_loop(loop=my_loop)
class TestClass(TestCase):
    async def test_3(self):
        assert asyncio.get_running_loop() is my_loop

Thank you!