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

Clean up old Djrill/Mandrill tests

Browse Source 
* Match other ESP test strategies for
  test_mandrill_backend and
  test_mandrill_integration

* Extract test_mandrill_session_sharing into
  SessionSharingTestCasesMixin for all
  requests-based ESP backends

* Move leftover Djrill feature tests into
  test_mandrill_djrill features (until they are
  handled as part of esp_extra or in normalized
  ESP template/merge features)

Closes #7
This commit is contained in:
medmunds
2016-03-15 18:06:17 -07:00
parent 12229ab116
commit abca7d9538
13 changed files with 1020 additions and 1046 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
anymail/tests/__init__.py
View File

@@ -2,10 +2,17 @@
# is required by the old (<=1.5) DjangoTestSuiteRunner.
from .test_mailgun_backend import *
from .test_mailgun_integration import *
from .test_mandrill_backend import *
from .test_mandrill_integration import *
from .test_mandrill_send import *
from .test_mandrill_send_template import *
from .test_mandrill_session_sharing import *
from .test_mandrill_subaccounts import *
from .test_postmark_backend import *
from .test_postmark_integration import *
from .test_sendgrid_backend import *
from .test_sendgrid_integration import *
# Djrill leftovers:
from .test_mandrill_djrill_features import *
from .test_mandrill_webhook import *

72
anymail/tests/mock_backend.py
View File

@@ -1,72 +0,0 @@
import json
import requests
import six
from mock import patch
from django.test import TestCase
from django.test.utils import override_settings
MANDRILL_SUCCESS_RESPONSE = b"""[{
"email": "to@example.com",
"status": "sent",
"_id": "abc123",
"reject_reason": null
}]"""
@override_settings(MANDRILL_API_KEY="FAKE_API_KEY_FOR_TESTING",
EMAIL_BACKEND="anymail.backends.mandrill.MandrillBackend")
class DjrillBackendMockAPITestCase(TestCase):
"""TestCase that uses Djrill EmailBackend with a mocked Mandrill API"""
class MockResponse(requests.Response):
"""requests.post return value mock sufficient for MandrillBackend"""
def __init__(self, status_code=200, raw=MANDRILL_SUCCESS_RESPONSE, encoding='utf-8'):
super(DjrillBackendMockAPITestCase.MockResponse, self).__init__()
self.status_code = status_code
self.encoding = encoding
self.raw = six.BytesIO(raw)
def setUp(self):
self.patch = patch('requests.Session.request', autospec=True)
self.mock_post = self.patch.start()
self.mock_post.return_value = self.MockResponse()
def tearDown(self):
self.patch.stop()
def assert_mandrill_called(self, endpoint):
"""Verifies the (mock) Mandrill API was called on endpoint.
endpoint is a Mandrill API, e.g., "/messages/send.json"
"""
# This assumes the last (or only) call to requests.post is the
# Mandrill API call of interest.
if self.mock_post.call_args is None:
raise AssertionError("Mandrill API was not called")
(args, kwargs) = self.mock_post.call_args
try:
post_url = kwargs.get('url', None) or args[2]
except IndexError:
raise AssertionError("requests.Session.request was called without an url (?!)")
if not post_url.endswith(endpoint):
raise AssertionError(
"requests.post was not called on %s\n(It was called on %s)"
% (endpoint, post_url))
def get_api_call_data(self):
"""Returns the data posted to the Mandrill API.
Fails test if API wasn't called.
"""
if self.mock_post.call_args is None:
raise AssertionError("Mandrill API was not called")
(args, kwargs) = self.mock_post.call_args
try:
post_data = kwargs.get('data', None) or args[4]
except IndexError:
raise AssertionError("requests.Session.request was called without data")
return json.loads(post_data)

79
anymail/tests/mock_requests_backend.py
View File

@@ -1,10 +1,13 @@
import json
from django.core import mail
from django.test import SimpleTestCase
import requests
import six
from mock import patch
from anymail.exceptions import AnymailAPIError
from .utils import AnymailTestMixin
UNSET = object()
@@ -25,9 +28,9 @@ class RequestsBackendMockAPITestCase(SimpleTestCase, AnymailTestMixin):
def setUp(self):
super(RequestsBackendMockAPITestCase, self).setUp()
self.patch = patch('requests.Session.request', autospec=True)
self.mock_request = self.patch.start()
self.addCleanup(self.patch.stop)
self.patch_request = patch('requests.Session.request', autospec=True)
self.mock_request = self.patch_request.start()
self.addCleanup(self.patch_request.stop)
self.set_mock_response()
def set_mock_response(self, status_code=200, raw=UNSET, encoding='utf-8'):
@@ -101,3 +104,73 @@ class RequestsBackendMockAPITestCase(SimpleTestCase, AnymailTestMixin):
def assert_esp_not_called(self, msg=None):
if self.mock_request.called:
raise AssertionError(msg or "ESP API was called and shouldn't have been")
# noinspection PyUnresolvedReferences
class SessionSharingTestCasesMixin(object):
"""Mixin that tests connection sharing in any RequestsBackendMockAPITestCase
(Contains actual test cases, so can't be included in RequestsBackendMockAPITestCase
itself, as that would re-run these tests several times for each backend, in
each TestCase for the backend.)
"""
def setUp(self):
super(SessionSharingTestCasesMixin, self).setUp()
self.patch_close = patch('requests.Session.close', autospec=True)
self.mock_close = self.patch_close.start()
self.addCleanup(self.patch_close.stop)
def test_connection_sharing(self):
"""RequestsBackend reuses one requests session when sending multiple messages"""
datatuple = (
('Subject 1', 'Body 1', 'from@example.com', ['to@example.com']),
('Subject 2', 'Body 2', 'from@example.com', ['to@example.com']),
)
mail.send_mass_mail(datatuple)
self.assertEqual(self.mock_request.call_count, 2)
session1 = self.mock_request.call_args_list[0][0] # arg[0] (self) is session
session2 = self.mock_request.call_args_list[1][0]
self.assertEqual(session1, session2)
self.assertEqual(self.mock_close.call_count, 1)
def test_caller_managed_connections(self):
"""Calling code can created long-lived connection that it opens and closes"""
connection = mail.get_connection()
connection.open()
mail.send_mail('Subject 1', 'body', 'from@example.com', ['to@example.com'], connection=connection)
session1 = self.mock_request.call_args[0]
self.assertEqual(self.mock_close.call_count, 0) # shouldn't be closed yet
mail.send_mail('Subject 2', 'body', 'from@example.com', ['to@example.com'], connection=connection)
self.assertEqual(self.mock_close.call_count, 0) # still shouldn't be closed
session2 = self.mock_request.call_args[0]
self.assertEqual(session1, session2) # should have reused same session
connection.close()
self.assertEqual(self.mock_close.call_count, 1)
def test_session_closed_after_exception(self):
self.set_mock_response(status_code=500)
with self.assertRaises(AnymailAPIError):
mail.send_mail('Subject', 'Message', 'from@example.com', ['to@example.com'])
self.assertEqual(self.mock_close.call_count, 1)
def test_session_closed_after_fail_silently_exception(self):
self.set_mock_response(status_code=500)
sent = mail.send_mail('Subject', 'Message', 'from@example.com', ['to@example.com'],
fail_silently=True)
self.assertEqual(sent, 0)
self.assertEqual(self.mock_close.call_count, 1)
def test_caller_managed_session_closed_after_exception(self):
connection = mail.get_connection()
connection.open()
self.set_mock_response(status_code=500)
with self.assertRaises(AnymailAPIError):
mail.send_mail('Subject', 'Message', 'from@example.com', ['to@example.com'],
connection=connection)
self.assertEqual(self.mock_close.call_count, 0) # wait for us to close it
connection.close()
self.assertEqual(self.mock_close.call_count, 1)

7
anymail/tests/test_mailgun_backend.py
View File

