获取 Celery 以使用 Django 测试数据库

问题描述 投票：0回答：5

我正在尝试为 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()
python django unit-testing celery django-celery
5个回答
5
投票

保证 Celery Worker 配置为使用与测试相同的测试数据库的一种方法是在测试本身内部生成 Celery Worker。这可以通过使用 

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
方法中手动销毁它们。

4
投票

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

  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 任务都将使用您的测试数据库。并希望一切顺利。

2
投票

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

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

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

0
投票

我发现将以下内容添加到

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):
    ...
0
投票

“问题是，如何让 celery 使用与 我剩下的测试？”

我通过使用 docker compose 运行测试来解决这个问题，使数据库名称可通过环境变量配置，并将数据库名称设置为 test_db （正常数据库名称为“db”）。

但是我不使用sqlite...

如果您需要 sqlite 的解决方案：使 Django 测试用例数据库对 Celery 可见

最新问题
© www.soinside.com 2019 - 2025. All rights reserved.