pacnpal/django-anymail
1
0
Fork 0
mirror of https://github.com/pacnpal/django-anymail.git synced 2025-12-20 03:41:05 -05:00
Code Issues Packages Projects Releases Wiki Activity

MailerSend: add support (new ESP for Anymail)

Browse Source 
Closes #298
This commit is contained in:
Mike Edmunds
2023-03-10 17:22:20 -08:00
committed by GitHub
parent c58640d438
commit 62bd0669af
15 changed files with 3093 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
tests/mock_requests_backend.py
View File

@@ -17,6 +17,7 @@ class RequestsBackendMockAPITestCase(AnymailTestMixin, SimpleTestCase):
"""TestCase that mocks API calls through requests"""
DEFAULT_RAW_RESPONSE = b"""{"subclass": "should override"}"""
DEFAULT_CONTENT_TYPE = None # e.g., "application/json"
DEFAULT_STATUS_CODE = 200 # most APIs use '200 OK' for success
class MockResponse(requests.Response):
@@ -26,6 +27,7 @@ class RequestsBackendMockAPITestCase(AnymailTestMixin, SimpleTestCase):
self,
status_code=200,
raw=b"RESPONSE",
content_type=None,
encoding="utf-8",
reason=None,
test_case=None,
@@ -35,6 +37,8 @@ class RequestsBackendMockAPITestCase(AnymailTestMixin, SimpleTestCase):
self.encoding = encoding
self.reason = reason or ("OK" if 200 <= status_code < 300 else "ERROR")
self.raw = BytesIO(raw)
if content_type is not None:
self.headers["Content-Type"] = content_type
self.test_case = test_case
@property
@@ -54,12 +58,32 @@ class RequestsBackendMockAPITestCase(AnymailTestMixin, SimpleTestCase):
self.set_mock_response()
def set_mock_response(
self, status_code=DEFAULT_STATUS_CODE, raw=UNSET, encoding="utf-8", reason=None
self,
status_code=UNSET,
raw=UNSET,
json_data=UNSET,
encoding="utf-8",
content_type=UNSET,
reason=None,
):
if status_code is UNSET:
status_code = self.DEFAULT_STATUS_CODE
if json_data is not UNSET:
assert raw is UNSET, "provide json_data or raw, not both"
raw = json.dumps(json_data).encode(encoding)
if content_type is UNSET:
content_type = "application/json"
if raw is UNSET:
raw = self.DEFAULT_RAW_RESPONSE
if content_type is UNSET:
content_type = self.DEFAULT_CONTENT_TYPE
mock_response = self.MockResponse(
status_code, raw=raw, encoding=encoding, reason=reason, test_case=self
status_code,
raw=raw,
content_type=content_type,
encoding=encoding,
reason=reason,
test_case=self,
)
self.mock_request.return_value = mock_response
return mock_response

920
tests/test_mailersend_backend.py Normal file
View File