@@ -16,7 +16,7 @@ from django.utils.timezone import get_fixed_timezone, override as override_curre
from anymail.exceptions import AnymailAPIError, AnymailSerializationError, AnymailUnsupportedFeature
from anymail.message import attach_inline_image_file
from .mock_requests_backend import RequestsBackendMockAPITestCase
from .mock_requests_backend import RequestsBackendMockAPITestCase, SessionSharingTestCasesMixin
from .utils import sample_image_content, sample_image_path, SAMPLE_IMAGE_FILENAME, AnymailTestMixin
@@ -439,6 +439,11 @@ class MailgunBackendRecipientsRefusedTests(MailgunBackendMockAPITestCase):
self.assertEqual(sent, 0)
class MailgunBackendSessionSharingTestCase(SessionSharingTestCasesMixin, MailgunBackendMockAPITestCase):
"""Requests session sharing tests"""
pass # tests are defined in the mixin
@override_settings(ANYMAIL_SEND_DEFAULTS={
'metadata': {'global': 'globalvalue', 'other': 'othervalue'},
'tags': ['globaltag'],

538
anymail/tests/test_mandrill_backend.py Normal file
View File

@@ -0,0 +1,538 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import unittest
from datetime import date, datetime
from decimal import Decimal
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from django.core import mail
from django.core.exceptions import ImproperlyConfigured
from django.test import SimpleTestCase
from django.test.utils import override_settings
from django.utils.timezone import get_fixed_timezone, override as override_current_timezone
from anymail.exceptions import (AnymailAPIError, AnymailRecipientsRefused,
AnymailSerializationError, AnymailUnsupportedFeature)
from anymail.message import attach_inline_image
from .mock_requests_backend import RequestsBackendMockAPITestCase, SessionSharingTestCasesMixin
from .utils import sample_image_content, sample_image_path, SAMPLE_IMAGE_FILENAME, AnymailTestMixin, decode_att
@override_settings(EMAIL_BACKEND='anymail.backends.mandrill.MandrillBackend',
ANYMAIL={'MANDRILL_API_KEY': 'test_api_key'})
class MandrillBackendMockAPITestCase(RequestsBackendMockAPITestCase):
DEFAULT_RAW_RESPONSE = b"""[{
"email": "to@example.com",
"status": "sent",
"_id": "abc123",
"reject_reason": null
}]"""
def setUp(self):
super(MandrillBackendMockAPITestCase, self).setUp()
# Simple message useful for many tests
self.message = mail.EmailMultiAlternatives('Subject', 'Text Body', 'from@example.com', ['to@example.com'])
class MandrillBackendStandardEmailTests(MandrillBackendMockAPITestCase):
"""Test backend support for Django mail wrappers"""
def test_send_mail(self):
mail.send_mail('Subject here', 'Here is the message.',
'from@example.com', ['to@example.com'], fail_silently=False)
self.assert_esp_called("/messages/send.json")
data = self.get_api_call_json()
self.assertEqual(data['key'], "test_api_key")
self.assertEqual(data['message']['subject'], "Subject here")
self.assertEqual(data['message']['text'], "Here is the message.")
self.assertNotIn('from_name', data['message'])
self.assertEqual(data['message']['from_email'], "from@example.com")
self.assertEqual(data['message']['to'], [{'email': 'to@example.com', 'name': '', 'type': 'to'}])
def test_name_addr(self):
"""Make sure RFC2822 name-addr format (with display-name) is allowed
(Test both sender and recipient addresses)
"""
msg = mail.EmailMessage(
'Subject', 'Message',
'From Name <from@example.com>',
['Recipient #1 <to1@example.com>', 'to2@example.com'],
cc=['Carbon Copy <cc1@example.com>', 'cc2@example.com'],
bcc=['Blind Copy <bcc1@example.com>', 'bcc2@example.com'])
msg.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['from_name'], "From Name")
self.assertEqual(data['message']['from_email'], "from@example.com")
self.assertEqual(data['message']['to'], [
{'email': 'to1@example.com', 'name': 'Recipient #1', 'type': 'to'},
{'email': 'to2@example.com', 'name': '', 'type': 'to'},
{'email': 'cc1@example.com', 'name': 'Carbon Copy', 'type': 'cc'},
{'email': 'cc2@example.com', 'name': '', 'type': 'cc'},
{'email': 'bcc1@example.com', 'name': 'Blind Copy', 'type': 'bcc'},
{'email': 'bcc2@example.com', 'name': '', 'type': 'bcc'},
])
def test_email_message(self):
email = mail.EmailMessage(
'Subject', 'Body goes here',
'from@example.com',
['to1@example.com', 'Also To <to2@example.com>'],
bcc=['bcc1@example.com', 'Also BCC <bcc2@example.com>'],
cc=['cc1@example.com', 'Also CC <cc2@example.com>'],
headers={'Reply-To': 'another@example.com',
'X-MyHeader': 'my value',
'Message-ID': 'mycustommsgid@example.com'})
email.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['subject'], "Subject")
self.assertEqual(data['message']['text'], "Body goes here")
self.assertEqual(data['message']['from_email'], "from@example.com")
self.assertEqual(data['message']['headers'],
{'Reply-To': 'another@example.com',
'X-MyHeader': 'my value',
'Message-ID': 'mycustommsgid@example.com'})
# Verify recipients correctly identified as "to", "cc", or "bcc"
self.assertEqual(data['message']['to'], [
{'email': 'to1@example.com', 'name': '', 'type': 'to'},
{'email': 'to2@example.com', 'name': 'Also To', 'type': 'to'},
{'email': 'cc1@example.com', 'name': '', 'type': 'cc'},
{'email': 'cc2@example.com', 'name': 'Also CC', 'type': 'cc'},
{'email': 'bcc1@example.com', 'name': '', 'type': 'bcc'},
{'email': 'bcc2@example.com', 'name': 'Also BCC', 'type': 'bcc'},
])
# Don't use Mandrill's bcc_address "logging" feature for bcc's:
self.assertNotIn('bcc_address', data['message'])
def test_html_message(self):
text_content = 'This is an important message.'
html_content = '<p>This is an <strong>important</strong> message.</p>'
email = mail.EmailMultiAlternatives('Subject', text_content,
'from@example.com', ['to@example.com'])
email.attach_alternative(html_content, "text/html")
email.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['text'], text_content)
self.assertEqual(data['message']['html'], html_content)
# Don't accidentally send the html part as an attachment:
self.assertFalse('attachments' in data['message'])
def test_html_only_message(self):
html_content = '<p>This is an <strong>important</strong> message.</p>'
email = mail.EmailMessage('Subject', html_content,
'from@example.com', ['to@example.com'])
email.content_subtype = "html" # Main content is now text/html
email.send()
data = self.get_api_call_json()
self.assertNotIn('text', data['message'])
self.assertEqual(data['message']['html'], html_content)
def test_reply_to(self):
# reply_to is new in Django 1.8 -- before that, you can simply include it in headers
try:
# noinspection PyArgumentList
email = mail.EmailMessage('Subject', 'Body goes here', 'from@example.com', ['to1@example.com'],
reply_to=['reply@example.com', 'Other <reply2@example.com>'],
headers={'X-Other': 'Keep'})
except TypeError:
# Pre-Django 1.8
raise unittest.SkipTest("Django version doesn't support EmailMessage(reply_to)")
email.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['headers']['Reply-To'],
'reply@example.com, Other <reply2@example.com>')
self.assertEqual(data['message']['headers']['X-Other'], 'Keep') # don't lose other headers
def test_attachments(self):
text_content = "* Item one\n* Item two\n* Item three"
self.message.attach(filename="test.txt", content=text_content, mimetype="text/plain")
# Should guess mimetype if not provided...
png_content = b"PNG\xb4 pretend this is the contents of a png file"
self.message.attach(filename="test.png", content=png_content)
# Should work with a MIMEBase object (also tests no filename)...
pdf_content = b"PDF\xb4 pretend this is valid pdf data"
mimeattachment = MIMEBase('application', 'pdf')
mimeattachment.set_payload(pdf_content)
self.message.attach(mimeattachment)
self.message.send()
data = self.get_api_call_json()
attachments = data['message']['attachments']
self.assertEqual(len(attachments), 3)
self.assertEqual(attachments[0]["type"], "text/plain")
self.assertEqual(attachments[0]["name"], "test.txt")
self.assertEqual(decode_att(attachments[0]["content"]).decode('ascii'), text_content)
self.assertEqual(attachments[1]["type"], "image/png") # inferred from filename
self.assertEqual(attachments[1]["name"], "test.png")
self.assertEqual(decode_att(attachments[1]["content"]), png_content)
self.assertEqual(attachments[2]["type"], "application/pdf")
self.assertEqual(attachments[2]["name"], "") # none
self.assertEqual(decode_att(attachments[2]["content"]), pdf_content)
# Make sure the image attachment is not treated as embedded:
self.assertFalse('images' in data['message'])
def test_unicode_attachment_correctly_decoded(self):
# Slight modification from the Django unicode docs:
# http://django.readthedocs.org/en/latest/ref/unicode.html#email
self.message.attach("Une pièce jointe.html", '<p>\u2019</p>', mimetype='text/html')
self.message.send()
data = self.get_api_call_json()
attachments = data['message']['attachments']
self.assertEqual(len(attachments), 1)
def test_embedded_images(self):
image_data = sample_image_content() # Read from a png file
cid = attach_inline_image(self.message, image_data)
html_content = '<p>This has an <img src="cid:%s" alt="inline" /> image.</p>' % cid
self.message.attach_alternative(html_content, "text/html")
self.message.send()
data = self.get_api_call_json()
self.assertEqual(len(data['message']['images']), 1)
self.assertEqual(data['message']['images'][0]["type"], "image/png")
self.assertEqual(data['message']['images'][0]["name"], cid)
self.assertEqual(decode_att(data['message']['images'][0]["content"]), image_data)
# Make sure neither the html nor the inline image is treated as an attachment:
self.assertFalse('attachments' in data['message'])
def test_attached_images(self):
image_filename = SAMPLE_IMAGE_FILENAME
image_path = sample_image_path(image_filename)
image_data = sample_image_content(image_filename)
self.message.attach_file(image_path) # option 1: attach as a file
image = MIMEImage(image_data) # option 2: construct the MIMEImage and attach it directly
self.message.attach(image)
self.message.send()
data = self.get_api_call_json()
attachments = data['message']['attachments']
self.assertEqual(len(attachments), 2)
self.assertEqual(attachments[0]["type"], "image/png")
self.assertEqual(attachments[0]["name"], image_filename)
self.assertEqual(decode_att(attachments[0]["content"]), image_data)
self.assertEqual(attachments[1]["type"], "image/png")
self.assertEqual(attachments[1]["name"], "") # unknown -- not attached as file
self.assertEqual(decode_att(attachments[1]["content"]), image_data)
# Make sure the image attachments are not treated as embedded:
self.assertFalse('images' in data['message'])
def test_multiple_html_alternatives(self):
# Multiple alternatives not allowed
self.message.attach_alternative("<p>First html is OK</p>", "text/html")
self.message.attach_alternative("<p>But not second html</p>", "text/html")
with self.assertRaises(AnymailUnsupportedFeature):
self.message.send()
def test_html_alternative(self):
# Only html alternatives allowed
self.message.attach_alternative("{'not': 'allowed'}", "application/json")
with self.assertRaises(AnymailUnsupportedFeature):
self.message.send()
def test_alternatives_fail_silently(self):
# Make sure fail_silently is respected
self.message.attach_alternative("{'not': 'allowed'}", "application/json")
sent = self.message.send(fail_silently=True)
self.assert_esp_not_called("API should not be called when send fails silently")
self.assertEqual(sent, 0)
def test_api_failure(self):
self.set_mock_response(status_code=400)
with self.assertRaises(AnymailAPIError):
sent = mail.send_mail('Subject', 'Body', 'from@example.com', ['to@example.com'])
self.assertEqual(sent, 0)
# Make sure fail_silently is respected
self.set_mock_response(status_code=400)
sent = mail.send_mail('Subject', 'Body', 'from@example.com', ['to@example.com'],
fail_silently=True)
self.assertEqual(sent, 0)
def test_api_error_includes_details(self):
"""AnymailAPIError should include ESP's error message"""
self.set_mock_response(status_code=400, raw=b"""{
"status": "error",
"code": 12,
"name": "Error_Name",
"message": "Helpful explanation from Mandrill"
}""")
with self.assertRaisesMessage(AnymailAPIError, "Helpful explanation from Mandrill"):
self.message.send()
# Non-JSON error response:
self.set_mock_response(status_code=500, raw=b"Invalid API key")
with self.assertRaisesMessage(AnymailAPIError, "Invalid API key"):
self.message.send()
# No content in the error response:
self.set_mock_response(status_code=502, raw=None)
with self.assertRaises(AnymailAPIError):
self.message.send()
class MandrillBackendAnymailFeatureTests(MandrillBackendMockAPITestCase):
"""Test backend support for Anymail added features"""
def test_metadata(self):
self.message.metadata = {'user_id': "12345", 'items': 6}
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['metadata'], {'user_id': "12345", 'items': 6})
def test_send_at(self):
utc_plus_6 = get_fixed_timezone(6 * 60)
utc_minus_8 = get_fixed_timezone(-8 * 60)
with override_current_timezone(utc_plus_6):
# Timezone-naive datetime assumed to be Django current_timezone
self.message.send_at = datetime(2022, 10, 11, 12, 13, 14, 567)
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['send_at'], "2022-10-11 06:13:14") # 12:13 UTC+6 == 06:13 UTC
# Timezone-aware datetime converted to UTC:
self.message.send_at = datetime(2016, 3, 4, 5, 6, 7, tzinfo=utc_minus_8)
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['send_at'], "2016-03-04 13:06:07") # 05:06 UTC-8 == 13:06 UTC
# Date-only treated as midnight in current timezone
self.message.send_at = date(2022, 10, 22)
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['send_at'], "2022-10-21 18:00:00") # 00:00 UTC+6 == 18:00-1d UTC
# POSIX timestamp
self.message.send_at = 1651820889 # 2022-05-06 07:08:09 UTC
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['send_at'], "2022-05-06 07:08:09")
# String passed unchanged (this is *not* portable between ESPs)
self.message.send_at = "2013-11-12 01:02:03"
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['send_at'], "2013-11-12 01:02:03")
def test_tags(self):
self.message.tags = ["receipt", "repeat-user"]
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['tags'], ["receipt", "repeat-user"])
def test_tracking(self):
# Test one way...
self.message.track_opens = True
self.message.track_clicks = False
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['track_opens'], True)
self.assertEqual(data['message']['track_clicks'], False)
# ...and the opposite way
self.message.track_opens = False
self.message.track_clicks = True
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['track_opens'], False)
self.assertEqual(data['message']['track_clicks'], True)
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()
self.assert_esp_called("/messages/send.json")
data = self.get_api_call_json()
self.assertNotIn('metadata', data['message'])
self.assertNotIn('send_at', data)
self.assertNotIn('tags', data['message'])
self.assertNotIn('track_opens', data['message'])
self.assertNotIn('track_clicks', data['message'])
# noinspection PyUnresolvedReferences
def test_send_attaches_anymail_status(self):
""" The anymail_status should be attached to the message when it is sent """
response_content = b'[{"email": "to1@example.com", "status": "sent", "_id": "abc123"}]'
self.set_mock_response(raw=response_content)
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],)
sent = msg.send()
self.assertEqual(sent, 1)
self.assertEqual(msg.anymail_status.status, {'sent'})
self.assertEqual(msg.anymail_status.message_id, 'abc123')
self.assertEqual(msg.anymail_status.recipients['to1@example.com'].status, 'sent')
self.assertEqual(msg.anymail_status.recipients['to1@example.com'].message_id, 'abc123')
self.assertEqual(msg.anymail_status.esp_response.content, response_content)
# noinspection PyUnresolvedReferences
def test_send_failed_anymail_status(self):
""" If the send fails, anymail_status should contain initial values"""
self.set_mock_response(status_code=500)
sent = self.message.send(fail_silently=True)
self.assertEqual(sent, 0)
self.assertIsNone(self.message.anymail_status.status)
self.assertIsNone(self.message.anymail_status.message_id)
self.assertEqual(self.message.anymail_status.recipients, {})
self.assertIsNone(self.message.anymail_status.esp_response)
# noinspection PyUnresolvedReferences
def test_send_unparsable_response(self):
"""If the send succeeds, but a non-JSON API response, should raise an API exception"""
mock_response = self.set_mock_response(status_code=200,
raw=b"yikes, this isn't a real response")
with self.assertRaises(AnymailAPIError):
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, mock_response)
def test_json_serialization_errors(self):
"""Try to provide more information about non-json-serializable data"""
self.message.metadata = {'total': Decimal('19.99')}
with self.assertRaises(AnymailSerializationError) as cm:
self.message.send()
print(self.get_api_call_data())
err = cm.exception
self.assertIsInstance(err, TypeError) # compatibility with json.dumps
self.assertIn("Don't know how to send this data to Mandrill", str(err)) # our added context
self.assertIn("Decimal('19.99') is not JSON serializable", str(err)) # original message
class MandrillBackendRecipientsRefusedTests(MandrillBackendMockAPITestCase):
"""Should raise AnymailRecipientsRefused when *all* recipients are rejected or invalid"""
def test_recipients_refused(self):
msg = mail.EmailMessage('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'reject@test.mandrillapp.com'])
self.set_mock_response(raw=b"""[
{"email": "invalid@localhost", "status": "invalid"},
{"email": "reject@test.mandrillapp.com", "status": "rejected"}
]""")
with self.assertRaises(AnymailRecipientsRefused):
msg.send()
def test_fail_silently(self):
self.set_mock_response(raw=b"""[
{"email": "invalid@localhost", "status": "invalid"},
{"email": "reject@test.mandrillapp.com", "status": "rejected"}
]""")
sent = mail.send_mail('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'reject@test.mandrillapp.com'],
fail_silently=True)
self.assertEqual(sent, 0)
def test_mixed_response(self):
"""If *any* recipients are valid or queued, no exception is raised"""
msg = mail.EmailMessage('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'valid@example.com',
'reject@test.mandrillapp.com', 'also.valid@example.com'])
self.set_mock_response(raw=b"""[
{"email": "invalid@localhost", "status": "invalid"},
{"email": "valid@example.com", "status": "sent"},
{"email": "reject@test.mandrillapp.com", "status": "rejected"},
{"email": "also.valid@example.com", "status": "queued"}
]""")
sent = msg.send()
self.assertEqual(sent, 1) # one message sent, successfully, to 2 of 4 recipients
status = msg.anymail_status
self.assertEqual(status.recipients['invalid@localhost'].status, 'invalid')
self.assertEqual(status.recipients['valid@example.com'].status, 'sent')
self.assertEqual(status.recipients['reject@test.mandrillapp.com'].status, 'rejected')
self.assertEqual(status.recipients['also.valid@example.com'].status, 'queued')
@override_settings(ANYMAIL_IGNORE_RECIPIENT_STATUS=True)
def test_settings_override(self):
"""No exception with ignore setting"""
self.set_mock_response(raw=b"""[
{"email": "invalid@localhost", "status": "invalid"},
{"email": "reject@test.mandrillapp.com", "status": "rejected"}
]""")
sent = mail.send_mail('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'reject@test.mandrillapp.com'])
self.assertEqual(sent, 1) # refused message is included in sent count
class MandrillBackendSessionSharingTestCase(SessionSharingTestCasesMixin, MandrillBackendMockAPITestCase):
"""Requests session sharing tests"""
pass # tests are defined in the mixin
@override_settings(ANYMAIL_SEND_DEFAULTS={
'metadata': {'global': 'globalvalue', 'other': 'othervalue'},
'tags': ['globaltag'],
'track_clicks': True,
'track_opens': True,
# 'esp_extra': {'globaloption': 'globalsetting'}, # Mandrill doesn't support esp_extra yet
})
class MandrillBackendSendDefaultsTests(MandrillBackendMockAPITestCase):
"""Tests backend support for global SEND_DEFAULTS"""
def test_send_defaults(self):
"""Test that global send defaults are applied"""
self.message.send()
data = self.get_api_call_json()
# All these values came from ANYMAIL_SEND_DEFAULTS:
self.assertEqual(data['message']['metadata'], {'global': 'globalvalue', 'other': 'othervalue'})
self.assertEqual(data['message']['tags'], ['globaltag'])
self.assertEqual(data['message']['track_clicks'], True)
self.assertEqual(data['message']['track_opens'], True)
# self.assertEqual(data['globaloption'], 'globalsetting')
def test_merge_message_with_send_defaults(self):
"""Test that individual message settings are *merged into* the global send defaults"""
self.message.metadata = {'message': 'messagevalue', 'other': 'override'}
self.message.tags = ['messagetag']
self.message.track_clicks = False
# self.message.esp_extra = {'messageoption': 'messagesetting'}
self.message.send()
data = self.get_api_call_json()
# All these values came from ANYMAIL_SEND_DEFAULTS + message.*:
self.assertEqual(data['message']['metadata'],
{'global': 'globalvalue', 'message': 'messagevalue', 'other': 'override'}) # merged
self.assertEqual(data['message']['tags'], ['globaltag', 'messagetag']) # tags concatenated
self.assertEqual(data['message']['track_clicks'], False) # message overrides
self.assertEqual(data['message']['track_opens'], True)
# self.assertEqual(data['globaloption'], 'globalsetting')
# self.assertEqual(data['messageoption'], 'messagesetting') # additional esp_extra
@override_settings(ANYMAIL_MANDRILL_SEND_DEFAULTS={
'tags': ['esptag'],
'metadata': {'esp': 'espvalue'},
'track_opens': False,
})
def test_esp_send_defaults(self):
"""Test that ESP-specific send defaults override individual global defaults"""
self.message.send()
data = self.get_api_call_json()
# All these values came from ANYMAIL_SEND_DEFAULTS plus ANYMAIL_MAILGUN_SEND_DEFAULTS:
self.assertEqual(data['message']['metadata'], {'esp': 'espvalue'}) # entire metadata overridden
self.assertEqual(data['message']['tags'], ['esptag']) # entire tags overridden
self.assertEqual(data['message']['track_clicks'], True) # we didn't override the global track_clicks
self.assertEqual(data['message']['track_opens'], False)
# self.assertEqual(data['globaloption'], 'globalsetting') # we didn't override the global esp_extra
@override_settings(EMAIL_BACKEND="anymail.backends.mandrill.MandrillBackend")
class MandrillBackendImproperlyConfiguredTests(SimpleTestCase, AnymailTestMixin):
"""Test backend without required settings"""
def test_missing_api_key(self):
with self.assertRaises(ImproperlyConfigured) as cm:
mail.send_mail('Subject', 'Message', 'from@example.com', ['to@example.com'])
errmsg = str(cm.exception)
self.assertRegex(errmsg, r'\bMANDRILL_API_KEY\b')
self.assertRegex(errmsg, r'\bANYMAIL_MANDRILL_API_KEY\b')

333
anymail/tests/test_mandrill_djrill_features.py Normal file
View File

@@ -0,0 +1,333 @@
from datetime import date
from django.core import mail
from django.test import override_settings
from anymail.exceptions import AnymailAPIError, AnymailSerializationError
from .test_mandrill_backend import MandrillBackendMockAPITestCase
class MandrillBackendDjrillFeatureTests(MandrillBackendMockAPITestCase):
"""Test backend support for features leftover from Djrill"""
# Most of these features should be moved to esp_extra.
# The template and merge_ta
def test_djrill_message_options(self):
self.message.url_strip_qs = True
self.message.important = True
self.message.auto_text = True
self.message.auto_html = True
self.message.inline_css = True
self.message.preserve_recipients = True
self.message.view_content_link = False
self.message.tracking_domain = "click.example.com"
self.message.signing_domain = "example.com"
self.message.return_path_domain = "support.example.com"
self.message.subaccount = "marketing-dept"
self.message.async = True
self.message.ip_pool = "Bulk Pool"
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['url_strip_qs'], True)
self.assertEqual(data['message']['important'], True)
self.assertEqual(data['message']['auto_text'], True)
self.assertEqual(data['message']['auto_html'], True)
self.assertEqual(data['message']['inline_css'], True)
self.assertEqual(data['message']['preserve_recipients'], True)
self.assertEqual(data['message']['view_content_link'], False)
self.assertEqual(data['message']['tracking_domain'], "click.example.com")
self.assertEqual(data['message']['signing_domain'], "example.com")
self.assertEqual(data['message']['return_path_domain'], "support.example.com")
self.assertEqual(data['message']['subaccount'], "marketing-dept")
self.assertEqual(data['async'], True)
self.assertEqual(data['ip_pool'], "Bulk Pool")
def test_google_analytics(self):
self.message.google_analytics_domains = ["example.com"]
self.message.google_analytics_campaign = "Email Receipts"
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['google_analytics_domains'], ["example.com"])
self.assertEqual(data['message']['google_analytics_campaign'], "Email Receipts")
def test_recipient_metadata(self):
self.message.recipient_metadata = {
# Anymail expands simple python dicts into the more-verbose
# rcpt/values structures the Mandrill API uses
"customer@example.com": {'cust_id': "67890", 'order_id': "54321"},
"guest@example.com": {'cust_id': "94107", 'order_id': "43215"}
}
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['recipient_metadata'],
[{'rcpt': "customer@example.com",
'values': {'cust_id': "67890", 'order_id': "54321"}},
{'rcpt': "guest@example.com",
'values': {'cust_id': "94107", 'order_id': "43215"}}
])
def test_no_subaccount_by_default(self):
mail.send_mail('Subject', 'Body', 'from@example.com', ['to@example.com'])
data = self.get_api_call_json()
self.assertFalse('subaccount' in data['message'])
@override_settings(ANYMAIL_MANDRILL_SEND_DEFAULTS={'subaccount': 'test_subaccount'})
def test_subaccount_setting(self):
mail.send_mail('Subject', 'Body', 'from@example.com', ['to@example.com'])
data = self.get_api_call_json()
self.assertEqual(data['message']['subaccount'], "test_subaccount")
@override_settings(ANYMAIL_MANDRILL_SEND_DEFAULTS={'subaccount': 'global_setting_subaccount'})
def test_subaccount_message_overrides_setting(self):
message = mail.EmailMessage('Subject', 'Body', 'from@example.com', ['to@example.com'])
message.subaccount = "individual_message_subaccount" # should override global setting
message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['subaccount'], "individual_message_subaccount")
def test_default_omits_options(self):
"""Make sure by default we don't send any Mandrill-specific options.
Options not specified by the caller should be omitted entirely from
the Mandrill API call (*not* sent as False or empty). This ensures
that your Mandrill account settings apply by default.
"""
self.message.send()
self.assert_esp_called("/messages/send.json")
data = self.get_api_call_json()
self.assertFalse('from_name' in data['message'])
self.assertFalse('bcc_address' in data['message'])
self.assertFalse('important' in data['message'])
self.assertFalse('auto_text' in data['message'])
self.assertFalse('auto_html' in data['message'])
self.assertFalse('inline_css' in data['message'])
self.assertFalse('url_strip_qs' in data['message'])
self.assertFalse('preserve_recipients' in data['message'])
self.assertFalse('view_content_link' in data['message'])
self.assertFalse('tracking_domain' in data['message'])
self.assertFalse('signing_domain' in data['message'])
self.assertFalse('return_path_domain' in data['message'])
self.assertFalse('subaccount' in data['message'])
self.assertFalse('google_analytics_domains' in data['message'])
self.assertFalse('google_analytics_campaign' in data['message'])
self.assertFalse('merge_language' in data['message'])
self.assertFalse('global_merge_vars' in data['message'])
self.assertFalse('merge_vars' in data['message'])
self.assertFalse('recipient_metadata' in data['message'])
# Options at top level of api params (not in message dict):
self.assertFalse('async' in data)
self.assertFalse('ip_pool' in data)
def test_dates_not_serialized(self):
"""Old versions of predecessor package Djrill accidentally serialized dates to ISO"""
self.message.metadata = {'SHIP_DATE': date(2015, 12, 2)}
with self.assertRaises(AnymailSerializationError):
self.message.send()
@override_settings(ANYMAIL_SEND_DEFAULTS={
'from_name': 'Djrill Test',
'important': True,
'auto_text': True,
'auto_html': True,
'inline_css': True,
'url_strip_qs': True,
'preserve_recipients': True,
'view_content_link': True,
'subaccount': 'example-subaccount',
'tracking_domain': 'example.com',
'signing_domain': 'example.com',
'return_path_domain': 'example.com',
'google_analytics_domains': ['example.com/test'],
'google_analytics_campaign': ['UA-00000000-1'],
'merge_language': 'mailchimp',
'global_merge_vars': {'TEST': 'djrill'},
'async': True,
'ip_pool': 'Pool1',
'invalid': 'invalid',
})
class MandrillBackendDjrillSendDefaultsTests(MandrillBackendMockAPITestCase):
"""Tests backend support for global SEND_DEFAULTS"""
def test_global_options(self):
"""Test that any global settings get passed through
"""
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['from_name'], 'Djrill Test')
self.assertTrue(data['message']['important'])
self.assertTrue(data['message']['auto_text'])
self.assertTrue(data['message']['auto_html'])
self.assertTrue(data['message']['inline_css'])
self.assertTrue(data['message']['url_strip_qs'])
self.assertTrue(data['message']['preserve_recipients'])
self.assertTrue(data['message']['view_content_link'])
self.assertEqual(data['message']['subaccount'], 'example-subaccount')
self.assertEqual(data['message']['tracking_domain'], 'example.com')
self.assertEqual(data['message']['signing_domain'], 'example.com')
self.assertEqual(data['message']['return_path_domain'], 'example.com')
self.assertEqual(data['message']['google_analytics_domains'], ['example.com/test'])
self.assertEqual(data['message']['google_analytics_campaign'], ['UA-00000000-1'])
self.assertEqual(data['message']['merge_language'], 'mailchimp')
self.assertEqual(data['message']['global_merge_vars'],
[{'name': 'TEST', 'content': 'djrill'}])
self.assertFalse('merge_vars' in data['message'])
self.assertFalse('recipient_metadata' in data['message'])
# Options at top level of api params (not in message dict):
self.assertTrue(data['async'])
self.assertEqual(data['ip_pool'], 'Pool1')
# Option that shouldn't be added
self.assertFalse('invalid' in data['message'])
def test_global_options_override(self):
"""Test that manually settings options overrides global settings
"""
self.message.from_name = "override"
self.message.important = False
self.message.auto_text = False
self.message.auto_html = False
self.message.inline_css = False
self.message.url_strip_qs = False
self.message.preserve_recipients = False
self.message.view_content_link = False
self.message.subaccount = "override"
self.message.tracking_domain = "override.example.com"
self.message.signing_domain = "override.example.com"
self.message.return_path_domain = "override.example.com"
self.message.google_analytics_domains = ['override.example.com']
self.message.google_analytics_campaign = ['UA-99999999-1']
self.message.merge_language = 'handlebars'
self.message.async = False
self.message.ip_pool = "Bulk Pool"
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['from_name'], 'override')
self.assertFalse(data['message']['important'])
self.assertFalse(data['message']['auto_text'])
self.assertFalse(data['message']['auto_html'])
self.assertFalse(data['message']['inline_css'])
self.assertFalse(data['message']['url_strip_qs'])
self.assertFalse(data['message']['preserve_recipients'])
self.assertFalse(data['message']['view_content_link'])
self.assertEqual(data['message']['subaccount'], 'override')
self.assertEqual(data['message']['tracking_domain'], 'override.example.com')
self.assertEqual(data['message']['signing_domain'], 'override.example.com')
self.assertEqual(data['message']['return_path_domain'], 'override.example.com')
self.assertEqual(data['message']['google_analytics_domains'], ['override.example.com'])
self.assertEqual(data['message']['google_analytics_campaign'], ['UA-99999999-1'])
self.assertEqual(data['message']['merge_language'], 'handlebars')
self.assertEqual(data['message']['global_merge_vars'], [{'name': 'TEST', 'content': 'djrill'}])
# Options at top level of api params (not in message dict):
self.assertFalse(data['async'])
self.assertEqual(data['ip_pool'], 'Bulk Pool')
def test_global_merge(self):
# Test that global settings merge in
self.message.global_merge_vars = {'GREETING': "Hello"}
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['global_merge_vars'],
[{'name': "GREETING", 'content': "Hello"},
{'name': 'TEST', 'content': 'djrill'}])
def test_global_merge_overwrite(self):
# Test that global merge settings are overwritten
self.message.global_merge_vars = {'TEST': "Hello"}
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['global_merge_vars'],
[{'name': 'TEST', 'content': 'Hello'}])
class MandrillBackendDjrillTemplateTests(MandrillBackendMockAPITestCase):
"""Test backend support for ESP templating features"""
# Holdovers from Djrill, until we design Anymail's normalized esp-template support
def test_merge_data(self):
# Anymail expands simple python dicts into the more-verbose name/content
# structures the Mandrill API uses
self.message.merge_language = "mailchimp"
self.message.global_merge_vars = {'GREETING': "Hello",
'ACCOUNT_TYPE': "Basic"}
self.message.merge_vars = {
"customer@example.com": {'GREETING': "Dear Customer",
'ACCOUNT_TYPE': "Premium"},
"guest@example.com": {'GREETING': "Dear Guest"},
}
self.message.send()
data = self.get_api_call_json()
self.assertEqual(data['message']['merge_language'], "mailchimp")
self.assertEqual(data['message']['global_merge_vars'],
[{'name': 'ACCOUNT_TYPE', 'content': "Basic"},
{'name': "GREETING", 'content': "Hello"}])
self.assertEqual(data['message']['merge_vars'],
[{'rcpt': "customer@example.com",
'vars': [{'name': 'ACCOUNT_TYPE', 'content': "Premium"},
{'name': "GREETING", 'content': "Dear Customer"}]},
{'rcpt': "guest@example.com",
'vars': [{'name': "GREETING", 'content': "Dear Guest"}]}
])
def test_send_template(self):
self.message.template_name = "PERSONALIZED_SPECIALS"
self.message.template_content = {
'HEADLINE': "<h1>Specials Just For *|FNAME|*</h1>",
'OFFER_BLOCK': "<p><em>Half off</em> all fruit</p>"
}
self.message.send()
self.assert_esp_called("/messages/send-template.json")
data = self.get_api_call_json()
self.assertEqual(data['template_name'], "PERSONALIZED_SPECIALS")
# Anymail expands simple python dicts into the more-verbose name/content
# structures the Mandrill API uses
self.assertEqual(data['template_content'],
[{'name': "HEADLINE",
'content': "<h1>Specials Just For *|FNAME|*</h1>"},
{'name': "OFFER_BLOCK",
'content': "<p><em>Half off</em> all fruit</p>"}])
def test_send_template_without_from_field(self):
self.message.template_name = "PERSONALIZED_SPECIALS"
self.message.use_template_from = True
self.message.send()
self.assert_esp_called("/messages/send-template.json")
data = self.get_api_call_json()
self.assertEqual(data['template_name'], "PERSONALIZED_SPECIALS")
self.assertFalse('from_email' in data['message'])
self.assertFalse('from_name' in data['message'])
def test_send_template_without_from_field_api_failure(self):
self.set_mock_response(status_code=400)
self.message.template_name = "PERSONALIZED_SPECIALS"
self.message.use_template_from = True
with self.assertRaises(AnymailAPIError):
self.message.send()
def test_send_template_without_subject_field(self):
self.message.template_name = "PERSONALIZED_SPECIALS"
self.message.use_template_subject = True
self.message.send()
self.assert_esp_called("/messages/send-template.json")
data = self.get_api_call_json()
self.assertEqual(data['template_name'], "PERSONALIZED_SPECIALS")
self.assertFalse('subject' in data['message'])
def test_no_template_content(self):
# Just a template, without any template_content to be merged
self.message.template_name = "WELCOME_MESSAGE"
self.message.send()
self.assert_esp_called("/messages/send-template.json")
data = self.get_api_call_json()
self.assertEqual(data['template_name'], "WELCOME_MESSAGE")
self.assertEqual(data['template_content'], []) # Mandrill requires this field
def test_non_template_send(self):
# Make sure the non-template case still uses /messages/send.json
self.message.send()
self.assert_esp_called("/messages/send.json")
data = self.get_api_call_json()
self.assertFalse('template_name' in data)
self.assertFalse('template_content' in data)
self.assertFalse('async' in data)

