是否可以对芹菜周期性任务运行单元测试?

Question

我正在尝试对我设置为每天运行的芹菜任务运行单元测试。
我尝试导入该函数并在测试中调用它，但这不起作用。

任务是：

@shared_task
def create_a_notification_if_a_product_is_in_or_out_of_season():
"""
Send a notification if a product is now in or out of season
"""
julian_date = date.today().timetuple().tm_yday + 1
active_products = Product.objects.filter(status='ACTIVE')
for products in active_products:
in_season_prd = ProductDescription.objects.filter(
product=products, 
early_start_julian=julian_date
)
for prd in in_season_prd:
notification = Notification()
notification.type = notification_choices.PRODUCT_IN_SEASON
notification.description = str(prd.product.name) + " will be in season from tomorrow."
notification.save()

这是我的一个测试示例：

def test_when_product_is_about_to_come_in_to_seasonality(self):
"""
Make a notification when a product is due to come in to seasonality tomorrow
"""
p = Product.objects.first()
p.status = "ACTIVE"
today = date.today().timetuple().tm_yday
p.early_start_julian = today + 1
create_a_notification_if_a_product_is_in_or_out_of_season()
updated_notifications = Notification.objects.all().count()
self.assertNotEqual(self.current_notifications, updated_notifications)

任何帮助将不胜感激！

谢谢

Answer 1

您可以apply()芹菜任务以同步执行它：

def test_when_product_is_about_to_come_in_to_seasonality(self):
"""
Make a notification when a product is due to come in to seasonality tomorrow
"""
p = Product.objects.first()
p.status = "ACTIVE"
today = date.today().timetuple().tm_yday
p.early_start_julian = today + 1
create_a_notification_if_a_product_is_in_or_out_of_season.apply()
updated_notifications = Notification.objects.all().count()
self.assertNotEqual(self.current_notifications, updated_notifications)

Answer 2

我认为您正在寻找CELERY_ALWAYS_EAGER设置。如果设置为True它将同步运行您的任务。可以在测试设置中设置它，也可以仅使用@override_settings(CELERY_ALWAYS_EAGER=True)修饰该测试