@@ -0,0 +1,920 @@
from calendar import timegm
from datetime import date, datetime
from decimal import Decimal
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from django.core import mail
from django.test import override_settings, tag
from django.utils.timezone import (
get_fixed_timezone,
override as override_current_timezone,
)
from anymail.exceptions import (
AnymailAPIError,
AnymailConfigurationError,
AnymailRecipientsRefused,
AnymailSerializationError,
AnymailUnsupportedFeature,
)
from anymail.message import attach_inline_image_file
from .mock_requests_backend import RequestsBackendMockAPITestCase
from .utils import (
SAMPLE_IMAGE_FILENAME,
decode_att,
sample_image_content,
sample_image_path,
)
@tag("mailersend")
@override_settings(
EMAIL_BACKEND="anymail.backends.mailersend.EmailBackend",
ANYMAIL={"MAILERSEND_API_TOKEN": "test_api_token"},
)
class MailerSendBackendMockAPITestCase(RequestsBackendMockAPITestCase):
"""TestCase that uses MailerSend EmailBackend with a mocked API"""
DEFAULT_STATUS_CODE = 202
DEFAULT_RAW_RESPONSE = b""
DEFAULT_CONTENT_TYPE = "text/html"
def setUp(self):
super().setUp()
# Simple message useful for many tests
self.message = mail.EmailMultiAlternatives(
"Subject", "Text Body", "from@example.com", ["to@example.com"]
)
def set_mock_success(self, message_id="1234567890abcdef"):
response = self.set_mock_response()
if message_id is not None:
response.headers["x-message-id"] = message_id
return response
def set_mock_rejected(
self, rejections, warning_type="ALL_SUPPRESSED", message_id="1234567890abcdef"
):
"""rejections should be a dict of {email: [reject_reason, ...], ...}"""
if warning_type == "ALL_SUPPRESSED":
message_id = None
response = self.set_mock_response(
json_data={
"warnings": [
{
"type": warning_type,
"recipients": [
{"email": email, "reasons": reasons}
for email, reasons in rejections.items()
],
}
]
}
)
if message_id is not None:
response.headers["x-message-id"] = message_id
return response
@tag("mailersend")
class MailerSendBackendStandardEmailTests(MailerSendBackendMockAPITestCase):
"""Test backend support for Django standard email features"""
def test_send_mail(self):
"""Test basic API for simple send"""
mail.send_mail(
"Subject here",
"Here is the message.",
"from@example.com",
["to@example.com"],
fail_silently=False,
)
self.assert_esp_called("/v1/email")
headers = self.get_api_call_headers()
self.assertEqual(headers["Authorization"], "Bearer test_api_token")
data = self.get_api_call_json()
self.assertEqual(data["subject"], "Subject here")
self.assertEqual(data["text"], "Here is the message.")
self.assertEqual(data["from"], {"email": "from@example.com"})
self.assertEqual(data["to"], [{"email": "to@example.com"}])
def test_name_addr(self):
"""Make sure RFC2822 name-addr format (with display-name) is allowed
(Test both sender and recipient addresses)
"""
msg = mail.EmailMessage(
"Subject",
"Message",
"From Name <from@example.com>",
["Recipient #1 <to1@example.com>", "to2@example.com"],
cc=["Carbon Copy <cc1@example.com>", "cc2@example.com"],
bcc=["Blind Copy <bcc1@example.com>", "bcc2@example.com"],
)
msg.send()
data = self.get_api_call_json()
self.assertEqual(
data["from"], {"email": "from@example.com", "name": "From Name"}
)
self.assertEqual(
data["to"],
[
{"email": "to1@example.com", "name": "Recipient #1"},
{"email": "to2@example.com"},
],
)
self.assertEqual(
data["cc"],
[
{"email": "cc1@example.com", "name": "Carbon Copy"},
{"email": "cc2@example.com"},
],
)
self.assertEqual(
data["bcc"],
[
{"email": "bcc1@example.com", "name": "Blind Copy"},
{"email": "bcc2@example.com"},
],
)
def test_custom_headers(self):
email = mail.EmailMessage(
"Subject",
"Body goes here",
"from@example.com",
["to1@example.com"],
headers={
"Reply-To": "another@example.com",
"In-Reply-To": "12345@example.com",
"X-MyHeader": "my value",
"Message-ID": "mycustommsgid@example.com",
"Precedence": "Bulk",
},
)
with self.assertRaisesMessage(AnymailUnsupportedFeature, "extra_headers"):
email.send()
def test_supported_custom_headers(self):
email = mail.EmailMessage(
"Subject",
"Body goes here",
"from@example.com",
["to1@example.com"],
headers={
"Reply-To": "another@example.com",
"In-Reply-To": "12345@example.com",
"Precedence": "Bulk",
},
)
email.send()
data = self.get_api_call_json()
self.assertEqual(data["reply_to"], {"email": "another@example.com"})
self.assertEqual(data["in_reply_to"], "12345@example.com")
self.assertIs(data["precedence_bulk"], True)
def test_html_message(self):
text_content = "This is an important message."
html_content = "<p>This is an <strong>important</strong> message.</p>"
email = mail.EmailMultiAlternatives(
"Subject", text_content, "from@example.com", ["to@example.com"]
)
email.attach_alternative(html_content, "text/html")
email.send()
data = self.get_api_call_json()
self.assertEqual(data["text"], text_content)
self.assertEqual(data["html"], html_content)
# Don't accidentally send the html part as an attachment:
self.assertNotIn("attachments", data)
def test_html_only_message(self):
html_content = "<p>This is an <strong>important</strong> message.</p>"
email = mail.EmailMessage(
"Subject", html_content, "from@example.com", ["to@example.com"]
)
email.content_subtype = "html" # Main content is now text/html
email.send()
data = self.get_api_call_json()
self.assertNotIn("text", data)
self.assertEqual(data["html"], html_content)
def test_reply_to(self):
email = mail.EmailMessage(
"Subject",
"Body goes here",
"from@example.com",
["to1@example.com"],
reply_to=["Reply Name <reply@example.com>"],
)
email.send()
data = self.get_api_call_json()
self.assertEqual(
data["reply_to"], {"email": "reply@example.com", "name": "Reply Name"}
)
def test_attachments(self):
text_content = "* Item one\n* Item two\n* Item three"
self.message.attach(
filename="test.txt", content=text_content, mimetype="text/plain"
)
# Should guess mimetype if not provided...
png_content = b"PNG\xb4 pretend this is the contents of a png file"
self.message.attach(filename="test.png", content=png_content)
# Should work with a MIMEBase object (also tests no filename)...
pdf_content = b"PDF\xb4 pretend this is valid pdf params"
mimeattachment = MIMEBase("application", "pdf")
mimeattachment.set_payload(pdf_content)
self.message.attach(mimeattachment)
self.message.send()
data = self.get_api_call_json()
attachments = data["attachments"]
self.assertEqual(len(attachments), 3)
self.assertEqual(attachments[0]["disposition"], "attachment")
self.assertEqual(attachments[0]["filename"], "test.txt")
self.assertEqual(
decode_att(attachments[0]["content"]).decode("ascii"), text_content
)
self.assertEqual(attachments[1]["disposition"], "attachment")
self.assertEqual(attachments[1]["filename"], "test.png")
self.assertEqual(decode_att(attachments[1]["content"]), png_content)
self.assertEqual(attachments[2]["disposition"], "attachment")
self.assertEqual(attachments[2]["filename"], "attachment.pdf") # generated
self.assertEqual(decode_att(attachments[2]["content"]), pdf_content)
def test_unicode_attachment_correctly_decoded(self):
# Slight modification from the Django unicode docs:
# http://django.readthedocs.org/en/latest/ref/unicode.html#email
self.message.attach(
"Une pièce jointe.html", "<p>\u2019</p>", mimetype="text/html"
)
self.message.send()
data = self.get_api_call_json()
attachments = data["attachments"]
self.assertEqual(len(attachments), 1)
def test_embedded_images(self):
image_filename = SAMPLE_IMAGE_FILENAME
image_path = sample_image_path(image_filename)
image_data = sample_image_content(image_filename)
cid = attach_inline_image_file(self.message, image_path)
html_content = (
'<p>This has an <img src="cid:%s" alt="inline" /> image.</p>' % cid
)
self.message.attach_alternative(html_content, "text/html")
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data["html"], html_content)
self.assertEqual(len(data["attachments"]), 1)
self.assertEqual(data["attachments"][0]["disposition"], "inline")
self.assertEqual(data["attachments"][0]["filename"], image_filename)
self.assertEqual(data["attachments"][0]["id"], cid)
self.assertEqual(decode_att(data["attachments"][0]["content"]), image_data)
def test_attached_images(self):
image_filename = SAMPLE_IMAGE_FILENAME
image_path = sample_image_path(image_filename)
image_data = sample_image_content(image_filename)
# option 1: attach as a file
self.message.attach_file(image_path)
# option 2: construct the MIMEImage and attach it directly
image = MIMEImage(image_data)
self.message.attach(image)
self.message.send()
data = self.get_api_call_json()
attachments = data["attachments"]
self.assertEqual(len(attachments), 2)
self.assertEqual(attachments[0]["disposition"], "attachment")
self.assertEqual(attachments[0]["filename"], image_filename)
self.assertEqual(decode_att(attachments[0]["content"]), image_data)
self.assertNotIn("id", attachments[0]) # not inline
self.assertEqual(attachments[1]["disposition"], "attachment")
self.assertEqual(attachments[1]["filename"], "attachment.png") # generated
self.assertEqual(decode_att(attachments[1]["content"]), image_data)
self.assertNotIn("id", attachments[0]) # not inline
def test_multiple_html_alternatives(self):
# Multiple text/html alternatives not allowed
self.message.attach_alternative("<p>First html is OK</p>", "text/html")
self.message.attach_alternative("<p>But not second html</p>", "text/html")
with self.assertRaises(AnymailUnsupportedFeature):
self.message.send()
def test_html_alternative(self):
# Only html alternatives allowed
self.message.attach_alternative("{'not': 'allowed'}", "application/json")
with self.assertRaises(AnymailUnsupportedFeature):
self.message.send()
def test_alternatives_fail_silently(self):
# Make sure fail_silently is respected
self.message.attach_alternative("{'not': 'allowed'}", "application/json")
sent = self.message.send(fail_silently=True)
self.assert_esp_not_called("API should not be called when send fails silently")
self.assertEqual(sent, 0)
def test_suppress_empty_address_lists(self):
"""Empty cc, bcc, and reply_to shouldn't generate empty headers"""
self.message.send()
data = self.get_api_call_json()
self.assertNotIn("cc", data)
self.assertNotIn("bcc", data)
self.assertNotIn("reply_to", data)
# MailerSend requires at least one "to" address
# def test_empty_to(self):
# self.message.to = []
# self.message.cc = ["cc@example.com"]
# self.message.send()
# data = self.get_api_call_json()
def test_api_failure(self):
raw_errors = {
"message": "Helpful ESP explanation",
"errors": {"some.field": ["The some.field must be valid."]},
}
self.set_mock_response(
status_code=422, reason="UNPROCESSABLE ENTITY", json_data=raw_errors
)
# Error string includes ESP response:
with self.assertRaisesMessage(AnymailAPIError, "Helpful ESP explanation"):
self.message.send()
def test_api_failure_fail_silently(self):
# Make sure fail_silently is respected
self.set_mock_response(status_code=422)
sent = self.message.send(fail_silently=True)
self.assertEqual(sent, 0)
@tag("mailersend")
class MailerSendBackendAnymailFeatureTests(MailerSendBackendMockAPITestCase):
"""Test backend support for Anymail added features"""
def test_envelope_sender(self):
self.message.envelope_sender = "bounce-handler@bounces.example.com"
with self.assertRaisesMessage(AnymailUnsupportedFeature, "envelope_sender"):
self.message.send()
def test_metadata(self):
self.message.metadata = {"user_id": "12345", "items": "mailer, send"}
with self.assertRaisesMessage(AnymailUnsupportedFeature, "metadata"):
self.message.send()
def test_send_at(self):
utc_plus_6 = get_fixed_timezone(6 * 60)
utc_minus_8 = get_fixed_timezone(-8 * 60)
with override_current_timezone(utc_plus_6):
# Timezone-aware datetime converted to UTC:
self.message.send_at = datetime(2016, 3, 4, 5, 6, 7, tzinfo=utc_minus_8)
self.message.send()
data = self.get_api_call_json()
self.assertEqual(
data["send_at"], timegm((2016, 3, 4, 13, 6, 7))
) # 05:06 UTC-8 == 13:06 UTC
# Timezone-naive datetime assumed to be Django current_timezone
self.message.send_at = datetime(
2022, 10, 11, 12, 13, 14, 567
) # microseconds should get stripped
self.message.send()
data = self.get_api_call_json()
self.assertEqual(
data["send_at"], timegm((2022, 10, 11, 6, 13, 14))
) # 12:13 UTC+6 == 06:13 UTC
# Date-only treated as midnight in current timezone
self.message.send_at = date(2022, 10, 22)
self.message.send()
data = self.get_api_call_json()
self.assertEqual(
data["send_at"], timegm((2022, 10, 21, 18, 0, 0))
) # 00:00 UTC+6 == 18:00-1d UTC
# POSIX timestamp
self.message.send_at = 1651820889 # 2022-05-06 07:08:09 UTC
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data["send_at"], 1651820889)
def test_tags(self):
self.message.tags = ["receipt", "repeat-user"]
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data["tags"], ["receipt", "repeat-user"])
def test_tracking(self):
# Test one way...
self.message.track_opens = True
self.message.track_clicks = False
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data["settings"]["track_opens"], True)
self.assertEqual(data["settings"]["track_clicks"], False)
# ...and the opposite way
self.message.track_opens = False
self.message.track_clicks = True
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data["settings"]["track_opens"], False)
self.assertEqual(data["settings"]["track_clicks"], True)
def test_template_id(self):
message = mail.EmailMultiAlternatives(
from_email="from@example.com", to=["to@example.com"]
)
message.template_id = "zyxwvut98765"
message.send()
data = self.get_api_call_json()
self.assertEqual(data["template_id"], "zyxwvut98765")
# With a template, MailerSend always ignores "text" and "html" params.
# For "subject", "from_email", and "reply_to" params, MailerSend ignores them
# if the corresponding field is set in the template's "Default settings",
# otherwise it uses the params. (And "subject" and "from_email" are required
# if no default setting for the template.)
# For "tags", MailerSend uses the param value if provided, otherwise
# the template default if any. (It does not attempt to combine them.)
@override_settings(ANYMAIL_MAILERSEND_BATCH_SEND_MODE="expose-to-list")
def test_merge_data_expose_to_list(self):
self.message.to = ["alice@example.com", "Bob <bob@example.com>"]
self.message.cc = ["cc@example.com"]
self.message.body = "Hi {{ name }}. Welcome to {{ group }} at {{ site }}."
self.message.merge_data = {
"alice@example.com": {"name": "Alice", "group": "Developers"},
"bob@example.com": {"name": "Bob"}, # and leave group undefined
"nobody@example.com": {"name": "Not a recipient for this message"},
}
self.message.merge_global_data = {"group": "Users", "site": "ExampleCo"}
self.message.send()
# BATCH_SEND_MODE="expose-to-list" uses 'email' API endpoint:
self.assert_esp_called("/v1/email")
data = self.get_api_call_json()
# personalization param covers all recipients:
self.assertEqual(
data["personalization"],
[
{
"email": "alice@example.com",
"data": {
"name": "Alice",
"group": "Developers",
"site": "ExampleCo",
},
},
{
"email": "bob@example.com",
"data": {"name": "Bob", "group": "Users", "site": "ExampleCo"},
},
# No personalization record for "nobody@example.com" -- including a
# personalization for email not found in "to" param causes API error.
],
)
@override_settings(ANYMAIL_MAILERSEND_BATCH_SEND_MODE="use-bulk-email")
def test_merge_data_use_bulk_email(self):
self.set_mock_response(
json_data={
"message": "The bulk email is being processed.",
"bulk_email_id": "12345abcde",
}
)
self.message.to = ["alice@example.com", "Bob <bob@example.com>"]
self.message.cc = ["cc@example.com"]
self.message.body = "Hi {{ name }}. Welcome to {{ group }} at {{ site }}."
self.message.merge_data = {
"alice@example.com": {"name": "Alice", "group": "Developers"},
"bob@example.com": {"name": "Bob"}, # and leave group undefined
"nobody@example.com": {"name": "Not a recipient for this message"},
}
self.message.merge_global_data = {"group": "Users", "site": "ExampleCo"}
self.message.send()
# BATCH_SEND_MODE="use-bulk-email" uses 'bulk-email' API endpoint:
self.assert_esp_called("/v1/bulk-email")
data = self.get_api_call_json()
self.assertEqual(len(data), 2) # batch of 2 separate emails
# "to" split to separate messages:
self.assertEqual(data[0]["to"], [{"email": "alice@example.com"}])
self.assertEqual(data[1]["to"], [{"email": "bob@example.com", "name": "Bob"}])
# "cc" appears in both:
self.assertEqual(data[0]["cc"], [{"email": "cc@example.com"}])
self.assertEqual(data[1]["cc"], [{"email": "cc@example.com"}])
# "personalization" param matches only single recipient:
self.assertEqual(
data[0]["personalization"],
[
{
"email": "alice@example.com",
"data": {
"name": "Alice",
"group": "Developers",
"site": "ExampleCo",
},
}
],
)
self.assertEqual(
data[1]["personalization"],
[
{
"email": "bob@example.com",
"data": {"name": "Bob", "group": "Users", "site": "ExampleCo"},
}
],
)
def test_merge_data_single_recipient(self):
# BATCH_SEND_MODE=None default uses 'email' for single recipient
self.message.to = ["Bob <bob@example.com>"]
self.message.cc = ["cc@example.com"]
self.message.body = "Hi {{ name }}. Welcome to {{ group }} at {{ site }}."
self.message.merge_data = {
"alice@example.com": {"name": "Alice", "group": "Developers"},
"bob@example.com": {"name": "Bob"}, # and leave group undefined
"nobody@example.com": {"name": "Not a recipient for this message"},
}
self.message.merge_global_data = {"group": "Users", "site": "ExampleCo"}
self.message.send()
self.assert_esp_called("/v1/email")
data = self.get_api_call_json()
self.assertEqual(
data["personalization"],
[
{
"email": "bob@example.com",
"data": {"name": "Bob", "group": "Users", "site": "ExampleCo"},
},
# No personalization record for merge_data emails not in "to" list.
],
)
def test_merge_data_ambiguous(self):
# Multiple recipients require a non-default MAILERSEND_BATCH_SEND_MODE
self.message.to = ["alice@example.com", "Bob <bob@example.com>"]
self.message.cc = ["cc@example.com"]
self.message.body = "Hi {{ name }}. Welcome to {{ group }} at {{ site }}."
# (even an empty merge_data dict should trigger Anymail batch send)
self.message.merge_data = {}
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "MAILERSEND_BATCH_SEND_MODE"
):
self.message.send()
def test_merge_metadata(self):
self.message.to = ["alice@example.com", "Bob <bob@example.com>"]
self.message.merge_metadata = {
"alice@example.com": {"order_id": 123},
"bob@example.com": {"order_id": 678, "tier": "premium"},
}
with self.assertRaisesMessage(AnymailUnsupportedFeature, "merge_metadata"):
self.message.send()
def test_default_omits_options(self):
"""Make sure by default we don't send any ESP-specific options.
Options not specified by the caller should be omitted entirely from
the API call (*not* sent as False or empty). This ensures
that your ESP account settings apply by default.
"""
self.message.send()
data = self.get_api_call_json()
self.assertNotIn("cc", data)
self.assertNotIn("bcc", data)
self.assertNotIn("reply_to", data)
self.assertNotIn("html", data)
self.assertNotIn("attachments", data)
self.assertNotIn("template_id", data)
self.assertNotIn("tags", data)
self.assertNotIn("variables", data)
self.assertNotIn("personalization", data)
self.assertNotIn("precedence_bulk", data)
self.assertNotIn("send_at", data)
self.assertNotIn("in_reply_to", data)
self.assertNotIn("settings", data)
def test_esp_extra(self):
self.message.track_clicks = True # test deep merge of "settings"
self.message.esp_extra = {
"variables": [
{
"email": "to@example.com",
"substitutions": [{"var": "order_id", "value": "12345"}],
}
],
"settings": {
"track_content": True,
},
}
self.message.track_clicks = True
self.message.send()
data = self.get_api_call_json()
self.assertEqual(
data["variables"],
[
{
"email": "to@example.com",
"substitutions": [{"var": "order_id", "value": "12345"}],
}
],
)
self.assertEqual(
data["settings"],
{
"track_content": True,
"track_clicks": True, # deep merge
},
)
def test_esp_extra_settings_overrides(self):
"""esp_extra can override batch_send_mode and api_token settings"""
self.message.merge_data = {} # trigger batch send
self.message.esp_extra = {
"api_token": "token-from-esp-extra",
"batch_send_mode": "use-bulk-email",
"hypothetical_future_mailersend_param": 123,
}
self.message.send()
self.assert_esp_called("/v1/bulk-email") # batch_send_mode from esp_extra
headers = self.get_api_call_headers()
self.assertEqual(headers["Authorization"], "Bearer token-from-esp-extra")
data = self.get_api_call_json()
self.assertEqual(len(data), 1) # payload burst for batch
self.assertNotIn("api_token", data[0]) # not in API payload
self.assertNotIn("batch_send_mode", data[0]) # not sent to API
# But other esp_extra params sent:
self.assertEqual(data[0]["hypothetical_future_mailersend_param"], 123)
def test_send_attaches_anymail_status(self):
"""The anymail_status should be attached to the message when it is sent"""
self.set_mock_success(message_id="12345abcde")
msg = mail.EmailMessage(
"Subject", "Message", "from@example.com", ["to1@example.com"]
)
sent = msg.send()
self.assertEqual(sent, 1)
self.assertEqual(msg.anymail_status.status, {"queued"})
self.assertEqual(msg.anymail_status.message_id, "12345abcde")
self.assertEqual(
msg.anymail_status.recipients["to1@example.com"].status, "queued"
)
self.assertEqual(
msg.anymail_status.recipients["to1@example.com"].message_id,
"12345abcde",
)
@override_settings(
ANYMAIL_IGNORE_RECIPIENT_STATUS=True # exception is tested later
)
def test_send_all_rejected(self):
"""The anymail_status should be 'rejected' when all recipients rejected"""
self.set_mock_rejected(
{"to1@example.com": ["blocklisted"], "to2@example.com": ["hard_bounced"]}
)
msg = mail.EmailMessage(
"Subject",
"Message",
"from@example.com",
["to1@example.com", "to2@example.com"],
)
msg.send()
self.assertEqual(msg.anymail_status.status, {"rejected"})
recipients = msg.anymail_status.recipients
self.assertEqual(recipients["to1@example.com"].status, "rejected")
self.assertIsNone(recipients["to1@example.com"].message_id)
self.assertEqual(recipients["to2@example.com"].status, "rejected")
self.assertIsNone(recipients["to2@example.com"].message_id)
def test_send_some_rejected(self):
"""
The anymail_status should identify which recipients are rejected.
"""
self.set_mock_rejected(
{"to1@example.com": ["blocklisted"]},
warning_type="SOME_SUPPRESSED",
message_id="12345abcde",
)
msg = mail.EmailMessage(
"Subject",
"Message",
"from@example.com",
["to1@example.com", "to2@example.com"],
)
msg.send()
self.assertEqual(msg.anymail_status.status, {"rejected", "queued"})
recipients = msg.anymail_status.recipients
self.assertEqual(recipients["to1@example.com"].status, "rejected")
self.assertIsNone(recipients["to1@example.com"].message_id)
self.assertEqual(recipients["to2@example.com"].status, "queued")
self.assertEqual(recipients["to2@example.com"].message_id, "12345abcde")
# noinspection PyUnresolvedReferences
@override_settings(ANYMAIL_MAILERSEND_BATCH_SEND_MODE="use-bulk-email")
def test_bulk_send_response(self):
self.set_mock_response(
json_data={
"message": "The bulk email is being processed.",
"bulk_email_id": "12345abcde",
}
)
self.message.merge_data = {} # trigger batch behavior
self.message.send()
# Unknown status for bulk send (until you poll the status API):
self.assertEqual(self.message.anymail_status.status, {"unknown"})
# Unknown message_id for bulk send, so provide batch id with "bulk:" prefix:
self.assertEqual(self.message.anymail_status.message_id, "bulk:12345abcde")
# noinspection PyUnresolvedReferences
def test_send_failed_anymail_status(self):
"""If the send fails, anymail_status should contain initial values"""
self.set_mock_response(status_code=400)
sent = self.message.send(fail_silently=True)
self.assertEqual(sent, 0)
self.assertIsNone(self.message.anymail_status.status)
self.assertIsNone(self.message.anymail_status.message_id)
self.assertEqual(self.message.anymail_status.recipients, {})
self.assertIsNone(self.message.anymail_status.esp_response)
# noinspection PyUnresolvedReferences
def test_unhandled_warnings(self):
# Non-suppression warnings should turn a 202 accepted response into an error
response_content = {"warnings": [{"type": "UNKNOWN_WARNING"}]}
self.set_mock_response(status_code=202, json_data=response_content)
with self.assertRaisesMessage(AnymailAPIError, "UNKNOWN_WARNING"):
self.message.send()
self.assertIsNone(self.message.anymail_status.status)
self.assertIsNone(self.message.anymail_status.message_id)
self.assertEqual(self.message.anymail_status.recipients, {})
self.assertEqual(
self.message.anymail_status.esp_response.json(), response_content
)
def test_json_serialization_errors(self):
"""Try to provide more information about non-json-serializable data"""
self.message.tags = [Decimal("19.99")] # yeah, don't do this
with self.assertRaises(AnymailSerializationError) as cm:
self.message.send()
print(self.get_api_call_json())
err = cm.exception
self.assertIsInstance(err, TypeError) # compatibility with json.dumps
# our added context:
self.assertIn("Don't know how to send this data to MailerSend", str(err))
# original message:
self.assertRegex(str(err), r"Decimal.*is not JSON serializable")
@tag("mailersend")
class MailerSendBackendRecipientsRefusedTests(MailerSendBackendMockAPITestCase):
"""
Should raise AnymailRecipientsRefused when *all* recipients are rejected or invalid
"""
def test_recipients_refused(self):
self.set_mock_rejected(
{
"invalid@localhost": ["hard_bounced"],
"reject@example.com": ["blocklisted"],
}
)
msg = mail.EmailMessage(
"Subject",
"Body",
"from@example.com",
["invalid@localhost", "reject@example.com"],
)
with self.assertRaises(AnymailRecipientsRefused):
msg.send()
def test_fail_silently(self):
self.set_mock_rejected(
{
"invalid@localhost": ["hard_bounced"],
"reject@example.com": ["blocklisted"],
}
)
sent = mail.send_mail(
"Subject",
"Body",
"from@example.com",
["invalid@localhost", "reject@example.com"],
fail_silently=True,
)
self.assertEqual(sent, 0)
def test_mixed_response(self):
"""If *any* recipients are valid or queued, no exception is raised"""
self.set_mock_rejected(
{
"invalid@localhost": ["hard_bounced"],
"reject@example.com": ["blocklisted"],
},
warning_type="SOME_SUPPRESSED",
)
msg = mail.EmailMessage(
"Subject",
"Body",
"from@example.com",
[
"invalid@localhost",
"valid@example.com",
"reject@example.com",
"also.valid@example.com",
],
)
sent = msg.send()
# one message sent, successfully, to 2 of 4 recipients:
self.assertEqual(sent, 1)
status = msg.anymail_status
self.assertEqual(status.recipients["invalid@localhost"].status, "rejected")
self.assertEqual(status.recipients["valid@example.com"].status, "queued")
self.assertEqual(status.recipients["reject@example.com"].status, "rejected")
self.assertEqual(status.recipients["also.valid@example.com"].status, "queued")
@override_settings(ANYMAIL_IGNORE_RECIPIENT_STATUS=True)
def test_settings_override(self):
"""No exception with ignore setting"""
self.set_mock_rejected(
{
"invalid@localhost": ["hard_bounced"],
"reject@example.com": ["blocklisted"],
},
)
sent = mail.send_mail(
"Subject",
"Body",
"from@example.com",
["invalid@localhost", "reject@example.com"],
)
self.assertEqual(sent, 1) # refused message is included in sent count
@tag("mailersend")
class MailerSendBackendConfigurationTests(MailerSendBackendMockAPITestCase):
"""Test various MailerSend client options"""
@override_settings(
# clear MAILERSEND_API_TOKEN from MailerSendBackendMockAPITestCase:
ANYMAIL={}
)
def test_missing_api_token(self):
with self.assertRaises(AnymailConfigurationError) as cm:
mail.send_mail("Subject", "Message", "from@example.com", ["to@example.com"])
errmsg = str(cm.exception)
# Make sure the error mentions the different places to set the key
self.assertRegex(errmsg, r"\bMAILERSEND_API_TOKEN\b")
self.assertRegex(errmsg, r"\bANYMAIL_MAILERSEND_API_TOKEN\b")
@override_settings(
ANYMAIL={
"MAILERSEND_API_URL": "https://api.dev.mailersend.com/v2",
"MAILERSEND_API_TOKEN": "test_api_key",
}
)
def test_mailersend_api_url(self):
mail.send_mail("Subject", "Message", "from@example.com", ["to@example.com"])
self.assert_esp_called("https://api.dev.mailersend.com/v2/email")
# can also override on individual connection
connection = mail.get_connection(api_url="https://api.mailersend.com/vNext")
mail.send_mail(
"Subject",
"Message",
"from@example.com",
["to@example.com"],
connection=connection,
)
self.assert_esp_called("https://api.mailersend.com/vNext/email")
@override_settings(ANYMAIL={"MAILERSEND_API_TOKEN": "bad_token"})
def test_invalid_api_key(self):
self.set_mock_response(
status_code=401,
reason="UNAUTHORIZED",
json_data={"message": "Unauthenticated."},
)
with self.assertRaisesMessage(AnymailAPIError, "Unauthenticated"):
self.message.send()