50
anymail/tests/test_mandrill_integration.py
View File

@@ -4,11 +4,13 @@ import os
import unittest
from django.core import mail
from django.test import TestCase
from django.test import SimpleTestCase
from django.test.utils import override_settings
from anymail.exceptions import AnymailAPIError, AnymailRecipientsRefused
from anymail.message import AnymailMessage
from .utils import AnymailTestMixin, sample_image_path
MANDRILL_TEST_API_KEY = os.getenv('MANDRILL_TEST_API_KEY')
@@ -17,7 +19,7 @@ MANDRILL_TEST_API_KEY = os.getenv('MANDRILL_TEST_API_KEY')
"Set MANDRILL_TEST_API_KEY environment variable to run integration tests")
@override_settings(MANDRILL_API_KEY=MANDRILL_TEST_API_KEY,
EMAIL_BACKEND="anymail.backends.mandrill.MandrillBackend")
class DjrillIntegrationTests(TestCase):
class MandrillBackendIntegrationTests(SimpleTestCase, AnymailTestMixin):
"""Mandrill API integration tests
These tests run against the **live** Mandrill API, using the
@@ -30,11 +32,12 @@ class DjrillIntegrationTests(TestCase):
"""
def setUp(self):
self.message = mail.EmailMultiAlternatives(
'Subject', 'Text content', 'from@example.com', ['to@example.com'])
super(MandrillBackendIntegrationTests, self).setUp()
self.message = mail.EmailMultiAlternatives('Anymail Mandrill integration test', 'Text content',
'from@example.com', ['to@example.com'])
self.message.attach_alternative('<p>HTML content</p>', "text/html")
def test_send_mail(self):
def test_simple_send(self):
# Example of getting the Mandrill send status and _id from the message
sent_count = self.message.send()
self.assertEqual(sent_count, 1)
@@ -50,14 +53,39 @@ class DjrillIntegrationTests(TestCase):
self.assertEqual(anymail_status.status, {sent_status}) # set of all recipient statuses
self.assertEqual(anymail_status.message_id, message_id) # because only a single recipient (else would be a set)
def test_all_options(self):
message = AnymailMessage(
subject="Anymail all-options integration test",
body="This is the text body",
from_email="Test From <from@example.com>",
to=["to1@example.com", "Recipient 2 <to2@example.com>"],
cc=["cc1@example.com", "Copy 2 <cc2@example.com>"],
bcc=["bcc1@example.com", "Blind Copy 2 <bcc2@example.com>"],
reply_to=["reply1@example.com", "Reply 2 <reply2@example.com>"],
headers={"X-Anymail-Test": "value"},
# no metadata, send_at, track_clicks support
tags=["tag 1"], # max one tag
track_opens=True,
)
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.send()
self.assertTrue(message.anymail_status.status.issubset({'queued', 'sent'}))
def test_invalid_from(self):
# Example of trying to send from an invalid address
# Mandrill returns a 500 response (which raises a MandrillAPIError)
self.message.from_email = 'webmaster@localhost' # Django default DEFAULT_FROM_EMAIL
try:
with self.assertRaises(AnymailAPIError) as cm:
self.message.send()
self.fail("This line will not be reached, because send() raised an exception")
except AnymailAPIError as err:
err = cm.exception
self.assertEqual(err.status_code, 500)
self.assertIn("email address is invalid", str(err))
@@ -102,9 +130,9 @@ class DjrillIntegrationTests(TestCase):
@override_settings(MANDRILL_API_KEY="Hey, that's not an API key!")
def test_invalid_api_key(self):
# Example of trying to send with an invalid MANDRILL_API_KEY
try:
with self.assertRaises(AnymailAPIError) as cm:
self.message.send()
self.fail("This line will not be reached, because send() raised an exception")
except AnymailAPIError as err:
err = cm.exception
self.assertEqual(err.status_code, 500)
# Make sure the exception message includes Mandrill's response:
self.assertIn("Invalid API key", str(err))

