使 Django 测试用例数据库对 Celery 可见

Question

当 Django 测试用例运行时，它会创建一个隔离的测试数据库，以便在每个测试完成时回滚数据库写入。我正在尝试使用 Celery 创建一个集成测试，但我不知道如何将 Celery 连接到这个临时测试数据库。在幼稚的设置中，保存在 Django 中的对象对 Celery 是不可见的，而保存在 Celery 中的对象会无限期地存在。

这是一个示例测试用例：

import json
from rest_framework.test import APITestCase
from myapp.models import MyModel
from myapp.util import get_result_from_response

class MyTestCase(APITestCase):
    @classmethod
    def setUpTestData(cls):
        # This object is not visible to Celery
        MyModel(id='test_object').save()

    def test_celery_integration(self):
        # This view spawns a Celery task
        # Task should see MyModel.objects.get(id='test_object'), but can't
        http_response = self.client.post('/', 'test_data', format='json')

        result = get_result_from_response(http_response)
        result.get()  # Wait for task to finish before ending test case
        # Objects saved by Celery task should be deleted, but persist

我有两个问题：

1. 如何使 Celery 可以看到 Django 测试用例的对象？

2. 如何确保 Celery 保存的所有对象在测试完成后自动回滚？

如果无法自动执行此操作，我愿意手动清理对象，但删除 tearDown 中的对象，即使是在 APISimpleTestCase 中似乎也会回滚。

Answer 1

这可以通过在 Django 测试用例中启动一个 Celery worker 来实现。

背景

Django 的内存数据库是 sqlite3。正如它在the description page for Sqlite in-memory databases 上所说，“[A]ll 共享内存数据库的所有数据库连接都需要在同一个进程中。”这意味着，只要 Django 使用内存中的测试数据库，并且 Celery 是在单独的进程中启动的，那么 Celery 和 Django 根本不可能共享一个测试数据库。

但是，使用celery.contrib.testing.worker.start_worker，可以在同一进程内的单独线程中启动 Celery 工作程序。该工作人员可以访问内存数据库。

这假设 Celery 已经在 the usual way 中与 Django 项目一起设置。

解决方案

因为 Django-Celery 涉及到一些跨线程通信，所以只有不在隔离事务中运行的测试用例才会起作用。测试用例必须直接从SimpleTestCase 或其等效的APISimpleTestCase 继承，并将databases 设置为'__all__' 或只是与测试交互的数据库。

关键是在TestCase的setUpClass方法中启动一个Celery worker，在tearDownClass方法中关闭它。关键函数是celery.contrib.testing.worker.start_worker，需要当前Celery应用的一个实例，大概是从mysite.celery.app获取的，返回一个PythonContextManager，里面有__enter__和__exit__方法，必须在@987654336中调用@ 和 tearDownClass，分别。可能有一种方法可以避免使用装饰器或其他东西手动输入和存在ContextManager，但我无法弄清楚。这是一个示例tests.py 文件：

from celery.contrib.testing.worker import start_worker
from django.test import SimpleTestCase

from mysite.celery import app

class BatchSimulationTestCase(SimpleTestCase):
    databases = '__all__'

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        # Start up celery worker
        cls.celery_worker = start_worker(app, perform_ping_check=False)
        cls.celery_worker.__enter__()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()

        # Close worker
        cls.celery_worker.__exit__(None, None, None)

    def test_my_function(self):
        # my_task.delay() or something

无论出于何种原因，测试工作者都会尝试使用名为'celery.ping' 的任务，可能是为了在工作者失败的情况下提供更好的错误消息。它正在寻找的任务是celery.contrib.testing.tasks.ping，在测试时不可用。将start_worker 的perform_ping_check 参数设置为False 会跳过对此的检查并避免相关的错误。

现在，运行测试时，无需启动单独的 Celery 进程。 Celery worker 将在 Django 测试过程中作为单独的线程启动。该工作人员可以查看任何内存数据库，包括默认的内存测试数据库。要控制工作人员的数量，start_worker 中提供了一些选项，但默认情况下似乎是单个工作人员。

Answer 2

我根据@drhagen 的解决方案找到了另一种解决方法：

在拨打start_worker(app)之前先拨打celery.contrib.testing.app.TestApp()

from celery.contrib.testing.worker import start_worker
from celery.contrib.testing.app import TestApp

from myapp.tasks import app, my_task


class TestTasks:
    def setup(self):
        TestApp()
        self.celery_worker = start_worker(app)
        self.celery_worker.__enter__()

    def teardown(self):
        self.celery_worker.__exit__(None, None, None)

Answer 3

对于您的单元测试，我建议您跳过 celery 依赖项，以下两个链接将为您提供启动单元测试所需的信息：

• http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.html
• http://docs.celeryproject.org/en/latest/userguide/testing.html

如果您真的想测试包括队列在内的 celery 函数调用，我可能会设置一个带有服务器、工作程序、队列组合的 dockercompose，并从 django-celery 文档中扩展自定义 CeleryTestRunner。但我看不出它有什么好处，因为测试系统离生产很远，无法具有代表性。