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

Amazon SES: remove deprecated v1 support

Browse Source 
- Remove deprecated amazon_sesv1 EmailBackend
- Remove deprecated amazon_sesv2 alias
  for amazon_ses EmailBackend
- Update docs
This commit is contained in:
Mike Edmunds
2024-03-12 13:17:49 -07:00
parent 1b78912b20
commit abb984485b
7 changed files with 81 additions and 1697 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
CHANGELOG.rst
View File

@@ -25,6 +25,23 @@ Release history
^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^
.. This extra heading level keeps the ToC from becoming unmanageably long .. This extra heading level keeps the ToC from becoming unmanageably long
vNext
-----
*Unreleased changes*
Breaking changes
~~~~~~~~~~~~~~~~
* **Amazon SES:** Drop support for the Amazon SES v1 API.
If your ``EMAIL_BACKEND`` setting uses ``amazon_sesv1``,
or if you are upgrading from Anymail 9.x or earlier directly to 11.0 or later, see
`Migrating to the SES v2 API <https://anymail.dev/en/latest/esps/amazon_ses/#amazon-ses-v2>`__.
(Anymail 10.0 switched to the SES v2 API by default. If your ``EMAIL_BACKEND``
setting has ``amazon_sesv2``, change that to just ``amazon_ses``.)
v10.3 v10.3
----- -----

449
anymail/backends/amazon_sesv1.py
View File

