Updates email method to have attachments

oppia · Nik-09 · Dec 30, 2024 · Dec 2, 2024 · Dec 4, 2024 · Dec 7, 2024
commit 5d63a5f6b110f141096d9c8e03de6e2644d0142b
@@ -20,6 +20,8 @@
 
 import base64
 import urllib
+import requests
+import logging
 
 from core import feconf
 from core import utils
@@ -43,7 +45,8 @@ def send_email_to_recipients(
     bcc: Optional[List[str]] = None,
     reply_to: Optional[str] = None,
     recipient_variables: Optional[
-        Dict[str, Dict[str, Union[str, float]]]] = None
+        Dict[str, Dict[str, Union[str, float]]]] = None,
+    attachments: Optional[List[str]] = None
 ) -> bool:
     """Send POST HTTP request to mailgun api. This method is adopted from
     the requests library's post method.
@@ -74,6 +77,8 @@ def send_email_to_recipients(
             More info about this format at:
             https://documentation.mailgun.com/en/
                 latest/user_manual.html#batch-sending.
+        attachments: list(str)|None. Optional argument. List of file paths
+            for attachments.
 
     Raises:
         Exception. The mailgun api key is not stored in
@@ -95,16 +100,17 @@ def send_email_to_recipients(
     # To send bulk emails we pass list of recipients in 'to' paarameter of
     # post data. Maximum limit of recipients per request is 1000.
     # For more detail check following link:
-    # https://documentation.mailgun.com/user_manual.html#batch-sending
+    # https://documentation.mailgun.com/docs/mailgun/user-manual/
+    # sending-messages/#batch-sending
     recipient_email_lists = [
         recipient_emails[i:i + 1000]
         for i in range(0, len(recipient_emails), 1000)]
     for email_list in recipient_email_lists:
         data = {
             'from': sender_email,
-            'subject': subject.encode('utf-8'),
-            'text': plaintext_body.encode('utf-8'),
-            'html': html_body.encode('utf-8'),
+            'subject': subject,
+            'text': plaintext_body,
+            'html': html_body,
             'to': email_list[0] if len(email_list) == 1 else email_list
         }
 
@@ -118,28 +124,35 @@ def send_email_to_recipients(
         # email to each recipient (This is intended to be a workaround for
         # sending individual emails).
         data['recipient_variables'] = recipient_variables or {}
-
-        # The b64encode accepts and returns bytes, so we first need to encode
-        # the MAILGUN_API_KEY to bytes, then decode the returned bytes back
-        # to string.
-        base64_mailgun_api_key = base64.b64encode(
-            b'api:%b' % mailgun_api_key.encode('utf-8')
-        ).strip().decode('utf-8')
-        auth_str = 'Basic %s' % base64_mailgun_api_key
-        header = {'Authorization': auth_str}
         server = 'https://api.mailgun.net/v3/%s/messages' % (
             feconf.MAILGUN_DOMAIN_NAME
         )
-        # The 'ascii' is used here, because only ASCII char are allowed in url,
-        # also the docs recommend this approach:
-        # https://docs.python.org/3.7/library/urllib.request.html#urllib-examples
-        encoded_url = urllib.parse.urlencode(data).encode('ascii')
-        req = urllib.request.Request(server, encoded_url, header)
-        resp = utils.url_open(req)
-        # The function url_open returns a file_like object that can be queried
-        # for the status code of the url query. If it is not 200, the mail query
-        # failed so we return False (this function did not complete
-        # successfully).
-        if resp.getcode() != 200:
+
+       # Adding attachments to the email.
+        files = [(
+            "attachment", 
+            (attachment["filename"], open(attachment["path"], "rb"))) 
+            for attachment in attachments
+        ] if attachments else None
+
+        try:
+            response = requests.post(
+                server,
+                auth=("api", mailgun_api_key),
+                data=data,
+                files=files
+            )
+            # Close file objects after the request.
+            if files:
+                for _, (_, file_obj) in files:
+                    file_obj.close()
+
+            if response.status_code != 200:
+                logging.error(
+                    "Failed to send email: %s - %s." 
+                    % (response.status_code, response.text))
+                return False
+        except requests.RequestException as e:
+            logging.error("Failed to send email: %s." %e)
             return False
     return True
@@ -19,6 +19,10 @@
 from __future__ import annotations
 
 import urllib
+import requests
+import unittest
+from unittest.mock import patch, Mock
+import os
 
 from core import feconf
 from core import utils
@@ -67,37 +71,95 @@ def getcode(self) -> int:
             """
             return 200 if self.url == self.expected_url else 500
 
