python - 运行连接到 Django 测试数据库的 Celery worker

标签 python django multithreading testing celery

背景:我正在开展一个项目,该项目使用 Celery 来安排将在未来特定时间运行的任务。这些任务插入最终状态机的状态向前发展。这是一个例子:

  • 计划在 2 天内向用户发送 future 提醒。
  • 当该计划任务运行时,会发送一封电子邮件,并且 FSM 会前进到下一个状态
  • 下一个状态是安排提醒在另外两天运行
  • 当这个任务运行时,它会发送另一封邮件,高级状态
  • 等等...

我目前正在按照 this SO answer 的建议使用 CELERY_ALWAYS_EAGER

在测试中使用该技术的问题在于,本应在单独线程中运行的任务代码与调度它的线程在同一线程中运行。这会导致 FSM 状态无法正确保存,并且难以测试。我无法确定究竟是什么导致了它,但似乎在调用堆栈的底部你正在保存到当前状态,但是当你返回调用堆栈时,之前的状态正在被保存。当代码没有按应有的方式运行时,我可能会花更多的时间来确定出了什么问题,但是尝试让代码按应有的方式运行并确保它按应有的方式运行似乎更合乎逻辑。

问题: 因此,我想知道是否有一种方法可以运行 django 在测试运行期间可以使用的完整 celery 设置。如果它可以自动运行,那将是理想的,但即使是一些手动干预也比手动测试行为要好。我在想如果我在测试中设置一个中断,运行 celery worker 连接到测试数据库,继续 django 测试是可能的。以前有人尝试过这样的事情吗?

最佳答案

您要做的不是单元测试,而是功能/集成测试。

我建议使用一些 BDD 框架(Behave、Lettuce)并从 CI 服务器(TravisCI 或 Jenkins)针对外部服务器(例如暂存环境)运行 BDD 测试。

所以,这个过程可能是:

  1. 将更改推送到 GitHub
  2. GitHub 在 CI 服务器上启动构建
  3. CI 服务器运行单元测试
  4. CI 服务器部署到集成环境(或暂存,如果您没有集成)
  5. CI 服务器针对新部署的代码运行集成端到端测试
  6. 如果一切成功,这个构建将被提升为“可以部署到生产”或类似的东西

关于python - 运行连接到 Django 测试数据库的 Celery worker,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/20638751/

相关文章:

python - Errno 2 没有这样的文件或目录 : 'ffprobe' : 'ffprobe'

multithreading - 使用 channel 在线程之间传递 Rust pnet 数据包

android - 检查所有 AsyncTask 是否已完成

mysql - 几小时不活动后出现 "MySQL server has gone away"错误

python - 无法在 __init__ 处向 ModelForm 添加字段

python - 我的Python多线程代码是否受到全局解释器锁的影响

python - 你如何找到低于某个值的最大 float ?

python - Python : calling programs that use curses, 等中的 Shell 脚本

python - 保存一个 numpy 矩阵

python - 添加到而不是覆盖模板 block