765
anymail/tests/test_mandrill_send.py
View File

@@ -1,765 +0,0 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
import os
import six
import unittest
from base64 import b64decode
from datetime import date, datetime
from decimal import Decimal
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from django.core import mail
from django.core.exceptions import ImproperlyConfigured
from django.test import TestCase
from django.test.utils import override_settings
from django.utils.timezone import get_fixed_timezone, override as override_current_timezone
from anymail.exceptions import (AnymailAPIError, AnymailRecipientsRefused,
AnymailSerializationError, AnymailUnsupportedFeature)
from anymail.message import attach_inline_image
from .mock_backend import DjrillBackendMockAPITestCase
def decode_att(att):
"""Returns the original data from base64-encoded attachment content"""
return b64decode(att.encode('ascii'))
class DjrillBackendTests(DjrillBackendMockAPITestCase):
"""Test Djrill backend support for Django mail wrappers"""
sample_image_filename = "sample_image.png"
def sample_image_pathname(self):
"""Returns path to an actual image file in the tests directory"""
test_dir = os.path.dirname(os.path.abspath(__file__))
path = os.path.join(test_dir, self.sample_image_filename)
return path
def sample_image_content(self):
"""Returns contents of an actual image file from the tests directory"""
filename = self.sample_image_pathname()
with open(filename, "rb") as f:
return f.read()
def test_send_mail(self):
mail.send_mail('Subject here', 'Here is the message.',
'from@example.com', ['to@example.com'], fail_silently=False)
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertEqual(data['message']['subject'], "Subject here")
self.assertEqual(data['message']['text'], "Here is the message.")
self.assertFalse('from_name' in data['message'])
self.assertEqual(data['message']['from_email'], "from@example.com")
self.assertEqual(len(data['message']['to']), 1)
self.assertEqual(data['message']['to'][0]['email'], "to@example.com")
def test_name_addr(self):
"""Make sure RFC2822 name-addr format (with display-name) is allowed
(Test both sender and recipient addresses)
"""
msg = mail.EmailMessage('Subject', 'Message',
'From Name <from@example.com>',
['Recipient #1 <to1@example.com>', 'to2@example.com'],
cc=['Carbon Copy <cc1@example.com>', 'cc2@example.com'],
bcc=['Blind Copy <bcc1@example.com>', 'bcc2@example.com'])
msg.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['from_name'], "From Name")
self.assertEqual(data['message']['from_email'], "from@example.com")
self.assertEqual(len(data['message']['to']), 6)
self.assertEqual(data['message']['to'][0]['name'], "Recipient #1")
self.assertEqual(data['message']['to'][0]['email'], "to1@example.com")
self.assertEqual(data['message']['to'][1]['name'], "")
self.assertEqual(data['message']['to'][1]['email'], "to2@example.com")
self.assertEqual(data['message']['to'][2]['name'], "Carbon Copy")
self.assertEqual(data['message']['to'][2]['email'], "cc1@example.com")
self.assertEqual(data['message']['to'][3]['name'], "")
self.assertEqual(data['message']['to'][3]['email'], "cc2@example.com")
self.assertEqual(data['message']['to'][4]['name'], "Blind Copy")
self.assertEqual(data['message']['to'][4]['email'], "bcc1@example.com")
self.assertEqual(data['message']['to'][5]['name'], "")
self.assertEqual(data['message']['to'][5]['email'], "bcc2@example.com")
def test_email_message(self):
email = mail.EmailMessage('Subject', 'Body goes here',
'from@example.com',
['to1@example.com', 'Also To <to2@example.com>'],
bcc=['bcc1@example.com', 'Also BCC <bcc2@example.com>'],
cc=['cc1@example.com', 'Also CC <cc2@example.com>'],
headers={'Reply-To': 'another@example.com',
'X-MyHeader': 'my value',
'Message-ID': 'mycustommsgid@example.com'})
email.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertEqual(data['message']['subject'], "Subject")
self.assertEqual(data['message']['text'], "Body goes here")
self.assertEqual(data['message']['from_email'], "from@example.com")
self.assertEqual(data['message']['headers'],
{'Reply-To': 'another@example.com',
'X-MyHeader': 'my value',
'Message-ID': 'mycustommsgid@example.com'})
# Verify recipients correctly identified as "to", "cc", or "bcc"
self.assertEqual(len(data['message']['to']), 6)
self.assertEqual(data['message']['to'][0]['email'], "to1@example.com")
self.assertEqual(data['message']['to'][0]['type'], "to")
self.assertEqual(data['message']['to'][1]['email'], "to2@example.com")
self.assertEqual(data['message']['to'][1]['type'], "to")
self.assertEqual(data['message']['to'][2]['email'], "cc1@example.com")
self.assertEqual(data['message']['to'][2]['type'], "cc")
self.assertEqual(data['message']['to'][3]['email'], "cc2@example.com")
self.assertEqual(data['message']['to'][3]['type'], "cc")
self.assertEqual(data['message']['to'][4]['email'], "bcc1@example.com")
self.assertEqual(data['message']['to'][4]['type'], "bcc")
self.assertEqual(data['message']['to'][5]['email'], "bcc2@example.com")
self.assertEqual(data['message']['to'][5]['type'], "bcc")
# Don't use Mandrill's bcc_address "logging" feature for bcc's:
self.assertNotIn('bcc_address', data['message'])
def test_html_message(self):
text_content = 'This is an important message.'
html_content = '<p>This is an <strong>important</strong> message.</p>'
email = mail.EmailMultiAlternatives('Subject', text_content,
'from@example.com', ['to@example.com'])
email.attach_alternative(html_content, "text/html")
email.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertEqual(data['message']['text'], text_content)
self.assertEqual(data['message']['html'], html_content)
# Don't accidentally send the html part as an attachment:
self.assertFalse('attachments' in data['message'])
def test_html_only_message(self):
html_content = '<p>This is an <strong>important</strong> message.</p>'
email = mail.EmailMessage('Subject', html_content,
'from@example.com', ['to@example.com'])
email.content_subtype = "html" # Main content is now text/html
email.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertNotIn('text', data['message'])
self.assertEqual(data['message']['html'], html_content)
def test_reply_to(self):
# reply_to is new in Django 1.8 -- before that, you can simply include it in headers
try:
# noinspection PyArgumentList
email = mail.EmailMessage('Subject', 'Body goes here', 'from@example.com', ['to1@example.com'],
reply_to=['reply@example.com', 'Other <reply2@example.com>'],
headers={'X-Other': 'Keep'})
except TypeError:
# Pre-Django 1.8
raise unittest.SkipTest("Django version doesn't support EmailMessage(reply_to)")
email.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertEqual(data['message']['headers']['Reply-To'],
'reply@example.com, Other <reply2@example.com>')
self.assertEqual(data['message']['headers']['X-Other'], 'Keep') # don't lose other headers
def test_attachments(self):
email = mail.EmailMessage('Subject', 'Body goes here', 'from@example.com', ['to1@example.com'])
text_content = "* Item one\n* Item two\n* Item three"
email.attach(filename="test.txt", content=text_content, mimetype="text/plain")
# Should guess mimetype if not provided...
png_content = b"PNG\xb4 pretend this is the contents of a png file"
email.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 data"
mimeattachment = MIMEBase('application', 'pdf')
mimeattachment.set_payload(pdf_content)
email.attach(mimeattachment)
# Attachment type that wasn't supported in early Mandrill releases:
ppt_content = b"PPT\xb4 pretend this is a valid ppt file"
email.attach(filename="presentation.ppt", content=ppt_content,
mimetype="application/vnd.ms-powerpoint")
email.send()
data = self.get_api_call_data()
attachments = data['message']['attachments']
self.assertEqual(len(attachments), 4)
self.assertEqual(attachments[0]["type"], "text/plain")
self.assertEqual(attachments[0]["name"], "test.txt")
self.assertEqual(decode_att(attachments[0]["content"]).decode('ascii'), text_content)
self.assertEqual(attachments[1]["type"], "image/png") # inferred from filename
self.assertEqual(attachments[1]["name"], "test.png")
self.assertEqual(decode_att(attachments[1]["content"]), png_content)
self.assertEqual(attachments[2]["type"], "application/pdf")
self.assertEqual(attachments[2]["name"], "") # none
self.assertEqual(decode_att(attachments[2]["content"]), pdf_content)
self.assertEqual(attachments[3]["type"], "application/vnd.ms-powerpoint")
self.assertEqual(attachments[3]["name"], "presentation.ppt")
self.assertEqual(decode_att(attachments[3]["content"]), ppt_content)
# Make sure the image attachment is not treated as embedded:
self.assertFalse('images' in data['message'])
def test_unicode_attachment_correctly_decoded(self):
msg = mail.EmailMessage(
subject='Subject',
body='Body goes here',
from_email='from@example.com',
to=['to1@example.com'],
)
# Slight modification from the Django unicode docs:
# http://django.readthedocs.org/en/latest/ref/unicode.html#email
msg.attach("Une pièce jointe.html", '<p>\u2019</p>', mimetype='text/html')
msg.send()
data = self.get_api_call_data()
attachments = data['message']['attachments']
self.assertEqual(len(attachments), 1)
def test_embedded_images(self):
text_content = 'This has an inline image.'
email = mail.EmailMultiAlternatives('Subject', text_content, 'from@example.com', ['to@example.com'])
image_data = self.sample_image_content() # Read from a png file
cid = attach_inline_image(email, image_data)
html_content = '<p>This has an <img src="cid:%s" alt="inline" /> image.</p>' % cid
email.attach_alternative(html_content, "text/html")
email.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['text'], text_content)
self.assertEqual(data['message']['html'], html_content)
self.assertEqual(len(data['message']['images']), 1)
self.assertEqual(data['message']['images'][0]["type"], "image/png")
self.assertEqual(data['message']['images'][0]["name"], cid)
self.assertEqual(decode_att(data['message']['images'][0]["content"]), image_data)
# Make sure neither the html nor the inline image is treated as an attachment:
self.assertFalse('attachments' in data['message'])
def test_attached_images(self):
image_data = self.sample_image_content()
email = mail.EmailMultiAlternatives('Subject', 'Message', 'from@example.com', ['to@example.com'])
email.attach_file(self.sample_image_pathname()) # option 1: attach as a file
image = MIMEImage(image_data) # option 2: construct the MIMEImage and attach it directly
email.attach(image)
email.send()
data = self.get_api_call_data()
attachments = data['message']['attachments']
self.assertEqual(len(attachments), 2)
self.assertEqual(attachments[0]["type"], "image/png")
self.assertEqual(attachments[0]["name"], self.sample_image_filename)
self.assertEqual(decode_att(attachments[0]["content"]), image_data)
self.assertEqual(attachments[1]["type"], "image/png")
self.assertEqual(attachments[1]["name"], "") # unknown -- not attached as file
self.assertEqual(decode_att(attachments[1]["content"]), image_data)
# Make sure the image attachments are not treated as embedded:
self.assertFalse('images' in data['message'])
def test_alternative_errors(self):
# Multiple alternatives not allowed
email = mail.EmailMultiAlternatives('Subject', 'Body',
'from@example.com', ['to@example.com'])
email.attach_alternative("<p>First html is OK</p>", "text/html")
email.attach_alternative("<p>But not second html</p>", "text/html")
with self.assertRaises(AnymailUnsupportedFeature):
email.send()
# Only html alternatives allowed
email = mail.EmailMultiAlternatives('Subject', 'Body',
'from@example.com', ['to@example.com'])
email.attach_alternative("{'not': 'allowed'}", "application/json")
with self.assertRaises(AnymailUnsupportedFeature):
email.send()
# Make sure fail_silently is respected
email = mail.EmailMultiAlternatives('Subject', 'Body',
'from@example.com', ['to@example.com'])
email.attach_alternative("{'not': 'allowed'}", "application/json")
sent = email.send(fail_silently=True)
self.assertFalse(self.mock_post.called,
msg="Mandrill API should not be called when send fails silently")
self.assertEqual(sent, 0)
def test_mandrill_api_failure(self):
self.mock_post.return_value = self.MockResponse(status_code=400)
with self.assertRaises(AnymailAPIError):
sent = mail.send_mail('Subject', 'Body', 'from@example.com',
['to@example.com'])
self.assertEqual(sent, 0)
# Make sure fail_silently is respected
self.mock_post.return_value = self.MockResponse(status_code=400)
sent = mail.send_mail('Subject', 'Body', 'from@example.com',
['to@example.com'], fail_silently=True)
self.assertEqual(sent, 0)
def test_api_error_includes_details(self):
"""MandrillAPIError should include Mandrill's error message"""
msg = mail.EmailMessage('Subject', 'Body', 'from@example.com', ['to@example.com'])
# JSON error response:
error_response = b"""{
"status": "error",
"code": 12,
"name": "Error_Name",
"message": "Helpful explanation from Mandrill"
}"""
self.mock_post.return_value = self.MockResponse(status_code=400, raw=error_response)
with self.assertRaisesMessage(AnymailAPIError, "Helpful explanation from Mandrill"):
msg.send()
# Non-JSON error response:
self.mock_post.return_value = self.MockResponse(status_code=500, raw=b"Invalid API key")
with self.assertRaisesMessage(AnymailAPIError, "Invalid API key"):
msg.send()
# No content in the error response:
self.mock_post.return_value = self.MockResponse(status_code=502, raw=None)
with self.assertRaises(AnymailAPIError):
msg.send()
class DjrillMandrillFeatureTests(DjrillBackendMockAPITestCase):
"""Test Djrill backend support for Mandrill-specific features"""
def setUp(self):
super(DjrillMandrillFeatureTests, self).setUp()
self.message = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
def test_tracking(self):
# First make sure we're not setting the API param if the track_click
# attr isn't there. (The Mandrill account option of True for html,
# False for plaintext can't be communicated through the API, other than
# by omitting the track_clicks API param to use your account default.)
self.message.send()
data = self.get_api_call_data()
self.assertFalse('track_clicks' in data['message'])
# Now re-send with the params set
self.message.track_opens = True
self.message.track_clicks = True
self.message.url_strip_qs = True
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['track_opens'], True)
self.assertEqual(data['message']['track_clicks'], True)
self.assertEqual(data['message']['url_strip_qs'], True)
def test_message_options(self):
self.message.important = True
self.message.auto_text = True
self.message.auto_html = True
self.message.inline_css = True
self.message.preserve_recipients = True
self.message.view_content_link = False
self.message.tracking_domain = "click.example.com"
self.message.signing_domain = "example.com"
self.message.return_path_domain = "support.example.com"
self.message.subaccount = "marketing-dept"
self.message.async = True
self.message.ip_pool = "Bulk Pool"
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['important'], True)
self.assertEqual(data['message']['auto_text'], True)
self.assertEqual(data['message']['auto_html'], True)
self.assertEqual(data['message']['inline_css'], True)
self.assertEqual(data['message']['preserve_recipients'], True)
self.assertEqual(data['message']['view_content_link'], False)
self.assertEqual(data['message']['tracking_domain'], "click.example.com")
self.assertEqual(data['message']['signing_domain'], "example.com")
self.assertEqual(data['message']['return_path_domain'], "support.example.com")
self.assertEqual(data['message']['subaccount'], "marketing-dept")
self.assertEqual(data['async'], True)
self.assertEqual(data['ip_pool'], "Bulk Pool")
def test_merge(self):
# Djrill expands simple python dicts into the more-verbose name/content
# structures the Mandrill API uses
self.message.merge_language = "mailchimp"
self.message.global_merge_vars = { 'GREETING': "Hello",
'ACCOUNT_TYPE': "Basic" }
self.message.merge_vars = {
"customer@example.com": { 'GREETING': "Dear Customer",
'ACCOUNT_TYPE': "Premium" },
"guest@example.com": { 'GREETING': "Dear Guest" },
}
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['merge_language'], "mailchimp")
self.assertEqual(data['message']['global_merge_vars'],
[ {'name': 'ACCOUNT_TYPE', 'content': "Basic"},
{'name': "GREETING", 'content': "Hello"} ])
self.assertEqual(data['message']['merge_vars'],
[ { 'rcpt': "customer@example.com",
'vars': [{ 'name': 'ACCOUNT_TYPE', 'content': "Premium" },
{ 'name': "GREETING", 'content': "Dear Customer"}] },
{ 'rcpt': "guest@example.com",
'vars': [{ 'name': "GREETING", 'content': "Dear Guest"}] }
])
def test_tags(self):
self.message.tags = ["receipt", "repeat-user"]
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['tags'], ["receipt", "repeat-user"])
def test_google_analytics(self):
self.message.google_analytics_domains = ["example.com"]
self.message.google_analytics_campaign = "Email Receipts"
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['google_analytics_domains'],
["example.com"])
self.assertEqual(data['message']['google_analytics_campaign'],
"Email Receipts")
def test_metadata(self):
self.message.metadata = { 'batch_num': "12345", 'type': "Receipts" }
self.message.recipient_metadata = {
# Djrill expands simple python dicts into the more-verbose
# rcpt/values structures the Mandrill API uses
"customer@example.com": { 'cust_id': "67890", 'order_id': "54321" },
"guest@example.com": { 'cust_id': "94107", 'order_id': "43215" }
}
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['metadata'], { 'batch_num': "12345",
'type': "Receipts" })
self.assertEqual(data['message']['recipient_metadata'],
[ { 'rcpt': "customer@example.com",
'values': { 'cust_id': "67890", 'order_id': "54321" } },
{ 'rcpt': "guest@example.com",
'values': { 'cust_id': "94107", 'order_id': "43215" } }
])
def test_send_at(self):
utc_plus_6 = get_fixed_timezone(6 * 60)
utc_minus_8 = get_fixed_timezone(-8 * 60)
with override_current_timezone(utc_plus_6):
# Timezone-naive datetime assumed to be Django current_timezone
self.message.send_at = datetime(2022, 10, 11, 12, 13, 14, 567)
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['send_at'], "2022-10-11 06:13:14") # 12:13 UTC+6 == 06:13 UTC
# Timezone-aware datetime converted to UTC:
self.message.send_at = datetime(2016, 3, 4, 5, 6, 7, tzinfo=utc_minus_8)
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['send_at'], "2016-03-04 13:06:07") # 05:06 UTC-8 == 13:06 UTC
# Date-only treated as midnight in current timezone
self.message.send_at = date(2022, 10, 22)
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['send_at'], "2022-10-21 18:00:00") # 00:00 UTC+6 == 18:00-1d UTC
# POSIX timestamp
self.message.send_at = 1651820889 # 2022-05-06 07:08:09 UTC
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['send_at'], "2022-05-06 07:08:09")
# String passed unchanged (this is *not* portable between ESPs)
self.message.send_at = "2013-11-12 01:02:03"
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['send_at'], "2013-11-12 01:02:03")
def test_default_omits_options(self):
"""Make sure by default we don't send any Mandrill-specific options.
Options not specified by the caller should be omitted entirely from
the Mandrill API call (*not* sent as False or empty). This ensures
that your Mandrill account settings apply by default.
"""
self.message.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertFalse('from_name' in data['message'])
self.assertFalse('bcc_address' in data['message'])
self.assertFalse('important' in data['message'])
self.assertFalse('track_opens' in data['message'])
self.assertFalse('track_clicks' in data['message'])
self.assertFalse('auto_text' in data['message'])
self.assertFalse('auto_html' in data['message'])
self.assertFalse('inline_css' in data['message'])
self.assertFalse('url_strip_qs' in data['message'])
self.assertFalse('tags' in data['message'])
self.assertFalse('preserve_recipients' in data['message'])
self.assertFalse('view_content_link' in data['message'])
self.assertFalse('tracking_domain' in data['message'])
self.assertFalse('signing_domain' in data['message'])
self.assertFalse('return_path_domain' in data['message'])
self.assertFalse('subaccount' in data['message'])
self.assertFalse('google_analytics_domains' in data['message'])
self.assertFalse('google_analytics_campaign' in data['message'])
self.assertFalse('metadata' in data['message'])
self.assertFalse('merge_language' in data['message'])
self.assertFalse('global_merge_vars' in data['message'])
self.assertFalse('merge_vars' in data['message'])
self.assertFalse('recipient_metadata' in data['message'])
self.assertFalse('images' in data['message'])
# Options at top level of api params (not in message dict):
self.assertFalse('send_at' in data)
self.assertFalse('async' in data)
self.assertFalse('ip_pool' in data)
# noinspection PyUnresolvedReferences
def test_send_attaches_anymail_status(self):
""" The anymail_status should be attached to the message when it is sent """
response = [{'email': 'to1@example.com', 'status': 'sent', '_id': 'abc123'}]
self.mock_post.return_value = self.MockResponse(raw=six.b(json.dumps(response)))
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],)
sent = msg.send()
self.assertEqual(sent, 1)
self.assertEqual(msg.anymail_status.status, {'sent'})
self.assertEqual(msg.anymail_status.message_id, 'abc123')
self.assertEqual(msg.anymail_status.recipients['to1@example.com'].status, 'sent')
self.assertEqual(msg.anymail_status.recipients['to1@example.com'].message_id, 'abc123')
self.assertEqual(msg.anymail_status.esp_response.json(), response)
# noinspection PyUnresolvedReferences
def test_send_failed_anymail_status(self):
""" If the send fails, anymail_status should contain initial values"""
self.mock_post.return_value = self.MockResponse(status_code=500)
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],)
sent = msg.send(fail_silently=True)
self.assertEqual(sent, 0)
self.assertIsNone(msg.anymail_status.status)
self.assertIsNone(msg.anymail_status.message_id)
self.assertEqual(msg.anymail_status.recipients, {})
self.assertIsNone(msg.anymail_status.esp_response)
# noinspection PyUnresolvedReferences
def test_send_unparsable_mandrill_response(self):
"""If the send succeeds, but a non-JSON API response, should raise an API exception"""
self.mock_post.return_value = self.MockResponse(status_code=200, raw=b"this isn't json")
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],)
with self.assertRaises(AnymailAPIError):
msg.send()
self.assertIsNone(msg.anymail_status.status)
self.assertIsNone(msg.anymail_status.message_id)
self.assertEqual(msg.anymail_status.recipients, {})
self.assertEqual(msg.anymail_status.esp_response, self.mock_post.return_value)
def test_json_serialization_errors(self):
"""Try to provide more information about non-json-serializable data"""
self.message.global_merge_vars = {'PRICE': Decimal('19.99')}
with self.assertRaises(AnymailSerializationError) as cm:
self.message.send()
err = cm.exception
self.assertTrue(isinstance(err, TypeError)) # Djrill 1.x re-raised TypeError from json.dumps
self.assertIn("Don't know how to send this data to Mandrill", str(err)) # our added context
self.assertIn("Decimal('19.99') is not JSON serializable", str(err)) # original message
def test_dates_not_serialized(self):
"""Pre-2.0 Djrill accidentally serialized dates to ISO"""
self.message.global_merge_vars = {'SHIP_DATE': date(2015, 12, 2)}
with self.assertRaises(AnymailSerializationError):
self.message.send()
class DjrillRecipientsRefusedTests(DjrillBackendMockAPITestCase):
"""Djrill raises AnymailRecipientsRefused when *all* recipients are rejected or invalid"""
def test_recipients_refused(self):
msg = mail.EmailMessage('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'reject@test.mandrillapp.com'])
self.mock_post.return_value = self.MockResponse(status_code=200, raw=b"""
[{ "email": "invalid@localhost", "status": "invalid" },
{ "email": "reject@test.mandrillapp.com", "status": "rejected" }]""")
with self.assertRaises(AnymailRecipientsRefused):
msg.send()
def test_fail_silently(self):
self.mock_post.return_value = self.MockResponse(status_code=200, raw=b"""
[{ "email": "invalid@localhost", "status": "invalid" },
{ "email": "reject@test.mandrillapp.com", "status": "rejected" }]""")
sent = mail.send_mail('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'reject@test.mandrillapp.com'],
fail_silently=True)
self.assertEqual(sent, 0)
def test_mixed_response(self):
"""If *any* recipients are valid or queued, no exception is raised"""
msg = mail.EmailMessage('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'valid@example.com',
'reject@test.mandrillapp.com', 'also.valid@example.com'])
self.mock_post.return_value = self.MockResponse(status_code=200, raw=b"""
[{ "email": "invalid@localhost", "status": "invalid" },
{ "email": "valid@example.com", "status": "sent" },
{ "email": "reject@test.mandrillapp.com", "status": "rejected" },
{ "email": "also.valid@example.com", "status": "queued" }]""")
sent = msg.send()
self.assertEqual(sent, 1) # one message sent, successfully, to 2 of 4 recipients
@override_settings(ANYMAIL_IGNORE_RECIPIENT_STATUS=True)
def test_settings_override(self):
"""Setting restores Djrill 1.x behavior"""
self.mock_post.return_value = self.MockResponse(status_code=200, raw=b"""
[{ "email": "invalid@localhost", "status": "invalid" },
{ "email": "reject@test.mandrillapp.com", "status": "rejected" }]""")
sent = mail.send_mail('Subject', 'Body', 'from@example.com',
['invalid@localhost', 'reject@test.mandrillapp.com'])
self.assertEqual(sent, 1) # refused message is included in sent count
@override_settings(ANYMAIL_SEND_DEFAULTS={
'from_name': 'Djrill Test',
'important': True,
'track_opens': True,
'track_clicks': True,
'auto_text': True,
'auto_html': True,
'inline_css': True,
'url_strip_qs': True,
'tags': ['djrill'],
'preserve_recipients': True,
'view_content_link': True,
'subaccount': 'example-subaccount',
'tracking_domain': 'example.com',
'signing_domain': 'example.com',
'return_path_domain': 'example.com',
'google_analytics_domains': ['example.com/test'],
'google_analytics_campaign': ['UA-00000000-1'],
'metadata': {'feature': 'global', 'plus': 'that'},
'merge_language': 'mailchimp',
'global_merge_vars': {'TEST': 'djrill'},
'async': True,
'ip_pool': 'Pool1',
'invalid': 'invalid',
})
class DjrillMandrillGlobalFeatureTests(DjrillBackendMockAPITestCase):
"""Test Djrill backend support for global ovveride Mandrill-specific features"""
def setUp(self):
super(DjrillMandrillGlobalFeatureTests, self).setUp()
self.message = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
def test_global_options(self):
"""Test that any global settings get passed through
"""
self.message.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertEqual(data['message']['from_name'], 'Djrill Test')
self.assertTrue(data['message']['important'])
self.assertTrue(data['message']['track_opens'])
self.assertTrue(data['message']['track_clicks'])
self.assertTrue(data['message']['auto_text'])
self.assertTrue(data['message']['auto_html'])
self.assertTrue(data['message']['inline_css'])
self.assertTrue(data['message']['url_strip_qs'])
self.assertEqual(data['message']['tags'], ['djrill'])
self.assertTrue(data['message']['preserve_recipients'])
self.assertTrue(data['message']['view_content_link'])
self.assertEqual(data['message']['subaccount'], 'example-subaccount')
self.assertEqual(data['message']['tracking_domain'], 'example.com')
self.assertEqual(data['message']['signing_domain'], 'example.com')
self.assertEqual(data['message']['return_path_domain'], 'example.com')
self.assertEqual(data['message']['google_analytics_domains'], ['example.com/test'])
self.assertEqual(data['message']['google_analytics_campaign'], ['UA-00000000-1'])
self.assertEqual(data['message']['metadata'], {'feature': 'global', 'plus': 'that'})
self.assertEqual(data['message']['merge_language'], 'mailchimp')
self.assertEqual(data['message']['global_merge_vars'],
[{'name': 'TEST', 'content': 'djrill'}])
self.assertFalse('merge_vars' in data['message'])
self.assertFalse('recipient_metadata' in data['message'])
# Options at top level of api params (not in message dict):
self.assertTrue(data['async'])
self.assertEqual(data['ip_pool'], 'Pool1')
# Option that shouldn't be added
self.assertFalse('invalid' in data['message'])
def test_global_options_override(self):
"""Test that manually settings options overrides global settings
"""
self.message.from_name = "override"
self.message.important = False
self.message.track_opens = False
self.message.track_clicks = False
self.message.auto_text = False
self.message.auto_html = False
self.message.inline_css = False
self.message.url_strip_qs = False
self.message.tags = ['override']
self.message.preserve_recipients = False
self.message.view_content_link = False
self.message.subaccount = "override"
self.message.tracking_domain = "override.example.com"
self.message.signing_domain = "override.example.com"
self.message.return_path_domain = "override.example.com"
self.message.google_analytics_domains = ['override.example.com']
self.message.google_analytics_campaign = ['UA-99999999-1']
self.message.metadata = {'feature': 'message', 'also': 'this'}
self.message.merge_language = 'handlebars'
self.message.async = False
self.message.ip_pool = "Bulk Pool"
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['from_name'], 'override')
self.assertFalse(data['message']['important'])
self.assertFalse(data['message']['track_opens'])
self.assertFalse(data['message']['track_clicks'])
self.assertFalse(data['message']['auto_text'])
self.assertFalse(data['message']['auto_html'])
self.assertFalse(data['message']['inline_css'])
self.assertFalse(data['message']['url_strip_qs'])
self.assertEqual(data['message']['tags'], ['djrill', 'override']) # tags are merged
self.assertFalse(data['message']['preserve_recipients'])
self.assertFalse(data['message']['view_content_link'])
self.assertEqual(data['message']['subaccount'], 'override')
self.assertEqual(data['message']['tracking_domain'], 'override.example.com')
self.assertEqual(data['message']['signing_domain'], 'override.example.com')
self.assertEqual(data['message']['return_path_domain'], 'override.example.com')
self.assertEqual(data['message']['google_analytics_domains'], ['override.example.com'])
self.assertEqual(data['message']['google_analytics_campaign'], ['UA-99999999-1'])
# metadata is merged:
self.assertEqual(data['message']['metadata'], {'feature': 'message', 'also': 'this', 'plus': 'that'})
self.assertEqual(data['message']['merge_language'], 'handlebars')
self.assertEqual(data['message']['global_merge_vars'],
[{'name': 'TEST', 'content': 'djrill'}])
# Options at top level of api params (not in message dict):
self.assertFalse(data['async'])
self.assertEqual(data['ip_pool'], 'Bulk Pool')
def test_global_merge(self):
# Test that global settings merge in
self.message.global_merge_vars = {'GREETING': "Hello"}
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['global_merge_vars'],
[{'name': "GREETING", 'content': "Hello"},
{'name': 'TEST', 'content': 'djrill'}])
def test_global_merge_overwrite(self):
# Test that global merge settings are overwritten
self.message.global_merge_vars = {'TEST': "Hello"}
self.message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['global_merge_vars'],
[{'name': 'TEST', 'content': 'Hello'}])
@override_settings(EMAIL_BACKEND="anymail.backends.mandrill.MandrillBackend")
class DjrillImproperlyConfiguredTests(TestCase):
"""Test Djrill backend without Djrill-specific settings in place"""
def test_missing_api_key(self):
with self.assertRaises(ImproperlyConfigured):
mail.send_mail('Subject', 'Message', 'from@example.com',
['to@example.com'])

