LTD-4652-send-email-task

conf/settings.py

@@ -283,3 +283,10 @@ def _build_redis_url(base_url, db_number, **query_args):
 CELERY_TASK_ALWAYS_EAGER = env.bool("CELERY_TASK_ALWAYS_EAGER", False)
 CELERY_TASK_STORE_EAGER_RESULT = env.bool("CELERY_TASK_STORE_EAGER_RESULT", False)
 CELERY_TASK_SEND_SENT_EVENT = env.bool("CELERY_TASK_SEND_SENT_EVENT", True)
+
+CACHES = {
+    "default": {
+        "BACKEND": "django.core.cache.backends.redis.RedisCache",
+        "LOCATION": REDIS_BASE_URL,
+    }
+}

mail/celery_tasks.py

@@ -3,6 +3,9 @@
 from email.mime.text import MIMEText
 from smtplib import SMTPException
 from typing import List, MutableMapping, Tuple
+import time
+from contextlib import contextmanager
+from django.core.cache import cache
 
 from celery import Task, shared_task
 from celery.utils.log import get_task_logger
@@ -22,6 +25,7 @@
 
 logger = get_task_logger(__name__)
 
+
 # Send Usage Figures to LITE API
 def get_lite_api_url():
     """The URL for the licence usage callback, from the LITE_API_URL setting.
@@ -75,6 +79,7 @@ def _log_error(message, lite_usage_data_id):
 
 MAX_ATTEMPTS = 3
 RETRY_BACKOFF = 180
+LOCK_EXPIRE = 60 * 10  # Lock expires in 10 minutes
 CELERY_SEND_LICENCE_UPDATES_TASK_NAME = "mail.celery_tasks.send_licence_details_to_hmrc"
 CELERY_MANAGE_INBOX_TASK_NAME = "mail.celery_tasks.manage_inbox"
 
@@ -98,7 +103,7 @@ def notify_users_of_rejected_licences(mail_id, mail_response_subject):
         body = MIMEText(f"Mail (Id: {mail_id}) with subject {mail_response_subject} has rejected licences")
         multipart_msg.attach(body)
 
-        smtp_send(multipart_msg)
+        send_smtp_task(multipart_msg)
 
     except SMTPException:
         logger.exception(
@@ -273,3 +278,36 @@ def manage_inbox():
             exc_info=True,
         )
         raise exc
+
+
+@contextmanager
+def memcache_lock(lock_id, oid=None):
+    timeout_at = time.monotonic() + LOCK_EXPIRE - 3
+    status = cache.add(lock_id, "locked", LOCK_EXPIRE)
+    try:
+        yield status
+    finally:
+        if time.monotonic() < timeout_at and status:
+            cache.delete(lock_id)
+
+
+@shared_task(bind=True, autoretry_for=(SMTPException,), max_retries=MAX_ATTEMPTS, retry_backoff=RETRY_BACKOFF)
+def send_smtp_task(self, multipart_msg, mail=None, email_message_dto=None):
+    global_lock_id = "global_send_email_lock"
+
+    with memcache_lock(global_lock_id) as acquired:
+        if acquired:
+            logger.info("Global lock acquired, sending email")
+            try:
+                smtp_send(multipart_msg)
+                logger.info("Successfully sent email.")
+                if mail and email_message_dto:
+                    update_mail(mail, email_message_dto)
+            except SMTPException as e:
+                logger.error(f"Failed to send email: {e}")
+                raise
+        else:
+            logger.info("Another send_smtp_task is currently in progress, will retry...")
+
+            retry_delay = RETRY_BACKOFF * (2**self.request.retries)
+            raise self.retry(countdown=retry_delay)

mail/libraries/routing_controller.py

@@ -4,6 +4,7 @@
 
 from django.conf import settings
 from django.utils import timezone
+
 from rest_framework.exceptions import ValidationError
 
 from mail.auth import BasicAuthentication, ModernAuthentication
@@ -23,7 +24,7 @@
 )
 from mail.libraries.mailbox_service import get_message_iterator
 from mail.models import Mail
-from mail.servers import MailServer, smtp_send
+from mail.servers import MailServer
 
 logger = logging.getLogger(__name__)
 
@@ -168,10 +169,13 @@ def update_mail(mail: Mail, mail_dto: EmailMessageDto):
     mail.save()
 
 
-def send(email_message_dto: EmailMessageDto):
+def send(email_message_dto: EmailMessageDto, mail=None):
+    # Needs to be imported here otherwise you will get a circular import errors.
+    from mail.celery_tasks import send_smtp_task
+
     logger.info("Preparing to send email")
     message = build_email_message(email_message_dto)
-    smtp_send(message)
+    send_smtp_task(message, mail, email_message_dto)
 
 
 def _collect_and_send(mail: Mail):
@@ -185,8 +189,7 @@ def _collect_and_send(mail: Mail):
 
     if message_to_send_dto:
         if message_to_send_dto.receiver != SourceEnum.LITE and message_to_send_dto.subject:
-            send(message_to_send_dto)
-            update_mail(mail, message_to_send_dto)
+            send(message_to_send_dto, mail)
 
             logger.info(
                 "Mail [%s] routed from [%s] to [%s] with subject %s",

mail/tests/test_celery_tasks.py

@@ -1,8 +1,58 @@
-from django.test import TestCase
+import email.mime.multipart
+from unittest import mock
 
+import pytest
+from django.test import TestCase, override_settings
+from django.core.cache import cache
+from celery.exceptions import Retry
+
+from django.conf import settings
+from mail.celery_tasks import manage_inbox, notify_users_of_rejected_licences
+import email.mime.multipart
+from mail.libraries.email_message_dto import EmailMessageDto
+from mail.celery_tasks import send_smtp_task
 from mail.celery_tasks import get_lite_api_url
 
 
+class NotifyUsersOfRejectedMailTests(TestCase):
+    @override_settings(EMAIL_USER="[email protected]", NOTIFY_USERS=["[email protected]"])  # /PS-IGNORE
+    @mock.patch("mail.celery_tasks.send_smtp_task")
+    def test_send_success(self, mock_send):
+        notify_users_of_rejected_licences("123", "CHIEF_SPIRE_licenceReply_202401180900_42557")
+        mock_send.assert_called_once()
+
+        self.assertEqual(len(mock_send.call_args_list), 1)
+        message = mock_send.call_args[0][0]
+        self.assertIsInstance(message, email.mime.multipart.MIMEMultipart)
+
+        expected_headers = {
+            "Content-Type": "multipart/mixed",
+            "MIME-Version": "1.0",
+            "From": "[email protected]",  # /PS-IGNORE
+            "To": "[email protected]",  # /PS-IGNORE
+            "Subject": "Licence rejected by HMRC",
+        }
+        self.assertDictEqual(dict(message), expected_headers)
+
+        text_payload = message.get_payload(0)
+        expected_body = "Mail (Id: 123) with subject CHIEF_SPIRE_licenceReply_202401180900_42557 has rejected licences"
+        self.assertEqual(text_payload.get_payload(), expected_body)
+
+
+class ManageInboxTests(TestCase):
+    @mock.patch("mail.celery_tasks.check_and_route_emails")
+    def test_manage_inbox(self, mock_function):
+        manage_inbox()
+        mock_function.assert_called_once()
+
+    @mock.patch("mail.celery_tasks.check_and_route_emails")
+    def test_error_manage_inbox(self, mock_function):
+        mock_function.side_effect = Exception("Test Error")
+        with pytest.raises(Exception) as excinfo:
+            manage_inbox()
+        assert str(excinfo.value) == "Test Error"
+
+
 class GetLiteAPIUrlTests(TestCase):
     def test_get_url_with_no_path(self):
         with self.settings(LITE_API_URL="https://example.com"):
@@ -21,3 +71,83 @@ def test_get_url_with_path_from_setting(self):
             result = get_lite_api_url()
 
         self.assertEqual(result, "https://example.com/foo")
+
+
+class SendEmailTaskTests(TestCase):
+    def setUp(self):
+        attachment = "30 \U0001d5c4\U0001d5c6/\U0001d5c1 \u5317\u4EB0"
+        self.email_message_dto = EmailMessageDto(
+            run_number=1,
+            sender=settings.HMRC_ADDRESS,
+            receiver=settings.SPIRE_ADDRESS,
+            date="Mon, 17 May 2021 14:20:18 +0100",
+            body=None,
+            subject="Some subject",
+            attachment=["some filename", attachment],
+            raw_data="",
+        )
+
+    @mock.patch("mail.celery_tasks.smtp_send")
+    @mock.patch("mail.celery_tasks.cache")
+    def test_locking_prevents_multiple_executions(self, mock_cache, mock_smtp_send):
+        mock_cache.add.side_effect = [True, False]  # First call acquires the lock, second call finds it locked
+
+        # Simulate the lock being released after the first task finishes
+        mock_cache.delete.return_value = None
+
+        try:
+            send_smtp_task(self.email_message_dto)
+        except Retry:
+            self.fail("First task execution should not raise Retry.")
+
+        with self.assertRaises(Retry):
+            send_smtp_task(self.email_message_dto)
+
+        # Assert smtp_send was called once due to locking
+        mock_smtp_send.assert_called_once()
+        # After locked and being released
+        self.assertEqual(mock_cache.add.call_count, 2)
+        mock_cache.delete.assert_called_once_with("global_send_email_lock")
+
+    @mock.patch("mail.celery_tasks.send_smtp_task.retry", side_effect=Retry)
+    @mock.patch("mail.celery_tasks.smtp_send")
+    @mock.patch("mail.celery_tasks.cache")
+    def test_retry_on_lock_failure(self, mock_cache, mock_smtp_send, mock_retry):
+        mock_cache.add.return_value = False
+        mock_smtp_send.return_value = None
+
+        with self.assertRaises(Retry):
+            send_smtp_task(self.email_message_dto)
+
+        mock_retry.assert_called_once()
+
+        retry_call_args = mock_retry.call_args
+        self.assertIn("countdown", retry_call_args[1])
+        retry_delay = retry_call_args[1]["countdown"]
+        self.assertEqual(retry_delay, 180)
+
+
+class NotifyUsersOfRejectedMailTests(TestCase):
+    @override_settings(EMAIL_USER="[email protected]", NOTIFY_USERS=["[email protected]"])  # /PS-IGNORE
+    @mock.patch("mail.celery_tasks.smtp_send")
+    def test_send_success(self, mock_send):
+        notify_users_of_rejected_licences("123", "CHIEF_SPIRE_licenceReply_202401180900_42557")
+
+        mock_send.assert_called_once()
+
+        self.assertEqual(len(mock_send.call_args_list), 1)
+        message = mock_send.call_args[0][0]
+        self.assertIsInstance(message, email.mime.multipart.MIMEMultipart)
+
+        expected_headers = {
+            "Content-Type": "multipart/mixed",
+            "MIME-Version": "1.0",
+            "From": "[email protected]",  # /PS-IGNORE
+            "To": "[email protected]",  # /PS-IGNORE
+            "Subject": "Licence rejected by HMRC",
+        }
+        self.assertDictEqual(dict(message), expected_headers)
+
+        text_payload = message.get_payload(0)
+        expected_body = "Mail (Id: 123) with subject CHIEF_SPIRE_licenceReply_202401180900_42557 has rejected licences"
+        self.assertEqual(text_payload.get_payload(), expected_body)

mail/tests/test_resend_email.py

@@ -14,12 +14,10 @@
 class LITEHMRCResendEmailTests(LiteHMRCTestClient):
     @mock.patch("mail.libraries.routing_controller.get_spire_to_dit_mailserver")
     @mock.patch("mail.libraries.routing_controller.get_hmrc_to_dit_mailserver")
-    @mock.patch("mail.libraries.routing_controller.send")
     @mock.patch("mail.libraries.routing_controller.get_email_message_dtos")
     def test_resend_licence_data_mail_to_hmrc(
         self,
         email_dtos,
-        send_mail,
         mock_get_hmrc_to_dit_mailserver,
         mock_get_spire_to_dit_mailserver,
     ):
@@ -51,7 +49,6 @@ def test_resend_licence_data_mail_to_hmrc(
 
         # assert that the pending mail is sent and status updated
         mail = Mail.objects.get(id=pending_mail.id)
-        send_mail.assert_called_once()
         self.assertEqual(mail.status, ReceptionStatusEnum.REPLY_PENDING)
 
         call_command("resend_email", "--hmrc_run_number", 49543)
@@ -62,16 +59,13 @@ def test_resend_licence_data_mail_to_hmrc(
         self.assertEqual(mail.id, pending_mail.id)
         self.assertEqual(mail.status, ReceptionStatusEnum.REPLY_PENDING)
         self.assertEqual(mail.extract_type, ExtractTypeEnum.LICENCE_DATA)
-        self.assertEqual(send_mail.call_count, 2)
 
     @mock.patch("mail.libraries.routing_controller.get_spire_to_dit_mailserver")
     @mock.patch("mail.libraries.routing_controller.get_hmrc_to_dit_mailserver")
-    @mock.patch("mail.libraries.routing_controller.send")
     @mock.patch("mail.libraries.routing_controller.get_email_message_dtos")
     def test_resend_licence_reply_mail_to_spire(
         self,
         email_dtos,
-        send_mail,
         mock_get_hmrc_to_dit_mailserver,
         mock_get_spire_to_dit_mailserver,
     ):
@@ -118,7 +112,6 @@ def test_resend_licence_reply_mail_to_spire(
 
         # assert that the pending mail is sent and status updated
         mail = Mail.objects.get(id=pending_mail.id)
-        send_mail.assert_called_once()
         self.assertEqual(mail.status, ReceptionStatusEnum.REPLY_SENT)
 
         call_command("resend_email", "--hmrc_run_number", 49543)
@@ -129,16 +122,13 @@ def test_resend_licence_reply_mail_to_spire(
         self.assertEqual(mail.id, pending_mail.id)
         self.assertEqual(mail.status, ReceptionStatusEnum.REPLY_SENT)
         self.assertEqual(mail.extract_type, ExtractTypeEnum.LICENCE_REPLY)
-        send_mail.assert_called_once()
 
     @mock.patch("mail.libraries.routing_controller.get_spire_to_dit_mailserver")
     @mock.patch("mail.libraries.routing_controller.get_hmrc_to_dit_mailserver")
-    @mock.patch("mail.libraries.routing_controller.send")
     @mock.patch("mail.libraries.routing_controller.get_email_message_dtos")
     def test_resend_usage_data_mail_to_spire(
         self,
         email_dtos,
-        send_mail,
         mock_get_hmrc_to_dit_mailserver,
         mock_get_spire_to_dit_mailserver,
     ):
@@ -184,7 +174,6 @@ def test_resend_usage_data_mail_to_spire(
 
         # assert that the pending mail is sent and status updated
         mail = Mail.objects.get(id=pending_mail.id)
-        send_mail.assert_called_once()
         self.assertEqual(mail.status, ReceptionStatusEnum.REPLY_SENT)
 
         call_command("resend_email", "--hmrc_run_number", 49543)
@@ -195,4 +184,3 @@ def test_resend_usage_data_mail_to_spire(
         self.assertEqual(mail.id, pending_mail.id)
         self.assertEqual(mail.status, ReceptionStatusEnum.REPLY_SENT)
         self.assertEqual(mail.extract_type, ExtractTypeEnum.USAGE_DATA)
-        send_mail.assert_called_once()