+    @patch('requests.post')
     def test_send_email_to_mailgun_without_bcc_reply_to_and_recipients(
-        self
+        self, mock_post: Mock
     ) -> None:
         """Test for sending HTTP POST request."""
         # Test sending email without bcc, reply_to or recipient_variables.
-        expected_query_url: MailgunQueryType = (
-            'https://api.mailgun.net/v3/domain/messages',
-            b'from=a%40a.com&'
-            b'subject=Hola+%F0%9F%98%82+-+invitation+to+collaborate&'
-            b'text=plaintext_body+%F0%9F%98%82&'
-            b'html=Hi+abc%2C%3Cbr%3E+%F0%9F%98%82&'
-            b'to=b%40b.com&'
-            b'recipient_variables=%7B%7D',
-            {'Authorization': 'Basic YXBpOmtleQ=='}
+
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_post.return_value = mock_response
+
+        sender_email = 'a@a.com'
+        recipient_emails = 'b@b.com'
+        subject = 'Hola 😂 - invitation to collaborate'
+        plaintext_body = 'plaintext_body 😂'
+        html_body = 'Hi abc,<br> 😂'
+        attachments = None
+
+        swap_domain = self.swap(feconf, 'MAILGUN_DOMAIN_NAME', 'domain')
+
+        with self.swap_api_key_secrets_return_secret, swap_domain:
+
+            resp = mailgun_email_services.send_email_to_recipients(
+                sender_email,
+                recipient_emails,
+                subject,
+                plaintext_body,
+                html_body
+            )
+
+        expected_data = {
+            'from': sender_email,
+            'subject': subject,
+            'text': plaintext_body,
+            'html': html_body,
+            'to': recipient_emails,
+            'recipient_variables': {}
+        }
+
+        mock_post.assert_called_once_with(
+            f'https://api.mailgun.net/v3/domain/messages',
+            auth=('api', 'key'),
+            data=expected_data,
+            files=attachments
         )
-        swapped_urlopen = lambda x: self.Response(x, expected_query_url)
+        self.assertTrue(resp)
+
+    @patch("requests.post")
+    def test_send_email_to_mailgun_with_file_attachments(self, mock_post):
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_post.return_value = mock_response
+
+        sender_email = "sender@example.com"
+        recipient_emails = ["recipient@example.com"]
+        subject = "Test Email with attachment"
+        plaintext_body = "This is a test email with an attachment."
+        html_body = 'Hi abc,<br> 😂'
+        file_path = "test_file.txt"
+        attachments = [{'filename': 'test_file.txt', 'path': file_path}]
+
+        with open(file_path, 'w') as f:
+            f.write("This is a test file.")
 
-        swap_urlopen_context = self.swap(
-            utils, 'url_open', swapped_urlopen)
-        swap_request_context = self.swap(
-            urllib.request, 'Request', self.swapped_request)
         swap_domain = self.swap(feconf, 'MAILGUN_DOMAIN_NAME', 'domain')
-        with self.swap_api_key_secrets_return_secret, swap_urlopen_context:
-            with swap_request_context, swap_domain:
-                resp = mailgun_email_services.send_email_to_recipients(
-                    'a@a.com',
-                    ['b@b.com'],
-                    'Hola 😂 - invitation to collaborate',
-                    'plaintext_body 😂',
-                    'Hi abc,<br> 😂')
-                self.assertTrue(resp)
+
+        with self.swap_api_key_secrets_return_secret, swap_domain:
+            resp = mailgun_email_services.send_email_to_recipients(
+                sender_email,
+                recipient_emails,
+                subject,
+                plaintext_body,
+                html_body,
+                attachments=attachments
+            )
+
+        mock_post.assert_called_once()
+        _, kwargs = mock_post.call_args
+
+        os.remove(file_path)
+        self.assertTrue(resp)
+        self.assertEqual(kwargs['auth'], ('api', 'key'))
+        self.assertEqual(kwargs['data']['from'], sender_email)
+        self.assertEqual(kwargs['data']['to'], recipient_emails[0])
+        self.assertEqual(kwargs['data']['subject'], subject)
+        self.assertEqual(kwargs['data']['text'], plaintext_body)
+        self.assertEqual(kwargs['data']['html'], html_body)
+        self.assertIn('files', kwargs)
+        self.assertEqual(kwargs['files'][0][1][0], 'test_file.txt')
 
     def test_send_email_to_mailgun_with_bcc_and_recipient(self) -> None:
         # Test sending email with single bcc and single recipient email.