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
# don't know what it'll be, but it should exist:
self.assertGreater(len(message_id), 0)
# set of all recipient statuses:
self.assertEqual(anymail_status.status, {sent_status})
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
