您如何对Celery任务进行单元测试？ [英] How do you unit test a Celery task?

问题描述

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)

希望有帮助！

这篇关于您如何对Celery任务进行单元测试？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！