大家
我可能遗漏了一些关于如何在一个测试过程中使用celery 的pytest fixture的内容，这个测试涉及到在一个http请求过程中执行一个任务。https://docs.celeryq.dev/en/v5.3.4/userguide/testing.html#pytest
使用'celery _session_app'和'celery _session_worker' fixture，我可以直接调用任务，并使用wait()或get()等待任务执行。

from django.core import mail
from django.test import TestCase
from polls.tasks import task_send_mail

@pytest.mark.usefixtures('celery_session_app')
@pytest.mark.usefixtures('celery_session_worker')
class SendMailTestCase(TestCase):
    def test_send_mail(self):
        task: AsyncResult = task_send_mail.delay(
            email_address='[email protected]',
            message='Hello'
        )

        # waits for task execution that just sends a simple mail using
        # from django.core.mail import send_mail
        task.get()

        # it passes because the code waited for mail send
        self.assertEqual(len(mail.outbox), 1)

字符串
但是，如果我在django视图中调用了同一个任务，那么这个任务将以同步的方式执行，并且测试失败，因为assert在邮件发送之前就已经被测试过了。

# view

from rest_framework import serializers
from rest_framework.generics import ListAPIView
from rest_framework.response import Response
from django.contrib.auth.models import Group
from polls.tasks import task_send_mail

class GroupSerializer(serializers.ModelSerializer):
    class Meta:
        model = Group
        fields = ['__all__']

class GroupView(ListAPIView):
    http_method_names = ["get"]
    queryset = Group.objects.all()
    serializer_class = GroupSerializer

    def get(self, request, *args, **kwargs):
        # task is sent to async execution
        task = task_send_mail.delay(
            email_address='[email protected]',
            message='hello'
        )
        return Response(
            status=200,
            data={}
        )

型
当pytest执行测试时...

from django.core import mail
from django.test import TestCase
from polls.tasks import task_send_mail

@pytest.mark.usefixtures('celery_session_app')
@pytest.mark.usefixtures('celery_session_worker')
class SendMailTestCase(TestCase):

    def test_request_url(self):
        response = self.client.get(
            reverse('groups')
        )
        self.assertEqual(
            response.status_code,
            200
        )
        self.assertEqual(
            len(mail.outbox),
            1
        )

型
它返回失败的结果AssertionError: 0 != 1，因为任务被发送到异步执行，并且在Assert时未完成。
我发现解决此问题的唯一方法是更改conftest.py文件中的celery 配置固定装置：

import pytest

@pytest.fixture(scope='session')
def celery_config():
    return {
        'task_always_eager': True
    }

型
是否有一种方法可以在不设置“task_always_eager”的情况下始终等待异步任务执行？
当我使用celery_session_app和celery_session_worker fixture作为TestCase类的装饰器时，我希望代码在保持代码执行之前，等待每个发送到执行的异步任务完成。