我有以下设置:
- 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 的输出来破解它，但这很丑陋。在这种情况下，推荐的单元测试方法是什么？您可以分享任何代码片段吗？谢谢!
最佳答案
我不确定是否值得使用单元测试显式测试传输机制(即通过 celery 发送任务参数)。就个人而言，我会按如下方式编写我的测试(可以分成几个单元测试):
- 使用项目 B 中的代码生成带有示例参数的任务。
- 使用与 Celery 相同的方法对任务参数进行编码(即 pickle 参数或将它们编码为 JSON)。
- 再次解码任务参数，检查没有发生损坏。
- 调用任务函数，确保它产生正确的结果。
- 对任务函数的结果执行相同的编码/解码序列。
使用此方法，您将能够测试它
- 任务生成按预期进行
- 任务参数和结果的编码和解码符合预期
如有必要，您仍然可以使用系统测试独立测试传输机制的功能。
关于python - 在单元测试中运行 Celery worker，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/17181923/