merge from main

tests/app/celery/test_reporting_tasks.py

@@ -103,7 +103,6 @@ def test_create_nightly_notification_status_triggers_relevant_tasks(
     mock_celery = mocker.patch(
         "app.celery.reporting_tasks.create_nightly_notification_status_for_service_and_day"
     ).apply_async
-
     for notification_type in NotificationType:
         template = create_template(sample_service, template_type=notification_type)
         create_notification(template=template, created_at=notification_date)

tests/app/celery/test_scheduled_tasks.py

@@ -1,17 +1,20 @@
+import json
 from collections import namedtuple
 from datetime import timedelta
 from unittest import mock
-from unittest.mock import ANY, call
+from unittest.mock import ANY, MagicMock, call
 
 import pytest
 
 from app.celery import scheduled_tasks
 from app.celery.scheduled_tasks import (
+    batch_insert_notifications,
     check_for_missing_rows_in_completed_jobs,
     check_for_services_with_high_failure_rates_or_sending_to_tv_numbers,
     check_job_status,
     delete_verify_codes,
     expire_or_delete_invitations,
+    process_delivery_receipts,
     replay_created_notifications,
     run_scheduled_jobs,
 )
@@ -308,10 +311,10 @@ def test_replay_created_notifications(notify_db_session, sample_service, mocker)
 
     replay_created_notifications()
     email_delivery_queue.assert_called_once_with(
-        [str(old_email.id)], queue="send-email-tasks"
+        [str(old_email.id)], queue="send-email-tasks", countdown=60
     )
     sms_delivery_queue.assert_called_once_with(
-        [str(old_sms.id)], queue="send-sms-tasks"
+        [str(old_sms.id)], queue="send-sms-tasks", countdown=60
     )
 
 
@@ -523,3 +526,101 @@ def test_check_for_services_with_high_failure_rates_or_sending_to_tv_numbers(
         technical_ticket=True,
     )
     mock_send_ticket_to_zendesk.assert_called_once()
+
+
+def test_batch_insert_with_valid_notifications(mocker):
+    mocker.patch("app.celery.scheduled_tasks.dao_batch_insert_notifications")
+    rs = MagicMock()
+    mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
+    notifications = [
+        {"id": 1, "notification_status": "pending"},
+        {"id": 2, "notification_status": "pending"},
+    ]
+    serialized_notifications = [json.dumps(n).encode("utf-8") for n in notifications]
+
+    pipeline_mock = MagicMock()
+
+    rs.pipeline.return_value.__enter__.return_value = pipeline_mock
+    rs.llen.return_value = len(notifications)
+    rs.lpop.side_effect = serialized_notifications
+
+    batch_insert_notifications()
+
+    rs.llen.assert_called_once_with("message_queue")
+    rs.lpop.assert_called_with("message_queue")
+
+
+def test_batch_insert_with_expired_notifications(mocker):
+    expired_time = utc_now() - timedelta(minutes=2)
+    mocker.patch(
+        "app.celery.scheduled_tasks.dao_batch_insert_notifications",
+        side_effect=Exception("DB Error"),
+    )
+    rs = MagicMock()
+    mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
+    notifications = [
+        {
+            "id": 1,
+            "notification_status": "pending",
+            "created_at": utc_now().isoformat(),
+        },
+        {
+            "id": 2,
+            "notification_status": "pending",
+            "created_at": expired_time.isoformat(),
+        },
+    ]
+    serialized_notifications = [json.dumps(n).encode("utf-8") for n in notifications]
+
+    pipeline_mock = MagicMock()
+
+    rs.pipeline.return_value.__enter__.return_value = pipeline_mock
+    rs.llen.return_value = len(notifications)
+    rs.lpop.side_effect = serialized_notifications
+
+    batch_insert_notifications()
+
+    rs.llen.assert_called_once_with("message_queue")
+    rs.rpush.assert_called_once()
+    requeued_notification = json.loads(rs.rpush.call_args[0][1])
+    assert requeued_notification["id"] == 1
+
+
+def test_batch_insert_with_malformed_notifications(mocker):
+    rs = MagicMock()
+    mocker.patch("app.celery.scheduled_tasks.redis_store", rs)
+    malformed_data = b"not_a_valid_json"
+    pipeline_mock = MagicMock()
+
+    rs.pipeline.return_value.__enter__.return_value = pipeline_mock
+    rs.llen.return_value = 1
+    rs.lpop.side_effect = [malformed_data]
+
+    with pytest.raises(json.JSONDecodeError):
+        batch_insert_notifications()
+
+    rs.llen.assert_called_once_with("message_queue")
+    rs.rpush.assert_not_called()
+
+
+def test_process_delivery_receipts_success(mocker):
+    dao_update_mock = mocker.patch(
+        "app.celery.scheduled_tasks.dao_update_delivery_receipts"
+    )
+    cloudwatch_mock = mocker.patch("app.celery.scheduled_tasks.AwsCloudwatchClient")
+    cloudwatch_mock.return_value.check_delivery_receipts.return_value = (
+        range(2000),
+        range(500),
+    )
+    current_app_mock = mocker.patch("app.celery.scheduled_tasks.current_app")
+    current_app_mock.return_value = MagicMock()
+    processor = MagicMock()
+    processor.process_delivery_receipts = process_delivery_receipts
+    processor.retry = MagicMock()
+
+    processor.process_delivery_receipts()
+    assert dao_update_mock.call_count == 3
+    dao_update_mock.assert_any_call(list(range(1000)), True)
+    dao_update_mock.assert_any_call(list(range(1000, 2000)), True)
+    dao_update_mock.assert_any_call(list(range(500)), False)
+    processor.retry.assert_not_called()

tests/app/celery/test_tasks.py

@@ -434,7 +434,7 @@ def test_should_send_template_to_correct_sms_task_and_persist(
     assert persisted_notification.personalisation == {}
     assert persisted_notification.notification_type == NotificationType.SMS
     mocked_deliver_sms.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks"
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
     )
 
 
@@ -475,7 +475,7 @@ def test_should_save_sms_if_restricted_service_and_valid_number(
     assert not persisted_notification.personalisation
     assert persisted_notification.notification_type == NotificationType.SMS
     provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks"
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
     )
 
 
@@ -608,7 +608,7 @@ def test_should_save_sms_template_to_and_persist_with_job_id(sample_job, mocker)
     assert persisted_notification.notification_type == NotificationType.SMS
 
     provider_tasks.deliver_sms.apply_async.assert_called_once_with(
-        [str(persisted_notification.id)], queue="send-sms-tasks"
+        [str(persisted_notification.id)], queue="send-sms-tasks", countdown=60
     )
 
 
@@ -948,7 +948,7 @@ def test_save_sms_uses_sms_sender_reply_to_text(mocker, notify_db_session):
     notification = _notification_json(template, to="2028675301")
     mocker.patch("app.celery.provider_tasks.deliver_sms.apply_async")
 
-    notification_id = uuid.uuid4()
+    notification_id = str(uuid.uuid4())
     save_sms(
         service.id,
         notification_id,