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()
.
# 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):
...