Merge pull request #3093 from alphagov/broadcast-tasks-onto-worker

Broadcast tasks onto worker

app/broadcast_message/rest.py

@@ -210,5 +210,5 @@ def _create_broadcast_event(broadcast_message):
 
     send_broadcast_event.apply_async(
         kwargs={'broadcast_event_id': str(event.id)},
-        queue=QueueNames.NOTIFY
+        queue=QueueNames.BROADCASTS
     )

app/celery/broadcast_message_tasks.py

@@ -23,7 +23,7 @@ def send_broadcast_event(broadcast_event_id):
     for provider in broadcast_event.service.get_available_broadcast_providers():
         send_broadcast_provider_message.apply_async(
             kwargs={'broadcast_event_id': broadcast_event_id, 'provider': provider},
-            queue=QueueNames.NOTIFY
+            queue=QueueNames.BROADCASTS
         )
 
 

app/celery/scheduled_tasks.py

@@ -309,4 +309,4 @@ def send_canary_to_cbc_proxy():
 def trigger_link_tests():
     if current_app.config['CBC_PROXY_ENABLED']:
         for cbc_name in current_app.config['ENABLED_CBCS']:
-            trigger_link_test.apply_async(kwargs={'provider': cbc_name}, queue=QueueNames.NOTIFY)
+            trigger_link_test.apply_async(kwargs={'provider': cbc_name}, queue=QueueNames.BROADCASTS)

app/config.py

@@ -33,6 +33,7 @@ class QueueNames(object):
     SANITISE_LETTERS = 'sanitise-letter-tasks'
     SAVE_API_EMAIL = 'save-api-email-tasks'
     SAVE_API_SMS = 'save-api-sms-tasks'
+    BROADCASTS = 'broadcast-tasks'
 
     @staticmethod
     def all_queues():
@@ -52,7 +53,8 @@ class QueueNames(object):
             QueueNames.LETTERS,
             QueueNames.SMS_CALLBACKS,
             QueueNames.SAVE_API_EMAIL,
-            QueueNames.SAVE_API_SMS
+            QueueNames.SAVE_API_SMS,
+            QueueNames.BROADCASTS,
         ]
 
 

manifest.yml.j2

@@ -64,6 +64,7 @@
   'notify-delivery-worker-letters': {'memory': '2G'},
   'notify-delivery-worker-retry-tasks': {},
   'notify-delivery-worker-internal': {},
+  'notify-delivery-worker-broadcasts': {},
   'notify-delivery-worker-receipts': {},
   'notify-delivery-worker-service-callbacks': {'disk_quota': '2G'},
   'notify-delivery-worker-save-api-notifications': {'disk_quota': '2G'},

scripts/paas_app_wrapper.sh

@@ -41,6 +41,10 @@ case $NOTIFY_APP_NAME in
     exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=11 \
     -Q notify-internal-tasks 2> /dev/null
     ;;
+  delivery-worker-broadcasts)
+    exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=2 \
+    -Q broadcast-tasks 2> /dev/null
+    ;;
   delivery-worker-receipts)
     exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=11 \
     -Q ses-callbacks,sms-callbacks 2> /dev/null

tests/app/broadcast_message/test_rest.py

@@ -392,7 +392,7 @@ def test_update_broadcast_message_status_stores_cancelled_by_and_cancelled_at(
 
     cancel_id = str(cancel_event.id)
 
-    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': cancel_id}, queue='notify-internal-tasks')
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': cancel_id}, queue='broadcast-tasks')
     assert response['status'] == BroadcastStatusType.CANCELLED
     assert response['cancelled_at'] is not None
     assert response['cancelled_by_id'] == str(canceller.id)
@@ -434,7 +434,7 @@ def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_
     assert len(bm.events) == 1
     alert_event = bm.events[0]
 
-    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='notify-internal-tasks')
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='broadcast-tasks')
 
     assert alert_event.service_id == sample_broadcast_service.id
     assert alert_event.transmitted_areas == bm.areas
