测试芹菜任务答案

【问题标题】：Testing celery task测试芹菜任务
【发布时间】：2017-10-14 04:18:01
【问题描述】：

我设置了多租户系统。设置了带有 RabbitMQ 的 Celery。我有一个系统可以在模型的预保存信号处触发电子邮件。代码如下：

@receiver(pre_save, sender=Student)
def invite_stu(sender, instance, **kwargs):
    user_email = instance.email
    subject = 'dummy'
    message = ''

    html_message = 'dummy'
    from_email = my email id 
    to_list = [user_email]
    send_html_mail(subject, message, html_message, settings.DEFAULT_FROM_EMAIL, to_list)

    # super(Employee, self).save()
    return instance

我正在为此编写一个测试用例：

class BaseSetup(TenantTestCase):
    def setup_tenant(self, tenant):
        """
        Add any additional setting to the tenant before it get saved. This is required if you have
        required fields.
        """
        user = User.objects.create(email="dummy@dummy.com", is_active=True)
        user.set_password('dummy')
        tenant.owner = user
        user.save()

    def setUp(self):
        self.sync_shared()
        tenant_model = get_tenant_model()
        app.conf.update(CELERY_ALWAYS_EAGER=True)
        app.conf.update(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True)
        print app.conf.CELERY_ALWAYS_EAGER
        test_schema_name = self.get_test_schema_name()
        test_tenant_domain_name = self.get_test_tenant_domain()
        self.tenant = tenant_model.objects.filter(schema_name=test_schema_name).first()
        self.c = TenantClient(self.tenant)

    def test_core_student_put_api(self):
        response = self.c.post('http://test.localhost:8000/login/',{'email':'admin@dummy.com','password':'dummy'})
        response_add_student=self.c.post('http://test.localhost:8000/student_admin/',{'email':'xxx@dummy.com','stu_number':'100','role':'STUDENT'},**{'HTTP_AUTHORIZATION':'JWT '+response.data['token']})

        self.assertEqual(response_put_employee.status_code, 201)

测试运行良好，甚至可以通过邮件功能进行测试。但是如果我通过测试运行芹菜，它不会启动和发送邮件，但如果我通过 API 添加一个学生，它会发送电子邮件。

有什么方法可以解决这个问题吗？谢谢！

【问题讨论】：

标签： django django-rest-framework rabbitmq celery

【解决方案1】：

我找到了解决方案。我的任务是一项定期任务，在 django 中进行测试会在测试期间创建一个新的测试数据库，我发现邮件队列用于存储在该数据库中，而 celery 无法从该测试数据库中获取。所以，我必须明确调用我的测试用例中的任务，它看起来像这样。

@override_settings(task_eager_propagates=True,task_always_eager=True,broker_url='memory://',backend='memory')
def test_core_student_put_api(self):
        response = response = self.c.post('http://test.localhost:8000/login/',{'email':'admin@dummy.com','password':'dummy'})
        response_add_student=self.c.post('http://test.localhost:8000/student_admin/',{'email':'xxx@dummy.com','stu_number':'100','role':'STUDENT'},**{'HTTP_AUTHORIZATION':'JWT '+response.data['token']})

        self.assertTrue(project_tasks.delay())
        self.assertEqual(response_put_employee.status_code, 200)

我必须按照 celery 文档中的建议覆盖 celery 设置以进行测试配置。我已经用 assertTrue 明确地调用了 project_tasks.delay()。我的任务包含一个简单的 send_mail() 命令。它也在从我的信号中获取正确的数据。invite_stu 信号调用 send_html_mail() ，它添加到测试数据库中的邮件队列中，当我明确调用我的周期性任务（这是一个周期性任务）时发送邮件。我在某处读到由于临时数据库问题，celery 无法为测试用例执行定期任务，应该对此进行更多研究，但目前这解决了我的问题。

【讨论】：