@@ -1,449 +0,0 @@
import warnings
from email.charset import QP, Charset
from email.mime.text import MIMEText
from ..exceptions import (
AnymailAPIError,
AnymailDeprecationWarning,
AnymailImproperlyInstalled,
)
from ..message import AnymailRecipientStatus
from ..utils import UNSET, get_anymail_setting
from .base import AnymailBaseBackend, BasePayload
try:
import boto3
from botocore.exceptions import BotoCoreError, ClientError, ConnectionError
except ImportError as err:
raise AnymailImproperlyInstalled(
missing_package="boto3", install_extra="amazon-ses"
) from err
# boto3 has several root exception classes; this is meant to cover all of them
BOTO_BASE_ERRORS = (BotoCoreError, ClientError, ConnectionError)
class EmailBackend(AnymailBaseBackend):
"""
Amazon SES v1 Email Backend (using boto3)
"""
esp_name = "Amazon SES"
def __init__(self, **kwargs):
"""Init options from Django settings"""
from .amazon_ses import _get_anymail_boto3_params
warnings.warn(
"anymail.backends.amazon_sesv1.EmailBackend is deprecated"
" and will be removed in the near future. Please migrate"
" to anymail.backends.amazon_ses.EmailBackend using Amazon SES v2.",
AnymailDeprecationWarning,
)
super().__init__(**kwargs)
# AMAZON_SES_CLIENT_PARAMS is optional
# (boto3 can find credentials several other ways)
self.session_params, self.client_params = _get_anymail_boto3_params(
kwargs=kwargs
)
self.configuration_set_name = get_anymail_setting(
"configuration_set_name",
esp_name=self.esp_name,
kwargs=kwargs,
allow_bare=False,
default=None,
)
self.message_tag_name = get_anymail_setting(
"message_tag_name",
esp_name=self.esp_name,
kwargs=kwargs,
allow_bare=False,
default=None,
)
self.client = None
def open(self):
if self.client:
return False # already exists
try:
self.client = boto3.session.Session(**self.session_params).client(
"ses", **self.client_params
)
except Exception:
if not self.fail_silently:
raise
else:
return True # created client
def close(self):
if self.client is None:
return
# self.client.close() # boto3 doesn't support (or require) client shutdown
self.client = None
def _send(self, message):
if self.client:
return super()._send(message)
elif self.fail_silently:
# (Probably missing boto3 credentials in open().)
return False
else:
class_name = self.__class__.__name__
raise RuntimeError(
"boto3 Session has not been opened in {class_name}._send. "
"(This is either an implementation error in {class_name}, "
"or you are incorrectly calling _send directly.)".format(
class_name=class_name
)
)
def build_message_payload(self, message, defaults):
# The SES SendRawEmail and SendBulkTemplatedEmail calls have
# very different signatures, so use a custom payload for each
if getattr(message, "template_id", UNSET) is not UNSET:
return AmazonSESSendBulkTemplatedEmailPayload(message, defaults, self)
else:
return AmazonSESSendRawEmailPayload(message, defaults, self)
def post_to_esp(self, payload, message):
try:
response = payload.call_send_api(self.client)
except BOTO_BASE_ERRORS as err:
# ClientError has a response attr with parsed json error response
# (other errors don't)
raise AnymailAPIError(
str(err),
backend=self,
email_message=message,
payload=payload,
response=getattr(err, "response", None),
) from err
return response
def parse_recipient_status(self, response, payload, message):
return payload.parse_recipient_status(response)
class AmazonSESBasePayload(BasePayload):
def init_payload(self):
self.params = {}
if self.backend.configuration_set_name is not None:
self.params["ConfigurationSetName"] = self.backend.configuration_set_name
def call_send_api(self, ses_client):
raise NotImplementedError()
def parse_recipient_status(self, response):
# response is the parsed (dict) JSON returned from the API call
raise NotImplementedError()
def set_esp_extra(self, extra):
# e.g., ConfigurationSetName, FromArn, SourceArn, ReturnPathArn
self.params.update(extra)
class AmazonSESSendRawEmailPayload(AmazonSESBasePayload):
def init_payload(self):
super().init_payload()
self.all_recipients = []
self.mime_message = self.message.message()
# Work around an Amazon SES bug where, if all of:
# - the message body (text or html) contains non-ASCII characters
# - the body is sent with `Content-Transfer-Encoding: 8bit`
# (which is Django email's default for most non-ASCII bodies)
# - you are using an SES ConfigurationSet with open or click tracking enabled
# then SES replaces the non-ASCII characters with question marks as it rewrites
# the message to add tracking. Forcing `CTE: quoted-printable` avoids the
# problem. (https://forums.aws.amazon.com/thread.jspa?threadID=287048)
for part in self.mime_message.walk():
if (
part.get_content_maintype() == "text"
and part["Content-Transfer-Encoding"] == "8bit"
):
content = part.get_payload()
del part["Content-Transfer-Encoding"]
qp_charset = Charset(part.get_content_charset("us-ascii"))
qp_charset.body_encoding = QP
# (can't use part.set_payload, because SafeMIMEText can undo
# this workaround)
MIMEText.set_payload(part, content, charset=qp_charset)
def call_send_api(self, ses_client):
# Set Destinations to make sure we pick up all recipients (including bcc).
# Any non-ASCII characters in recipient domains must be encoded with Punycode.
# (Amazon SES doesn't support non-ASCII recipient usernames.)
self.params["Destinations"] = [email.address for email in self.all_recipients]
self.params["RawMessage"] = {"Data": self.mime_message.as_bytes()}
return ses_client.send_raw_email(**self.params)
def parse_recipient_status(self, response):
try:
message_id = response["MessageId"]
except (KeyError, TypeError) as err:
raise AnymailAPIError(
"%s parsing Amazon SES send result %r" % (str(err), response),
backend=self.backend,
email_message=self.message,
payload=self,
) from None
recipient_status = AnymailRecipientStatus(
message_id=message_id, status="queued"
)
return {
recipient.addr_spec: recipient_status for recipient in self.all_recipients
}
# Standard EmailMessage attrs...
# These all get rolled into the RFC-5322 raw mime directly via
# EmailMessage.message()
def _no_send_defaults(self, attr):
# Anymail global send defaults don't work for standard attrs, because the
# merged/computed value isn't forced back into the EmailMessage.
if attr in self.defaults:
self.unsupported_feature(
"Anymail send defaults for '%s' with Amazon SES" % attr
)
def set_from_email_list(self, emails):
# Although Amazon SES will send messages with any From header, it can only parse
# Source if the From header is a single email. Explicit Source avoids an
# "Illegal address" error:
if len(emails) > 1:
self.params["Source"] = emails[0].addr_spec
# (else SES will look at the (single) address in the From header)
def set_recipients(self, recipient_type, emails):
self.all_recipients += emails
# included in mime_message
assert recipient_type in ("to", "cc", "bcc")
self._no_send_defaults(recipient_type)
def set_subject(self, subject):
# included in mime_message
self._no_send_defaults("subject")
def set_reply_to(self, emails):
# included in mime_message
self._no_send_defaults("reply_to")
def set_extra_headers(self, headers):
# included in mime_message
self._no_send_defaults("extra_headers")
def set_text_body(self, body):
# included in mime_message
self._no_send_defaults("body")
def set_html_body(self, body):
# included in mime_message
self._no_send_defaults("body")
def set_alternatives(self, alternatives):
# included in mime_message
self._no_send_defaults("alternatives")
def set_attachments(self, attachments):
# included in mime_message
self._no_send_defaults("attachments")
# Anymail-specific payload construction
def set_envelope_sender(self, email):
self.params["Source"] = email.addr_spec
def set_spoofed_to_header(self, header_to):
# django.core.mail.EmailMessage.message() has already set
# self.mime_message["To"] = header_to
# and performed any necessary header sanitization.
#
# The actual "to" is already in self.all_recipients,
# which is used as the SendRawEmail Destinations later.
#
# So, nothing to do here, except prevent the default
# "unsupported feature" error.
pass
def set_metadata(self, metadata):
# Amazon SES has two mechanisms for adding custom data to a message:
# * Custom message headers are available to webhooks (SNS notifications),
# but not in CloudWatch metrics/dashboards or Kinesis Firehose streams.
# Custom headers can be sent only with SendRawEmail.
# * "Message Tags" are available to CloudWatch and Firehose, and to SNS
# notifications for SES *events* but not SES *notifications*. (Got that?)
# Message Tags also allow *very* limited characters in both name and value.
# Message Tags can be sent with any SES send call.
# (See "How do message tags work?" in
# https://aws.amazon.com/blogs/ses/introducing-sending-metrics/
# and https://forums.aws.amazon.com/thread.jspa?messageID=782922.)
# To support reliable retrieval in webhooks, just use custom headers for
# metadata.
self.mime_message["X-Metadata"] = self.serialize_json(metadata)
def set_tags(self, tags):
# See note about Amazon SES Message Tags and custom headers in set_metadata
# above. To support reliable retrieval in webhooks, use custom headers for tags.
# (There are no restrictions on number or content for custom header tags.)
for tag in tags:
# creates multiple X-Tag headers, one per tag:
self.mime_message.add_header("X-Tag", tag)
# Also *optionally* pass a single Message Tag if the AMAZON_SES_MESSAGE_TAG_NAME
# Anymail setting is set (default no). The AWS API restricts tag content in this
# case. (This is useful for dashboard segmentation; use esp_extra["Tags"] for
# anything more complex.)
if tags and self.backend.message_tag_name is not None:
if len(tags) > 1:
self.unsupported_feature(
"multiple tags with the AMAZON_SES_MESSAGE_TAG_NAME setting"
)
self.params.setdefault("Tags", []).append(
{"Name": self.backend.message_tag_name, "Value": tags[0]}
)
def set_template_id(self, template_id):
raise NotImplementedError(
"AmazonSESSendRawEmailPayload should not have been used with template_id"
)
def set_merge_data(self, merge_data):
self.unsupported_feature("merge_data without template_id")
def set_merge_global_data(self, merge_global_data):
self.unsupported_feature("global_merge_data without template_id")
class AmazonSESSendBulkTemplatedEmailPayload(AmazonSESBasePayload):
def init_payload(self):
super().init_payload()
# late-bind recipients and merge_data in call_send_api
self.recipients = {"to": [], "cc": [], "bcc": []}
self.merge_data = {}
def call_send_api(self, ses_client):
# include any 'cc' or 'bcc' in every destination
cc_and_bcc_addresses = {}
if self.recipients["cc"]:
cc_and_bcc_addresses["CcAddresses"] = [
cc.address for cc in self.recipients["cc"]
]
if self.recipients["bcc"]:
cc_and_bcc_addresses["BccAddresses"] = [
bcc.address for bcc in self.recipients["bcc"]
]
# set up destination and data for each 'to'
self.params["Destinations"] = [
{
"Destination": dict(ToAddresses=[to.address], **cc_and_bcc_addresses),
"ReplacementTemplateData": self.serialize_json(
self.merge_data.get(to.addr_spec, {})
),
}
for to in self.recipients["to"]
]
return ses_client.send_bulk_templated_email(**self.params)
def parse_recipient_status(self, response):
try:
# response["Status"] should be a list in Destinations (to) order
anymail_statuses = [
AnymailRecipientStatus(
message_id=status.get("MessageId", None),
status="queued" if status.get("Status") == "Success" else "failed",
)
for status in response["Status"]
]
except (KeyError, TypeError) as err:
raise AnymailAPIError(
"%s parsing Amazon SES send result %r" % (str(err), response),
backend=self.backend,
email_message=self.message,
payload=self,
) from None
to_addrs = [to.addr_spec for to in self.recipients["to"]]
if len(anymail_statuses) != len(to_addrs):
raise AnymailAPIError(
"Sent to %d destinations, but only %d statuses in Amazon SES"
" send result %r" % (len(to_addrs), len(anymail_statuses), response),
backend=self.backend,
email_message=self.message,
payload=self,
)
return dict(zip(to_addrs, anymail_statuses))
def set_from_email(self, email):
# this will RFC2047-encode display_name if needed:
self.params["Source"] = email.address
def set_recipients(self, recipient_type, emails):
# late-bound in call_send_api
assert recipient_type in ("to", "cc", "bcc")
self.recipients[recipient_type] = emails
def set_subject(self, subject):
# (subject can only come from template; you can use substitution vars in that)
if subject:
self.unsupported_feature("overriding template subject")
def set_reply_to(self, emails):
if emails:
self.params["ReplyToAddresses"] = [email.address for email in emails]
def set_extra_headers(self, headers):
self.unsupported_feature("extra_headers with template")
def set_text_body(self, body):
if body:
self.unsupported_feature("overriding template body content")
def set_html_body(self, body):
if body:
self.unsupported_feature("overriding template body content")
def set_attachments(self, attachments):
if attachments:
self.unsupported_feature("attachments with template")
# Anymail-specific payload construction
def set_envelope_sender(self, email):
self.params["ReturnPath"] = email.addr_spec
def set_metadata(self, metadata):
# no custom headers with SendBulkTemplatedEmail
self.unsupported_feature("metadata with template")
def set_tags(self, tags):
# no custom headers with SendBulkTemplatedEmail, but support
# AMAZON_SES_MESSAGE_TAG_NAME if used (see tags/metadata in
# AmazonSESSendRawEmailPayload for more info)
if tags:
if self.backend.message_tag_name is not None:
if len(tags) > 1:
self.unsupported_feature(
"multiple tags with the AMAZON_SES_MESSAGE_TAG_NAME setting"
)
self.params["DefaultTags"] = [
{"Name": self.backend.message_tag_name, "Value": tags[0]}
]
else:
self.unsupported_feature(
"tags with template (unless using the"
" AMAZON_SES_MESSAGE_TAG_NAME setting)"
)
def set_template_id(self, template_id):
self.params["Template"] = template_id
def set_merge_data(self, merge_data):
# late-bound in call_send_api
self.merge_data = merge_data
def set_merge_global_data(self, merge_global_data):
self.params["DefaultTemplateData"] = self.serialize_json(merge_global_data)

14
anymail/backends/amazon_sesv2.py
View File

@@ -1,14 +0,0 @@
import warnings
from ..exceptions import AnymailDeprecationWarning
from .amazon_ses import EmailBackend as AmazonSESV2EmailBackend
class EmailBackend(AmazonSESV2EmailBackend):
def __init__(self, **kwargs):
warnings.warn(
"Anymail now uses Amazon SES v2 by default. Please change"
" 'amazon_sesv2' to 'amazon_ses' in your EMAIL_BACKEND setting.",
AnymailDeprecationWarning,
)
super().__init__(**kwargs)

