运行芹菜工作人员在单元测试

Question

我有以下设置：

• Django的芹菜项目中的寄存器任务foo
• 项目B：使用芹菜的send_task调用foo
• 项目A和项目B具有相同的配置：SQS，msgpack 用于序列化，gzip等
• 每个项目都存在于不同的github版本库中

我在单元测试的项目A中的“foo”调用，根本不使用Celery，只是foo(1,2,3)并声明结果。我知道它的工作原理。

我有单元测试，在B工程send_task发送正确的参数。

什么我没有测试，需要在你的建议是这两个项目之间的整合。我想有一个单元测试，将：

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

似乎有可能通过使用python的子进程和解析worker的输出来破解这个，但这很丑。在这种情况下单元测试的建议方法是什么？任何代码片段，你可以分享？谢谢！

Answer 1

我不知道这是否是值得的明确测试输送机构使用单元测试（即任务参数通过芹菜发送）。就个人而言，我会按照以下方式编写我的测试（可以拆分为几个单元测试）：

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

使用这种方法，你将能够测试

• 的任务生成的作品如预期
• 任务参数的编码&解码和预期结果的作品

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