让Celery使用Django测试数据库

17 投票

5 回答

4557 浏览

提问于 2025-04-17 20:23

我正在为我的Django应用中的一些celery任务编写单元测试。这些任务需要一个模型的ID作为参数，执行一些操作，然后更新这个模型。在运行开发服务器和celery工作进程时，一切都很顺利，但在运行我的测试时，我发现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()

django 单元测试任务调度 celery 对象传递测试数据库模型操作测试覆盖

5 个回答

“我的问题是，怎么让celery使用和我其他测试一样的临时数据库？”

我解决这个问题的方法是通过docker compose来运行我的测试，把数据库名称设置为可以通过环境变量来配置，然后把数据库名称设为test_db（正常的数据库名称是'db'）。

不过我不使用sqlite……

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

回答于 2025-04-17 由 Python大师

分享举报

我发现把下面的内容加到 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):
    ...

回答于 2025-04-17 由 Python大师

分享举报

你的代码看起来没有明显的问题。你不需要运行一个celery工作进程。根据这些设置，celery会同步执行任务，实际上不会把任何东西发送到你的消息队列。

而且，你也不容易用活的celery工作进程来测试，因为每个测试都是在一个事务中进行的，即使它们连接的是同一个数据库（其实并不是），这些事务总是会被测试回滚，工作进程根本无法获取到。

如果你真的需要这样做，可以看看这个stackoverflow的回答。

回答于 2025-04-17 由 Python大师

分享举报

我之前也遇到过类似的问题。下面的解决方案不是很完美，但确实有效。

创建一个单独的Django设置文件，它会从你的主设置文件中继承。我们可以把它叫做integration_testing.py。
你的文件应该像这样：
from .settings import *

DATABASES = { 'default': { 'ENGINE': '<你的数据库引擎>', 'NAME': 'test_<你的数据库名称>', 'USER': <你的数据库用户>, 'PASSWORD': <你的数据库密码>, 'HOST': <你的主机名>, 'PORT': <你的端口号>, }
创建一个shell脚本，用来设置你的环境并启动celery工作进程：

#!/usr/bin/env bash

export DJANGO_SETTINGS_MODULE="YOURPROJECTNAME.settings.integration_testing"

celery purge -A YOURPROJECTNAME -f && celery worker -A YOURPROJECTNAME -l debug
如果你是这样配置celery的，上面的步骤就能正常工作：

app = Celery('YOURPROJECTNAME')

app.config_from_object('django.conf:settings', namespace='CELERY')
在后台运行这个脚本。
确保所有涉及到Celery的测试都继承自TransactionTestCase（或者在django-rest-framework中使用APITransactionTestCase）。
运行你的单元测试，这些测试会使用celery。任何celery任务现在都会使用你的测试数据库。希望一切顺利。

回答于 2025-04-17 由 Python大师

分享举报

为了确保Celery工作进程使用和测试相同的测试数据库，有一种方法是在测试内部启动Celery工作进程。这可以通过使用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方法中手动销毁它们。

回答于 2025-04-17 由 Python大师

分享举报

让Celery使用Django测试数据库

5 个回答

撰写回答