84
anymail/tests/test_mandrill_send_template.py
View File

@@ -1,84 +0,0 @@
from django.core import mail
from anymail.exceptions import AnymailAPIError
from .mock_backend import DjrillBackendMockAPITestCase
class DjrillMandrillSendTemplateTests(DjrillBackendMockAPITestCase):
"""Test Djrill backend support for Mandrill send-template features"""
def test_send_template(self):
msg = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
msg.template_name = "PERSONALIZED_SPECIALS"
msg.template_content = {
'HEADLINE': "<h1>Specials Just For *|FNAME|*</h1>",
'OFFER_BLOCK': "<p><em>Half off</em> all fruit</p>"
}
msg.send()
self.assert_mandrill_called("/messages/send-template.json")
data = self.get_api_call_data()
self.assertEqual(data['template_name'], "PERSONALIZED_SPECIALS")
# Djrill expands simple python dicts into the more-verbose name/content
# structures the Mandrill API uses
self.assertEqual(data['template_content'],
[ {'name': "HEADLINE",
'content': "<h1>Specials Just For *|FNAME|*</h1>"},
{'name': "OFFER_BLOCK",
'content': "<p><em>Half off</em> all fruit</p>"} ]
)
def test_send_template_without_from_field(self):
msg = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
msg.template_name = "PERSONALIZED_SPECIALS"
msg.use_template_from = True
msg.send()
self.assert_mandrill_called("/messages/send-template.json")
data = self.get_api_call_data()
self.assertEqual(data['template_name'], "PERSONALIZED_SPECIALS")
self.assertFalse('from_email' in data['message'])
self.assertFalse('from_name' in data['message'])
def test_send_template_without_from_field_api_failure(self):
self.mock_post.return_value = self.MockResponse(status_code=400)
msg = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
msg.template_name = "PERSONALIZED_SPECIALS"
msg.use_template_from = True
with self.assertRaises(AnymailAPIError):
msg.send()
def test_send_template_without_subject_field(self):
msg = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
msg.template_name = "PERSONALIZED_SPECIALS"
msg.use_template_subject = True
msg.send()
self.assert_mandrill_called("/messages/send-template.json")
data = self.get_api_call_data()
self.assertEqual(data['template_name'], "PERSONALIZED_SPECIALS")
self.assertFalse('subject' in data['message'])
def test_no_template_content(self):
# Just a template, without any template_content to be merged
msg = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
msg.template_name = "WELCOME_MESSAGE"
msg.send()
self.assert_mandrill_called("/messages/send-template.json")
data = self.get_api_call_data()
self.assertEqual(data['template_name'], "WELCOME_MESSAGE")
self.assertEqual(data['template_content'], []) # Mandrill requires this field
def test_non_template_send(self):
# Make sure the non-template case still uses /messages/send.json
msg = mail.EmailMessage('Subject', 'Text Body',
'from@example.com', ['to@example.com'])
msg.send()
self.assert_mandrill_called("/messages/send.json")
data = self.get_api_call_data()
self.assertFalse('template_name' in data)
self.assertFalse('template_content' in data)
self.assertFalse('async' in data)

