我正在使用一个面向软件的架构,它有多个 celery worker(我们称它们为 worker1
、worker2
和 worker3
)。所有三个工作人员都是独立的实体(即,独立的代码库、独立的 repos、独立的 celery 实例、独立的机器),它们都没有连接到 Django 应用程序。
通过基于 Django、MySQL 支持的 RESTful API 与这三个工作人员中的每一个进行通信。
在开发中,这些服务都在一个 vagrant box 上,每一个都作为一个独立的机器运行在一个单独的端口上。我们有一个 RabbitMQ 代理来处理所有 Celery 任务。
通过这些服务的典型路径可能如下所示:worker1
从设备获取消息,进行一些处理,在 worker2
上排队执行任务,后者执行进一步处理并向 API
发送 POST,它写入 MySQL 数据库并触发 worker3
上的任务,后者执行一些其他处理并向 发送另一个 POST >API
导致 MySQL 写入。
服务通信良好,但每次我们对任何服务进行更改时都要测试此流程非常烦人。我真的很想进行一些完整的集成测试(即从发送到 worker1
的消息开始并通过整个链),但我不确定从哪里开始。我面临的主要问题是:
如果我在 worker1
上排队,我怎么知道整个流程何时结束?当我什至不知道结果是否已经到来时,我如何对结果做出合理的断言?
我如何处理数据库设置/拆除?我想在每次测试结束时删除测试期间创建的所有条目,但如果我从 Django 应用程序外部开始测试,我不确定如何有效地清除它。每次测试后手动删除并重新创建它似乎开销太大。
最佳答案
Celery 允许同步运行任务,所以第一步是:将整个流程分成单独的任务、假请求和断言结果:
原始流程:
device --- worker1 --- worker2 --- django --- worker3 --- django
一级集成测试:
1. |- worker1 -|
2. |- worker2 -|
3. |- django -|
4. |- worker3 -|
5. |- django -|
为每个测试创建假请求或同步调用并断言结果。将这些测试放在相应的存储库中。例如,在 worker1 的测试中,您可以模拟 worker2 并测试它是否已使用正确的参数调用。然后,在另一个测试中,您将调用 worker2 和模拟请求来检查它是否调用了正确的 API。等等。
测试整个流程会很困难,因为所有任务都是独立的实体。我现在想出的唯一方法是对 worker1 进行一次假调用,设置合理的超时并等待数据库中的最终结果。这种测试只告诉你它是否有效。它不会告诉您问题出在哪里。
关于python - 集成测试多个 Celery Worker 和一个数据库支持的 Django API,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/23431351/