import json 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.exceptions import AnymailAPIError, 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_ses.EmailBackend") class AmazonSESBackendMockAPITestCase(AnymailTestMixin, SimpleTestCase): """TestCase that uses the Amazon SES EmailBackend with a mocked boto3 client""" def setUp(self): super().setUp() # 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_ses.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" '] self.message.cc = ["cc1@example.com", "Also cc "] self.message.bcc = ["bcc1@example.com", "BCC 2 "] self.message.send() params = self.get_send_params() self.assertEqual( params["Destinations"], [ "to1@example.com", '"Recipient, second" ', "cc1@example.com", "Also cc ", "bcc1@example.com", "BCC 2 ", ], ) # 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" '] # 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?= \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?= ", "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 = ( '

This has an inline image.

' % 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.inline_attachments 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("

First html is OK

", "text/html") self.message.attach_alternative("

And so is second

", "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

First html is OK

\n", raw_mime) self.assertIn(b"\n\n

And so is second

\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("

Это html body

", "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" ', ["to@example.com"], ) with self.assertRaises(BadHeaderError): mail.send_mail( "Subject", "Body", "from@example.com", ['"Display-Name\rInjected" '], ) 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 "] self.message.extra_headers["To"] = "Spoofed " self.message.send() params = self.get_send_params() raw_mime = params["RawMessage"]["Data"] self.assertEqual(params["Destinations"], ["Envelope "]) self.assertIn(b"\nTo: Spoofed \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." ', to=["alice@example.com", "罗伯特 "], cc=["cc@example.com"], reply_to=["reply1@example.com", "Reply 2 "], 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." ') 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?= "], "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 "], ) 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_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.assertRegex( config.user_agent_extra, r"django-anymail/\d(\.\w+){1,}-amazon-ses" ) @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_ses.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")