142
docs/esps/amazon_ses.rst
View File

@@ -6,16 +6,11 @@ Amazon SES
Anymail integrates with the `Amazon Simple Email Service`_ (SES) using the `Boto 3`_ Anymail integrates with the `Amazon Simple Email Service`_ (SES) using the `Boto 3`_
AWS SDK for Python, and supports sending, tracking, and inbound receiving capabilities. AWS SDK for Python, and supports sending, tracking, and inbound receiving capabilities.
.. versionchanged:: 10.0 .. versionchanged:: 11.0
.. note:: Anymail supports only the newer Amazon SES v2 API. (Anymail 10.x supported both
SES v1 and v2, and used v2 by default. Anymail 9.x and earlier used SES v1.)
AWS has two versions of the SES API available for sending email. Anymail 10.0 See :ref:`amazon-ses-v2` below if you are upgrading from an earlier Anymail version.
uses the newer SES v2 API by default, and this is recommended for new projects.
If you integrated Amazon SES using an earlier Anymail release, you may need to
update your IAM permissions. See :ref:`amazon-ses-v2` below. Or if you are not
ready to switch, see :ref:`amazon-ses-v1` below.
.. sidebar:: Alternatives .. sidebar:: Alternatives
@@ -67,72 +62,6 @@ setting to customize the Boto session.
.. _Credentials: https://boto3.readthedocs.io/en/stable/guide/configuration.html#configuring-credentials .. _Credentials: https://boto3.readthedocs.io/en/stable/guide/configuration.html#configuring-credentials
.. _amazon-ses-v2:
Migrating to the SES v2 API
---------------------------
.. versionchanged:: 10.0
Anymail 10.0 uses Amazon's updated SES v2 API to send email. Earlier Anymail releases
used the original Amazon SES API (v1) by default. Although the capabilities of the two
SES versions are virtually identical, Amazon is implementing improvements (such as
increased maximum message size) only in the v2 API.
(The upgrade for SES v2 affects only sending email. There are no changes required
for status tracking webhooks or receiving inbound email.)
Migrating to SES v2 requires minimal code changes:
1. Update your :ref:`IAM permissions <amazon-ses-iam-permissions>` to grant Anymail
access to the SES v2 sending actions: ``ses:SendEmail`` for ordinary sends, and/or
``ses:SendBulkEmail`` to send using SES templates. (The IAM action
prefix is just ``ses`` for both the v1 and v2 APIs.)
If you run into unexpected IAM authorization failures, see the note about
:ref:`misleading IAM permissions errors <amazon-ses-iam-errors>` below.
2. If your code uses Anymail's :attr:`~anymail.message.AnymailMessage.esp_extra`
to pass additional SES API parameters, or examines the raw
:attr:`~anymail.message.AnymailStatus.esp_response` after sending a message,
you may need to update it for the v2 API. Many parameters have different names
in the v2 API compared to the equivalent v1 calls, and the response formats are
slightly different.
Among v1 parameters commonly used, ``ConfigurationSetName`` is unchanged in v2,
but v1's ``Tags`` and most ``*Arn`` parameters have been renamed in v2.
See AWS's docs for SES v1 `SendRawEmail`_ vs. v2 `SendEmail`_, or if you are sending
with SES templates, compare v1 `SendBulkTemplatedEmail`_ to v2 `SendBulkEmail`_.
.. _SendRawEmail:
https://docs.aws.amazon.com/ses/latest/APIReference/API_SendRawEmail.html
.. _SendBulkTemplatedEmail:
https://docs.aws.amazon.com/ses/latest/APIReference/API_SendBulkTemplatedEmail.html
.. _amazon-ses-v1:
Using SES v1 (deprecated)
~~~~~~~~~~~~~~~~~~~~~~~~~
New projects should use Anymail's default Amazon SES v2 integration. If you have an
existing project that is not ready to switch to v2, Anymail's original SES v1 support
is still available. In your settings.py, change the :setting:`!EMAIL_BACKEND` from:
.. code-block:: python
EMAIL_BACKEND = "anymail.backends.amazon_ses.EmailBackend" # default SES v2
to this:
.. code-block:: python
EMAIL_BACKEND = "anymail.backends.amazon_sesv1.EmailBackend" # SES v1
# ^^
Note that SES v1 support is deprecated and will be removed in a future Anymail release
(likely in late 2023).
.. _amazon-ses-quirks: .. _amazon-ses-quirks:
@@ -793,10 +722,8 @@ This IAM policy covers all of those:
}] }]
} }
(To send using the deprecated ``amazon_sesv1`` EmailBackend, (Anymail does not need access to ``ses:SendRawEmail``
you will also need to allow ``ses:SendRawEmail`` for ordinary, or ``ses:SendBulkTemplatedEmail``. Those are SES v1 actions.)
non-templated sends, and/or ``ses:SendBulkTemplatedEmail`` for
templated/merge sends.)
.. _amazon-ses-iam-errors: .. _amazon-ses-iam-errors:
@@ -854,3 +781,60 @@ for any features you aren't using, and you may want to add additional restrictio
https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-permissions.html https://docs.aws.amazon.com/ses/latest/DeveloperGuide/receiving-email-permissions.html
.. _IAM condition context keys: .. _IAM condition context keys:
https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html
.. _amazon-ses-v1:
.. _amazon-ses-v2:
Migrating to the SES v2 API
---------------------------
.. versionchanged:: 10.0
Anymail 10.0 and later use Amazon's updated SES v2 API to send email. Earlier Anymail releases
used the original Amazon SES API (v1) by default. Although the capabilities of the two
SES versions are virtually identical, Amazon is implementing improvements (such as
increased maximum message size) only in the v2 API.
(The upgrade for SES v2 affects only sending email. There are no changes required
for status tracking webhooks or receiving inbound email.)
Migrating to SES v2 requires minimal code changes:
1. Update your :ref:`IAM permissions <amazon-ses-iam-permissions>` to grant Anymail
access to the SES v2 sending actions: ``ses:SendEmail`` for ordinary sends, and/or
``ses:SendBulkEmail`` to send using SES templates. (The IAM action
prefix is just ``ses`` for both the v1 and v2 APIs.)
Access to ``ses:SendRawEmail`` or ``ses:SendBulkTemplatedEmail`` can be removed.
(Those actions are only needed for SES v1.)
If you run into unexpected IAM authorization failures, see the note about
:ref:`misleading IAM permissions errors <amazon-ses-iam-errors>` above.
2. If your code uses Anymail's :attr:`~anymail.message.AnymailMessage.esp_extra`
to pass additional SES API parameters, or examines the raw
:attr:`~anymail.message.AnymailStatus.esp_response` after sending a message,
you may need to update it for the v2 API. Many parameters have different names
in the v2 API compared to the equivalent v1 calls, and the response formats are
slightly different.
Among v1 parameters commonly used, ``ConfigurationSetName`` is unchanged in v2,
but v1's ``Tags`` and most ``*Arn`` parameters have been renamed in v2.
See AWS's docs for SES v1 `SendRawEmail`_ vs. v2 `SendEmail`_, or if you are sending
with SES templates, compare v1 `SendBulkTemplatedEmail`_ to v2 `SendBulkEmail`_.
(If you do not use :attr:`!esp_extra` or :attr:`!esp_response`, you can
safely ignore this.)
3. If your settings.py :setting:`!EMAIL_BACKEND` setting refers to ``amazon_sesv1``
or ``amazon_sesv2``, change that to just ``amazon_ses``:
.. code-block:: python
EMAIL_BACKEND = "anymail.backends.amazon_ses.EmailBackend"
.. _SendRawEmail:
https://docs.aws.amazon.com/ses/latest/APIReference/API_SendRawEmail.html
.. _SendBulkTemplatedEmail:
https://docs.aws.amazon.com/ses/latest/APIReference/API_SendBulkTemplatedEmail.html

21
tests/test_amazon_ses_backend.py
View File

