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
Files
2cf14c3653e891dcffc794590c397dabb2f180cb
django-anymail/tests/test_mailgun_integration.py
medmunds 3f63fdd713 Mailgun: Fix lost attachments with non-ASCII filenames. 
Workaround requests/requests#4652 (urllib3/urllib3#303), where
uploaded files in multipart/form-data are improperly given RFC 2231
encoded filenames. That format is not accepted by Mailgun's API (and is
prohibited by RFC 7578), resulting in the attachments being silently
dropped.

Fix is to patch up the multipart/form-data before posting to remove
the RFC 2231 encoding.

Fixes #125
2018-10-10 15:02:13 -07:00

185 lines
8.8 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import os
import logging
import unittest
from datetime import datetime, timedelta
from time import mktime, sleep
import requests
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
MAILGUN_TEST_API_KEY = os.getenv('MAILGUN_TEST_API_KEY')
MAILGUN_TEST_DOMAIN = os.getenv('MAILGUN_TEST_DOMAIN')
@unittest.skipUnless(RUN_LIVE_TESTS, "RUN_LIVE_TESTS disabled in this environment")
@unittest.skipUnless(MAILGUN_TEST_API_KEY and MAILGUN_TEST_DOMAIN,
"Set MAILGUN_TEST_API_KEY and MAILGUN_TEST_DOMAIN environment variables "
"to run Mailgun integration tests")
@override_settings(ANYMAIL={'MAILGUN_API_KEY': MAILGUN_TEST_API_KEY,
'MAILGUN_SENDER_DOMAIN': MAILGUN_TEST_DOMAIN,
'MAILGUN_SEND_DEFAULTS': {'esp_extra': {'o:testmode': 'yes'}}},
EMAIL_BACKEND="anymail.backends.mailgun.EmailBackend")
class MailgunBackendIntegrationTests(SimpleTestCase, AnymailTestMixin):
"""Mailgun API integration tests
These tests run against the **live** Mailgun API, using the
environment variable `MAILGUN_TEST_API_KEY` as the API key
and `MAILGUN_TEST_DOMAIN` as the sender domain.
If those variables are not set, these tests won't run.
"""
def setUp(self):
super(MailgunBackendIntegrationTests, self).setUp()
self.message = AnymailMessage('Anymail Mailgun integration test', 'Text content',
'from@example.com', ['test+to1@anymail.info'])
self.message.attach_alternative('<p>HTML content</p>', "text/html")
def fetch_mailgun_events(self, message_id, event=None,
initial_delay=2, retry_delay=2, max_retries=5):
"""Return list of Mailgun events related to message_id"""
url = "https://api.mailgun.net/v3/%s/events" % MAILGUN_TEST_DOMAIN
auth = ("api", MAILGUN_TEST_API_KEY)
# Despite the docs, Mailgun's events API actually expects the message-id
# without the <...> brackets (so, not exactly "as returned by the messages API")
# https://documentation.mailgun.com/api-events.html#filter-field
params = {'message-id': message_id[1:-1]} # strip <...>
if event is not None:
params['event'] = event
# It can take a few seconds for the events to show up
# in Mailgun's logs, so retry a few times if necessary:
sleep(initial_delay)
response = None
for retry in range(max_retries):
if retry > 0:
sleep(retry_delay)
response = requests.get(url, auth=auth, params=params)
if 200 == response.status_code:
items = response.json()["items"]
if len(items) > 0:
return items
# else no matching events found yet, so try again after delay
elif 500 <= response.status_code < 600:
# server error (hopefully transient); try again after delay
pass
elif 403 == response.status_code:
# "forbidden": this may be related to API throttling; try again after delay
pass
else:
response.raise_for_status()
# Max retries exceeded:
if response is not None and 200 != response.status_code:
logging.warning("Ignoring Mailgun events API error %d:\n%s"
% (response.status_code, response.text))
return None
def test_simple_send(self):
# Example of getting the Mailgun send status and message id from the message
sent_count = self.message.send()
self.assertEqual(sent_count, 1)
anymail_status = self.message.anymail_status
sent_status = anymail_status.recipients['test+to1@anymail.info'].status
message_id = anymail_status.recipients['test+to1@anymail.info'].message_id
self.assertEqual(sent_status, 'queued') # Mailgun always queues
self.assertGreater(len(message_id), 0) # don't know what it'll be, but it should exist
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):
send_at = datetime.now().replace(microsecond=0) + timedelta(minutes=2)
send_at_timestamp = mktime(send_at.timetuple()) # python3: send_at.timestamp()
message = AnymailMessage(
subject="Anymail Mailgun all-options integration test",
body="This is the text body",
from_email="Test From <from@example.com>, also-from@example.com",
to=["test+to1@anymail.info", "Recipient 2 <test+to2@anymail.info>"],
cc=["test+cc1@anymail.info", "Copy 2 <test+cc2@anymail.info>"],
bcc=["test+bcc1@anymail.info", "Blind Copy 2 <test+bcc2@anymail.info>"],
reply_to=["reply1@example.com", "Reply 2 <reply2@example.com>"],
headers={"X-Anymail-Test": "value"},
metadata={"meta1": "simple string", "meta2": 2},
send_at=send_at,
tags=["tag 1", "tag 2"],
track_clicks=False,
track_opens=True,
)
message.attach("attachment1.txt", "Here is some\ntext for you", "text/plain")
message.attach("vedhæftet fil.csv", "ID,Name\n1,3", "text/csv")
cid = message.attach_inline_image_file(sample_image_path(), domain=MAILGUN_TEST_DOMAIN)
message.attach_alternative(
"<div>This is the <i>html</i> body <img src='cid:%s'></div>" % cid,
"text/html")
message.send()
self.assertEqual(message.anymail_status.status, {'queued'}) # Mailgun always queues
message_id = message.anymail_status.message_id
events = self.fetch_mailgun_events(message_id, event="accepted")
if events is None:
self.skipTest("No Mailgun 'accepted' event after 30sec -- can't complete this test")
return
event = events.pop()
self.assertCountEqual(event["tags"], ["tag 1", "tag 2"]) # don't care about order
self.assertEqual(event["user-variables"],
{"meta1": "simple string", "meta2": "2"}) # all metadata values become strings
self.assertEqual(event["message"]["scheduled-for"], send_at_timestamp)
self.assertIn(event["recipient"], ['test+to1@anymail.info', 'test+to2@anymail.info',
'test+cc1@anymail.info', 'test+cc2@anymail.info',
'test+bcc1@anymail.info', 'test+bcc2@anymail.info'])
headers = event["message"]["headers"]
self.assertEqual(headers["from"], "Test From <from@example.com>, also-from@example.com")
self.assertEqual(headers["to"],
"test+to1@anymail.info, Recipient 2 <test+to2@anymail.info>")
self.assertEqual(headers["subject"], "Anymail Mailgun all-options integration test")
attachments = event["message"]["attachments"]
self.assertEqual(len(attachments), 2) # because inline image shouldn't be an attachment
self.assertEqual(attachments[0]["filename"], "attachment1.txt")
self.assertEqual(attachments[0]["content-type"], "text/plain")
self.assertEqual(attachments[1]["filename"], "vedhæftet fil.csv")
self.assertEqual(attachments[1]["content-type"], "text/csv")
# No other fields are verifiable from the event data.
# (We could try fetching the message from event["storage"]["url"]
# to verify content and other headers.)
# As of Anymail 0.10, this test is no longer possible, because
# Anymail now raises AnymailInvalidAddress without even calling Mailgun
# def test_invalid_from(self):
# self.message.from_email = 'webmaster'
# with self.assertRaises(AnymailAPIError) as cm:
# self.message.send()
# err = cm.exception
# self.assertEqual(err.status_code, 400)
# self.assertIn("'from' parameter is not a valid address", str(err))
@override_settings(ANYMAIL={'MAILGUN_API_KEY': "Hey, that's not an API key",
'MAILGUN_SENDER_DOMAIN': MAILGUN_TEST_DOMAIN,
'MAILGUN_SEND_DEFAULTS': {'esp_extra': {'o:testmode': 'yes'}}})
def test_invalid_api_key(self):
with self.assertRaises(AnymailAPIError) as cm:
self.message.send()
err = cm.exception
self.assertEqual(err.status_code, 401)
# Mailgun doesn't offer any additional explanation in its response body
# self.assertIn("Forbidden", str(err))
Reference in New Issue View Git Blame Copy Permalink