База данных 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().
# 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):
...