@@ -9,11 +9,7 @@ from django.core.mail import BadHeaderError
from django.test import SimpleTestCase, override_settings, tag from django.test import SimpleTestCase, override_settings, tag
from anymail import __version__ as ANYMAIL_VERSION from anymail import __version__ as ANYMAIL_VERSION
from anymail.exceptions import ( from anymail.exceptions import AnymailAPIError, AnymailUnsupportedFeature
AnymailAPIError,
AnymailDeprecationWarning,
AnymailUnsupportedFeature,
)
from anymail.inbound import AnymailInboundMessage from anymail.inbound import AnymailInboundMessage
from anymail.message import AnymailMessage, attach_inline_image_file from anymail.message import AnymailMessage, attach_inline_image_file
@@ -973,18 +969,3 @@ class AmazonSESBackendConfigurationTests(AmazonSESBackendMockAPITestCase):
self.message.send() self.message.send()
params = self.get_send_params() params = self.get_send_params()
self.assertEqual(params["ConfigurationSetName"], "CustomConfigurationSet") self.assertEqual(params["ConfigurationSetName"], "CustomConfigurationSet")
@override_settings(EMAIL_BACKEND="anymail.backends.amazon_sesv2.EmailBackend")
def test_sesv2_warning(self):
# Default SES v2 backend is still available as "amazon_sesv2",
# but using that should warn to switch to just "amazon_ses".
with self.assertWarnsMessage(
AnymailDeprecationWarning,
"Please change 'amazon_sesv2' to 'amazon_ses' in your EMAIL_BACKEND setting.",
):
self.message.send()
def test_no_warning_default(self):
# Default SES backend does not have "amazon_sesv2" warning.
with self.assertDoesNotWarn(AnymailDeprecationWarning):
self.message.send()

928
tests/test_amazon_ses_backendv1.py
View File