353
tests/test_mailersend_inbound.py Normal file
View File

@@ -0,0 +1,353 @@
import json
from datetime import datetime, timezone
from textwrap import dedent
from unittest.mock import ANY
from django.core.exceptions import ImproperlyConfigured
from django.test import override_settings, tag
from anymail.exceptions import AnymailConfigurationError
from anymail.inbound import AnymailInboundMessage
from anymail.signals import AnymailInboundEvent
from anymail.webhooks.mailersend import MailerSendInboundWebhookView
from .test_mailersend_webhooks import (
TEST_WEBHOOK_SIGNING_SECRET,
MailerSendWebhookTestCase,
)
from .utils import sample_image_content
from .webhook_cases import WebhookBasicAuthTestCase
@tag("mailersend")
@override_settings(ANYMAIL_MAILERSEND_INBOUND_SECRET=TEST_WEBHOOK_SIGNING_SECRET)
class MailerSendInboundSecurityTestCase(
MailerSendWebhookTestCase, WebhookBasicAuthTestCase
):
should_warn_if_no_auth = False # because we check webhook signature
def call_webhook(self):
return self.client_post_signed(
"/anymail/mailersend/inbound/",
{"type": "inbound.message", "data": {"raw": "..."}},
secret=TEST_WEBHOOK_SIGNING_SECRET,
)
# Additional tests are in WebhookBasicAuthTestCase
def test_verifies_correct_signature(self):
response = self.client_post_signed(
"/anymail/mailersend/inbound/",
{"type": "inbound.message", "data": {"raw": "..."}},
secret=TEST_WEBHOOK_SIGNING_SECRET,
)
self.assertEqual(response.status_code, 200)
def test_verifies_missing_signature(self):
response = self.client.post(
"/anymail/mailersend/inbound/",
content_type="application/json",
data=json.dumps({"type": "inbound.message", "data": {"raw": "..."}}),
)
self.assertEqual(response.status_code, 400)
def test_verifies_bad_signature(self):
# This also verifies that the error log references the correct setting to check.
with self.assertLogs() as logs:
response = self.client_post_signed(
"/anymail/mailersend/inbound/",
{"type": "inbound.message", "data": {"raw": "..."}},
secret="wrong signing key",
)
# SuspiciousOperation causes 400 response (even in test client):
self.assertEqual(response.status_code, 400)
self.assertIn("check Anymail MAILERSEND_INBOUND_SECRET", logs.output[0])
@tag("mailersend")
class MailerSendInboundSettingsTestCase(MailerSendWebhookTestCase):
def test_requires_inbound_secret(self):
with self.assertRaisesMessage(
ImproperlyConfigured, "MAILERSEND_INBOUND_SECRET"
):
self.client_post_signed(
"/anymail/mailersend/inbound/",
{
"type": "inbound.message",
"data": {"object": "message", "raw": "..."},
},
)
@override_settings(
ANYMAIL={
"MAILERSEND_INBOUND_SECRET": "inbound secret",
"MAILERSEND_WEBHOOK_SIGNING_SECRET": "webhook secret",
}
)
def test_webhook_signing_secret_is_different(self):
response = self.client_post_signed(
"/anymail/mailersend/inbound/",
{
"type": "inbound.message",
"data": {"object": "message", "raw": "..."},
},
secret="inbound secret",
)
self.assertEqual(response.status_code, 200)
@override_settings(ANYMAIL_MAILERSEND_INBOUND_SECRET="settings secret")
def test_inbound_secret_view_params(self):
"""Webhook signing secret can be provided as a view param"""
view = MailerSendInboundWebhookView.as_view(inbound_secret="view-level secret")
view_instance = view.view_class(**view.view_initkwargs)
self.assertEqual(view_instance.signing_secret, b"view-level secret")
@tag("mailersend")
@override_settings(ANYMAIL_MAILERSEND_INBOUND_SECRET=TEST_WEBHOOK_SIGNING_SECRET)
class MailerSendInboundTestCase(MailerSendWebhookTestCase):
# Since Anymail just parses the raw MIME message through the Python email
# package, there aren't really a lot of different cases to test here.
# (We don't need to re-test the whole email.parser.)
def test_inbound(self):
# This is an actual (sanitized) inbound payload received from MailerSend:
raw_event = {
"type": "inbound.message",
"inbound_id": "[inbound-route-id-redacted]",
"url": "https://test.anymail.dev/anymail/mailersend/inbound/",
"created_at": "2023-03-04T02:22:16.417935Z",
"data": {
"object": "message",
"id": "6402ab57f79d39d7e10f2523",
"recipients": {
"rcptTo": [{"email": "envelope-recipient@example.com"}],
"to": {
"raw": "Recipient <to@example.com>",
"data": [{"email": "to@example.com", "name": "Recipient"}],
},
},
"from": {
"email": "sender@example.org",
"name": "Sender Name",
"raw": "Sender Name <sender@example.org>",
},
"sender": {"email": "envelope-sender@example.org"},
"subject": "Testing inbound \ud83c\udf0e",
"date": "Fri, 3 Mar 2023 18:22:03 -0800",
"headers": {
"X-Envelope-From": "<envelope-sender@example.org>",
# Multiple-instance headers appear as arrays:
"Received": [
"from example.org (mail.example.org [10.10.10.10])\r\n"
" by inbound.mailersend.net with ESMTPS id ...\r\n"
" Sat, 04 Mar 2023 02:22:15 +0000 (UTC)",
"by mail.example.org with SMTP id ...\r\n"
" for <envelope-recipient@example.com>;\r\n"
" Fri, 03 Mar 2023 18:22:15 -0800 (PST)",
],
"DKIM-Signature": "v=1; a=rsa-sha256; c=relaxed/relaxed; ...",
"MIME-Version": "1.0",
"From": "Sender Name <sender@example.org>",
"Date": "Fri, 3 Mar 2023 18:22:03 -0800",
"Message-ID": "<AzjSdSHsmvXUeZGTPQ@mail.example.org>",
"Subject": "=?UTF-8?Q?Testing_inbound_=F0=9F=8C=8E?=",
"To": "Recipient <to@example.com>",
"Content-Type": 'multipart/mixed; boundary="000000000000e5575c05f609bab6"',
},
"text": "This is a *test*!\r\n\r\n[image: sample_image.png]\r\n",
"html": (
"<p>This is a <b>test</b>!</p>"
'<img src="cid:ii_letc8ro50" alt="sample_image.png">'
),
"raw": dedent(
"""\
X-Envelope-From: <envelope-sender@example.org>
Received: from example.org (mail.example.org [10.10.10.10])
by inbound.mailersend.net with ESMTPS id ...
Sat, 04 Mar 2023 02:22:15 +0000 (UTC)
Received: by mail.example.org with SMTP id ...
for <envelope-recipient@example.com>;
Fri, 03 Mar 2023 18:22:15 -0800 (PST)
DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; ...
MIME-Version: 1.0
From: Sender Name <sender@example.org>
Date: Fri, 3 Mar 2023 18:22:03 -0800
Message-ID: <AzjSdSHsmvXUeZGTPQ@mail.example.org>
Subject: =?UTF-8?Q?Testing_inbound_=F0=9F=8C=8E?=
To: Recipient <to@example.com>
Content-Type: multipart/mixed; boundary="000000000000e5575c05f609bab6"
--000000000000e5575c05f609bab6
Content-Type: multipart/related; boundary="000000000000e5575b05f609bab5"
--000000000000e5575b05f609bab5
Content-Type: multipart/alternative; boundary="000000000000e5575a05f609bab4"
--000000000000e5575a05f609bab4
Content-Type: text/plain; charset="UTF-8"
This is a *test*!
[image: sample_image.png]
--000000000000e5575a05f609bab4
Content-Type: text/html; charset="UTF-8"
<p>This is a <b>test</b>!</p>
<img src="cid:ii_letc8ro50" alt="sample_image.png">
--000000000000e5575a05f609bab4--
--000000000000e5575b05f609bab5
Content-Type: image/png; name="sample_image.png"
Content-Disposition: inline; filename="sample_image.png"
Content-Transfer-Encoding: base64
Content-ID: <ii_letc8ro50>
iVBORw0KGgoAAAANSUhEUgAAACAAAAAgCAYAAABzenr0AAAABHNCSVQICAgIfAhkiAAAAAlwSFlz
AAALEgAACxIB0t1+/AAAABR0RVh0Q3JlYXRpb24gVGltZQAzLzEvMTNoZNRjAAAAHHRFWHRTb2Z0
d2FyZQBBZG9iZSBGaXJld29ya3MgQ1M1cbXjNgAAAZ1JREFUWIXtl7FKA0EQhr+TgIFgo5BXyBUp
fIGksLawUNAXWFFfwCJgBAtfIJFMLXgQn8BSwdpCiPcKAdOIoI2x2Dmyd7kYwXhp9odluX/uZv6d
nZu7DXowxiKZi0IAUHKCvxcsoAIEpST4IawVGb0Hb0BlpcigefACvAAvwAsoTTGGlwwzBAyivLUP
EZrOM10AhGOH2wWugVVlHoAdhJHrPC8DNR0JGsAAQ9mxNzBOMNjS4Qrq69U5EKmf12ywWVsQI4QI
IbCn3Gnmnk7uk1bokfooI7QRDlQIGCdzPwiYh0idtXNs2zq3UqwVEiDcu/R0DVjUnFpItuPSscfA
FXCGSfEAdZ2fVeQ68OjYWwi3ycVvMhABGwgfKXZScHeZ+4c6VzN8FbuYukvOykCs+z8PJ0xqIXYE
d4ALoKlVH2IIgUHWwd/6gNAFPjPcCPvKNTDcYAj1lXzKc7GIRrSZI6yJzcQ+dtV9bD+IkHThBj34
4j9/yYxupaQbXPJLNqsGFgeZ6qwpLP1b4AV4AV5AoKfjpR5OwR6VKwULCAC+AQV4W9Ps4uZQAAAA
AElFTkSuQmCC
--000000000000e5575b05f609bab5--
--000000000000e5575c05f609bab6
Content-Type: text/csv; charset="US-ASCII"; name="sample_data.csv"
Content-Disposition: attachment; filename="sample_data.csv"
Content-Transfer-Encoding: quoted-printable
Product,Price
Widget,33.20
--000000000000e5575c05f609bab6--"""
).replace("\n", "\r\n"),
"attachments": [
{
"file_name": "sample_image.png",
"content_type": "image/png",
"content_disposition": "inline",
"content_id": "ii_letc8ro50",
"size": 579,
"content": (
"iVBORw0KGgoAAAANSUhEUgAAACAAAAAgCAYAAABzenr0AAAABHNCSVQICAgIfAhki"
"AAAAAlwSFlzAAALEgAACxIB0t1+/AAAABR0RVh0Q3JlYXRpb24gVGltZQAzLzEvMT"
"NoZNRjAAAAHHRFWHRTb2Z0d2FyZQBBZG9iZSBGaXJld29ya3MgQ1M1cbXjNgAAAZ1"
"JREFUWIXtl7FKA0EQhr+TgIFgo5BXyBUpfIGksLawUNAXWFFfwCJgBAtfIJFMLXgQ"
"n8BSwdpCiPcKAdOIoI2x2Dmyd7kYwXhp9odluX/uZv6dnZu7DXowxiKZi0IAUHKCv"
"xcsoAIEpST4IawVGb0Hb0BlpcigefACvAAvwAsoTTGGlwwzBAyivLUPEZrOM10AhG"
"OH2wWugVVlHoAdhJHrPC8DNR0JGsAAQ9mxNzBOMNjS4Qrq69U5EKmf12ywWVsQI4Q"
"IIbCn3Gnmnk7uk1bokfooI7QRDlQIGCdzPwiYh0idtXNs2zq3UqwVEiDcu/R0DVjU"
"nFpItuPSscfAFXCGSfEAdZ2fVeQ68OjYWwi3ycVvMhABGwgfKXZScHeZ+4c6VzN8F"
"buYukvOykCs+z8PJ0xqIXYEd4ALoKlVH2IIgUHWwd/6gNAFPjPcCPvKNTDcYAj1lX"
"zKc7GIRrSZI6yJzcQ+dtV9bD+IkHThBj344j9/yYxupaQbXPJLNqsGFgeZ6qwpLP1"
"b4AV4AV5AoKfjpR5OwR6VKwULCAC+AQV4W9Ps4uZQAAAAAElFTkSuQmCC"
),
},
{
"file_name": "sample_data.csv",
"content_type": "text/csv",
"content_disposition": "attachment",
"size": 26,
"content": "UHJvZHVjdCxQcmljZQpXaWRnZXQsMzMuMjA=",
},
],
"spf_check": {"code": "+", "value": None},
"dkim_check": False,
"created_at": "2023-03-04T02:22:15.525000Z",
},
}
response = self.client_post_signed("/anymail/mailersend/inbound/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.inbound_handler,
sender=MailerSendInboundWebhookView,
event=ANY,
esp_name="MailerSend",
)
# AnymailInboundEvent
event = kwargs["event"]
self.assertIsInstance(event, AnymailInboundEvent)
self.assertEqual(event.event_type, "inbound")
self.assertEqual(
event.timestamp,
# "2023-03-04T02:22:15.525000Z"
datetime(2023, 3, 4, 2, 22, 15, microsecond=525000, tzinfo=timezone.utc),
)
self.assertEqual(event.event_id, "6402ab57f79d39d7e10f2523")
self.assertIsInstance(event.message, AnymailInboundMessage)
# (The raw_event subject contains a "\N{EARTH GLOBE AMERICAS}" (🌎)
# character in the escaped form "\ud83c\udf0e", which won't compare equal
# until unescaped. Passing through json dumps/loads resolves the escapes.)
self.assertEqual(event.esp_event, json.loads(json.dumps(raw_event)))
# AnymailInboundMessage - convenience properties
message = event.message
self.assertEqual(message.from_email.display_name, "Sender Name")
self.assertEqual(message.from_email.addr_spec, "sender@example.org")
self.assertEqual(str(message.to[0]), "Recipient <to@example.com>")
self.assertEqual(message.subject, "Testing inbound 🌎")
self.assertEqual(message.date.isoformat(" "), "2023-03-03 18:22:03-08:00")
self.assertEqual(
message.text, "This is a *test*!\r\n\r\n[image: sample_image.png]\r\n"
)
self.assertHTMLEqual(
message.html,
"<p>This is a <b>test</b>!</p>"
'<img src="cid:ii_letc8ro50" alt="sample_image.png">',
)
self.assertEqual(message.envelope_sender, "envelope-sender@example.org")
self.assertEqual(message.envelope_recipient, "envelope-recipient@example.com")
# MailerSend inbound doesn't provide these:
self.assertIsNone(message.stripped_text)
self.assertIsNone(message.stripped_html)
self.assertIsNone(message.spam_detected)
self.assertIsNone(message.spam_score)
# AnymailInboundMessage - other headers
self.assertEqual(message["Message-ID"], "<AzjSdSHsmvXUeZGTPQ@mail.example.org>")
self.assertEqual(
message.get_all("Received"),
[
"from example.org (mail.example.org [10.10.10.10]) by inbound.mailersend.net"
" with ESMTPS id ... Sat, 04 Mar 2023 02:22:15 +0000 (UTC)",
"by mail.example.org with SMTP id ... for <envelope-recipient@example.com>;"
" Fri, 03 Mar 2023 18:22:15 -0800 (PST)",
],
)
inlines = message.inline_attachments
self.assertEqual(len(inlines), 1)
inline = inlines["ii_letc8ro50"]
self.assertEqual(inline.get_filename(), "sample_image.png")
self.assertEqual(inline.get_content_type(), "image/png")
self.assertEqual(inline.get_content_bytes(), sample_image_content())
attachments = message.attachments
self.assertEqual(len(attachments), 1)
self.assertEqual(attachments[0].get_filename(), "sample_data.csv")
self.assertEqual(attachments[0].get_content_type(), "text/csv")
self.assertEqual(
attachments[0].get_content_text(), "Product,Price\r\nWidget,33.20"
)
def test_misconfigured_inbound(self):
errmsg = (
"You seem to have set MailerSend's *activity.sent* webhook"
" to Anymail's MailerSend *inbound* webhook URL."
)
with self.assertRaisesMessage(AnymailConfigurationError, errmsg):
self.client_post_signed(
"/anymail/mailersend/inbound/",
{
"type": "activity.sent",
"data": {"object": "activity", "type": "sent"},
},
)

