python - 让 Celery 使用 Django 测试数据库

Question

我正在尝试为我的 Django 应用程序中的一些 celery 任务编写一些单元测试。这些任务将模型 ID 作为参数，执行一些操作并更新模型。在运行 devserver 和 celery worker 时，一切都很好，但是在运行我的测试时，很明显 celery 任务没有使用作为测试运行的一部分创建和销毁的 django 测试数据库。问题是，我怎样才能让 celery 使用与其余测试相同的临时数据库？

如您所见，我正在使用针对类似问题的每个答案中建议的设置覆盖。

更新:发现如果我只是将对象本身传递给任务，而不是将对象 id 传递给任务并让任务从数据库中获取它，测试工作正常，显然对任务。所以至少现在，这将是我的解决办法。

在我的测试中:

class JobTest(TestCase):

    @override_settings(CELERY_ALWAYS_EAGER=True,
                       CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
                       BROKER_BACKEND='memory')
    def test_Job_Complete(self):
        job = models.Job()
        job.save()
        tasks.do_a_thing(job.id)
        self.assertTrue(job.complete)

在我的任务中:

@celery.task
def do_a_thing(job_id):
    job = models.Job.objects.get(pk=job_id)
    bunch_of_things(job)
    job.complete = True
    job.save()

Answer 1

保证 Celery worker 配置为使用与测试相同的测试数据库的一种方法是 spawn the Celery worker inside the test itself .这可以通过使用 start_worker

来完成

from celery.contrib.testing.worker import start_worker
from myproject.celery import app
def setUpClass(self):
   start_worker(app)

测试用例的方法。

您还必须使用来自 Django 的 SimpleTestCase 或来自 Rest 的 APISimpleTestCase 而不是普通的 TestCase 以便 Celery 线程和测试线程可以看到彼此对测试数据库所做的更改。更改在测试结束时仍会被销毁，但它们不会在测试之间被销毁，除非您在 tearDown 方法中手动销毁它们。