让 Celery 在没有 task_always_eager 的情况下使用 Django 的测试数据库

Posted 2023-02-24

【中文标题】让 Celery 在没有 task_always_eager 的情况下使用 Django 的测试数据库

【英文标题】：Make Celery use Django's test database without task_always_eager

【发布时间】：2017-06-22 20:01:52

【问题描述】：

在使用 Celery 任务的 Django 应用程序中运行测试时，我无法完全测试需要从数据库获取数据的任务，因为它们没有连接到 Django 创建的测试数据库。

将 Celery 中的 task_always_eager 设置为 True 部分解决了这个问题，但正如 documentation for testing 所说，这并不能完全反映代码将如何在真正的 Celery worker 上运行并且不适合测试。

在不设置task_always_eager = True的情况下运行Django测试时如何让Celery任务使用Django测试数据库？

【参考方案1】：

一个简单的解决方案是use celery.contrib.testing.worker.start_worker to spawn a Celery worker within the Django test process。因为它存在于同一个进程中，所以它可以访问默认的内存中测试数据库，但因为它存在于自己的线程中，所以它并不急切，不需要或不推荐使用 task_always_eager 标志。

【参考方案2】：

Short = 您必须像在生产中一样运行 celery worker

简单：

使用专用测试数据库（与生产环境一样） 配置 celery 以使用它 在运行测试之前手动启动 celery worker

高级：

使用自动创建的测试数据库（可能是 sqlite） 在您的测试 setUp() 中运行 celery worker 将 celery 配置为使用自动创建的测试数据库（将 django.conf.settings.DATABASE 从测试进程复制到 celery）

而且您始终必须为 celery 提供消息代理。

我有一个测试，需要专门的 celery worker 来检查我在 celery 任务和调用代码之间传递消息的代码： https://gist.github.com/Sovetnikov/a7ad982fc77e8dfbc528bfc20fcf3b1e 这个 python 模块是二合一的 - 一个具有独立配置的 TestUnit 和 celery worker runner。

我的代码没有使用任何数据库，但您可以轻松地将其调整为您的需要。只需将 django.conf.settings.DATABASE（作为 json 或 pickle 或任何你喜欢的方法）传递给 celery 启动代码，并将 Django DATABASE 配置为指向测试数据库。

其他信息：

这个案例有完整的解决方案https://github.com/RentMethod/celerytest（我试过 它的一些旧版本并且没有运气，因为它使用线程，使用 python GIL ...而且我认为它过于复杂）

示例代码，如何在单个模块中配置 DATABASE 设置和初始化 django 本身 https://gist.github.com/Sovetnikov/369a8d05ba2b6482fa20769bc498f122