您如何对Celery任务进行单元测试？

本文介绍了您如何对Celery任务进行单元测试？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

Celery文档在Django中测试Celery的说明，但是如果您不使用Django，则不会说明如何测试Celery任务。

The Celery documentation mentions testing Celery within Django but doesn't explain how to test a Celery task if you are not using Django. How do you do this?

推荐答案

可以使用那里的任何unittest库同步测试任务。在处理芹菜任务时，我通常会进行2次不同的测试。第一个（正如我建议的那样）是完全同步的，应该确保算法能够执行应做的工作。第二部分使用整个系统（包括代理），并确保我没有序列化问题或任何其他分发，通信问题。

It is possible to test tasks synchronously using any unittest lib out there. I normaly do 2 different test sessions when working with celery tasks. The first one (as I'm suggesting bellow) is completely synchronous and should be the one that makes sure the algorithm does what it should do. The second session uses the whole system (including the broker) and makes sure I'm not having serialization issues or any other distribution, comunication problem.

所以：

from celery import Celery celery = Celery() @celery.task def add(x, y): return x + y

您的测试：

from nose.tools import eq_ def test_add_task(): rst = add.apply(args=(4, 4)).get() eq_(rst, 8)

希望有帮助！