[Python] Django testing with pytest-asyncio

Discussão em 'Python' iniciado por Stack, Abril 15, 2025.

  1. Stack

    Stack Membro Participativo

    I've just learned how to use pytest for tests in Django/DRF-projects and successfully implemented a bunch of test functions. But, though they work OK, they work pretty slowly; and, for the sake of increasing their work speed, as well as for my personal educational and "sport" interest, I decided to try to implement them asynchronously. For that sake, I have found pytest-asyncio package.

    While I used pytest-asyncio the way it was described in its docs, when trying to activate tests they returned me an error (django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async). I've read about this error and learned that Django ORM operations block the event loop and thus they have to operate via some workaround (threading or asgiref.sync_to_async). While I faced this error connected not only with Django ORM operations, but also with my response request methods (get/post), still I've tried to write this workaround and looked whether the reaction would have changed.

    It hasn't changed. And now nothing I have googled and read through helps me.

    The testing module tests downloading image files functionality and is set with API client and user (the first one has authorised and non-authorised versions). I'll mention here only one test of the whole bunch.

    @pytest_asyncio.fixture(loop_scope="module")
    async def current_loop():
    yield asyncio.get_running_loop()


    @pytest.mark.django_db
    @pytest.fixture
    async def user() -> User:
    """ Creating test user """

    yield User.objects.create_superuser(username='test.user', password='user.password')


    @pytest.mark.django_db
    @pytest_asyncio.fixture
    async def auth_client(user: User) -> Iterator[APIClient]:
    """ Authorised client session """

    client = await sync_to_async(APIClient())()
    await sync_to_async(client.force_login(user))()
    yield client


    @pytest.mark.django_db
    @pytest_asyncio.fixture
    async def non_auth_client(user: User) -> Iterator[APIClient]:
    """ Non-authorised client """

    client = await sync_to_async(APIClient())()
    yield client


    @pytest_asyncio.fixture
    async def model(current_loop):
    model_case = await sync_to_async(document_models.LicensesAndCertificatesImages)()
    yield model_case
    # teardown
    for model_object in model_case.objects.all():
    model_object.delete()


    @pytest_asyncio.fixture
    async def post_url():
    post_url_case = reverse('admin_site:documents_licenses-upload')
    yield post_url_case


    @pytest_asyncio.fixture
    async def get_list_url():
    get_list_url_case = reverse('admin_site:documents_licenses-list')
    return get_list_url_case


    @pytest.mark.django_db
    @pytest_asyncio.fixture
    async def prepare_models_objects(user, model):
    """ Filling DB with model objects for GET-requests testing """

    uploaded_by = await sync_to_async(get_user_model().objects.first())()

    for i in range(5):
    file = SimpleUploadedFile(
    name=f'file_{i}.png',
    content=b'correct_file_content',
    content_type='image/png',
    )
    await model.objects.create(document=file, uploaded_by=uploaded_by)
    yield


    """ Tests """


    @pytest.mark.django_db
    async def test_get_list(prepare_models_objects, model, auth_client, get_list_url):
    """ Check if GET-request receives License and Sertificates list """

    response: Response = await sync_to_async(auth_client.get(path=get_list_url))()

    assert response.status_code == 200
    assert len(response.data) == 5
    assert list(response.data[0].keys()) == ['id', 'document']


    After pytest activation command every test returns me django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async. In sync-version the test works well (meaning that all other tests and fixtures are also in their sync-versions):

    @pytest.mark.django_db
    def test_get_list(prepare_models_objects, model, auth_client, get_list_url):
    response: Response = auth_client.get(path=get_list_url)

    assert response.status_code == 200
    assert len(response.data) == 5
    assert list(response.data[0].keys()) == ['id', 'document']


    Can someone explain me what I am doing wrong?

    Continue reading...

Compartilhe esta Página