Почему в асинхронных тестах Django «течёт» БД и как это исправить

Разбираем, почему django_db не изолирует асинхронные тесты Django, как фиксить протечки БД без transaction=True с помощью sync_to_async. Примеры pytest-django

Асинхронные тесты Django с доступом к базе данных нередко удивляют даже опытные команды: строки, созданные в одном тесте, «просачиваются» в следующий, несмотря на использование маркера django_db из pytest-django. Если ваш стек WebSocket или ASGI опирается на асинхронные пути, это руководство объясняет, почему так происходит и как это исправить, не переходя на более медленные прогоны с полноценными транзакциями.

Воспроизводим проблему

Ниже приведены тесты на pytest-django и pytest-asyncio. Ожидание простое: внутри одного теста данные видны; между тестами они очищаются откатом. Именно это в режиме по умолчанию должен обеспечивать django_db.

import pytest
from cameras.models import CameraGroup as GroupUnit
@pytest.mark.django_db
@pytest.mark.asyncio
class AsyncSuite:
    async def test_a(self):  # ОК
        await GroupUnit.objects.acreate(name="alpha1")
        assert await GroupUnit.objects.acount() == 1
    async def test_b(self):  # ПАДАЕТ
        # При откате записи из test_a не должны быть видны
        assert await GroupUnit.objects.acount() == 0
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
class AsyncSuiteTxn:
    async def test_a(self):  # ОК
        await GroupUnit.objects.acreate(name="alpha2")
        assert await GroupUnit.objects.acount() == 1
    async def test_b(self):  # ОК
        # Здесь изоляция сохраняется, но тесты медленнее
        assert await GroupUnit.objects.acount() == 0
@pytest.mark.django_db
class SyncSuite:
    def test_a(self):  # ОК
        GroupUnit.objects.create(name="alpha3")
        assert GroupUnit.objects.count() == 1
    def test_b(self):  # ОК
        # Синхронный путь сохраняет изоляцию за счёт отката
        assert GroupUnit.objects.count() == 0

Что происходит

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

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

Иными словами, транзакционная изоляция — страховка, которая очищает изменения между тестами — не сработает для полностью асинхронных вызовов ORM. В результате объекты, созданные через acreate и родственные асинхронные методы queryset, могут сохраняться и всплывать в последующих тестах. Использование django_db(transaction=True) действительно работает, но при этом выполняются реальные коммиты и уборка идёт медленнее, что увеличивает общее время прогона.

Решение, которое сохраняет изоляцию без включения полноценных транзакций

Выполните часть, которой нужны транзакционные семантики, в синхронной функции и вызовите её из асинхронного теста через asgiref.sync.sync_to_async. Собственные асинхронные тесты Django тоже точечно используют sync_to_async и async_to_sync.

import pytest
from asgiref.sync import sync_to_async, async_to_sync
from cameras.models import CameraGroup as GroupUnit
@pytest.mark.django_db
@pytest.mark.asyncio
class AsyncSuiteFixed:
    async def test_a(self):
        # Вызовы ORM выполняются синхронно, чтобы воспользоваться откатом транзакции
        await sync_to_async(GroupUnit.objects.create)(name="alpha1")
        assert await sync_to_async(GroupUnit.objects.count)() == 1
    async def test_b(self):
        # Изоляция восстановлена, потому что предыдущие изменения были внутри синхронной транзакции
        assert await sync_to_async(GroupUnit.objects.count)() == 0
# Либо местами можно сменить направление
# прогнав всё тело асинхронного теста через async_to_sync.
# @async_to_sync
# async def test_bulk_like_case():
#     ... и здесь использовать sync_to_async(...) при обращениях к ORM ...

Почему это важно

Асинхронные наборы тестов без корректной изоляции хрупки. Протечки данных между тестами приводят к плавающим сбоям, маскируют настоящие дефекты и замедляют отладку. Если команда проверяет WebSocket-потоки или другую ASGI-логику, важно добиться надёжной изоляции при работе с Django ORM — это даёт детерминизм и скорость.

Практические выводы

Если вам нужна транзакционная семантика в асинхронных тестах, выполняйте части, взаимодействующие с базой, синхронно через sync_to_async. Так сохраняется поведение с откатом, знакомое по синхронным тестам, без издержек на запуск всего класса с transaction=True. Если хотите путь попроще и готовы к более долгим прогонам, включение django_db(transaction=True) тоже сработает, как указано выше, но ожидайте более медленного выполнения из‑за реальных коммитов.

В любом случае осознанно выбирайте, где исполняются вызовы ORM. Те места, которым нужна опора на транзакции, держите внутри синхронной границы и вызывайте её из асинхронных тестов через sync_to_async. Так тесты остаются быстрыми, изолированными и предсказуемыми.