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

Amazon SES support

Browse Source 
Integrate Amazon SES.

Closes #54.
This commit is contained in:
Mike Edmunds
2018-04-11 10:35:23 -07:00
committed by GitHub
parent d079a506a1
commit ef69fa3bf7
15 changed files with 3156 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

664
tests/test_amazon_ses_backend.py Normal file
View File

@@ -0,0 +1,664 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
from datetime import datetime
from email.mime.application import MIMEApplication
from unittest import skipIf
import botocore.config
import botocore.exceptions
import six
from django.core import mail
from django.core.mail import BadHeaderError
from django.test import SimpleTestCase
from django.test.utils import override_settings
from mock import ANY, patch
from anymail.exceptions import AnymailAPIError, AnymailUnsupportedFeature
from anymail.inbound import AnymailInboundMessage
from anymail.message import attach_inline_image_file, AnymailMessage
from .utils import (
AnymailTestMixin, SAMPLE_IMAGE_FILENAME, python_has_broken_mime_param_handling,
sample_image_content, sample_image_path)
@override_settings(EMAIL_BACKEND='anymail.backends.amazon_ses.EmailBackend')
class AmazonSESBackendMockAPITestCase(SimpleTestCase, AnymailTestMixin):
"""TestCase that uses the Amazon SES EmailBackend with a mocked boto3 client"""
def setUp(self):
super(AmazonSESBackendMockAPITestCase, self).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)
self.mock_client = self.mock_session.return_value.client # boto3.session.Session().client
self.mock_client_instance = self.mock_client.return_value # boto3.session.Session().client('ses', ...)
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"):
mock_operation = getattr(self.mock_client_instance, operation_name)
mock_operation.side_effect = botocore.exceptions.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")
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, six.binary_type) # 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)
# 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_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)
@skipIf(python_has_broken_mime_param_handling(),
"This Python has a buggy email package that crashes on non-ASCII "
"characters in RFC2231-encoded MIME header parameters")
def test_attachments(self):
text_content = "• Item one\n• Item two\n• Item three" # those are \u2022 bullets ("\N{BULLET}")
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")
self.assertEqual(attachments[1].get_content_disposition(), "attachment") # not inline
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.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("<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):
"""Anymail works around a Python 2 email header bug that adds 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_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_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()
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-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(ANYMAIL_AMAZON_SES_MESSAGE_TAG_NAME="Campaign") # only way to use tags with template_id
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"},
tags=["WelcomeVariantA"], # (only with AMAZON_SES_MESSAGE_TAG_NAME when using template)
envelope_sender="bounces@example.com",
esp_extra={'SourceArn': "arn:aws:ses:us-east-1:123456789012:identity/example.com"},
)
message.send()
self.assert_esp_not_called(operation_name="send_raw_email") # templates use a different API call...
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'],
{"ToAddresses": ['=?utf-8?b?572X5Lyv54m5?= <bob@example.com>'], # SES requires RFC2047
"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(message.anymail_status.message_id,
{"1111111111111111-bbbbbbbb-3333-7777", None}) # different for each recipient
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")
self.get_send_params(operation_name="send_raw_email") # fails if send_raw_email not called
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('Destinations', 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()
self.assertNotIn("X-Metadata", sent_message) # custom headers not added if not needed
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)
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()
self.assertEqual(session_params, {}) # no additional params passed to boto3.session.Session()
client_params = self.get_client_params()
config = client_params.pop("config") # Anymail adds a default config, which doesn't support ==
self.assertEqual(client_params, {}) # no additional params passed to session.client('ses')
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
"aws_access_key_id": "test-access-key-id", # safer: `os.getenv("MY_SPECIAL_AWS_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()
config = client_params.pop("config") # botocore.config.Config doesn't support ==
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"""
boto_config = botocore.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()
config = client_params.pop("config") # botocore.config.Config doesn't support ==
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()
client_params.pop("config") # Anymail adds a default config, which doesn't support ==
self.assertEqual(client_params, {}) # no additional params passed to session.client('ses')
@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")

311
tests/test_amazon_ses_inbound.py Normal file
View File

@@ -0,0 +1,311 @@
from __future__ import unicode_literals
import json
from base64 import b64encode
from datetime import datetime
from textwrap import dedent
import botocore.exceptions
from django.utils.timezone import utc
from mock import ANY, patch
from anymail.exceptions import AnymailAPIError, AnymailConfigurationError
from anymail.inbound import AnymailInboundMessage
from anymail.signals import AnymailInboundEvent
from anymail.webhooks.amazon_ses import AmazonSESInboundWebhookView
from .test_amazon_ses_webhooks import AmazonSESWebhookTestsMixin
from .webhook_cases import WebhookTestCase
class AmazonSESInboundTests(WebhookTestCase, AmazonSESWebhookTestsMixin):
def setUp(self):
super(AmazonSESInboundTests, self).setUp()
# Mock boto3.session.Session().client('s3').download_fileobj
# (We could also use botocore.stub.Stubber, but mock works well with our test structure)
self.patch_boto3_session = patch('anymail.webhooks.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)
def mock_download_fileobj(bucket, key, fileobj):
fileobj.write(self.mock_s3_downloadables[bucket][key])
self.mock_s3_downloadables = {} # bucket: key: bytes
self.mock_client = self.mock_session.return_value.client # boto3.session.Session().client
self.mock_s3 = self.mock_client.return_value # boto3.session.Session().client('s3', ...)
self.mock_s3.download_fileobj.side_effect = mock_download_fileobj
TEST_MIME_MESSAGE = dedent("""\
Return-Path: <bounce-handler@mail.example.org>
Received: from mail.example.org by inbound-smtp.us-east-1.amazonaws.com...
MIME-Version: 1.0
Received: by 10.1.1.1 with HTTP; Fri, 30 Mar 2018 10:21:49 -0700 (PDT)
From: "Sender, Inc." <from@example.org>
Date: Fri, 30 Mar 2018 10:21:50 -0700
Message-ID: <CAEPk3RKsi@mail.example.org>
Subject: Test inbound message
To: Recipient <inbound@example.com>, someone-else@example.org
Content-Type: multipart/alternative; boundary="94eb2c05e174adb140055b6339c5"
--94eb2c05e174adb140055b6339c5
Content-Type: text/plain; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable
It's a body=E2=80=A6
--94eb2c05e174adb140055b6339c5
Content-Type: text/html; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable
<div dir=3D"ltr">It's a body=E2=80=A6</div>
--94eb2c05e174adb140055b6339c5--
""").replace("\n", "\r\n")
def test_inbound_sns_utf8(self):
raw_ses_event = {
"notificationType": "Received",
"mail": {
"timestamp": "2018-03-30T17:21:51.636Z",
"source": "envelope-from@example.org",
"messageId": "jili9m351il3gkburn7o2f0u6788stij94c8ld01", # assigned by Amazon SES
"destination": ["inbound@example.com", "someone-else@example.org"],
"headersTruncated": False,
"headers": [
# (omitting a few headers that Amazon SES adds on receipt)
{"name": "Return-Path", "value": "<bounce-handler@mail.example.org>"},
{"name": "Received", "value": "from mail.example.org by inbound-smtp.us-east-1.amazonaws.com..."},
{"name": "MIME-Version", "value": "1.0"},
{"name": "Received", "value": "by 10.1.1.1 with HTTP; Fri, 30 Mar 2018 10:21:49 -0700 (PDT)"},
{"name": "From", "value": '"Sender, Inc." <from@example.org>'},
{"name": "Date", "value": "Fri, 30 Mar 2018 10:21:50 -0700"},
{"name": "Message-ID", "value": "<CAEPk3RKsi@mail.example.org>"},
{"name": "Subject", "value": "Test inbound message"},
{"name": "To", "value": "Recipient <inbound@example.com>, someone-else@example.org"},
{"name": "Content-Type", "value": 'multipart/alternative; boundary="94eb2c05e174adb140055b6339c5"'},
],
"commonHeaders": {
"returnPath": "bounce-handler@mail.example.org",
"from": ['"Sender, Inc." <from@example.org>'],
"date": "Fri, 30 Mar 2018 10:21:50 -0700",
"to": ["Recipient <inbound@example.com>", "someone-else@example.org"],
"messageId": "<CAEPk3RKsi@mail.example.org>",
"subject": "Test inbound message",
},
},
"receipt": {
"timestamp": "2018-03-30T17:21:51.636Z",
"processingTimeMillis": 357,
"recipients": ["inbound@example.com"],
"spamVerdict": {"status": "PASS"},
"virusVerdict": {"status": "PASS"},
"spfVerdict": {"status": "PASS"},
"dkimVerdict": {"status": "PASS"},
"dmarcVerdict": {"status": "PASS"},
"action": {
"type": "SNS",
"topicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"encoding": "UTF8",
},
},
"content": self.TEST_MIME_MESSAGE,
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e",
"TopicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"Subject": "Amazon SES Email Receipt Notification",
"Message": json.dumps(raw_ses_event),
"Timestamp": "2018-03-30T17:17:36.516Z",
"SignatureVersion": "1",
"Signature": "EXAMPLE_SIGNATURE==",
"SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-12345abcde.pem",
"UnsubscribeURL": "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn...",
}
response = self.post_from_sns('/anymail/amazon_ses/inbound/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=AmazonSESInboundWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertIsInstance(event, AnymailInboundEvent)
self.assertEqual(event.event_type, 'inbound')
self.assertEqual(event.timestamp, datetime(2018, 3, 30, 17, 21, 51, microsecond=636000, tzinfo=utc))
self.assertEqual(event.event_id, "jili9m351il3gkburn7o2f0u6788stij94c8ld01")
self.assertIsInstance(event.message, AnymailInboundMessage)
self.assertEqual(event.esp_event, raw_ses_event)
message = event.message
self.assertIsInstance(message, AnymailInboundMessage)
self.assertEqual(message.envelope_sender, 'envelope-from@example.org')
self.assertEqual(message.envelope_recipient, 'inbound@example.com')
self.assertEqual(str(message.from_email), '"Sender, Inc." <from@example.org>')
self.assertEqual([str(to) for to in message.to],
['Recipient <inbound@example.com>', 'someone-else@example.org'])
self.assertEqual(message.subject, 'Test inbound message')
self.assertEqual(message.text, "It's a body\N{HORIZONTAL ELLIPSIS}\r\n")
self.assertEqual(message.html, """<div dir="ltr">It's a body\N{HORIZONTAL ELLIPSIS}</div>\r\n""")
self.assertIs(message.spam_detected, False)
def test_inbound_sns_base64(self):
"""Should handle 'Base 64' content option on received email SNS action"""
raw_ses_event = {
# (omitting some fields that aren't used by Anymail)
"notificationType": "Received",
"mail": {
"source": "envelope-from@example.org",
"timestamp": "2018-03-30T17:21:51.636Z",
"messageId": "jili9m351il3gkburn7o2f0u6788stij94c8ld01", # assigned by Amazon SES
"destination": ["inbound@example.com", "someone-else@example.org"],
},
"receipt": {
"recipients": ["inbound@example.com"],
"action": {
"type": "SNS",
"topicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"encoding": "BASE64",
},
"spamVerdict": {"status": "FAIL"},
},
"content": b64encode(self.TEST_MIME_MESSAGE.encode('ascii')).decode('ascii'),
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e",
"TopicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"Message": json.dumps(raw_ses_event),
}
response = self.post_from_sns('/anymail/amazon_ses/inbound/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=AmazonSESInboundWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertIsInstance(event, AnymailInboundEvent)
self.assertEqual(event.event_type, 'inbound')
self.assertEqual(event.timestamp, datetime(2018, 3, 30, 17, 21, 51, microsecond=636000, tzinfo=utc))
self.assertEqual(event.event_id, "jili9m351il3gkburn7o2f0u6788stij94c8ld01")
self.assertIsInstance(event.message, AnymailInboundMessage)
self.assertEqual(event.esp_event, raw_ses_event)
message = event.message
self.assertIsInstance(message, AnymailInboundMessage)
self.assertEqual(message.envelope_sender, 'envelope-from@example.org')
self.assertEqual(message.envelope_recipient, 'inbound@example.com')
self.assertEqual(str(message.from_email), '"Sender, Inc." <from@example.org>')
self.assertEqual([str(to) for to in message.to],
['Recipient <inbound@example.com>', 'someone-else@example.org'])
self.assertEqual(message.subject, 'Test inbound message')
self.assertEqual(message.text, "It's a body\N{HORIZONTAL ELLIPSIS}\r\n")
self.assertEqual(message.html, """<div dir="ltr">It's a body\N{HORIZONTAL ELLIPSIS}</div>\r\n""")
self.assertIs(message.spam_detected, True)
def test_inbound_s3(self):
"""Should handle 'S3' receipt action"""
self.mock_s3_downloadables["InboundEmailBucket-KeepPrivate"] = {
"inbound/fqef5sop459utgdf4o9lqbsv7jeo73pejig34301": self.TEST_MIME_MESSAGE.encode('ascii')
}
raw_ses_event = {
# (omitting some fields that aren't used by Anymail)
"notificationType": "Received",
"mail": {
"source": "envelope-from@example.org",
"timestamp": "2018-03-30T17:21:51.636Z",
"messageId": "fqef5sop459utgdf4o9lqbsv7jeo73pejig34301", # assigned by Amazon SES
"destination": ["inbound@example.com", "someone-else@example.org"],
},
"receipt": {
"recipients": ["inbound@example.com"],
"action": {
"type": "S3",
"topicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"bucketName": "InboundEmailBucket-KeepPrivate",
"objectKeyPrefix": "inbound",
"objectKey": "inbound/fqef5sop459utgdf4o9lqbsv7jeo73pejig34301"
},
"spamVerdict": {"status": "GRAY"},
},
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e",
"TopicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"Message": json.dumps(raw_ses_event),
}
response = self.post_from_sns('/anymail/amazon_ses/inbound/', raw_sns_message)
self.assertEqual(response.status_code, 200)
self.mock_client.assert_called_once_with('s3', config=ANY)
self.mock_s3.download_fileobj.assert_called_once_with(
"InboundEmailBucket-KeepPrivate", "inbound/fqef5sop459utgdf4o9lqbsv7jeo73pejig34301", ANY)
kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=AmazonSESInboundWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertIsInstance(event, AnymailInboundEvent)
self.assertEqual(event.event_type, 'inbound')
self.assertEqual(event.timestamp, datetime(2018, 3, 30, 17, 21, 51, microsecond=636000, tzinfo=utc))
self.assertEqual(event.event_id, "fqef5sop459utgdf4o9lqbsv7jeo73pejig34301")
self.assertIsInstance(event.message, AnymailInboundMessage)
self.assertEqual(event.esp_event, raw_ses_event)
message = event.message
self.assertIsInstance(message, AnymailInboundMessage)
self.assertEqual(message.envelope_sender, 'envelope-from@example.org')
self.assertEqual(message.envelope_recipient, 'inbound@example.com')
self.assertEqual(str(message.from_email), '"Sender, Inc." <from@example.org>')
self.assertEqual([str(to) for to in message.to],
['Recipient <inbound@example.com>', 'someone-else@example.org'])
self.assertEqual(message.subject, 'Test inbound message')
# rstrip below because the Python 3 EmailBytesParser converts \r\n to \n, but the Python 2 version doesn't
self.assertEqual(message.text.rstrip(), "It's a body\N{HORIZONTAL ELLIPSIS}")
self.assertEqual(message.html.rstrip(), """<div dir="ltr">It's a body\N{HORIZONTAL ELLIPSIS}</div>""")
self.assertIsNone(message.spam_detected)
def test_inbound_s3_failure_message(self):
"""Issue a helpful error when S3 download fails"""
# Boto's error: "An error occurred (403) when calling the HeadObject operation: Forbidden")
self.mock_s3.download_fileobj.side_effect = botocore.exceptions.ClientError(
{'Error': {'Code': 403, 'Message': 'Forbidden'}}, operation_name='HeadObject')
raw_ses_event = {
"notificationType": "Received",
"receipt": {
"action": {"type": "S3", "bucketName": "YourBucket", "objectKey": "inbound/the_object_key"}
},
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e",
"TopicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"Message": json.dumps(raw_ses_event),
}
with self.assertRaisesMessage(
AnymailAPIError,
"Anymail AmazonSESInboundWebhookView couldn't download S3 object 'YourBucket:inbound/the_object_key'"
) as cm:
self.post_from_sns('/anymail/amazon_ses/inbound/', raw_sns_message)
self.assertIsInstance(cm.exception, botocore.exceptions.ClientError) # both Boto and Anymail exception class
self.assertIn("ClientError: An error occurred (403) when calling the HeadObject operation: Forbidden",
str(cm.exception)) # original Boto message included
def test_incorrect_tracking_event(self):
"""The inbound webhook should warn if it receives tracking events"""
raw_sns_message = {
"Type": "Notification",
"MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e",
"TopicArn": "arn:...:111111111111:SES_Tracking",
"Message": '{"notificationType": "Delivery"}',
}
with self.assertRaisesMessage(
AnymailConfigurationError,
"You seem to have set an Amazon SES *sending* event or notification to publish to an SNS Topic "
"that posts to Anymail's *inbound* webhook URL. (SNS TopicArn arn:...:111111111111:SES_Tracking)"
):
self.post_from_sns('/anymail/amazon_ses/inbound/', raw_sns_message)

162
tests/test_amazon_ses_integration.py Normal file
View File

@@ -0,0 +1,162 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import os
import unittest
import warnings
from django.test import SimpleTestCase
from django.test.utils import override_settings
from anymail.exceptions import AnymailAPIError
from anymail.message import AnymailMessage
from .utils import AnymailTestMixin, sample_image_path, RUN_LIVE_TESTS
try:
ResourceWarning
except NameError:
ResourceWarning = Warning # Python 2
AMAZON_SES_TEST_ACCESS_KEY_ID = os.getenv("AMAZON_SES_TEST_ACCESS_KEY_ID")
AMAZON_SES_TEST_SECRET_ACCESS_KEY = os.getenv("AMAZON_SES_TEST_SECRET_ACCESS_KEY")
AMAZON_SES_TEST_REGION_NAME = os.getenv("AMAZON_SES_TEST_REGION_NAME", "us-east-1")
@unittest.skipUnless(RUN_LIVE_TESTS, "RUN_LIVE_TESTS disabled in this environment")
@unittest.skipUnless(AMAZON_SES_TEST_ACCESS_KEY_ID and AMAZON_SES_TEST_SECRET_ACCESS_KEY,
"Set AMAZON_SES_TEST_ACCESS_KEY_ID and AMAZON_SES_TEST_SECRET_ACCESS_KEY "
"environment variables to run Amazon SES integration tests")
@override_settings(
EMAIL_BACKEND="anymail.backends.amazon_ses.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": AMAZON_SES_TEST_ACCESS_KEY_ID,
"aws_secret_access_key": AMAZON_SES_TEST_SECRET_ACCESS_KEY,
"region_name": AMAZON_SES_TEST_REGION_NAME,
# Can supply any other boto3.client params, including botocore.config.Config as dict
"config": {"retries": {"max_attempts": 2}},
},
"AMAZON_SES_CONFIGURATION_SET_NAME": "TestConfigurationSet", # actual config set in Anymail test account
})
class AmazonSESBackendIntegrationTests(SimpleTestCase, AnymailTestMixin):
"""Amazon SES API integration tests
These tests run against the **live** Amazon SES API, using the environment
variables `AMAZON_SES_TEST_ACCESS_KEY_ID` and `AMAZON_SES_TEST_SECRET_ACCESS_KEY`
as AWS credentials. If those variables are not set, these tests won't run.
(You can also set the environment variable `AMAZON_SES_TEST_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
Amazon SES also doesn't support arbitrary senders (so no from@example.com).
We've set up @test-ses.anymail.info as a validated sending domain for these tests.
You may need to change the from_email to your own address when testing.
"""
def setUp(self):
super(AmazonSESBackendIntegrationTests, self).setUp()
self.message = AnymailMessage('Anymail Amazon SES integration test', 'Text content',
'test@test-ses.anymail.info', ['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 might not be a real problem worth warning, but in any case it's not our problem.)
# https://www.google.com/search?q=unittest+boto3+ResourceWarning+unclosed+ssl.SSLSocket
# 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
self.assertEqual(sent_status, 'queued') # Amazon SES always queues (or raises an error)
self.assertRegex(message_id, r'[0-9a-f-]+') # Amazon SES message ids are groups of hex chars
self.assertEqual(anymail_status.status, {sent_status}) # set of all recipient statuses
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='"Test From" <test@test-ses.anymail.info>',
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\nYour order {{order}} shipped {{ship_date}}."
# })
message = AnymailMessage(
template_id='TestTemplate',
from_email='"Test From" <test@test-ses.anymail.info>',
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", # default
'ship_date': "today"
},
)
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": AMAZON_SES_TEST_REGION_NAME,
}
})
def test_invalid_aws_credentials(self):
with self.assertRaises(AnymailAPIError) as cm:
self.message.send()
err = cm.exception
# Make sure the exception message includes AWS's response:
self.assertIn("The security token included in the request is invalid", str(err))

