[pytest-django] Database is not properly rolled back after async tests

[JakeUrban]

When using pytest-django's pytest.mark.django_db marker in conjunction with pytest.mark.asyncio, any writes to the database are not rolled back when the test completes and affect subsequent tests.

To test, I created a fresh django app, installed pytest-django and pytest-asyncio, created a simple model, and wrote two tests:

models.py:

from django.db import models

class MyModel(models.Model):
    my_field = models.TextField(null=True, blank=True)

test_app.py:

import pytest
from asgiref.sync import sync_to_async

from app.models import MyModel


@pytest.mark.django_db
@pytest.mark.asyncio
async def test_save_model_to_db():
    await sync_to_async(MyModel.objects.create)()


@pytest.mark.django_db
def test_check_if_model_present():
    assert MyModel.objects.count() == 0

Running pytest resulted in the first test succeeding and the second failing:

    @pytest.mark.django_db
    def test_check_if_model_present():
>       assert MyModel.objects.count() == 0
E       assert 1 == 0

At first I thought it might be due to how Django runs synchronous code from an asynchronous context. Maybe it was creating the instance in another transaction that pytest wasn't rolling back.

So, I updated the test code to set DJANGO_ALLOW_ASYNC_UNSAFE to "true" so I could remove the sync_to_async() wrapper around the create() call, ensuring the instance would be included created in the same thread and within pytest's transaction.

Unfortunately, the test failed for the same reason. This could be an issue with pytest-asyncio or compatibility with pytest-django.

[JakeUrban]

Looks like there is an issue filed on pytest-django too: pytest-dev/pytest-django#580
This also could be a django-specific, non-pytest related problem: https://code.djangoproject.com/ticket/32409

[pcraciunoiu]

Is there any update on this? I'm definitely seeing data persist between tests after converting them to async with pytest fixtures. E.g. if one test changes a user's password, the next test starts with that changed password.

It seems that switching to pytest.mark.django_db(transaction=True) fixes the leak, but that slows down the tests A LOT (like 3x), so it's not a good solution for large test suites.

Has anyone found a better way to roll back the data between tests?

[scastlara]

@pcraciunoiu I am using this "fix" in my codebase: django/channels#1091 (comment)

Even if the thread is about Django channels, that was good enough to fix this issue for me (i am not using channels at all, just Django with pytest-asyncio and pytest-django).

[seifertm]

My current understanding of the issue is that database transactions are not properly rolled by when using asgiref.sync.sync_to_async. This is also the general issue discussed in the Django channels thread linked by @scastlara.

sync_to_async executes the wrapped function in a thread pool, in order to prevent blocking the event loop. Under the hood, it seems to retrieve the running event loop and to submit the wrapped callable via loop.run_in_executor.

Pytest-asyncio neither sets nor modifies the default event loop executor. It's currently unclear to me how pytest-asyncio can address this issue.

By default, pytest-asyncio provides a new event loop for each test case. The code provided by the OP of this (pytest-asyncio) does not overwrite the event_loop fixture. Hence, the tests use the default function-scoped event loop. As such, I don't see how sync_to_async is relevant for the test.

If anyone can provide a more complete reproducer and/or explain how the issue is related to pytest-asyncio, I'm happy to look into it.

[saxelsen]

I just encountered this issue, but seems to be resolved by explicitly marking the test cases with a transaction: @pytest.mark.django_db(transaction=True)

Worked for me, at least.

[pcraciunoiu]

@saxelsen that does work, but as noted above, it slows down tests A LOT. Not really a good solution if you have lots of tests.

It seems the fix from the django-channels thread broke with asgiref v3.8, so I'm pinning that to continue using it, for now.

Has anyone found a better solution? Without this patch + without marking them transactional, the tests just hang for me.

[scastlara]

Has anybody found a solution for this? With asgiref 3.8, the fix mentioned here #226 (comment) does not work, and without it, the database does not teardown correctly.

[scastlara]

Alright! We (@sacha-c and me) worked on this hack and it works for our tests.

We share it here, but please if you are going to use this, don't be mad at us if something bigger breaks!

@pytest.fixture(autouse=True)
def fix_async_db(request):
    """
    If you don't use this fixture for async tests that use the ORM/database
    you won't get proper teardown of the database.
    This is a bug somehwere in pytest-django, pytest-asyncio or django itself.

    Nobody knows how to solve it, or who should solve it.
    Workaround here: https://github.com/django/channels/issues/1091#issuecomment-701361358
    More info:
    https://github.com/pytest-dev/pytest-django/issues/580
    https://code.djangoproject.com/ticket/32409
    https://github.com/pytest-dev/pytest-asyncio/issues/226


    The actual implementation of this workaround constists on ensuring
    Django always returns the same database connection independently of the thread
    the code that requests a db connection is in.

    We were unable to use better patching methods (the target is asgiref/local.py),
    so we resorted to mocking the _lock_storage context manager so that it returns a Mock.
    That mock contains the default connection of the main thread (instead of the connection
    of the running thread).

    This only works because our tests only ever use the default connection, which is the only thing our mock returns.

    We apologize in advance for the shitty implementation.
    """
    if request.node.get_closest_marker("asyncio") is None or request.node.get_closest_marker("django_db") is None:
        # Only run for async tests that use the database
        yield
        return

    main_thread_local = connections._connections
    for conn in connections.all():
        conn.inc_thread_sharing()

    main_thread_default_conn = main_thread_local._storage.default
    main_thread_storage = main_thread_local._lock_storage

    @contextlib.contextmanager
    def _lock_storage():
        yield mock.Mock(default=main_thread_default_conn)

    try:
        with patch.object(main_thread_default_conn, "close"):
            object.__setattr__(main_thread_local, "_lock_storage", _lock_storage)
            yield
    finally:
        object.__setattr__(main_thread_local, "_lock_storage", main_thread_storage)