184
tests/test_mailersend_integration.py Normal file
View File

@@ -0,0 +1,184 @@
import os
import unittest
from datetime import datetime, timedelta
from email.utils import formataddr
from django.test import SimpleTestCase, override_settings, tag
from anymail.exceptions import AnymailAPIError
from anymail.message import AnymailMessage
from .utils import AnymailTestMixin, sample_image_path
ANYMAIL_TEST_MAILERSEND_API_TOKEN = os.getenv("ANYMAIL_TEST_MAILERSEND_API_TOKEN")
ANYMAIL_TEST_MAILERSEND_DOMAIN = os.getenv("ANYMAIL_TEST_MAILERSEND_DOMAIN")
@tag("mailersend", "live")
@unittest.skipUnless(
ANYMAIL_TEST_MAILERSEND_API_TOKEN and ANYMAIL_TEST_MAILERSEND_DOMAIN,
"Set ANYMAIL_TEST_MAILERSEND_API_TOKEN and ANYMAIL_TEST_MAILERSEND_DOMAIN"
" environment variables to run MailerSend integration tests",
)
@override_settings(
ANYMAIL={
"MAILERSEND_API_TOKEN": ANYMAIL_TEST_MAILERSEND_API_TOKEN,
},
EMAIL_BACKEND="anymail.backends.mailersend.EmailBackend",
)
class MailerSendBackendIntegrationTests(AnymailTestMixin, SimpleTestCase):
"""MailerSend API integration tests
These tests run against the **live** MailerSend API, using the
environment variable `ANYMAIL_TEST_MAILERSEND_API_TOKEN` as the API token
and `ANYMAIL_TEST_MAILERSEND_DOMAIN` as the sender domain.
If those variables are not set, these tests won't run.
"""
def setUp(self):
super().setUp()
self.from_email = f"from@{ANYMAIL_TEST_MAILERSEND_DOMAIN}"
self.message = AnymailMessage(
"Anymail MailerSend integration test",
"Text content",
self.from_email,
["test+to1@anymail.dev"],
)
self.message.attach_alternative("<p>HTML content</p>", "text/html")
def test_simple_send(self):
# Example of getting the MailerSend send status and message id from the message
sent_count = self.message.send()
self.assertEqual(sent_count, 1)
anymail_status = self.message.anymail_status
sent_status = anymail_status.recipients["test+to1@anymail.dev"].status
message_id = anymail_status.recipients["test+to1@anymail.dev"].message_id
self.assertEqual(sent_status, "queued") # MailerSend queues
# don't know what it'll be, but it should exist (and not be a bulk id):
self.assertGreater(len(message_id), 0)
self.assertFalse(message_id.startswith("bulk:"))
# set of all recipient statuses:
self.assertEqual(anymail_status.status, {sent_status})
self.assertEqual(anymail_status.message_id, message_id)
def test_all_options(self):
send_at = datetime.now() + timedelta(minutes=2)
from_email = formataddr(("Test From, with comma", self.from_email))
message = AnymailMessage(
subject="Anymail MailerSend all-options integration test",
body="This is the text body",
from_email=from_email,
to=[
"test+to1@anymail.dev",
"Recipient 2 <test+to2@anymail.dev>",
"test+bounce@anymail.dev", # will be rejected
],
cc=["test+cc1@anymail.dev", "Copy 2 <test+cc2@anymail.dev>"],
bcc=["test+bcc1@anymail.dev", "Blind Copy 2 <test+bcc2@anymail.dev>"],
# MailerSend only supports single reply_to:
reply_to=["Reply <reply@example.com>"],
# MailerSend supports very limited extra headers:
headers={"Precedence": "bulk", "In-Reply-To": "earlier-id@anymail.dev"},
send_at=send_at,
tags=["tag 1", "tag 2"],
track_clicks=False,
track_opens=True,
)
message.attach("attachment1.txt", "Here is some\ntext for you", "text/plain")
message.attach("vedhæftet fil.csv", "ID,Name\n1,3", "text/csv")
cid = message.attach_inline_image_file(
sample_image_path(), domain=ANYMAIL_TEST_MAILERSEND_DOMAIN
)
message.attach_alternative(
f"<div>This is the <i>html</i> body <img src='cid:{cid}'></div>",
"text/html",
)
message.send()
# First two recipients should be queued, other should be bounced
self.assertEqual(message.anymail_status.status, {"queued", "rejected"})
recipient_status = message.anymail_status.recipients
self.assertEqual(recipient_status["test+to1@anymail.dev"].status, "queued")
self.assertEqual(recipient_status["test+to2@anymail.dev"].status, "queued")
self.assertEqual(recipient_status["test+bounce@anymail.dev"].status, "rejected")
def test_stored_template(self):
message = AnymailMessage(
# id of a real template named in Anymail's MailerSend test account:
template_id="vywj2lpokkm47oqz",
to=[
"test+to1@anymail.dev",
"test+to2@anymail.dev",
"test+bounce@anymail.dev", # will be rejected
],
merge_data={
"test+to1@anymail.dev": {"name": "First Recipient", "order": 12345},
"test+to2@anymail.dev": {"name": "Second Recipient", "order": 67890},
"test+bounce@anymail.dev": {"name": "Bounces", "order": 3},
},
merge_global_data={"date": "yesterday"},
esp_extra={
# CAREFUL: with "expose-to-list", all recipients will see
# every other recipient's email address. (See docs!)
"batch_send_mode": "expose-to-list"
},
)
message.from_email = None # use template From
message.send()
recipient_status = message.anymail_status.recipients
self.assertEqual(recipient_status["test+to1@anymail.dev"].status, "queued")
self.assertEqual(recipient_status["test+to2@anymail.dev"].status, "queued")
self.assertEqual(recipient_status["test+bounce@anymail.dev"].status, "rejected")
self.assertFalse(
recipient_status["test+to1@anymail.dev"].message_id.startswith("bulk:")
)
self.assertIsNone(recipient_status["test+bounce@anymail.dev"].message_id)
def test_batch_send_mode_bulk(self):
# Same test as above, but with batch_send_mode "use-bulk-email".
# (Uses different API; status is handled very differently.)
message = AnymailMessage(
# id of a real template named in Anymail's MailerSend test account:
template_id="vywj2lpokkm47oqz",
to=[
"test+to1@anymail.dev",
"test+to2@anymail.dev",
"test+bounce@anymail.dev", # will be rejected
],
merge_data={
"test+to1@anymail.dev": {"name": "First Recipient", "order": 12345},
"test+to2@anymail.dev": {"name": "Second Recipient", "order": 67890},
"test+bounce@anymail.dev": {"name": "Bounces", "order": 3},
},
merge_global_data={"date": "yesterday"},
esp_extra={"batch_send_mode": "use-bulk-email"},
)
message.from_email = None # use template From
message.send()
recipient_status = message.anymail_status.recipients
# With use-bulk-email, must poll bulk-email status API to determine status:
self.assertEqual(recipient_status["test+to1@anymail.dev"].status, "unknown")
self.assertEqual(recipient_status["test+to2@anymail.dev"].status, "unknown")
self.assertEqual(recipient_status["test+bounce@anymail.dev"].status, "unknown")
# With use-bulk-email, message_id will be MailerSend's bulk_email_id
# rather than an actual message_id. Anymail adds "bulk:" to differentiate:
self.assertTrue(
recipient_status["test+to1@anymail.dev"].message_id.startswith("bulk:")
)
self.assertTrue(
recipient_status["test+bounce@anymail.dev"].message_id.startswith("bulk:")
)
@override_settings(
ANYMAIL={
"MAILERSEND_API_TOKEN": "Hey, that's not an API token",
}
)
def test_invalid_api_key(self):
with self.assertRaisesMessage(AnymailAPIError, "Unauthenticated"):
self.message.send()

