在unittest中运行芹菜工人

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 使用的相同方法对任务参数进行编码（即腌制参数或将它们编码为 JSON）。
• 再次解码任务参数，检查是否发生损坏。
• 调用任务函数，确保它产生正确的结果。
• 对任务函数的结果执行相同的编码/解码序列。

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

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

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