from django.test import TestCase
from celery.task.base import Task
# For recent versions, Task has been moved to celery.task.app:
# from celery.app.task import Task
# See http://docs.celeryproject.org/en/latest/reference/celery.app.task.html
class CeleryTestCaseBase(TestCase):
def setUp(self):
super(CeleryTestCaseBase, self).setUp()
self.applied_tasks = []
self.task_apply_async_orig = Task.apply_async
@classmethod
def new_apply_async(task_class, args=None, kwargs=None, **options):
self.handle_apply_async(task_class, args, kwargs, **options)
# monkey patch the regular apply_sync with our method
Task.apply_async = new_apply_async
def tearDown(self):
super(CeleryTestCaseBase, self).tearDown()
# Reset the monkey patch to the original method
Task.apply_async = self.task_apply_async_orig
def handle_apply_async(self, task_class, args=None, kwargs=None, **options):
self.applied_tasks.append((task_class, tuple(args), kwargs))
def assert_task_sent(self, task_class, *args, **kwargs):
was_sent = any(task_class == task[0] and args == task[1] and kwargs == task[2]
for task in self.applied_tasks)
self.assertTrue(was_sent, 'Task not called w/class %s and args %s' % (task_class, args))
def assert_task_not_sent(self, task_class):
was_sent = any(task_class == task[0] for task in self.applied_tasks)
self.assertFalse(was_sent, 'Task was not expected to be called, but was. Applied tasks: %s' % self.applied_tasks)
下面是一个“从头开始”的例子,说明如何在测试用例中使用它:
mymodule.py
from my_tasks import SomeTask
def run_some_task(should_run):
if should_run:
SomeTask.delay(1, some_kwarg=2)
from django.test import TestCase
from django.test.utils import override_settings
from myapp.tasks import mytask
class AddTestCase(TestCase):
@override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_BACKEND='memory')
def test_mytask(self):
result = mytask.delay()
self.assertTrue(result.successful())
这是我的测试基类的一个摘录,它去掉了
apply_async
方法,并记录了对它的调用(包括Task.delay
),这有点恶心,但它满足了我过去几个月的使用需求。下面是一个“从头开始”的例子,说明如何在测试用例中使用它:
mymodule.py
test_mymodule.py
尝试设置:
(感谢asksol的评论。)
我喜欢在需要芹菜结果才能完成的测试上使用override_settings装饰符。
如果要将此应用于所有测试,可以使用http://docs.celeryproject.org/en/2.5/django/unit-testing.html中描述的芹菜测试运行器,它基本上设置了除(
BROKER_BACKEND = 'memory'
)之外的相同设置。在“设置”中:
看看芹菜SuiteRunner的来源,很清楚发生了什么。
相关问题 更多 >
编程相关推荐