集成测试多个芹菜工人和数据库支持的DjangoAPI-IntegrationTestingMultipleCeleryWorkersandaDBBackedDjangoAPI

3  

Celery allows to run task synchronously, so the first step is: Divide whole flow into separate tasks, fake requests and assert results:

Celery允许同步运行任务，因此第一步是：将整个流程划分为单独的任务，伪造请求和断言结果：

Original flow:

原始流程：

device --- worker1 --- worker2 --- django --- worker3 --- django

First-level integration tests:

一级集成测试：

1.      |- worker1 -|
2.                  |- worker2 -|
3.                              |- django -|
4.                                         |- worker3 -|
5.                                                     |- django -|

For each test create fake request or synchronous call and assert results. Place these tests in corresponding repository. For example in test for worker1, you can mock worker2 and test that it has been called with proper arguments. Then, in another test, you will call worker2 and mock request to check, that it calls right API. And so on.

对于每个测试，创建假请求或同步调用并断言结果。将这些测试放在相应的存储库中例如，在test1的测试中，您可以模拟worker2并测试它是否已使用适当的参数调用。然后，在另一个测试中，您将调用worker2和模拟请求来检查它是否正确调用API。等等。

Testing whole flow, will be difficult, since all tasks are separate entities. The only way I've came up with now is to make one fake call to worker1, set reasonable timeout and wait for final result in database. This kind test only tells you if it works or not. It won't show you, where is the problem.

测试整个流程将很困难，因为所有任务都是独立的实体。我现在想出的唯一方法就是对worker1进行一次虚假调用，设置合理的超时并等待数据库中的最终结果。此类测试仅告诉您它是否有效。它不会告诉你，问题出在哪里。

0  

To work with the full setup, you can set up a Celery results backend. See the Celery 'next steps' documentation for the basics.

要使用完整设置，您可以设置Celery结果后端。有关基础知识，请参阅Celery的“后续步骤”文档。

worker1 could then report the task handle of what it has passed on to worker2. The result returned by worker2 would be the task id of what it has passed on to worker3. And the result returned by worker3 would mean the whole sequence is finished and you can check the outcomes. The results could also already report interesting bits of those outcomes right away to make the checking easier.

然后，worker1可以报告传递给worker2的任务句柄。 worker2返回的结果将是传递给worker3的任务ID。 worker3返回的结果意味着整个序列已经完成，您可以检查结果。结果也可以立即报告这些结果的有趣部分，以使检查更容易。

This might look somewhat like this in Celery:

这可能在Celery中看起来像这样：

worker1_result = mytask.delay(someargs)  # executed by worker1
worker2_result = worker1_result.get()  # waits for worker1 to finish
worker3_result = worker2_result.get()  # waits for worker2 to finish
outcome = worker3_result.get()  # waits for worker3 to finish

(The details probably need to be different; I have not used this myself yet. I am not sure whether task results are serializable and hence themselves suitable as task function return values.)

（细节可能需要不同;我自己还没有使用过。我不确定任务结果是否可序列化，因此它们本身适合作为任务函数返回值。）