@@ -472,7 +472,7 @@ def test_update_broadcast_message_status_creates_event_with_correct_content_if_b
     assert len(bm.events) == 1
     alert_event = bm.events[0]
 
-    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='notify-internal-tasks')
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='broadcast-tasks')
 
     assert alert_event.transmitted_content == {"body": "tailor made emergency broadcast content"}
 
@@ -553,7 +553,7 @@ def test_update_broadcast_message_status_allows_platform_admin_to_approve_own_me
     assert response['approved_by_id'] == str(user.id)
     mock_task.assert_called_once_with(
         kwargs={'broadcast_event_id': str(bm.events[0].id)},
-        queue='notify-internal-tasks'
+        queue='broadcast-tasks'
     )
 
 
@@ -584,7 +584,7 @@ def test_update_broadcast_message_status_allows_trial_mode_services_to_approve_o
     assert response['approved_at'] is not None
     assert response['created_by_id'] == str(t.created_by_id)
     assert response['approved_by_id'] == str(t.created_by_id)
-    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': ANY}, queue='notify-internal-tasks')
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': ANY}, queue='broadcast-tasks')
 
 
 def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_service(

tests/app/celery/test_broadcast_message_tasks.py

@@ -35,8 +35,8 @@ def test_send_broadcast_event_queues_up_for_active_providers(mocker, notify_api,
         send_broadcast_event(event.id)
 
     assert mock_send_broadcast_provider_message.apply_async.call_args_list == [
-        call(kwargs={'broadcast_event_id': event.id, 'provider': 'ee'}, queue='notify-internal-tasks'),
-        call(kwargs={'broadcast_event_id': event.id, 'provider': 'vodafone'}, queue='notify-internal-tasks')
+        call(kwargs={'broadcast_event_id': event.id, 'provider': 'ee'}, queue='broadcast-tasks'),
+        call(kwargs={'broadcast_event_id': event.id, 'provider': 'vodafone'}, queue='broadcast-tasks')
     ]
 
 
@@ -63,7 +63,7 @@ def test_send_broadcast_event_only_sends_to_one_provider_if_set_on_service(
         send_broadcast_event(event.id)
 
     assert mock_send_broadcast_provider_message.apply_async.call_args_list == [
-        call(kwargs={'broadcast_event_id': event.id, 'provider': 'vodafone'}, queue='notify-internal-tasks')
+        call(kwargs={'broadcast_event_id': event.id, 'provider': 'vodafone'}, queue='broadcast-tasks')
     ]
 
 

tests/app/celery/test_scheduled_tasks.py

@@ -603,8 +603,8 @@ def test_trigger_link_tests_calls_for_all_providers(
         trigger_link_tests()
 
     assert mock_trigger_link_test.apply_async.call_args_list == [
-        call(kwargs={'provider': 'ee'}, queue='notify-internal-tasks'),
-        call(kwargs={'provider': 'vodafone'}, queue='notify-internal-tasks')
+        call(kwargs={'provider': 'ee'}, queue='broadcast-tasks'),
+        call(kwargs={'provider': 'vodafone'}, queue='broadcast-tasks')
     ]
 
 

tests/app/test_config.py

@@ -60,7 +60,7 @@ def test_load_config_if_cloudfoundry_not_available(reload_config):
 def test_queue_names_all_queues_correct():
     # Need to ensure that all_queues() only returns queue names used in API
     queues = QueueNames.all_queues()
-    assert len(queues) == 16
+    assert len(queues) == 17
     assert set([
         QueueNames.PRIORITY,
         QueueNames.PERIODIC,
@@ -77,5 +77,6 @@ def test_queue_names_all_queues_correct():
         QueueNames.LETTERS,
         QueueNames.SMS_CALLBACKS,
         QueueNames.SAVE_API_EMAIL,
-        QueueNames.SAVE_API_SMS
+        QueueNames.SAVE_API_SMS,
+        QueueNames.BROADCASTS,
     ]) == set(queues)