import logging
import os
import unittest
from datetime import datetime, timedelta
from email.utils import formataddr
from time import sleep
import requests
from django.test import SimpleTestCase, override_settings, tag
from anymail.exceptions import AnymailAPIError
from anymail.message import AnymailMessage
from .utils import AnymailTestMixin, sample_image_path
ANYMAIL_TEST_MAILGUN_API_KEY = os.getenv('ANYMAIL_TEST_MAILGUN_API_KEY')
ANYMAIL_TEST_MAILGUN_DOMAIN = os.getenv('ANYMAIL_TEST_MAILGUN_DOMAIN')
@tag('mailgun', 'live')
@unittest.skipUnless(ANYMAIL_TEST_MAILGUN_API_KEY and ANYMAIL_TEST_MAILGUN_DOMAIN,
"Set ANYMAIL_TEST_MAILGUN_API_KEY and ANYMAIL_TEST_MAILGUN_DOMAIN environment variables "
"to run Mailgun integration tests")
@override_settings(ANYMAIL={'MAILGUN_API_KEY': ANYMAIL_TEST_MAILGUN_API_KEY,
'MAILGUN_SENDER_DOMAIN': ANYMAIL_TEST_MAILGUN_DOMAIN,
'MAILGUN_SEND_DEFAULTS': {'esp_extra': {'o:testmode': 'yes'}}},
EMAIL_BACKEND="anymail.backends.mailgun.EmailBackend")
class MailgunBackendIntegrationTests(AnymailTestMixin, SimpleTestCase):
"""Mailgun API integration tests
These tests run against the **live** Mailgun API, using the
environment variable `ANYMAIL_TEST_MAILGUN_API_KEY` as the API key
and `ANYMAIL_TEST_MAILGUN_DOMAIN` as the sender domain.
If those variables are not set, these tests won't run.
"""
def setUp(self):
super().setUp()
self.from_email = 'from@%s' % ANYMAIL_TEST_MAILGUN_DOMAIN
self.message = AnymailMessage('Anymail Mailgun integration test', 'Text content',
self.from_email, ['test+to1@anymail.dev'])
self.message.attach_alternative('HTML content
', "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" % ANYMAIL_TEST_MAILGUN_DOMAIN
auth = ("api", ANYMAIL_TEST_MAILGUN_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.dev'].status
message_id = anymail_status.recipients['test+to1@anymail.dev'].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 = send_at.timestamp()
from_email = formataddr(("Test From, with comma", self.from_email))
message = AnymailMessage(
subject="Anymail Mailgun all-options integration test",
body="This is the text body",
from_email=from_email,
to=["test+to1@anymail.dev", "Recipient 2 "],
cc=["test+cc1@anymail.dev", "Copy 2 "],
bcc=["test+bcc1@anymail.dev", "Blind Copy 2 "],
reply_to=["reply1@example.com", "Reply 2 "],
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=ANYMAIL_TEST_MAILGUN_DOMAIN)
message.attach_alternative(
"This is the
html body
