python - 是否可以对 celery 周期性任务进行单元测试？

Question

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

任务是:

@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()你的 celery 任务同步执行:

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)