465
tests/test_mailersend_webhooks.py Normal file
View File

@@ -0,0 +1,465 @@
import hashlib
import hmac
import json
from datetime import datetime, timezone
from unittest.mock import ANY
from django.core.exceptions import ImproperlyConfigured
from django.test import override_settings, tag
from anymail.exceptions import AnymailConfigurationError
from anymail.signals import AnymailTrackingEvent
from anymail.webhooks.mailersend import MailerSendTrackingWebhookView
from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase
TEST_WEBHOOK_SIGNING_SECRET = "TEST_WEBHOOK_SIGNING_SECRET"
def mailersend_signature(data, secret):
"""Generate a MailerSend webhook signature for data with secret"""
# https://developers.mailersend.com/api/v1/webhooks.html#security
return hmac.new(
key=secret.encode("ascii"),
msg=data,
digestmod=hashlib.sha256,
).hexdigest()
class MailerSendWebhookTestCase(WebhookTestCase):
def client_post_signed(self, url, json_data, secret=TEST_WEBHOOK_SIGNING_SECRET):
"""Return self.client.post(url, serialized json_data) signed with secret"""
# MailerSend for some reason backslash-escapes all forward slashes ("/")
# in its webhook payloads ("https:\/\/www..."). This is unnecessary, but
# harmless. We emulate it here to make sure it won't cause problems.
data = json.dumps(json_data).replace("/", "\\/").encode("ascii")
signature = mailersend_signature(data, secret)
return self.client.post(
url, content_type="application/json", data=data, HTTP_SIGNATURE=signature
)
@tag("mailersend")
class MailerSendWebhookSettingsTestCase(MailerSendWebhookTestCase):
def test_requires_signing_secret(self):
with self.assertRaisesMessage(
ImproperlyConfigured, "MAILERSEND_SIGNING_SECRET"
):
self.client_post_signed(
"/anymail/mailersend/tracking/", {"data": {"type": "sent"}}
)
@override_settings(
ANYMAIL={
"MAILERSEND_SIGNING_SECRET": "webhook secret",
"MAILERSEND_INBOUND_SECRET": "inbound secret",
}
)
def test_inbound_secret_is_different(self):
response = self.client_post_signed(
"/anymail/mailersend/tracking/",
{"data": {"type": "sent"}},
secret="webhook secret",
)
self.assertEqual(response.status_code, 200)
@override_settings(ANYMAIL_MAILERSEND_SIGNING_SECRET="settings secret")
def test_signing_secret_view_params(self):
"""Webhook signing secret can be provided as a view param"""
view = MailerSendTrackingWebhookView.as_view(signing_secret="view-level secret")
view_instance = view.view_class(**view.view_initkwargs)
self.assertEqual(view_instance.signing_secret, b"view-level secret")
@tag("mailersend")
@override_settings(ANYMAIL_MAILERSEND_SIGNING_SECRET=TEST_WEBHOOK_SIGNING_SECRET)
class MailerSendWebhookSecurityTestCase(
MailerSendWebhookTestCase, WebhookBasicAuthTestCase
):
should_warn_if_no_auth = False # because we check webhook signature
def call_webhook(self):
return self.client_post_signed(
"/anymail/mailersend/tracking/",
{"data": {"type": "sent"}},
secret=TEST_WEBHOOK_SIGNING_SECRET,
)
# Additional tests are in WebhookBasicAuthTestCase
def test_verifies_correct_signature(self):
response = self.client_post_signed(
"/anymail/mailersend/tracking/",
{"data": {"type": "sent"}},
secret=TEST_WEBHOOK_SIGNING_SECRET,
)
self.assertEqual(response.status_code, 200)
def test_verifies_missing_signature(self):
response = self.client.post(
"/anymail/mailersend/tracking/",
content_type="application/json",
data=json.dumps({"data": {"type": "sent"}}),
)
self.assertEqual(response.status_code, 400)
def test_verifies_bad_signature(self):
# This also verifies that the error log references the correct setting to check.
with self.assertLogs() as logs:
response = self.client_post_signed(
"/anymail/mailersend/tracking/",
{"data": {"type": "sent"}},
secret="wrong signing key",
)
# SuspiciousOperation causes 400 response (even in test client):
self.assertEqual(response.status_code, 400)
self.assertIn("check Anymail MAILERSEND_SIGNING_SECRET", logs.output[0])
@tag("mailersend")
@override_settings(ANYMAIL_MAILERSEND_SIGNING_SECRET=TEST_WEBHOOK_SIGNING_SECRET)
class MailerSendTestCase(MailerSendWebhookTestCase):
def test_sent_event(self):
# This is an actual, complete (sanitized) "sent" event as received from
# MailerSend. (For brevity, later tests omit several payload fields that
# Anymail doesn't use.)
raw_event = {
"type": "activity.sent",
"domain_id": "[domain-id-redacted]",
"created_at": "2023-02-27T21:09:49.520507Z",
"webhook_id": "[webhook-id-redacted]",
"url": "https://test.anymail.dev/anymail/mailersend/tracking/",
"data": {
"object": "activity",
"id": "63fd1c1d31b9c750540fe85c",
"type": "sent",
"created_at": "2023-02-27T21:09:49.506000Z",
"email": {
"object": "email",
"id": "63fd1c1de225707fa905f0a8",
"created_at": "2023-02-27T21:09:49.141000Z",
"from": "sender@mailersend.anymail.dev",
"subject": "Test webhooks",
"status": "sent",
"tags": ["tag1", "Tag 2"],
"message": {
"object": "message",
"id": "63fd1c1d5f010335ed07066b",
"created_at": "2023-02-27T21:09:49.061000Z",
},
"recipient": {
"object": "recipient",
"id": "63f3bb1965d98aa98c07d6b7",
"email": "recipient@example.com",
"created_at": "2023-02-20T18:25:29.162000Z",
},
},
"morph": None,
"template_id": "",
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "sent")
# event.timestamp comes from data.created_at:
self.assertEqual(
event.timestamp,
# "2023-02-27T21:09:49.506000Z"
datetime(2023, 2, 27, 21, 9, 49, microsecond=506000, tzinfo=timezone.utc),
)
# event.message_id matches the message.anymail_status.message_id when the
# message was sent. It comes from data.email.message.id:
self.assertEqual(event.message_id, "63fd1c1d5f010335ed07066b")
# event.event_id is unique for each event, and comes from data.id:
self.assertEqual(event.event_id, "63fd1c1d31b9c750540fe85c")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.tags, ["tag1", "Tag 2"])
self.assertEqual(event.metadata, {}) # MailerSend doesn't support metadata
self.assertEqual(event.esp_event, raw_event)
# You can construct the sent Message-ID header (which is different from the
# event.message_id, and is unique per recipient) from esp_event.data.email.id:
sent_message_id = f"<{event.esp_event['data']['email']['id']}@mailersend.net>"
self.assertEqual(sent_message_id, "<63fd1c1de225707fa905f0a8@mailersend.net>")
def test_delivered_event(self):
raw_event = {
"type": "activity.delivered",
"data": {
"object": "activity",
"id": "63fd1c1fcfbe46145d003a7b",
"type": "delivered",
"created_at": "2023-02-27T21:09:51.865000Z",
"email": {
"status": "delivered",
"message": {
"id": "63fd1c1d5f010335ed07066b",
},
"recipient": {
"email": "recipient@example.com",
},
},
"morph": None,
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "delivered")
self.assertEqual(
event.timestamp,
# "2023-02-27T21:09:51.865000Z"
datetime(2023, 2, 27, 21, 9, 51, microsecond=865000, tzinfo=timezone.utc),
)
self.assertEqual(event.message_id, "63fd1c1d5f010335ed07066b")
self.assertEqual(event.event_id, "63fd1c1fcfbe46145d003a7b")
self.assertEqual(event.recipient, "recipient@example.com")
def test_hard_bounced_event(self):
raw_event = {
"type": "activity.hard_bounced",
"data": {
"id": "63fd251d5c00f8e52001fce6",
"type": "hard_bounced",
"created_at": "2023-02-27T21:48:13.593000Z",
"email": {
"status": "rejected",
"message": {
"id": "63fd25194d5edba3da09e044",
},
"recipient": {
"email": "invalid@example.com",
},
},
"morph": {
"object": "recipient_bounce",
"reason": "Host or domain name not found",
},
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.recipient, "invalid@example.com")
self.assertEqual(event.reject_reason, "bounced")
self.assertEqual(event.description, "Host or domain name not found")
self.assertIsNone(event.mta_response) # raw MTA info not provided
def test_soft_bounced_event(self):
raw_event = {
"type": "activity.soft_bounced",
"data": {
"object": "activity",
"id": "62f114f8165fe0d8db0288e5",
"type": "soft_bounced",
"created_at": "2022-08-08T13:51:52.747000Z",
"email": {
"status": "rejected",
"tags": None,
"message": {
"id": "62fb66bef54a112e920b5493",
},
"recipient": {
"email": "notauser@example.com",
},
},
"morph": {
"object": "recipient_bounce",
"reason": "Unknown reason",
},
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.recipient, "notauser@example.com")
self.assertEqual(event.reject_reason, "bounced")
self.assertEqual(event.description, "Unknown reason")
self.assertIsNone(event.mta_response) # raw MTA info not provided
def test_spam_complaint_event(self):
raw_event = {
"type": "activity.spam_complaint",
"data": {
"id": "62f114f8165fe0d8db0288e5",
"type": "spam_complaint",
"created_at": "2022-08-08T13:51:52.747000Z",
"email": {
"status": "delivered",
"message": {
"id": "62fb66bef54a112e920b5493",
},
"recipient": {
"email": "recipient@example.com",
},
},
"morph": {
"object": "spam_complaint",
"reason": None,
},
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "complained")
self.assertEqual(event.recipient, "recipient@example.com")
def test_unsubscribed_event(self):
raw_event = {
"type": "activity.unsubscribed",
"data": {
"id": "63fd21c23f2bdd360e07d6b2",
"type": "unsubscribed",
"created_at": "2023-02-27T21:33:54.791000Z",
"email": {
"status": "delivered",
"message": {
"id": "63fd1c1d5f010335ed07066b",
},
"recipient": {
"email": "recipient@example.com",
},
},
"morph": {
"object": "recipient_unsubscribe",
"reason": "option_3",
"readable_reason": "I get too many emails",
},
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "unsubscribed")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.description, "I get too many emails")
def test_opened_event(self):
raw_event = {
"type": "activity.opened",
"data": {
"id": "63fd1e05532bd6cc700a793b",
"type": "opened",
"created_at": "2023-02-27T21:17:57.025000Z",
"email": {
"status": "delivered",
"message": {
"id": "63fd1c1d5f010335ed07066b",
},
"recipient": {
"email": "recipient@example.com",
},
},
"morph": {
"object": "open",
"id": "63fd1e05532bd6cc700a793a",
"created_at": "2023-02-27T21:17:57.018000Z",
"ip": "10.10.10.10",
},
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
self.assertEqual(event.recipient, "recipient@example.com")
def test_clicked_event(self):
raw_event = {
"type": "activity.clicked",
"data": {
"id": "63fd1d23afa3c770b00da7d3",
"type": "clicked",
"created_at": "2023-02-27T21:14:11.691000Z",
"email": {
"status": "delivered",
"message": {
"id": "63fd1c1d5f010335ed07066b",
},
"recipient": {
"email": "recipient@example.com",
},
},
"morph": {
"object": "click",
"id": "63fd1d23afa3c770b00da7d2",
"created_at": "2023-02-27T21:14:11.679000Z",
"ip": "10.10.10.10",
"url": "https://example.com/test",
},
},
}
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailerSendTrackingWebhookView,
event=ANY,
esp_name="MailerSend",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "clicked")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.click_url, "https://example.com/test")
def test_misconfigured_inbound(self):
errmsg = (
"You seem to have set MailerSend's *inbound* route endpoint"
" to Anymail's MailerSend *activity tracking* webhook URL."
)
with self.assertRaisesMessage(AnymailConfigurationError, errmsg):
self.client_post_signed(
"/anymail/mailersend/tracking/",
{
"type": "inbound.message",
"data": {"object": "message", "raw": "..."},
},
)