База данных Pytest-django не была откатана с помощью pytest-asyncio

Что касается контекста, я пытаюсь протестировать подключение к WebSocket, установленное через Django, поэтому мне нужно настроить несколько асинхронных тестов с поддержкой базы данных.
Для этого я настроил pytest-django, и у меня возникли некоторые проблемы с пониманием поведения декоратора django_db в асинхронных контекстах.

django_db имеет параметр transaction, значение которого по умолчанию равно False. При этом параметре предполагается, что тестовый класс будет действовать как django.test.TestCase, который запускает все тесты в транзакции, которая откатывается в конце теста.

Однако в асинхронном контексте объекты, созданные в рамках теста, по-видимому, сохраняются в других тестах. Установка @pytest.mark.django_db(transaction=True) действительно приводит к успешному выполнению тестов, но увеличивает продолжительность тестирования по мере внесения фактических изменений в базу данных.

Вот несколько примеров:

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

Я неправильно понял django_db, или есть что-то несовместимое с pytest-asyncio ?

Вместо использования django_db(Transaction=True) вы можете использовать служебную функцию sync_to_async:

Транзакции пока не работают в асинхронном режиме. Если у вас есть фрагмент кода, который требует управления транзакциями, мы рекомендуем вам написать этот фрагмент как единую синхронную функцию и вызвать его с помощью sync_to_async().

В кодовой базе Django и их асинхронных тестах они используют @async_to_sync и sync_to_async() в некоторых местах:

# 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):
    ...
Вернуться на верх