mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-20 03:41:05 -05:00
34af81aee6
* Restructure runtests.py as suggested in https://docs.djangoproject.com/en/1.9/topics/testing/advanced/#using-the-django-test-runner-to-test-reusable-applications * Load version-specific Django settings modules (rather than trying to make a single settings.configure() work with all supported versions). * Use `django-admin startproject` defaults for all settings modules. (Tests compatibility with typical apps, middleware, and other settings.) * Set up tests-specific url config; switch to literal urls in test cases. (Eliminates url `reverse` from tests.) * Make runtests.py executable Closes #18
243 lines
12 KiB
Python
243 lines
12 KiB
Python
import json
|
|
from datetime import datetime
|
|
|
|
import hashlib
|
|
import hmac
|
|
from django.core.exceptions import ImproperlyConfigured
|
|
from django.test import override_settings
|
|
from django.utils.timezone import utc
|
|
from mock import ANY
|
|
|
|
from anymail.signals import AnymailTrackingEvent
|
|
from anymail.webhooks.mailgun import MailgunTrackingWebhookView
|
|
|
|
from .webhook_cases import WebhookTestCase, WebhookBasicAuthTestsMixin
|
|
|
|
TEST_API_KEY = 'TEST_API_KEY'
|
|
|
|
|
|
def mailgun_sign(data, api_key=TEST_API_KEY):
|
|
"""Add a Mailgun webhook signature to data dict"""
|
|
# Modifies the dict in place
|
|
data.setdefault('timestamp', '1234567890')
|
|
data.setdefault('token', '1234567890abcdef1234567890abcdef')
|
|
data['signature'] = hmac.new(key=api_key.encode('ascii'),
|
|
msg='{timestamp}{token}'.format(**data).encode('ascii'),
|
|
digestmod=hashlib.sha256).hexdigest()
|
|
return data
|
|
|
|
|
|
def querydict_to_postdict(qd):
|
|
"""Converts a Django QueryDict to a TestClient.post(data)-style dict
|
|
|
|
Single-value fields appear as normal
|
|
Multi-value fields appear as a list (differs from QueryDict.dict)
|
|
"""
|
|
return {
|
|
key: values if len(values) > 1 else values[0]
|
|
for key, values in qd.lists()
|
|
}
|
|
|
|
|
|
class MailgunWebhookSettingsTestCase(WebhookTestCase):
|
|
def test_requires_api_key(self):
|
|
with self.assertRaises(ImproperlyConfigured):
|
|
self.client.post('/anymail/mailgun/tracking/',
|
|
data=mailgun_sign({'event': 'delivered'}))
|
|
|
|
|
|
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
|
|
class MailgunWebhookSecurityTestCase(WebhookTestCase, WebhookBasicAuthTestsMixin):
|
|
should_warn_if_no_auth = False # because we check webhook signature
|
|
|
|
def call_webhook(self):
|
|
return self.client.post('/anymail/mailgun/tracking/',
|
|
data=mailgun_sign({'event': 'delivered'}))
|
|
|
|
# Additional tests are in WebhookBasicAuthTestsMixin
|
|
|
|
def test_verifies_correct_signature(self):
|
|
response = self.client.post('/anymail/mailgun/tracking/',
|
|
data=mailgun_sign({'event': 'delivered'}))
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_verifies_missing_signature(self):
|
|
response = self.client.post('/anymail/mailgun/tracking/',
|
|
data={'event': 'delivered'})
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_verifies_bad_signature(self):
|
|
data = mailgun_sign({'event': 'delivered'}, api_key="wrong API key")
|
|
response = self.client.post('/anymail/mailgun/tracking/', data=data)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
|
|
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
|
|
class MailgunDeliveryTestCase(WebhookTestCase):
|
|
|
|
def test_delivered_event(self):
|
|
raw_event = mailgun_sign({
|
|
'domain': 'example.com',
|
|
'message-headers': json.dumps([
|
|
["Sender", "from=example.com"],
|
|
["Date", "Thu, 21 Apr 2016 17:55:29 +0000"],
|
|
["X-Mailgun-Sid", "WyIxZmY4ZSIsICJtZWRtdW5kc0BnbWFpbC5jb20iLCAiZjFjNzgyIl0="],
|
|
["Received", "by luna.mailgun.net with HTTP; Thu, 21 Apr 2016 17:55:29 +0000"],
|
|
["Message-Id", "<20160421175529.19495.89030.B3AE3728@example.com>"],
|
|
["To", "recipient@example.com"],
|
|
["From", "from@example.com"],
|
|
["Subject", "Webhook testing"],
|
|
["Mime-Version", "1.0"],
|
|
["Content-Type", ["multipart/alternative", {"boundary": "74fb561763da440d8e6a034054974251"}]]
|
|
]),
|
|
'X-Mailgun-Sid': 'WyIxZmY4ZSIsICJtZWRtdW5kc0BnbWFpbC5jb20iLCAiZjFjNzgyIl0=',
|
|
'token': '06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0',
|
|
'timestamp': '1461261330',
|
|
'Message-Id': '<20160421175529.19495.89030.B3AE3728@example.com>',
|
|
'recipient': 'recipient@example.com',
|
|
'event': 'delivered',
|
|
})
|
|
response = self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
|
|
event=ANY, esp_name='Mailgun')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "delivered")
|
|
self.assertEqual(event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=utc))
|
|
self.assertEqual(event.message_id, "<20160421175529.19495.89030.B3AE3728@example.com>")
|
|
self.assertEqual(event.event_id, "06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(querydict_to_postdict(event.esp_event), raw_event)
|
|
|
|
def test_dropped_bounce(self):
|
|
raw_event = mailgun_sign({
|
|
'code': '605',
|
|
'domain': 'example.com',
|
|
'description': 'Not delivering to previously bounced address',
|
|
'attachment-count': '1',
|
|
'Message-Id': '<20160421180324.70521.79375.96884DDB@example.com>',
|
|
'reason': 'hardfail',
|
|
'event': 'dropped',
|
|
'message-headers': json.dumps([
|
|
["X-Mailgun-Sid", "WyI3Y2VjMyIsICJib3VuY2VAZXhhbXBsZS5jb20iLCAiZjFjNzgyIl0="],
|
|
["Received", "by luna.mailgun.net with HTTP; Thu, 21 Apr 2016 18:03:24 +0000"],
|
|
["Message-Id", "<20160421180324.70521.79375.96884DDB@example.com>"],
|
|
["To", "bounce@example.com"],
|
|
["From", "from@example.com"],
|
|
["Subject", "Webhook testing"],
|
|
["Mime-Version", "1.0"],
|
|
["Content-Type", ["multipart/alternative", {"boundary": "a5b51388a4e3455d8feb8510bb8c9fa2"}]]
|
|
]),
|
|
'recipient': 'bounce@example.com',
|
|
'timestamp': '1461261330',
|
|
'X-Mailgun-Sid': 'WyI3Y2VjMyIsICJib3VuY2VAZXhhbXBsZS5jb20iLCAiZjFjNzgyIl0=',
|
|
'token': 'a3fe1fa1640349ac552b84ddde373014b4c41645830c8dd3fc',
|
|
})
|
|
response = self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
|
|
event=ANY, esp_name='Mailgun')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "rejected")
|
|
self.assertEqual(event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=utc))
|
|
self.assertEqual(event.message_id, "<20160421180324.70521.79375.96884DDB@example.com>")
|
|
self.assertEqual(event.event_id, "a3fe1fa1640349ac552b84ddde373014b4c41645830c8dd3fc")
|
|
self.assertEqual(event.recipient, "bounce@example.com")
|
|
self.assertEqual(event.reject_reason, "bounced")
|
|
self.assertEqual(event.description, 'Not delivering to previously bounced address')
|
|
self.assertEqual(querydict_to_postdict(event.esp_event), raw_event)
|
|
|
|
def test_dropped_spam(self):
|
|
raw_event = mailgun_sign({
|
|
'code': '607',
|
|
'description': 'Not delivering to a user who marked your messages as spam',
|
|
'reason': 'hardfail',
|
|
'event': 'dropped',
|
|
'recipient': 'complaint@example.com',
|
|
# (omitting some fields that aren't relevant to the test)
|
|
})
|
|
response = self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
|
|
event=ANY, esp_name='Mailgun')
|
|
event = kwargs['event']
|
|
self.assertEqual(event.event_type, "rejected")
|
|
self.assertEqual(event.reject_reason, "spam")
|
|
self.assertEqual(event.description, 'Not delivering to a user who marked your messages as spam')
|
|
|
|
def test_dropped_timed_out(self):
|
|
raw_event = mailgun_sign({
|
|
'code': '499',
|
|
'description': 'Unable to connect to MX servers: [example.com]',
|
|
'reason': 'old',
|
|
'event': 'dropped',
|
|
'recipient': 'complaint@example.com',
|
|
# (omitting some fields that aren't relevant to the test)
|
|
})
|
|
response = self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
|
|
event=ANY, esp_name='Mailgun')
|
|
event = kwargs['event']
|
|
self.assertEqual(event.event_type, "rejected")
|
|
self.assertEqual(event.reject_reason, "timed_out")
|
|
self.assertEqual(event.description, 'Unable to connect to MX servers: [example.com]')
|
|
|
|
def test_invalid_mailbox(self):
|
|
raw_event = mailgun_sign({
|
|
'code': '550',
|
|
'error': "550 5.1.1 The email account that you tried to reach does not exist. Please try "
|
|
" 5.1.1 double-checking the recipient's email address for typos or "
|
|
" 5.1.1 unnecessary spaces.",
|
|
'event': 'bounced',
|
|
'recipient': 'noreply@example.com',
|
|
# (omitting some fields that aren't relevant to the test)
|
|
})
|
|
response = self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
|
|
event=ANY, esp_name='Mailgun')
|
|
event = kwargs['event']
|
|
self.assertEqual(event.event_type, "bounced")
|
|
self.assertEqual(event.reject_reason, "bounced")
|
|
self.assertIn("The email account that you tried to reach does not exist", event.mta_response)
|
|
|
|
def test_metadata(self):
|
|
# Metadata fields are interspersed with other data, but also in message-headers
|
|
raw_event = mailgun_sign({
|
|
'event': 'delivered',
|
|
'message-headers': json.dumps([
|
|
["X-Mailgun-Variables", "{\"custom1\": \"value1\", \"custom2\": \"{\\\"key\\\":\\\"value\\\"}\"}"],
|
|
]),
|
|
'custom1': 'value',
|
|
'custom2': '{"key":"value"}', # you can store JSON, but you'll need to unpack it yourself
|
|
})
|
|
self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler)
|
|
event = kwargs['event']
|
|
self.assertEqual(event.metadata, {"custom1": "value1", "custom2": '{"key":"value"}'})
|
|
|
|
def test_tags(self):
|
|
# Most events include multiple 'tag' fields for message's tags
|
|
raw_event = mailgun_sign({
|
|
'tag': ['tag1', 'tag2'], # Django TestClient encodes list as multiple field values
|
|
'event': 'opened',
|
|
})
|
|
self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler)
|
|
event = kwargs['event']
|
|
self.assertEqual(event.tags, ["tag1", "tag2"])
|
|
|
|
def test_x_tags(self):
|
|
# Delivery events don't include 'tag', but do include 'X-Mailgun-Tag' fields
|
|
raw_event = mailgun_sign({
|
|
'X-Mailgun-Tag': ['tag1', 'tag2'],
|
|
'event': 'delivered',
|
|
})
|
|
self.client.post('/anymail/mailgun/tracking/', data=raw_event)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler)
|
|
event = kwargs['event']
|
|
self.assertEqual(event.tags, ["tag1", "tag2"])
|