72
anymail/tests/test_mandrill_session_sharing.py
View File

@@ -1,72 +0,0 @@
from decimal import Decimal
from mock import patch
from django.core import mail
from .mock_backend import DjrillBackendMockAPITestCase
class DjrillSessionSharingTests(DjrillBackendMockAPITestCase):
"""Test Djrill backend sharing of single Mandrill API connection"""
@patch('requests.Session.close', autospec=True)
def test_connection_sharing(self, mock_close):
"""Djrill reuses one requests session when sending multiple messages"""
datatuple = (
('Subject 1', 'Body 1', 'from@example.com', ['to@example.com']),
('Subject 2', 'Body 2', 'from@example.com', ['to@example.com']),
)
mail.send_mass_mail(datatuple)
self.assertEqual(self.mock_post.call_count, 2)
session1 = self.mock_post.call_args_list[0][0] # arg[0] (self) is session
session2 = self.mock_post.call_args_list[1][0]
self.assertEqual(session1, session2)
self.assertEqual(mock_close.call_count, 1)
@patch('requests.Session.close', autospec=True)
def test_caller_managed_connections(self, mock_close):
"""Calling code can created long-lived connection that it opens and closes"""
connection = mail.get_connection()
connection.open()
mail.send_mail('Subject 1', 'body', 'from@example.com', ['to@example.com'], connection=connection)
session1 = self.mock_post.call_args[0]
self.assertEqual(mock_close.call_count, 0) # shouldn't be closed yet
mail.send_mail('Subject 2', 'body', 'from@example.com', ['to@example.com'], connection=connection)
self.assertEqual(mock_close.call_count, 0) # still shouldn't be closed
session2 = self.mock_post.call_args[0]
self.assertEqual(session1, session2) # should have reused same session
connection.close()
self.assertEqual(mock_close.call_count, 1)
def test_session_closed_after_exception(self):
# fail loud case:
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],)
msg.global_merge_vars = {'PRICE': Decimal('19.99')} # will cause JSON serialization error
with patch('requests.Session.close', autospec=True) as mock_close:
with self.assertRaises(TypeError):
msg.send()
self.assertEqual(mock_close.call_count, 1)
# fail silently case (EmailMessage caches backend on send, so must create new one):
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],)
msg.global_merge_vars = {'PRICE': Decimal('19.99')}
with patch('requests.Session.close', autospec=True) as mock_close:
sent = msg.send(fail_silently=True)
self.assertEqual(sent, 0)
self.assertEqual(mock_close.call_count, 1)
# caller-supplied connection case:
with patch('requests.Session.close', autospec=True) as mock_close:
connection = mail.get_connection()
connection.open()
msg = mail.EmailMessage('Subject', 'Message', 'from@example.com', ['to1@example.com'],
connection=connection)
msg.global_merge_vars = {'PRICE': Decimal('19.99')}
with self.assertRaises(TypeError):
msg.send()
self.assertEqual(mock_close.call_count, 0) # wait for us to close it
connection.close()
self.assertEqual(mock_close.call_count, 1)

