擅长:python、mysql、java
<p>可以使用任何unittest库同步测试任务。我通常在处理芹菜任务时做两个不同的测试。第一个(我建议如下)是完全同步的,应该是确保算法完成它应该做的事情的那个。第二个会话使用整个系统(包括代理),并确保我没有序列化问题或任何其他分发、通信问题。</p>
<p>所以:</p>
<pre><code>from celery import Celery
celery = Celery()
@celery.task
def add(x, y):
return x + y
</code></pre>
<p>你的测试:</p>
<pre><code>from nose.tools import eq_
def test_add_task():
rst = add.apply(args=(4, 4)).get()
eq_(rst, 8)
</code></pre>
<p>希望能有帮助!</p>