【发布时间】：2013-12-17 15:54:46
【问题描述】：
背景：我正在做一个项目，该项目使用 Celery 来安排将在未来某个时间运行的任务。这些任务将最终状态机的状态向前推进。这是一个例子：
- 计划在 2 天内向用户发送未来提醒。
- 当该计划任务运行时，会发送一封电子邮件，并且 FSM 会进入下一个状态
- 下一个状态是安排提醒在另外两天运行
- 当这个任务运行时，它会发送另一封邮件，提前状态
- 等等……
按照this SO answer 的建议，我目前正在使用 CELERY_ALWAYS_EAGER
在测试中使用该技术的问题在于，本来应该在单独的线程中运行的任务代码与调度它的线程在同一个线程中运行。这会导致 FSM 状态无法正确保存，并使其难以测试。我无法确定究竟是什么原因造成的，但似乎在调用堆栈的底部您正在保存到当前状态，但是当您返回调用堆栈时，之前的状态正在被保存。当代码未按应有的方式运行时，我可能会花更多的时间来确定出了什么问题，但是尝试让代码按应有的方式运行并确保它按应有的方式运行似乎更合乎逻辑。
问题：因此，我想知道是否有一种方法可以运行 django 在测试运行期间可以使用的完整 celery 设置。如果它可以自动运行，那将是理想的，但即使是一些手动干预也比必须手动测试行为要好。我想如果我在测试中设置中断，运行 celery worker 以连接到测试数据库，继续 django 测试，可能会有一些事情发生。有没有人尝试过这样的事情？
【问题讨论】：
-
我猜另一种方法是编写一些不使用 Django 测试运行器的单元测试，因此使用主（开发）数据库，手动设置和拆卸数据。
-
是的，这可能是最简单的方法。我这样做是为了针对不实用的大型数据库运行测试。
-
我不知道这是否正是您的问题，但我有一些竞争条件，即 Celery 任务不通过 Django 中间件，因此数据库操作不是事务性的。这可能是导致奇怪的数据库行为的原因
-
@geekazoid 这真的很有趣。您是否在普通模型或 celery 相关模型（即任务、返回等）中看到了这一点
标签： python django multithreading testing celery