セロリの定期タスクで単体テストを実行することはできますか？

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)