27
anymail/tests/test_mandrill_subaccounts.py
View File

@@ -1,27 +0,0 @@
from django.core import mail
from django.test.utils import override_settings
from .mock_backend import DjrillBackendMockAPITestCase
class DjrillMandrillSubaccountTests(DjrillBackendMockAPITestCase):
"""Test Djrill backend support for Mandrill subaccounts"""
def test_no_subaccount_by_default(self):
mail.send_mail('Subject', 'Body', 'from@example.com', ['to@example.com'])
data = self.get_api_call_data()
self.assertFalse('subaccount' in data['message'])
@override_settings(ANYMAIL_MANDRILL_SEND_DEFAULTS={'subaccount': 'test_subaccount'})
def test_subaccount_setting(self):
mail.send_mail('Subject', 'Body', 'from@example.com', ['to@example.com'])
data = self.get_api_call_data()
self.assertEqual(data['message']['subaccount'], "test_subaccount")
@override_settings(ANYMAIL_MANDRILL_SEND_DEFAULTS={'subaccount': 'global_setting_subaccount'})
def test_subaccount_message_overrides_setting(self):
message = mail.EmailMessage('Subject', 'Body', 'from@example.com', ['to@example.com'])
message.subaccount = "individual_message_subaccount" # should override global setting
message.send()
data = self.get_api_call_data()
self.assertEqual(data['message']['subaccount'], "individual_message_subaccount")

