unit testing与Django芹菜？
我想为我们的django-芹菜项目提出一个testing方法。 我已经阅读了文档中的注释，但并没有给我一个好的想法。 我并不担心在实际的守护进程中testing任务，只是我的代码的function。 主要是我想知道：
- 我们如何绕过
task.delay()在testing过程中（我尝试设置CELERY_ALWAYS_EAGER = True但没有区别）？ - 我们如何使用推荐的testing设置（如果这是最好的方式），而不实际改变我们的settings.py？
- 我们仍然可以使用
manage.py test还是必须使用自定义运行器？
总的来说，使用芹菜testing的任何提示或技巧将会非常有帮助。
尝试设置：
BROKER_BACKEND = 'memory'
（感谢asksol的评论）
我喜欢在需要芹菜结果的testing上使用override_settings装饰器来完成。
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())
如果你想应用到所有的testing你可以使用芹菜testing亚军http://docs.celeryproject.org/en/2.5/django/unit-testing.html基本上设置这些相同的设置除外（; BROKER_BACKEND = 'memory' ）。
在设置中：
TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'
看看CeleryTestSuiteRunner的来源，这很清楚发生了什么事情。
这是我的testing基类的一个片段，它存根apply_async方法，并logging到它的调用（其中包括Task.delay 。这是一个粗略的，但它的devise，以满足我的需求在过去几个月我一直使用它。
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)
下面是你在testing用例中如何使用它的一个“头顶”例子：
mymodule.py
from my_tasks import SomeTask def run_some_task(should_run): if should_run: SomeTask.delay(1, some_kwarg=2)
test_mymodule.py
class RunSomeTaskTest(CeleryTestCaseBase): def test_should_run(self): run_some_task(should_run=True) self.assert_task_sent(SomeTask, 1, some_kwarg=2) def test_should_not_run(self): run_some_task(should_run=False) self.assert_task_not_sent(SomeTask)
因为我仍然看到这在search结果中出现，设置覆盖
TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'
按照芹菜文档为我工作
