让Celery使用Django Test DB

Question

我正在尝试在我的Django应用程序中为一些芹菜任务编写一些单元测试.这些任务将模型ID作为参数,执行一些操作并更新模型.当运行devserver和celery worker时,一切都很好,但是在运行我的测试时,很明显celery任务没有使用在测试运行过程中创建和销毁的django测试数据库.问题是,如何让芹菜使用与我的其他测试相同的临时数据库？

正如您所看到的,我正在使用针对类似问题的每个答案中建议的设置覆盖.

更新:发现不是将对象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 工作器配置为使用与测试相同的测试数据库的一种方法是在测试本身内部生成 Celery 工作器。这可以通过使用来完成start_worker

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

的方法TestCase。

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

Answer 2

你的代码没有明显的问题。你不需要运行芹菜工人。通过这些设置，celery 将同步运行任务，并且实际上不会向消息队列发送任何内容。

无论如何，您都无法轻松地使用实时 celery 工作人员运行测试，因为每个测试都包含在一个事务中，因此即使它们连接到同一个数据库（它们不是），事务也总是会被测试回滚，并且永远无法用于工人。

如果您确实需要这样做，请查看此 stackoverflow 答案。