@@ -1,928 +0,0 @@
import json
import warnings
from datetime import datetime
from email.mime.application import MIMEApplication
from unittest.mock import ANY, patch
from django.core import mail
from django.core.mail import BadHeaderError
from django.test import SimpleTestCase, override_settings, tag
from anymail import __version__ as ANYMAIL_VERSION
from anymail.exceptions import (
AnymailAPIError,
AnymailDeprecationWarning,
AnymailUnsupportedFeature,
)
from anymail.inbound import AnymailInboundMessage
from anymail.message import AnymailMessage, attach_inline_image_file
from .utils import (
SAMPLE_IMAGE_FILENAME,
AnymailTestMixin,
sample_image_content,
sample_image_path,
)
@tag("amazon_ses")
@override_settings(EMAIL_BACKEND="anymail.backends.amazon_sesv1.EmailBackend")
class AmazonSESBackendMockAPITestCase(AnymailTestMixin, SimpleTestCase):
"""TestCase that uses the Amazon SES EmailBackend with a mocked boto3 client"""
def setUp(self):
super().setUp()
# Silence the "amazon_sesv1.EmailBackend is deprecated" warning for these tests.
# (Tests can still verify the warning with assertWarns.)
warnings.simplefilter("ignore", category=AnymailDeprecationWarning)
# Mock boto3.session.Session().client('ses').send_raw_email (and any other
# client operations). (We could also use botocore.stub.Stubber, but mock works
# well with our test structure.)
self.patch_boto3_session = patch(
"anymail.backends.amazon_sesv1.boto3.session.Session", autospec=True
)
self.mock_session = self.patch_boto3_session.start() # boto3.session.Session
self.addCleanup(self.patch_boto3_session.stop)
#: boto3.session.Session().client
self.mock_client = self.mock_session.return_value.client
#: boto3.session.Session().client('ses', ...)
self.mock_client_instance = self.mock_client.return_value
self.set_mock_response()
# Simple message useful for many tests
self.message = mail.EmailMultiAlternatives(
"Subject", "Text Body", "from@example.com", ["to@example.com"]
)
DEFAULT_SEND_RESPONSE = {
"MessageId": "1111111111111111-bbbbbbbb-3333-7777-aaaa-eeeeeeeeeeee-000000",
"ResponseMetadata": {
"RequestId": "aaaaaaaa-2222-1111-8888-bbbb3333bbbb",
"HTTPStatusCode": 200,
"HTTPHeaders": {
"x-amzn-requestid": "aaaaaaaa-2222-1111-8888-bbbb3333bbbb",
"content-type": "text/xml",
"content-length": "338",
"date": "Sat, 17 Mar 2018 03:33:33 GMT",
},
"RetryAttempts": 0,
},
}
def set_mock_response(self, response=None, operation_name="send_raw_email"):
mock_operation = getattr(self.mock_client_instance, operation_name)
mock_operation.return_value = response or self.DEFAULT_SEND_RESPONSE
return mock_operation.return_value
def set_mock_failure(self, response, operation_name="send_raw_email"):
from botocore.exceptions import ClientError
mock_operation = getattr(self.mock_client_instance, operation_name)
mock_operation.side_effect = ClientError(
response, operation_name=operation_name
)
def get_session_params(self):
if self.mock_session.call_args is None:
raise AssertionError("boto3 Session was not created")
(args, kwargs) = self.mock_session.call_args
if args:
raise AssertionError(
"boto3 Session created with unexpected positional args %r" % args
)
return kwargs
def get_client_params(self, service="ses"):
"""Returns kwargs params passed to mock boto3 client constructor
Fails test if boto3 client wasn't constructed with named service
"""
if self.mock_client.call_args is None:
raise AssertionError("boto3 client was not created")
(args, kwargs) = self.mock_client.call_args
if len(args) != 1:
raise AssertionError(
"boto3 client created with unexpected positional args %r" % args
)
if args[0] != service:
raise AssertionError(
"boto3 client created with service %r, not %r" % (args[0], service)
)
return kwargs
def get_send_params(self, operation_name="send_raw_email"):
"""Returns kwargs params passed to the mock send API.
Fails test if API wasn't called.
"""
self.mock_client.assert_called_with("ses", config=ANY)
mock_operation = getattr(self.mock_client_instance, operation_name)
if mock_operation.call_args is None:
raise AssertionError("API was not called")
(args, kwargs) = mock_operation.call_args
return kwargs
def get_sent_message(self):
"""Returns a parsed version of the send_raw_email RawMessage.Data param"""
params = self.get_send_params(
operation_name="send_raw_email"
# (other operations don't have raw mime param)
)
raw_mime = params["RawMessage"]["Data"]
parsed = AnymailInboundMessage.parse_raw_mime_bytes(raw_mime)
return parsed
def assert_esp_not_called(self, msg=None, operation_name="send_raw_email"):
mock_operation = getattr(self.mock_client_instance, operation_name)
if mock_operation.called:
raise AssertionError(msg or "ESP API was called and shouldn't have been")
@tag("amazon_ses")
class AmazonSESBackendStandardEmailTests(AmazonSESBackendMockAPITestCase):
"""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,
)
params = self.get_send_params()
# send_raw_email takes a fully-formatted MIME message.
# This is a simple (if inexact) way to check for expected headers and body:
raw_mime = params["RawMessage"]["Data"]
self.assertIsInstance(raw_mime, bytes) # SendRawEmail expects Data as bytes
self.assertIn(b"\nFrom: from@example.com\n", raw_mime)
self.assertIn(b"\nTo: to@example.com\n", raw_mime)
self.assertIn(b"\nSubject: Subject here\n", raw_mime)
self.assertIn(b"\n\nHere is the message", raw_mime)
# Destinations must include all recipients:
self.assertEqual(params["Destinations"], ["to@example.com"])
# Since the SES backend generates the MIME message using Django's
# EmailMessage.message().to_string(), there's not really a need
# to exhaustively test all the various standard email features.
# (EmailMessage.message() is well tested in the Django codebase.)
# Instead, just spot-check a few things...
def test_destinations(self):
self.message.to = ["to1@example.com", '"Recipient, second" <to2@example.com>']
self.message.cc = ["cc1@example.com", "Also cc <cc2@example.com>"]
self.message.bcc = ["bcc1@example.com", "BCC 2 <bcc2@example.com>"]
self.message.send()
params = self.get_send_params()
self.assertEqual(
params["Destinations"],
[
"to1@example.com",
'"Recipient, second" <to2@example.com>',
"cc1@example.com",
"Also cc <cc2@example.com>",
"bcc1@example.com",
"BCC 2 <bcc2@example.com>",
],
)
# Bcc's shouldn't appear in the message itself:
self.assertNotIn(b"bcc", params["RawMessage"]["Data"])
def test_non_ascii_headers(self):
self.message.subject = "Thử tin nhắn" # utf-8 in subject header
self.message.to = ['"Người nhận" <to@example.com>'] # utf-8 in display name
self.message.cc = ["cc@thư.example.com"] # utf-8 in domain
self.message.send()
params = self.get_send_params()
raw_mime = params["RawMessage"]["Data"]
# Non-ASCII headers must use MIME encoded-word syntax:
self.assertIn(b"\nSubject: =?utf-8?b?VGjhu60gdGluIG5o4bqvbg==?=\n", raw_mime)
# Non-ASCII display names as well:
self.assertIn(
b"\nTo: =?utf-8?b?TmfGsOG7nWkgbmjhuq1u?= <to@example.com>\n", raw_mime
)
# Non-ASCII address domains must use Punycode:
self.assertIn(b"\nCc: cc@xn--th-e0a.example.com\n", raw_mime)
# SES doesn't support non-ASCII in the username@ part
# (RFC 6531 "SMTPUTF8" extension)
# Destinations must include all recipients:
self.assertEqual(
params["Destinations"],
[
"=?utf-8?b?TmfGsOG7nWkgbmjhuq1u?= <to@example.com>",
"cc@xn--th-e0a.example.com",
],
)
def test_attachments(self):
# These are \u2022 bullets ("\N{BULLET}") below:
text_content = "• Item one\n• Item two\n• Item three"
self.message.attach(
filename="Une pièce jointe.txt", # utf-8 chars in filename
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 = MIMEApplication(pdf_content, "pdf") # application/pdf
mimeattachment["Content-Disposition"] = "attachment"
self.message.attach(mimeattachment)
self.message.send()
sent_message = self.get_sent_message()
attachments = sent_message.attachments
self.assertEqual(len(attachments), 3)
self.assertEqual(attachments[0].get_content_type(), "text/plain")
self.assertEqual(attachments[0].get_filename(), "Une pièce jointe.txt")
self.assertEqual(attachments[0].get_param("charset"), "utf-8")
self.assertEqual(attachments[0].get_content_text(), text_content)
self.assertEqual(attachments[1].get_content_type(), "image/png")
# not inline:
self.assertEqual(attachments[1].get_content_disposition(), "attachment")
self.assertEqual(attachments[1].get_filename(), "test.png")
self.assertEqual(attachments[1].get_content_bytes(), png_content)
self.assertEqual(attachments[2].get_content_type(), "application/pdf")
self.assertIsNone(attachments[2].get_filename()) # no filename specified
self.assertEqual(attachments[2].get_content_bytes(), pdf_content)
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, domain="example.com")
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()
sent_message = self.get_sent_message()
self.assertEqual(sent_message.html, html_content)
inlines = sent_message.content_id_map
self.assertEqual(len(inlines), 1)
self.assertEqual(inlines[cid].get_content_type(), "image/png")
self.assertEqual(inlines[cid].get_filename(), image_filename)
self.assertEqual(inlines[cid].get_content_bytes(), image_data)
# Make sure neither the html nor the inline image is treated as an attachment:
params = self.get_send_params()
raw_mime = params["RawMessage"]["Data"]
self.assertNotIn(b"\nContent-Disposition: attachment", raw_mime)
def test_multiple_html_alternatives(self):
# Multiple alternatives *are* allowed
self.message.attach_alternative("<p>First html is OK</p>", "text/html")
self.message.attach_alternative("<p>And so is second</p>", "text/html")
self.message.send()
params = self.get_send_params()
raw_mime = params["RawMessage"]["Data"]
# just check the alternative smade it into the message
# (assume that Django knows how to format them properly)
self.assertIn(b"\n\n<p>First html is OK</p>\n", raw_mime)
self.assertIn(b"\n\n<p>And so is second</p>\n", raw_mime)
def test_alternative(self):
# Non-HTML alternatives *are* allowed
self.message.attach_alternative('{"is": "allowed"}', "application/json")
self.message.send()
params = self.get_send_params()
raw_mime = params["RawMessage"]["Data"]
# just check the alternative made it into the message
# (assume that Django knows how to format it properly)
self.assertIn(b"\nContent-Type: application/json\n", raw_mime)
def test_multiple_from(self):
# Amazon allows multiple addresses in the From header,
# but must specify which is Source
self.message.from_email = "from1@example.com, from2@example.com"
self.message.send()
params = self.get_send_params()
raw_mime = params["RawMessage"]["Data"]
self.assertIn(b"\nFrom: from1@example.com, from2@example.com\n", raw_mime)
self.assertEqual(params["Source"], "from1@example.com")
def test_commas_in_subject(self):
"""
There used to be a Python email header bug that added unwanted spaces
after commas in long subjects
"""
self.message.subject = (
"100,000,000 isn't a number you'd really want"
" to break up in this email subject, right?"
)
self.message.send()
sent_message = self.get_sent_message()
self.assertEqual(sent_message["Subject"], self.message.subject)
def test_body_avoids_cte_8bit(self):
"""Anymail works around an Amazon SES bug that can corrupt non-ASCII bodies."""
# (see detailed comments in the backend code)
self.message.body = "Это text body"
self.message.attach_alternative("<p>Это html body</p>", "text/html")
self.message.send()
sent_message = self.get_sent_message()
# Make sure none of the text parts use `Content-Transfer-Encoding: 8bit`.
# (Technically, either quoted-printable or base64 would be OK, but base64 text
# parts have a reputation for triggering spam filters, so just require
# quoted-printable.)
text_part_encodings = [
(part.get_content_type(), part["Content-Transfer-Encoding"])
for part in sent_message.walk()
if part.get_content_maintype() == "text"
]
self.assertEqual(
text_part_encodings,
[
("text/plain", "quoted-printable"),
("text/html", "quoted-printable"),
],
)
def test_api_failure(self):
error_response = {
"Error": {
"Type": "Sender",
"Code": "MessageRejected",
"Message": "Email address is not verified. The following identities"
" failed the check in region US-EAST-1: to@example.com",
},
"ResponseMetadata": {
"RequestId": "aaaaaaaa-2222-1111-8888-bbbb3333bbbb",
"HTTPStatusCode": 400,
"HTTPHeaders": {
"x-amzn-requestid": "aaaaaaaa-2222-1111-8888-bbbb3333bbbb",
"content-type": "text/xml",
"content-length": "277",
"date": "Sat, 17 Mar 2018 04:44:44 GMT",
},
"RetryAttempts": 0,
},
}
self.set_mock_failure(error_response)
with self.assertRaises(AnymailAPIError) as cm:
self.message.send()
err = cm.exception
# AWS error is included in Anymail message:
self.assertIn(
"Email address is not verified. The following identities failed "
"the check in region US-EAST-1: to@example.com",
str(err),
)
# Raw AWS response is available on the exception:
self.assertEqual(err.response, error_response)
def test_api_failure_fail_silently(self):
# Make sure fail_silently is respected
self.set_mock_failure(
{
"Error": {
"Type": "Sender",
"Code": "InvalidParameterValue",
"Message": "That is not allowed",
}
}
)
sent = self.message.send(fail_silently=True)
self.assertEqual(sent, 0)
def test_session_failure_fail_silently(self):
# Make sure fail_silently is respected if boto3.Session creation fails
# (e.g., due to invalid or missing credentials)
from botocore.exceptions import NoCredentialsError
self.mock_session.side_effect = NoCredentialsError()
sent = self.message.send(fail_silently=True)
self.assertEqual(sent, 0)
def test_prevents_header_injection(self):
# Since we build the raw MIME message, we're responsible for preventing header
# injection. django.core.mail.EmailMessage.message() implements most of that
# (for the SMTP backend); spot check some likely cases just to be sure...
with self.assertRaises(BadHeaderError):
mail.send_mail(
"Subject\r\ninjected", "Body", "from@example.com", ["to@example.com"]
)
with self.assertRaises(BadHeaderError):
mail.send_mail(
"Subject",
"Body",
'"Display-Name\nInjected" <from@example.com>',
["to@example.com"],
)
with self.assertRaises(BadHeaderError):
mail.send_mail(
"Subject",
"Body",
"from@example.com",
['"Display-Name\rInjected" <to@example.com>'],
)
with self.assertRaises(BadHeaderError):
mail.EmailMessage(
"Subject",
"Body",
"from@example.com",
["to@example.com"],
headers={"X-Header": "custom header value\r\ninjected"},
).send()
@tag("amazon_ses")
class AmazonSESBackendAnymailFeatureTests(AmazonSESBackendMockAPITestCase):
"""Test backend support for Anymail added features"""
def test_envelope_sender(self):
self.message.envelope_sender = "bounce-handler@bounces.example.com"
self.message.send()
params = self.get_send_params()
self.assertEqual(params["Source"], "bounce-handler@bounces.example.com")
def test_spoofed_to(self):
# Amazon SES is one of the few ESPs that actually permits the To header
# to differ from the envelope recipient...
self.message.to = ["Envelope <envelope-to@example.com>"]
self.message.extra_headers["To"] = "Spoofed <spoofed-to@elsewhere.example.org>"
self.message.send()
params = self.get_send_params()
raw_mime = params["RawMessage"]["Data"]
self.assertEqual(params["Destinations"], ["Envelope <envelope-to@example.com>"])
self.assertIn(b"\nTo: Spoofed <spoofed-to@elsewhere.example.org>\n", raw_mime)
self.assertNotIn(b"envelope-to@example.com", raw_mime)
def test_metadata(self):
# (that \n is a header-injection test)
self.message.metadata = {
"User ID": 12345,
"items": "Correct horse,Battery,\nStaple",
"Cart-Total": "22.70",
}
self.message.send()
# Metadata is passed as JSON in a message header field:
sent_message = self.get_sent_message()
self.assertJSONEqual(
sent_message["X-Metadata"],
'{"User ID": 12345,'
' "items": "Correct horse,Battery,\\nStaple",'
' "Cart-Total": "22.70"}',
)
def test_send_at(self):
# Amazon SES does not support delayed sending
self.message.send_at = datetime(2016, 3, 4, 5, 6, 7)
with self.assertRaisesMessage(AnymailUnsupportedFeature, "send_at"):
self.message.send()
def test_tags(self):
self.message.tags = ["Transactional", "Cohort 12/2017"]
self.message.send()
# Tags are added as multiple X-Tag message headers:
sent_message = self.get_sent_message()
self.assertCountEqual(
sent_message.get_all("X-Tag"), ["Transactional", "Cohort 12/2017"]
)
# Tags are *not* by default used as Amazon SES "Message Tags":
params = self.get_send_params()
self.assertNotIn("Tags", params)
@override_settings(ANYMAIL_AMAZON_SES_MESSAGE_TAG_NAME="Campaign")
def test_amazon_message_tags(self):
"""
The Anymail AMAZON_SES_MESSAGE_TAG_NAME setting enables a single Message Tag
"""
self.message.tags = ["Welcome"]
self.message.send()
params = self.get_send_params()
self.assertEqual(params["Tags"], [{"Name": "Campaign", "Value": "Welcome"}])
# Multiple Anymail tags are not supported when using this feature
self.message.tags = ["Welcome", "Variation_A"]
with self.assertRaisesMessage(
AnymailUnsupportedFeature,
"multiple tags with the AMAZON_SES_MESSAGE_TAG_NAME setting",
):
self.message.send()
def test_tracking(self):
# Amazon SES doesn't support overriding click/open-tracking settings
# on individual messages through any standard API params.
# (You _can_ use a ConfigurationSet to control this; see esp_extra below.)
self.message.track_clicks = True
with self.assertRaisesMessage(AnymailUnsupportedFeature, "track_clicks"):
self.message.send()
delattr(self.message, "track_clicks")
self.message.track_opens = True
with self.assertRaisesMessage(AnymailUnsupportedFeature, "track_opens"):
self.message.send()
def test_merge_data(self):
# Amazon SES only supports merging when using templates (see below)
self.message.merge_data = {}
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "merge_data without template_id"
):
self.message.send()
delattr(self.message, "merge_data")
self.message.merge_global_data = {"group": "Users", "site": "ExampleCo"}
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "global_merge_data without template_id"
):
self.message.send()
@override_settings(
# only way to use tags with template_id:
ANYMAIL_AMAZON_SES_MESSAGE_TAG_NAME="Campaign"
)
def test_template(self):
"""With template_id, Anymail switches to SES SendBulkTemplatedEmail"""
# SendBulkTemplatedEmail uses a completely different API call and payload
# structure, so this re-tests a bunch of Anymail features that were handled
# differently above. (See test_amazon_ses_integration for a more realistic
# template example.)
raw_response = {
"Status": [
{
"Status": "Success",
"MessageId": "1111111111111111-bbbbbbbb-3333-7777",
},
{"Status": "AccountThrottled"},
],
"ResponseMetadata": self.DEFAULT_SEND_RESPONSE["ResponseMetadata"],
}
self.set_mock_response(raw_response, operation_name="send_bulk_templated_email")
message = AnymailMessage(
template_id="welcome_template",
from_email='"Example, Inc." <from@example.com>',
to=["alice@example.com", "罗伯特 <bob@example.com>"],
cc=["cc@example.com"],
reply_to=["reply1@example.com", "Reply 2 <reply2@example.com>"],
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"},
},
merge_global_data={"group": "Users", "site": "ExampleCo"},
# (only works with AMAZON_SES_MESSAGE_TAG_NAME when using template):
tags=["WelcomeVariantA"],
envelope_sender="bounces@example.com",
esp_extra={
"SourceArn": "arn:aws:ses:us-east-1:123456789012:identity/example.com"
},
)
message.send()
# templates use a different API call...
self.assert_esp_not_called(operation_name="send_raw_email")
params = self.get_send_params(operation_name="send_bulk_templated_email")
self.assertEqual(params["Template"], "welcome_template")
self.assertEqual(params["Source"], '"Example, Inc." <from@example.com>')
destinations = params["Destinations"]
self.assertEqual(len(destinations), 2)
self.assertEqual(
destinations[0]["Destination"],
{"ToAddresses": ["alice@example.com"], "CcAddresses": ["cc@example.com"]},
)
self.assertEqual(
json.loads(destinations[0]["ReplacementTemplateData"]),
{"name": "Alice", "group": "Developers"},
)
self.assertEqual(
destinations[1]["Destination"],
{
# SES requires RFC2047:
"ToAddresses": ["=?utf-8?b?572X5Lyv54m5?= <bob@example.com>"],
"CcAddresses": ["cc@example.com"],
},
)
self.assertEqual(
json.loads(destinations[1]["ReplacementTemplateData"]), {"name": "Bob"}
)
self.assertEqual(
json.loads(params["DefaultTemplateData"]),
{"group": "Users", "site": "ExampleCo"},
)
self.assertEqual(
params["ReplyToAddresses"],
["reply1@example.com", "Reply 2 <reply2@example.com>"],
)
self.assertEqual(
params["DefaultTags"], [{"Name": "Campaign", "Value": "WelcomeVariantA"}]
)
self.assertEqual(params["ReturnPath"], "bounces@example.com")
self.assertEqual(
params["SourceArn"],
"arn:aws:ses:us-east-1:123456789012:identity/example.com", # esp_extra
)
self.assertEqual(message.anymail_status.status, {"queued", "failed"})
self.assertEqual(
# different for each recipient
message.anymail_status.message_id,
{"1111111111111111-bbbbbbbb-3333-7777", None},
)
self.assertEqual(
message.anymail_status.recipients["alice@example.com"].status, "queued"
)
self.assertEqual(
message.anymail_status.recipients["bob@example.com"].status, "failed"
)
self.assertEqual(
message.anymail_status.recipients["alice@example.com"].message_id,
"1111111111111111-bbbbbbbb-3333-7777",
)
self.assertIsNone(
message.anymail_status.recipients["bob@example.com"].message_id
)
self.assertEqual(message.anymail_status.esp_response, raw_response)
def test_template_unsupported(self):
"""A lot of options are not compatible with SendBulkTemplatedEmail"""
message = AnymailMessage(template_id="welcome_template", to=["to@example.com"])
message.subject = "nope, can't change template subject"
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "overriding template subject"
):
message.send()
message.subject = None
message.body = "nope, can't change text body"
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "overriding template body content"
):
message.send()
message.content_subtype = "html"
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "overriding template body content"
):
message.send()
message.body = None
message.attach("attachment.txt", "this is an attachment", "text/plain")
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "attachments with template"
):
message.send()
message.attachments = []
message.extra_headers = {"X-Custom": "header"}
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "extra_headers with template"
):
message.send()
message.extra_headers = {}
message.metadata = {"meta": "data"}
with self.assertRaisesMessage(
AnymailUnsupportedFeature, "metadata with template"
):
message.send()
message.metadata = None
message.tags = ["tag 1", "tag 2"]
with self.assertRaisesMessage(AnymailUnsupportedFeature, "tags with template"):
message.send()
message.tags = None
def test_send_anymail_message_without_template(self):
# Make sure SendRawEmail is used for non-template_id messages
message = AnymailMessage(
from_email="from@example.com", to=["to@example.com"], subject="subject"
)
message.send()
self.assert_esp_not_called(operation_name="send_bulk_templated_email")
# fails if send_raw_email not called:
self.get_send_params(operation_name="send_raw_email")
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()
params = self.get_send_params()
self.assertNotIn("ConfigurationSetName", params)
self.assertNotIn("DefaultTags", params)
self.assertNotIn("DefaultTemplateData", params)
self.assertNotIn("FromArn", params)
self.assertNotIn("Message", params)
self.assertNotIn("ReplyToAddresses", params)
self.assertNotIn("ReturnPath", params)
self.assertNotIn("ReturnPathArn", params)
self.assertNotIn("Source", params)
self.assertNotIn("SourceArn", params)
self.assertNotIn("Tags", params)
self.assertNotIn("Template", params)
self.assertNotIn("TemplateArn", params)
self.assertNotIn("TemplateData", params)
sent_message = self.get_sent_message()
# custom headers not added if not needed:
self.assertNotIn("X-Metadata", sent_message)
self.assertNotIn("X-Tag", sent_message)
def test_esp_extra(self):
# Values in esp_extra are merged into the Amazon SES SendRawEmail parameters
self.message.esp_extra = {
# E.g., if you've set up a configuration set
# that disables open/click tracking:
"ConfigurationSetName": "NoTrackingConfigurationSet",
}
self.message.send()
params = self.get_send_params()
self.assertEqual(params["ConfigurationSetName"], "NoTrackingConfigurationSet")
def test_send_attaches_anymail_status(self):
"""The anymail_status should be attached to the message when it is sent"""
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,
"1111111111111111-bbbbbbbb-3333-7777-aaaa-eeeeeeeeeeee-000000",
)
self.assertEqual(
msg.anymail_status.recipients["to1@example.com"].status, "queued"
)
self.assertEqual(
msg.anymail_status.recipients["to1@example.com"].message_id,
"1111111111111111-bbbbbbbb-3333-7777-aaaa-eeeeeeeeeeee-000000",
)
self.assertEqual(msg.anymail_status.esp_response, self.DEFAULT_SEND_RESPONSE)
# Amazon SES doesn't report rejected addresses at send time in a form that can be
# distinguished from other API errors. If SES rejects *any* recipient you'll get
# an AnymailAPIError, and the message won't be sent to *all* recipients.
# noinspection PyUnresolvedReferences
def test_send_unparsable_response(self):
"""
If the send succeeds, but result is unexpected format,
should raise an API exception
"""
response_content = {"wrong": "format"}
self.set_mock_response(response_content)
with self.assertRaisesMessage(
AnymailAPIError, "parsing Amazon SES send result"
):
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, response_content)
@tag("amazon_ses")
class AmazonSESBackendConfigurationTests(AmazonSESBackendMockAPITestCase):
"""Test configuration options"""
def test_deprecation_warning(self):
with self.assertWarnsMessage(
AnymailDeprecationWarning,
"anymail.backends.amazon_sesv1.EmailBackend is deprecated",
):
self.message.send()
def test_boto_default_config(self):
"""By default, boto3 gets credentials from the environment or its config files
See http://boto3.readthedocs.io/en/stable/guide/configuration.html
"""
self.message.send()
session_params = self.get_session_params()
# no additional params passed to boto3.session.Session():
self.assertEqual(session_params, {})
client_params = self.get_client_params()
# Ignore botocore.config.Config, which doesn't support ==
config = client_params.pop("config")
# no additional params passed to session.client('ses'):
self.assertEqual(client_params, {})
self.assertIn(
f"django-anymail/{ANYMAIL_VERSION}-amazon-ses",
config.user_agent_extra,
)
@override_settings(
ANYMAIL={
"AMAZON_SES_CLIENT_PARAMS": {
# Example for testing; it's not a good idea to hardcode credentials in
# your code. Safer: `os.getenv("MY_SPECIAL_AWS_KEY_ID")` etc.
"aws_access_key_id": "test-access-key-id",
"aws_secret_access_key": "test-secret-access-key",
"region_name": "ap-northeast-1",
# config can be given as dict of botocore.config.Config params
"config": {
"read_timeout": 30,
"retries": {"max_attempts": 2},
},
}
}
)
def test_client_params_in_setting(self):
"""
The Anymail AMAZON_SES_CLIENT_PARAMS setting specifies
boto3 session.client() params for Anymail
"""
self.message.send()
client_params = self.get_client_params()
# Ignore botocore.config.Config, which doesn't support ==
config = client_params.pop("config")
self.assertEqual(
client_params,
{
"aws_access_key_id": "test-access-key-id",
"aws_secret_access_key": "test-secret-access-key",
"region_name": "ap-northeast-1",
},
)
self.assertEqual(config.read_timeout, 30)
self.assertEqual(config.retries, {"max_attempts": 2})
def test_client_params_in_connection_init(self):
"""
You can also supply credentials specifically
for a particular EmailBackend connection instance
"""
from botocore.config import Config
boto_config = Config(connect_timeout=30)
conn = mail.get_connection(
"anymail.backends.amazon_sesv1.EmailBackend",
client_params={
"aws_session_token": "test-session-token",
"config": boto_config,
},
)
conn.send_messages([self.message])
client_params = self.get_client_params()
# Ignore botocore.config.Config, which doesn't support ==
config = client_params.pop("config")
self.assertEqual(client_params, {"aws_session_token": "test-session-token"})
self.assertEqual(config.connect_timeout, 30)
@override_settings(
ANYMAIL={"AMAZON_SES_SESSION_PARAMS": {"profile_name": "anymail-testing"}}
)
def test_session_params_in_setting(self):
"""
The Anymail AMAZON_SES_SESSION_PARAMS setting
specifies boto3.session.Session() params for Anymail
"""
self.message.send()
session_params = self.get_session_params()
self.assertEqual(session_params, {"profile_name": "anymail-testing"})
client_params = self.get_client_params()
# Ignore botocore.config.Config, which doesn't support ==
client_params.pop("config")
# no additional params passed to session.client('ses'):
self.assertEqual(client_params, {})
@override_settings(
ANYMAIL={"AMAZON_SES_CONFIGURATION_SET_NAME": "MyConfigurationSet"}
)
def test_config_set_setting(self):
"""You can supply a default ConfigurationSetName"""
self.message.send()
params = self.get_send_params()
self.assertEqual(params["ConfigurationSetName"], "MyConfigurationSet")
# override on individual message using esp_extra
self.message.esp_extra = {"ConfigurationSetName": "CustomConfigurationSet"}
self.message.send()
params = self.get_send_params()
self.assertEqual(params["ConfigurationSetName"], "CustomConfigurationSet")

207
tests/test_amazon_ses_integrationv1.py
View File

@@ -1,207 +0,0 @@
import os
import unittest
import warnings
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_AMAZON_SES_ACCESS_KEY_ID = os.getenv(
"ANYMAIL_TEST_AMAZON_SES_ACCESS_KEY_ID"
)
ANYMAIL_TEST_AMAZON_SES_SECRET_ACCESS_KEY = os.getenv(
"ANYMAIL_TEST_AMAZON_SES_SECRET_ACCESS_KEY"
)
ANYMAIL_TEST_AMAZON_SES_REGION_NAME = os.getenv(
"ANYMAIL_TEST_AMAZON_SES_REGION_NAME", "us-east-1"
)
ANYMAIL_TEST_AMAZON_SES_DOMAIN = os.getenv("ANYMAIL_TEST_AMAZON_SES_DOMAIN")
@unittest.skipUnless(
ANYMAIL_TEST_AMAZON_SES_ACCESS_KEY_ID
and ANYMAIL_TEST_AMAZON_SES_SECRET_ACCESS_KEY
and ANYMAIL_TEST_AMAZON_SES_DOMAIN,
"Set ANYMAIL_TEST_AMAZON_SES_ACCESS_KEY_ID and"
" ANYMAIL_TEST_AMAZON_SES_SECRET_ACCESS_KEY and ANYMAIL_TEST_AMAZON_SES_DOMAIN"
" environment variables to run Amazon SES integration tests",
)
@override_settings(
EMAIL_BACKEND="anymail.backends.amazon_sesv1.EmailBackend",
ANYMAIL={
"AMAZON_SES_CLIENT_PARAMS": {
# This setting provides Anymail-specific AWS credentials to boto3.client(),
# overriding any credentials in the environment or boto config. It's often
# *not* the best approach. See the Anymail and boto3 docs for other options.
"aws_access_key_id": ANYMAIL_TEST_AMAZON_SES_ACCESS_KEY_ID,
"aws_secret_access_key": ANYMAIL_TEST_AMAZON_SES_SECRET_ACCESS_KEY,
"region_name": ANYMAIL_TEST_AMAZON_SES_REGION_NAME,
# Can supply any other boto3.client params,
# including botocore.config.Config as dict
"config": {"retries": {"max_attempts": 2}},
},
# actual config set in Anymail test account:
"AMAZON_SES_CONFIGURATION_SET_NAME": "TestConfigurationSet",
},
)
@tag("amazon_ses", "live")
class AmazonSESBackendIntegrationTests(AnymailTestMixin, SimpleTestCase):
"""Amazon SES API integration tests
These tests run against the **live** Amazon SES API, using the environment
variables `ANYMAIL_TEST_AMAZON_SES_ACCESS_KEY_ID` and
`ANYMAIL_TEST_AMAZON_SES_SECRET_ACCESS_KEY` as AWS credentials.
If those variables are not set, these tests won't run.
(You can also set the environment variable `ANYMAIL_TEST_AMAZON_SES_REGION_NAME`
to test SES using a region other than the default "us-east-1".)
Amazon SES doesn't offer a test mode -- it tries to send everything you ask.
To avoid stacking up a pile of undeliverable @example.com
emails, the tests use Amazon's @simulator.amazonses.com addresses.
https://docs.aws.amazon.com/ses/latest/DeveloperGuide/mailbox-simulator.html
"""
def setUp(self):
super().setUp()
self.from_email = "test@%s" % ANYMAIL_TEST_AMAZON_SES_DOMAIN
self.message = AnymailMessage(
"Anymail Amazon SES integration test",
"Text content",
self.from_email,
["success@simulator.amazonses.com"],
)
self.message.attach_alternative("<p>HTML content</p>", "text/html")
# boto3 relies on GC to close connections. Python 3 warns about unclosed
# ssl.SSLSocket during cleanup. We don't care. (It may be a false positive,
# or it may be a botocore problem, but it's not *our* problem.)
# https://github.com/boto/boto3/issues/454#issuecomment-586033745
# Filter in TestCase.setUp because unittest resets the warning filters
# for each test. https://stackoverflow.com/a/26620811/647002
warnings.filterwarnings(
"ignore", message=r"unclosed <ssl\.SSLSocket", category=ResourceWarning
)
def test_simple_send(self):
# Example of getting the Amazon SES 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[
"success@simulator.amazonses.com"
].status
message_id = anymail_status.recipients[
"success@simulator.amazonses.com"
].message_id
# Amazon SES always queues (or raises an error):
self.assertEqual(sent_status, "queued")
# Amazon SES message ids are groups of hex chars:
self.assertRegex(message_id, r"[0-9a-f-]+")
# 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):
message = AnymailMessage(
subject="Anymail Amazon SES all-options integration test",
body="This is the text body",
from_email=formataddr(("Test From, with comma", self.from_email)),
to=[
"success+to1@simulator.amazonses.com",
"Recipient 2 <success+to2@simulator.amazonses.com>",
],
cc=[
"success+cc1@simulator.amazonses.com",
"Copy 2 <success+cc2@simulator.amazonses.com>",
],
bcc=[
"success+bcc1@simulator.amazonses.com",
"Blind Copy 2 <success+bcc2@simulator.amazonses.com>",
],
reply_to=["reply1@example.com", "Reply 2 <reply2@example.com>"],
headers={"X-Anymail-Test": "value"},
metadata={"meta1": "simple_string", "meta2": 2},
tags=["Re-engagement", "Cohort 12/2017"],
)
message.attach("attachment1.txt", "Here is some\ntext for you", "text/plain")
message.attach("attachment2.csv", "ID,Name\n1,Amy Lina", "text/csv")
cid = message.attach_inline_image_file(sample_image_path())
message.attach_alternative(
"<p><b>HTML:</b> with <a href='http://example.com'>link</a>"
"and image: <img src='cid:%s'></div>" % cid,
"text/html",
)
message.attach_alternative(
"Amazon SES SendRawEmail actually supports multiple alternative parts",
"text/x-note-for-email-geeks",
)
message.send()
self.assertEqual(message.anymail_status.status, {"queued"})
def test_stored_template(self):
# Using a template created like this:
# boto3.client('ses').create_template(Template={
# "TemplateName": "TestTemplate",
# "SubjectPart": "Your order {{order}} shipped",
# "HtmlPart": "<h1>Dear {{name}}:</h1>"
# "<p>Your order {{order}} shipped {{ship_date}}.</p>",
# "TextPart": "Dear {{name}}:\r\n"
# "Your order {{order}} shipped {{ship_date}}."
# })
message = AnymailMessage(
template_id="TestTemplate",
from_email=formataddr(("Test From", self.from_email)),
to=[
"First Recipient <success+to1@simulator.amazonses.com>",
"success+to2@simulator.amazonses.com",
],
merge_data={
"success+to1@simulator.amazonses.com": {
"order": 12345,
"name": "Test Recipient",
},
"success+to2@simulator.amazonses.com": {"order": 6789},
},
merge_global_data={"name": "Customer", "ship_date": "today"}, # default
)
message.send()
recipient_status = message.anymail_status.recipients
self.assertEqual(
recipient_status["success+to1@simulator.amazonses.com"].status, "queued"
)
self.assertRegex(
recipient_status["success+to1@simulator.amazonses.com"].message_id,
r"[0-9a-f-]+",
)
self.assertEqual(
recipient_status["success+to2@simulator.amazonses.com"].status, "queued"
)
self.assertRegex(
recipient_status["success+to2@simulator.amazonses.com"].message_id,
r"[0-9a-f-]+",
)
@override_settings(
ANYMAIL={
"AMAZON_SES_CLIENT_PARAMS": {
"aws_access_key_id": "test-invalid-access-key-id",
"aws_secret_access_key": "test-invalid-secret-access-key",
"region_name": ANYMAIL_TEST_AMAZON_SES_REGION_NAME,
}
}
)
def test_invalid_aws_credentials(self):
# Make sure the exception message includes AWS's response:
with self.assertRaisesMessage(
AnymailAPIError, "The security token included in the request is invalid"
):
self.message.send()