如何使用 djcelery 正确测试预定测试
How to properly test scheduled test using djcelery
我有一个使用 djcelery 的周期性任务。此任务正在查看数据库,并在必要时进行相同的更改。
如您所知,"python manage.py test" 创建了一个测试数据库,那么我如何才能 运行 这个测试数据库中的这个周期性任务?
我先尝试运行:pythonmanage.py测试
然后当我尝试 运行 以下内容时
python manage.py celery worker -l info &
pythonmanage.pycelery beat -l info &
测试结束
您不需要 运行 芹菜工人或芹菜节拍,那不是很好的单元测试。您正在尝试测试任务,因此您应该隔离任务并对其进行测试。您需要使用 CELERY_ALWAYS_EAGER 模式 运行 测试,或者只是 运行 从没有异步模式的测试内部完成您的任务。
为了避免使用异步,只需像这样调用您的任务:
@celery.task
def task_to_test():
print('stuff')
def test:
task_to_test()
如果你做不到。覆盖您的设置以使用始终渴望。
from tasks import task_to_test
@override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_MEMORY='memory',
CACHES = {'default': {'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',}})
class MyTests(TestCase):
def test():
task_to_test.delay()
我有一个使用 djcelery 的周期性任务。此任务正在查看数据库,并在必要时进行相同的更改。 如您所知,"python manage.py test" 创建了一个测试数据库,那么我如何才能 运行 这个测试数据库中的这个周期性任务?
我先尝试运行:pythonmanage.py测试
然后当我尝试 运行 以下内容时
python manage.py celery worker -l info &
pythonmanage.pycelery beat -l info &
测试结束
您不需要 运行 芹菜工人或芹菜节拍,那不是很好的单元测试。您正在尝试测试任务,因此您应该隔离任务并对其进行测试。您需要使用 CELERY_ALWAYS_EAGER 模式 运行 测试,或者只是 运行 从没有异步模式的测试内部完成您的任务。
为了避免使用异步,只需像这样调用您的任务:
@celery.task
def task_to_test():
print('stuff')
def test:
task_to_test()
如果你做不到。覆盖您的设置以使用始终渴望。
from tasks import task_to_test
@override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True,
BROKER_MEMORY='memory',
CACHES = {'default': {'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',}})
class MyTests(TestCase):
def test():
task_to_test.delay()