Coverage for website/pushnotifications/tasks.py: 0.00%

1import logging 

2from datetime import timedelta 

3 

4from django.utils import timezone 

5 

6from celery import shared_task 

7 

8from pushnotifications.models import ScheduledMessage 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13@shared_task 

14def send_scheduled_messages(interval_time): 

15 """Send a scheduled push notifications.""" 

16 interval = int(interval_time) 

17 now = timezone.now() 

18 

19 logger.info("Start sending scheduled notifications") 

20 

21 before_time = timezone.now() + timedelta(seconds=interval / 2) 

22 messages = ScheduledMessage.objects.filter(sent__isnull=True, time__lte=before_time) 

23 

24 for message in messages: 

25 if (timezone.now() - now).seconds < interval: 

26 logger.info("Sending push notification %d", message.pk) 

27 message.executed = timezone.now() 

28 message.send()