Add scheduled task to send scheduled notifcations.

Fix the query to fetch scheduled notifications.

app/celery/scheduled_tasks.py

@@ -14,13 +14,14 @@ from app.dao.jobs_dao import dao_set_scheduled_jobs_to_pending, dao_get_jobs_old
 from app.dao.notifications_dao import (
     delete_notifications_created_more_than_a_week_ago,
     dao_timeout_notifications,
-    is_delivery_slow_for_provider
+    is_delivery_slow_for_provider,
-)
+    dao_get_scheduled_notifications)
 from app.dao.provider_details_dao import (
     get_current_provider,
     dao_toggle_sms_provider
 )
 from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
+from app.notifications.process_notifications import send_notification_to_queue
 from app.statsd_decorators import statsd
 from app.celery.tasks import process_job
 
@@ -46,6 +47,18 @@ def run_scheduled_jobs():
         raise
 
 
+@notify_celery.task(name='send-scheduled-notifications')
+@statsd(namespace="tasks")
+def send_scheduled_notifications():
+    try:
+        for notification in dao_get_scheduled_notifications():
+            send_notification_to_queue(notification, notification.service.research_mode)
+
+    except SQLAlchemyError as e:
+        current_app.logger.exception("Failed to send scheduled notifications")
+        raise
+
+
 @notify_celery.task(name="delete-verify-codes")
 @statsd(namespace="tasks")
 def delete_verify_codes():

app/config.py

@@ -109,6 +109,11 @@ class Config(object):
             'schedule': crontab(minute=1),
             'options': {'queue': 'periodic'}
         },
+        'send-scheduled-notifications': {
+            'task': 'send-scheduled-notifications',
+            'schedule': crontab(minute=1),
+            'options': {'queue': 'periodic'}
+        },
         'delete-verify-codes': {
             'task': 'delete-verify-codes',
             'schedule': timedelta(minutes=63),

app/dao/notifications_dao.py

@@ -473,6 +473,10 @@ def dao_created_scheduled_notification(scheduled_notification):
 
 @statsd(namespace="dao")
 def dao_get_scheduled_notifications():
-    scheduled_notifications = ScheduledNotification.query.filter(
+    notifications = Notification.query.join(
-        ScheduledNotification.scheduled_for < datetime.utcnow()).all()
+        ScheduledNotification
-    return scheduled_notifications
+    ).filter(
+        ScheduledNotification.scheduled_for < datetime.utcnow(),
+        Notification.status == NOTIFICATION_CREATED).all()
+
+    return notifications

tests/app/celery/test_scheduled_tasks.py

@@ -5,7 +5,7 @@ from functools import partial
 
 from flask import current_app
 from freezegun import freeze_time
-from app.celery.scheduled_tasks import s3
+from app.celery.scheduled_tasks import s3, send_scheduled_notifications
 from app.celery import scheduled_tasks
 from app.celery.scheduled_tasks import (
     delete_verify_codes,
@@ -30,8 +30,8 @@ from tests.app.db import create_notification, create_service
 from tests.app.conftest import (
     sample_job as create_sample_job,
     sample_notification_history as create_notification_history,
-    create_custom_template
+    create_custom_template,
-)
+    sample_notification)
 from tests.conftest import set_config_values
 from unittest.mock import call, patch, PropertyMock
 
@@ -409,3 +409,22 @@ def test_switch_providers_on_slow_delivery_does_not_switch_based_on_older_notifi
         switch_current_sms_provider_on_slow_delivery()
         current_provider = get_current_provider('sms')
         assert starting_provider.identifier == current_provider.identifier
+
+
+@freeze_time("2017-05-01 14:00:00")
+def test_should_send_all_scheduled_notifications_to_deliver_queue(notify_db,
+                                                                  notify_db_session,
+                                                                  sample_template, mocker):
+    mocked = mocker.patch('app.celery.provider_tasks.deliver_sms')
+    message_to_deliver = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
+                                             template=sample_template, scheduled_for="2017-05-01 13:50:00")
+    sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
+                        template=sample_template, scheduled_for="2017-05-01 10:50:00", status='delivered')
+    sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
+                        template=sample_template)
+    sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
+                        template=sample_template, scheduled_for="2017-05-01 14:30:00")
+
+    send_scheduled_notifications()
+
+    mocked.apply_async.assert_called_once_with([str(message_to_deliver.id)], queue='send-sms')

tests/app/dao/test_notification_dao.py

@@ -1708,9 +1708,12 @@ def test_dao_created_scheduled_notification(sample_notification):
 
 def test_dao_get_scheduled_notifications(notify_db, notify_db_session, sample_template):
     notification_1 = sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
-                                         template=sample_template, scheduled_for='2017-05-05 14:00:00')
+                                         template=sample_template, scheduled_for='2017-05-05 14:00:00',
+                                         status='created')
     sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
-                        template=sample_template)
+                        template=sample_template, scheduled_for='2017-05-04 14:00:00', status='delivered')
+    sample_notification(notify_db=notify_db, notify_db_session=notify_db_session,
+                        template=sample_template, status='created')
     scheduled_notifications = dao_get_scheduled_notifications()
     assert len(scheduled_notifications) == 1
-    assert scheduled_notifications[0].notification_id == notification_1.id
+    assert scheduled_notifications[0].id == notification_1.id