python - 在单元测试中运行 Celery worker

Question

我有以下设置:

• Django-Celery 项目 A 注册任务 foo
• 项目 B:使用 Celery 的 send_task 调用 foo
• 项目A和项目B配置相同:SQS、msgpack用于序列化、gzip 等。
• 每个项目都位于不同的 github 存储库中

我已经对项目 A 中对“foo”的调用进行了单元测试，完全没有使用 Celery，只是 foo(1,2,3) 并断言结果。我知道它有效。

我已经对项目 B 中的 send_task 发送了正确的参数进行了单元测试。

我没有测试的是两个项目之间的集成，需要您提供建议。我想要一个单元测试:

• 在项目 A 的上下文中启动一个 worker
• 使用项目B的代码发送任务
• 断言第一步开始的工作人员使用我在第二步发送的参数获取任务，并且 foo 函数返回预期结果。

似乎可以通过使用 python 的子进程并解析 worker 的输出来破解它，但这很丑陋。在这种情况下，推荐的单元测试方法是什么？您可以分享任何代码片段吗？谢谢!

Answer 1

我不确定是否值得使用单元测试显式测试传输机制(即通过 celery 发送任务参数)。就个人而言，我会按如下方式编写我的测试(可以分成几个单元测试):

• 使用项目 B 中的代码生成带有示例参数的任务。
• 使用与 Celery 相同的方法对任务参数进行编码(即 pickle 参数或将它们编码为 JSON)。
• 再次解码任务参数，检查没有发生损坏。
• 调用任务函数，确保它产生正确的结果。
• 对任务函数的结果执行相同的编码/解码序列。

使用此方法，您将能够测试它

• 任务生成按预期进行
• 任务参数和结果的编码和解码符合预期

如有必要，您仍然可以使用系统测试独立测试传输机制的功能。