我有三个芹菜任务:
@celery_app.task
def load_rawdata_on_monday():
if not load_rawdata(): # run synchronously
notify_report_was_not_updated.delay()
@celery_app.task
def load_rawdata():
# load and process file from FTP
return False # some error happened
@celery_app.task
def notify_rawdata_was_not_updated():
pass # send email by Django
如果load_rawdata任务(函数)返回False，我需要测试是否发送了电子邮件。为此，我编写了一些不工作的测试:
@override_settings(EMAIL_BACKEND='django.core.mail.backends.memcache.EmailBackend')
@override_settings(CELERY_ALWAYS_EAGER=False)
@patch('load_rawdata', MagicMock(return_value=False))
def test_load_rawdata_on_monday():
load_rawdata_on_monday()
assert len(mail.outbox) == 1, "Inbox is not empty"
assert mail.outbox[0].subject == 'Subject here'
assert mail.outbox[0].body == 'Here is the message.'
assert mail.outbox[0].from_email == 'from@example.com'
assert mail.outbox[0].to == ['to@example.com']
似乎notify_rawdata_was_not_updated仍在异步运行。如何编写合适的测试?
看起来可能会发生两件事:
- 你应该使用
apply()方法来同步运行你的任务。 CELERY_ALWAYS_EAGER设置应该是活动的，以允许后续的任务调用也被执行。
@override_settings(EMAIL_BACKEND='django.core.mail.backends.memcache.EmailBackend')
@override_settings(CELERY_ALWAYS_EAGER=True)
@patch('load_rawdata', MagicMock(return_value=False))
def test_load_rawdata_on_monday():
load_rawdata_on_monday.apply()
assert len(mail.outbox) == 1, "Inbox is not empty"
assert mail.outbox[0].subject == 'Subject here'
assert mail.outbox[0].body == 'Here is the message.'
assert mail.outbox[0].from_email == 'from@example.com'
assert mail.outbox[0].to == ['to@example.com']
虽然@tinom9关于使用apply()方法是正确的，但notify_rawdata_was_not_updated仍然异步运行的问题与您的任务定义有关:
@celery_app.task
def load_rawdata_on_monday():
if not load_rawdata():
notify_report_was_not_updated.delay() # delay is an async invocation
试试这个:
@celery_app.task
def load_rawdata_on_monday():
if not load_rawdata():
notify_report_was_not_updated.apply() # run on local thread
，对于测试，调用load_rawdata_on_monday()而不调用.delay()或.apply()应该仍然在本地执行任务并阻塞，直到任务结果返回。只要确保你正确地处理返回值，一些芹菜调用方法，如apply()返回一个celery.result. eagerresult实例，而delay()或apply_async()返回一个celery.result. asyncresult实例，这可能不会给你想要的结果，如果期待False，当你检查if not load_rawdata()或者在其他任何你试图获得函数的返回值而不是任务本身的地方。
我应该用CELERY_TASK_ALWAYS_EAGER而不是CELERY_ALWAYS_EAGER