538
tests/test_amazon_ses_webhooks.py Normal file
View File

@@ -0,0 +1,538 @@
import json
import warnings
from datetime import datetime
import botocore.exceptions
from django.test import override_settings
from django.utils.timezone import utc
from mock import ANY, patch
from anymail.exceptions import AnymailConfigurationError, AnymailInsecureWebhookWarning
from anymail.signals import AnymailTrackingEvent
from anymail.webhooks.amazon_ses import AmazonSESTrackingWebhookView
from .webhook_cases import WebhookBasicAuthTestsMixin, WebhookTestCase
class AmazonSESWebhookTestsMixin(object):
def post_from_sns(self, path, raw_sns_message, **kwargs):
# noinspection PyUnresolvedReferences
return self.client.post(
path,
content_type='text/plain; charset=UTF-8', # SNS posts JSON as text/plain
data=json.dumps(raw_sns_message),
HTTP_X_AMZ_SNS_MESSAGE_ID=raw_sns_message["MessageId"],
HTTP_X_AMZ_SNS_MESSAGE_TYPE=raw_sns_message["Type"],
# Anymail doesn't use other x-amz-sns-* headers
**kwargs)
class AmazonSESWebhookSecurityTests(WebhookTestCase, AmazonSESWebhookTestsMixin, WebhookBasicAuthTestsMixin):
def call_webhook(self):
return self.post_from_sns('/anymail/amazon_ses/tracking/',
{"Type": "Notification", "MessageId": "123", "Message": "{}"})
# Most actual tests are in WebhookBasicAuthTestsMixin
def test_verifies_missing_auth(self):
# Must handle missing auth header slightly differently from Anymail default 400 SuspiciousOperation:
# SNS will only send basic auth after missing auth responds 401 WWW-Authenticate: Basic realm="..."
self.clear_basic_auth()
response = self.call_webhook()
self.assertEqual(response.status_code, 401)
self.assertEqual(response["WWW-Authenticate"], 'Basic realm="Anymail WEBHOOK_SECRET"')
class AmazonSESNotificationsTests(WebhookTestCase, AmazonSESWebhookTestsMixin):
def test_bounce_event(self):
# This test includes a complete Amazon SES example event. (Later tests omit some payload for brevity.)
# https://docs.aws.amazon.com/ses/latest/DeveloperGuide/notification-examples.html#notification-examples-bounce
raw_ses_event = {
"notificationType": "Bounce",
"bounce": {
"bounceType": "Permanent",
"reportingMTA": "dns; email.example.com",
"bouncedRecipients": [{
"emailAddress": "jane@example.com",
"status": "5.1.1",
"action": "failed",
"diagnosticCode": "smtp; 550 5.1.1 <jane@example.com>... User unknown",
}],
"bounceSubType": "General",
"timestamp": "2016-01-27T14:59:44.101Z", # when bounce sent (by receiving ISP)
"feedbackId": "00000138111222aa-44455566-cccc-cccc-cccc-ddddaaaa068a-000000", # unique id for bounce
"remoteMtaIp": "127.0.2.0",
},
"mail": {
"timestamp": "2016-01-27T14:59:38.237Z", # when message sent
"source": "john@example.com",
"sourceArn": "arn:aws:ses:us-west-2:888888888888:identity/example.com",
"sourceIp": "127.0.3.0",
"sendingAccountId": "123456789012",
"messageId": "00000138111222aa-33322211-cccc-cccc-cccc-ddddaaaa0680-000000",
"destination": ["jane@example.com", "mary@example.com", "richard@example.com"],
"headersTruncated": False,
"headers": [
{"name": "From", "value": '"John Doe" <john@example.com>'},
{"name": "To", "value": '"Jane Doe" <jane@example.com>, "Mary Doe" <mary@example.com>,'
' "Richard Doe" <richard@example.com>'},
{"name": "Message-ID", "value": "custom-message-ID"},
{"name": "Subject", "value": "Hello"},
{"name": "Content-Type", "value": 'text/plain; charset="UTF-8"'},
{"name": "Content-Transfer-Encoding", "value": "base64"},
{"name": "Date", "value": "Wed, 27 Jan 2016 14:05:45 +0000"},
{"name": "X-Tag", "value": "tag 1"},
{"name": "X-Tag", "value": "tag 2"},
{"name": "X-Metadata", "value": '{"meta1":"string","meta2":2}'},
],
"commonHeaders": {
"from": ["John Doe <john@example.com>"],
"date": "Wed, 27 Jan 2016 14:05:45 +0000",
"to": ["Jane Doe <jane@example.com>, Mary Doe <mary@example.com>,"
" Richard Doe <richard@example.com>"],
"messageId": "custom-message-ID",
"subject": "Hello",
},
},
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", # unique id for SNS event
"TopicArn": "arn:aws:sns:us-east-1:1234567890:SES_Events",
"Subject": "Amazon SES Email Event Notification",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
"SignatureVersion": "1",
"Signature": "EXAMPLE-SIGNATURE==",
"SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-12345abcde.pem",
"UnsubscribeURL": "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn...",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.esp_event, raw_ses_event)
self.assertEqual(event.timestamp, datetime(2018, 3, 26, 17, 58, 59, microsecond=675000, tzinfo=utc)) # SNS
self.assertEqual(event.message_id, "00000138111222aa-33322211-cccc-cccc-cccc-ddddaaaa0680-000000")
self.assertEqual(event.event_id, "19ba9823-d7f2-53c1-860e-cb10e0d13dfc")
self.assertEqual(event.recipient, "jane@example.com")
self.assertEqual(event.reject_reason, "bounced")
self.assertEqual(event.description, "Permanent: General")
self.assertEqual(event.mta_response, "smtp; 550 5.1.1 <jane@example.com>... User unknown")
self.assertEqual(event.tags, ["tag 1", "tag 2"])
self.assertEqual(event.metadata, {"meta1": "string", "meta2": 2})
# For brevity, remaining tests omit some event fields that aren't used by Anymail
def test_multiple_bounce_event(self):
"""Amazon SES notification can cover multiple recipients"""
raw_ses_event = {
"notificationType": "Bounce",
"bounce": {
"bounceType": "Permanent",
"bounceSubType": "General",
"bouncedRecipients": [
{"emailAddress": "jane@example.com"},
{"emailAddress": "richard@example.com"}
],
},
"mail": {
"messageId": "00000137860315fd-34208509-5b74-41f3-95c5-22c1edc3c924-000000",
"destination": ["jane@example.com", "mary@example.com", "richard@example.com"],
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
# tracking handler should be called twice -- once for each bounced recipient
# (but not for the third, non-bounced recipient)
self.assertEqual(self.tracking_handler.call_count, 2)
_, kwargs = self.tracking_handler.call_args_list[0]
event = kwargs['event']
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.recipient, "jane@example.com")
self.assertEqual(event.description, "Permanent: General")
self.assertIsNone(event.mta_response)
_, kwargs = self.tracking_handler.call_args_list[1]
event = kwargs['event']
self.assertEqual(event.esp_event, raw_ses_event)
self.assertEqual(event.recipient, "richard@example.com")
def test_complaint_event(self):
raw_ses_event = {
"notificationType": "Complaint",
"complaint": {
"userAgent": "AnyCompany Feedback Loop (V0.01)",
"complainedRecipients": [{"emailAddress": "richard@example.com"}],
"complaintFeedbackType": "abuse",
},
"mail": {
"messageId": "000001378603177f-7a5433e7-8edb-42ae-af10-f0181f34d6ee-000000",
"destination": ["jane@example.com", "mary@example.com", "richard@example.com"],
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertEqual(event.event_type, "complained")
self.assertEqual(event.recipient, "richard@example.com")
self.assertEqual(event.reject_reason, "spam")
self.assertEqual(event.description, "abuse")
self.assertEqual(event.user_agent, "AnyCompany Feedback Loop (V0.01)")
def test_delivery_event(self):
raw_ses_event = {
"notificationType": "Delivery",
"mail": {
"timestamp": "2016-01-27T14:59:38.237Z",
"messageId": "0000014644fe5ef6-9a483358-9170-4cb4-a269-f5dcdf415321-000000",
"destination": ["jane@example.com", "mary@example.com", "richard@example.com"],
},
"delivery": {
"timestamp": "2016-01-27T14:59:38.237Z",
"recipients": ["jane@example.com"],
"processingTimeMillis": 546,
"reportingMTA": "a8-70.smtp-out.amazonses.com",
"smtpResponse": "250 ok: Message 64111812 accepted",
"remoteMtaIp": "127.0.2.0"
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertEqual(event.event_type, "delivered")
self.assertEqual(event.recipient, "jane@example.com")
self.assertEqual(event.mta_response, "250 ok: Message 64111812 accepted")
def test_send_event(self):
raw_ses_event = {
"eventType": "Send",
"mail": {
"timestamp": "2016-10-14T05:02:16.645Z",
"messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
"destination": ["recipient@example.com"],
"tags": {
"ses:configuration-set": ["ConfigSet"],
"ses:source-ip": ["192.0.2.0"],
"ses:from-domain": ["example.com"],
"ses:caller-identity": ["ses_user"],
"myCustomTag1": ["myCustomTagValue1"],
"myCustomTag2": ["myCustomTagValue2"]
}
},
"send": {}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "sent")
self.assertEqual(event.esp_event, raw_ses_event)
self.assertEqual(event.timestamp, datetime(2018, 3, 26, 17, 58, 59, microsecond=675000, tzinfo=utc)) # SNS
self.assertEqual(event.message_id, "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000")
self.assertEqual(event.event_id, "19ba9823-d7f2-53c1-860e-cb10e0d13dfc")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.tags, []) # Anymail doesn't load Amazon SES "Message Tags"
self.assertEqual(event.metadata, {})
def test_reject_event(self):
raw_ses_event = {
"eventType": "Reject",
"mail": {
"timestamp": "2016-10-14T17:38:15.211Z",
"messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
"destination": ["recipient@example.com"],
},
"reject": {
"reason": "Bad content"
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertEqual(event.event_type, "rejected")
self.assertEqual(event.reject_reason, "blocked")
self.assertEqual(event.description, "Bad content")
self.assertEqual(event.recipient, "recipient@example.com")
def test_open_event(self):
raw_ses_event = {
"eventType": "Open",
"mail": {
"destination": ["recipient@example.com"],
"messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
},
"open": {
"ipAddress": "192.0.2.1",
"timestamp": "2017-08-09T22:00:19.652Z",
"userAgent": "Mozilla/5.0 (iPhone; CPU iPhone OS 10_3_3 like Mac OS X)..."
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertEqual(event.event_type, "opened")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.user_agent, "Mozilla/5.0 (iPhone; CPU iPhone OS 10_3_3 like Mac OS X)...")
def test_click_event(self):
raw_ses_event = {
"eventType": "Click",
"click": {
"ipAddress": "192.0.2.1",
"link": "https://docs.aws.amazon.com/ses/latest/DeveloperGuide/",
"linkTags": {
"samplekey0": ["samplevalue0"],
"samplekey1": ["samplevalue1"],
},
"timestamp": "2017-08-09T23:51:25.570Z",
"userAgent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36..."
},
"mail": {
"destination": ["recipient@example.com"],
"messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertEqual(event.event_type, "clicked")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.user_agent, "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...")
self.assertEqual(event.click_url, "https://docs.aws.amazon.com/ses/latest/DeveloperGuide/")
def test_rendering_failure_event(self):
raw_ses_event = {
"eventType": "Rendering Failure",
"mail": {
"messageId": "c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
"destination": ["recipient@example.com"],
},
"failure": {
"errorMessage": "Attribute 'attributeName' is not present in the rendering data.",
"templateName": "MyTemplate"
}
}
raw_sns_message = {
"Type": "Notification",
"MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc",
"Message": json.dumps(raw_ses_event) + "\n",
"Timestamp": "2018-03-26T17:58:59.675Z",
}
response = self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=AmazonSESTrackingWebhookView,
event=ANY, esp_name='Amazon SES')
event = kwargs['event']
self.assertEqual(event.event_type, "failed")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.description, "Attribute 'attributeName' is not present in the rendering data.")
def test_incorrect_received_event(self):
"""The tracking webhook should warn if it receives inbound events"""
raw_sns_message = {
"Type": "Notification",
"MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e",
"TopicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound",
"Message": '{"notificationType": "Received"}',
}
with self.assertRaisesMessage(
AnymailConfigurationError,
"You seem to have set an Amazon SES *inbound* receipt rule to publish to an SNS Topic that posts "
"to Anymail's *tracking* webhook URL. (SNS TopicArn arn:aws:sns:us-east-1:111111111111:SES_Inbound)"
):
self.post_from_sns('/anymail/amazon_ses/tracking/', raw_sns_message)
class AmazonSESSubscriptionManagementTests(WebhookTestCase, AmazonSESWebhookTestsMixin):
# Anymail will automatically respond to SNS subscription notifications
# if Anymail is configured to require basic auth via WEBHOOK_SECRET.
# (Note that WebhookTestCase sets up ANYMAIL WEBHOOK_SECRET.)
def setUp(self):
super(AmazonSESSubscriptionManagementTests, self).setUp()
# Mock boto3.session.Session().client('sns').confirm_subscription (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.webhooks.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)
self.mock_client = self.mock_session.return_value.client # boto3.session.Session().client
self.mock_client_instance = self.mock_client.return_value # boto3.session.Session().client('sns', ...)
self.mock_client_instance.confirm_subscription.return_value = {
'SubscriptionArn': 'arn:aws:sns:us-west-2:123456789012:SES_Notifications:aaaaaaa-...'
}
SNS_SUBSCRIPTION_CONFIRMATION = {
"Type": "SubscriptionConfirmation",
"MessageId": "165545c9-2a5c-472c-8df2-7ff2be2b3b1b",
"Token": "EXAMPLE_TOKEN",
"TopicArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications",
"Message": "You have chosen to subscribe ...\nTo confirm..., visit the SubscribeURL included in this message.",
"SubscribeURL": "https://sns.us-west-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=...",
"Timestamp": "2012-04-26T20:45:04.751Z",
"SignatureVersion": "1",
"Signature": "EXAMPLE-SIGNATURE==",
"SigningCertURL": "https://sns.us-west-2.amazonaws.com/SimpleNotificationService-12345abcde.pem"
}
def test_sns_subscription_auto_confirmation(self):
"""Anymail webhook will auto-confirm SNS topic subscriptions"""
response = self.post_from_sns('/anymail/amazon_ses/tracking/', self.SNS_SUBSCRIPTION_CONFIRMATION)
self.assertEqual(response.status_code, 200)
# auto-confirmed:
self.mock_client.assert_called_once_with('sns', config=ANY)
self.mock_client_instance.confirm_subscription.assert_called_once_with(
TopicArn="arn:aws:sns:us-west-2:123456789012:SES_Notifications",
Token="EXAMPLE_TOKEN", AuthenticateOnUnsubscribe="true")
# didn't notify receivers:
self.assertEqual(self.tracking_handler.call_count, 0)
self.assertEqual(self.inbound_handler.call_count, 0)
def test_sns_subscription_confirmation_failure(self):
"""Auto-confirmation allows error through if confirm call fails"""
self.mock_client_instance.confirm_subscription.side_effect = botocore.exceptions.ClientError({
'Error': {
'Type': 'Sender',
'Code': 'InternalError',
'Message': 'Gremlins!',
},
'ResponseMetadata': {
'RequestId': 'aaaaaaaa-2222-1111-8888-bbbb3333bbbb',
'HTTPStatusCode': 500,
}
}, operation_name="confirm_subscription")
with self.assertRaisesMessage(botocore.exceptions.ClientError, "Gremlins!"):
self.post_from_sns('/anymail/amazon_ses/tracking/', self.SNS_SUBSCRIPTION_CONFIRMATION)
# didn't notify receivers:
self.assertEqual(self.tracking_handler.call_count, 0)
self.assertEqual(self.inbound_handler.call_count, 0)
@override_settings(ANYMAIL={}) # clear WEBHOOK_SECRET setting from base WebhookTestCase
def test_sns_subscription_confirmation_auth_disabled(self):
"""Anymail *won't* auto-confirm SNS subscriptions if WEBHOOK_SECRET isn't in use"""
warnings.simplefilter("ignore", AnymailInsecureWebhookWarning) # (this gets tested elsewhere)
with self.assertLogs('django.security.AnymailWebhookValidationFailure') as cm:
response = self.post_from_sns('/anymail/amazon_ses/tracking/', self.SNS_SUBSCRIPTION_CONFIRMATION)
self.assertEqual(response.status_code, 400) # bad request
self.assertEqual(
["Anymail received an unexpected SubscriptionConfirmation request for Amazon SNS topic "
"'arn:aws:sns:us-west-2:123456789012:SES_Notifications'. (Anymail can automatically confirm "
"SNS subscriptions if you set a WEBHOOK_SECRET and use that in your SNS notification url. Or "
"you can manually confirm this subscription in the SNS dashboard with token 'EXAMPLE_TOKEN'.)"],
[record.getMessage() for record in cm.records])
# *didn't* try to confirm the subscription:
self.assertEqual(self.mock_client_instance.confirm_subscription.call_count, 0)
# didn't notify receivers:
self.assertEqual(self.tracking_handler.call_count, 0)
self.assertEqual(self.inbound_handler.call_count, 0)
def test_sns_confirmation_success_notification(self):
"""Anymail ignores the 'Successfully validated' notification after confirming an SNS subscription"""
response = self.post_from_sns('/anymail/amazon_ses/tracking/', {
"Type": "Notification",
"MessageId": "7fbca0d9-eeab-5285-ae27-f3f57f2e84b0",
"TopicArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications",
"Message": "Successfully validated SNS topic for Amazon SES event publishing.",
"Timestamp": "2018-03-21T16:58:45.077Z",
"SignatureVersion": "1",
"Signature": "EXAMPLE_SIGNATURE==",
"SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-12345abcde.pem",
"UnsubscribeURL": "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe...",
})
self.assertEqual(response.status_code, 200)
# didn't notify receivers:
self.assertEqual(self.tracking_handler.call_count, 0)
self.assertEqual(self.inbound_handler.call_count, 0)
def test_sns_unsubscribe_confirmation(self):
"""Anymail ignores the UnsubscribeConfirmation SNS message after deleting a subscription"""
response = self.post_from_sns('/anymail/amazon_ses/tracking/', {
"Type": "UnsubscribeConfirmation",
"MessageId": "47138184-6831-46b8-8f7c-afc488602d7d",
"Token": "EXAMPLE_TOKEN",
"TopicArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications",
"Message": "You have chosen to deactivate subscription ...\nTo cancel ... visit the SubscribeURL...",
"SubscribeURL": "https://sns.us-west-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=...",
"Timestamp": "2012-04-26T20:06:41.581Z",
"SignatureVersion": "1",
"Signature": "EXAMPLE_SIGNATURE==",
"SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-12345abcde.pem",
})
self.assertEqual(response.status_code, 200)
# *didn't* try to use the Token to re-enable the subscription:
self.assertEqual(self.mock_client_instance.confirm_subscription.call_count, 0)
# didn't notify receivers:
self.assertEqual(self.tracking_handler.call_count, 0)
self.assertEqual(self.inbound_handler.call_count, 0)
@override_settings(ANYMAIL_AMAZON_SES_AUTO_CONFIRM_SNS_SUBSCRIPTIONS=False)
def test_disable_auto_confirmation(self):
"""The ANYMAIL setting AMAZON_SES_AUTO_CONFIRM_SNS_SUBSCRIPTIONS will disable this feature"""
response = self.post_from_sns('/anymail/amazon_ses/tracking/', self.SNS_SUBSCRIPTION_CONFIRMATION)
self.assertEqual(response.status_code, 200)
# *didn't* try to subscribe:
self.assertEqual(self.mock_session.call_count, 0)
self.assertEqual(self.mock_client.call_count, 0)
# didn't notify receivers:
self.assertEqual(self.tracking_handler.call_count, 0)
self.assertEqual(self.inbound_handler.call_count, 0)