Pytest-django database not rolled back with pytest-asyncio

For context, I am trying to test a WebSocket connection made through Django, so I need to set up some async tests with database support.
To do so, I have set up pytest-django and I have some trouble understanding the django_db decorator's behavior in async contexts.

django_db has a parameter transaction which defaults to False. With this setting, the test class is supposed to act as django.test.TestCase which runs all tests in a transaction which is rolled back at the end of the test.

However, in an async context, the objects created within a test seem to persist in other tests. Setting @pytest.mark.django_db(transaction=True) does make the tests pass, but increases test duration as actual modifications are made to the database.

Here are quick examples:

import pytest
from cameras.models import CameraGroup as MyObject


@pytest.mark.django_db
@pytest.mark.asyncio
class TestAsync:
    async def test_1(self):  # OK
        await MyObject.objects.acreate(name="same-name1")
        assert await MyObject.objects.acount() == 1

    async def test_2(self):  # FAILS
        # This should not see test_1's object if rollback worked
        assert await MyObject.objects.acount() == 0


@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
class TestAsyncWithTransaction:
    async def test_1(self):  # OK
        await MyObject.objects.acreate(name="same-name2")
        assert await MyObject.objects.acount() == 1

    async def test_2(self):  # OK
        # This should not see test_1's object if rollback worked
        assert await MyObject.objects.acount() == 0


@pytest.mark.django_db
class TestSync:
    def test_1(self):  # OK
        MyObject.objects.create(name="same-name3")
        assert MyObject.objects.count() == 1

    def test_2(self):  # OK
        # This should not see test_1's object if rollback worked
        assert MyObject.objects.count() == 0

Have I misunderstood django_db, or is there something incompatible with pytest-asyncio ?

Instead of using django_db(Transaction=True), you can use the sync_to_async utility function:

Transactions do not yet work in async mode. If you have a piece of code that needs transactions behavior, we recommend you write that piece as a single synchronous function and call it using sync_to_async().

In the Django codebase and their async tests, they're using @async_to_sync and sync_to_async() in some spots:

# used within a test, for a specific part of the code
await sync_to_async(SimpleModel.objects.create)(
    field=4,
    created=datetime(2022, 1, 1, 0, 0, 0),
)

@skipUnlessDBFeature("has_bulk_insert")
@async_to_sync
async def test_abulk_create(self):
    ...
Вернуться на верх