7
anymail/tests/test_postmark_backend.py
View File

@@ -16,7 +16,7 @@ from anymail.exceptions import (AnymailAPIError, AnymailSerializationError,
AnymailUnsupportedFeature, AnymailRecipientsRefused)
from anymail.message import attach_inline_image_file
from .mock_requests_backend import RequestsBackendMockAPITestCase
from .mock_requests_backend import RequestsBackendMockAPITestCase, SessionSharingTestCasesMixin
from .utils import sample_image_content, sample_image_path, SAMPLE_IMAGE_FILENAME, AnymailTestMixin, decode_att
@@ -517,6 +517,11 @@ class PostmarkBackendRecipientsRefusedTests(PostmarkBackendMockAPITestCase):
self.assertEqual(status.recipients['spam@example.com'].status, 'rejected')
class PostmarkBackendSessionSharingTestCase(SessionSharingTestCasesMixin, PostmarkBackendMockAPITestCase):
"""Requests session sharing tests"""
pass # tests are defined in the mixin
@override_settings(ANYMAIL_SEND_DEFAULTS={
'tags': ['globaltag'],
'track_opens': True,

7
anymail/tests/test_sendgrid_backend.py
View File

@@ -18,7 +18,7 @@ from django.utils.timezone import get_fixed_timezone, override as override_curre
from anymail.exceptions import AnymailAPIError, AnymailSerializationError, AnymailUnsupportedFeature
from anymail.message import attach_inline_image_file
from .mock_requests_backend import RequestsBackendMockAPITestCase
from .mock_requests_backend import RequestsBackendMockAPITestCase, SessionSharingTestCasesMixin
from .utils import sample_image_content, sample_image_path, SAMPLE_IMAGE_FILENAME, AnymailTestMixin
@@ -487,6 +487,11 @@ class SendGridBackendRecipientsRefusedTests(SendGridBackendMockAPITestCase):
pass
class SendGridBackendSessionSharingTestCase(SessionSharingTestCasesMixin, SendGridBackendMockAPITestCase):
"""Requests session sharing tests"""
pass # tests are defined in the mixin
@override_settings(ANYMAIL_SEND_DEFAULTS={
'metadata': {'global': 'globalvalue', 'other': 'othervalue'},
'tags': ['globaltag'],