获取 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)

TestCase的方法。

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

Answer 2

我也遇到过类似的问题。以下解决方案不干净，但它有效。

1. 创建一个单独的 Django 设置文件，该文件继承自您的主文件 一。我们就叫它integration_testing.py吧。

2. 您的文件应如下所示：
   from .settings import *

   DATABASES = { 'default': { 'ENGINE': '<your engine>', 'NAME': 'test_<your database name>', 'USER': <your db user>, 'PASSWORD': <your db password>, 'HOST': <your hostname>, 'PORT': <your port number>, }

3. 创建一个 shell 脚本来设置你的环境并启动 芹菜工人：

   #!/usr/bin/env bash

   export DJANGO_SETTINGS_MODULE="YOURPROJECTNAME.settings.integration_testing"

   celery purge -A YOURPROJECTNAME -f && celery worker -A YOURPROJECTNAME -l debug

4. 如果您以这种方式配置 celery，上述方法有效：

   app = Celery('YOURPROJECTNAME')

   app.config_from_object('django.conf:settings', namespace='CELERY')

5. 在后台运行脚本。

6. 使所有涉及 Celery 的测试都继承自 TransactionTestCase（或 django-rest-framework 中的 APITransactionTestCase）

7. 运行使用 celery 的单元测试。任何 celery 任务现在都将使用您的测试数据库。并希望一切顺利。

Answer 3

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

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

如果您确实需要这样做，请查看this stackoverflow answer。

Answer 4

我发现将以下内容添加到 conftest.py 有效：

from django.conf import settings

...

@pytest.fixture(scope="session")
def celery_worker_parameters(django_db_setup):
    assert settings.DATABASES["default"]["NAME"].startswith("test_")
    return {}

诀窍是在此处添加 django_db_setup 固定装置，以便在工作器上启用它。

这是标记为的测试的测试：

@pytest.mark.django_db(transaction=True)
@pytest.mark.celery()
def test_something(celery_worker):
    ...