mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-20 03:41:05 -05:00
48de044c9b
Handle Postmark SubscriptionChange events as Anymail unsubscribe, subscribe, or bounce Anymail tracking events.
316 lines
16 KiB
Python
316 lines
16 KiB
Python
import json
|
|
from datetime import datetime
|
|
from unittest.mock import ANY
|
|
|
|
from django.test import tag
|
|
from django.utils.timezone import get_fixed_timezone, utc
|
|
|
|
from anymail.exceptions import AnymailConfigurationError
|
|
from anymail.signals import AnymailTrackingEvent
|
|
from anymail.webhooks.postmark import PostmarkTrackingWebhookView
|
|
from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase
|
|
|
|
|
|
@tag('postmark')
|
|
class PostmarkWebhookSecurityTestCase(WebhookBasicAuthTestCase):
|
|
def call_webhook(self):
|
|
return self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps({}))
|
|
|
|
# Actual tests are in WebhookBasicAuthTestCase
|
|
|
|
|
|
@tag('postmark')
|
|
class PostmarkDeliveryTestCase(WebhookTestCase):
|
|
def test_bounce_event(self):
|
|
raw_event = {
|
|
"RecordType": "Bounce",
|
|
"ID": 901542550,
|
|
"Type": "HardBounce",
|
|
"TypeCode": 1,
|
|
"ServerID": 23,
|
|
"Name": "Hard bounce",
|
|
"MessageID": "2706ee8a-737c-4285-b032-ccd317af53ed",
|
|
"Description": "The server was unable to deliver your message (ex: unknown user, mailbox not found).",
|
|
"Details": "smtp;550 5.1.1 The email account that you tried to reach does not exist.",
|
|
"Email": "bounce@example.com",
|
|
"From": "sender@example.com",
|
|
"BouncedAt": "2016-04-27T16:28:50.3963933-04:00",
|
|
"DumpAvailable": True,
|
|
"Inactive": True,
|
|
"CanActivate": True,
|
|
"Subject": "Postmark event test",
|
|
"Content": "..."
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "bounced")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2016, 4, 27, 16, 28, 50, microsecond=396393,
|
|
tzinfo=get_fixed_timezone(-4*60)))
|
|
self.assertEqual(event.message_id, "2706ee8a-737c-4285-b032-ccd317af53ed")
|
|
self.assertEqual(event.event_id, "901542550")
|
|
self.assertEqual(event.recipient, "bounce@example.com")
|
|
self.assertEqual(event.reject_reason, "bounced")
|
|
self.assertEqual(event.description,
|
|
"The server was unable to deliver your message (ex: unknown user, mailbox not found).")
|
|
self.assertEqual(event.mta_response,
|
|
"smtp;550 5.1.1 The email account that you tried to reach does not exist.")
|
|
|
|
def test_delivered_event(self):
|
|
raw_event = {
|
|
"RecordType": "Delivery",
|
|
"ServerId": 23,
|
|
"MessageID": "883953f4-6105-42a2-a16a-77a8eac79483",
|
|
"Recipient": "recipient@example.com",
|
|
"Tag": "welcome-email",
|
|
"Metadata": {
|
|
"cohort": "2014-08-A",
|
|
"userid": "12345", # Postmark metadata is always converted to string
|
|
},
|
|
"DeliveredAt": "2014-08-01T13:28:10.2735393-04:00",
|
|
"Details": "Test delivery webhook details"
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "delivered")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2014, 8, 1, 13, 28, 10, microsecond=273539,
|
|
tzinfo=get_fixed_timezone(-4*60)))
|
|
self.assertEqual(event.message_id, "883953f4-6105-42a2-a16a-77a8eac79483")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(event.tags, ["welcome-email"])
|
|
self.assertEqual(event.metadata, {"cohort": "2014-08-A", "userid": "12345"})
|
|
|
|
def test_open_event(self):
|
|
raw_event = {
|
|
"RecordType": "Open",
|
|
"FirstOpen": True,
|
|
"Client": {"Name": "Gmail", "Company": "Google", "Family": "Gmail"},
|
|
"OS": {"Name": "unknown", "Company": "unknown", "Family": "unknown"},
|
|
"Platform": "Unknown",
|
|
"UserAgent": "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0",
|
|
"ReadSeconds": 0,
|
|
"Geo": {},
|
|
"MessageID": "f4830d10-9c35-4f0c-bca3-3d9b459821f8",
|
|
"ReceivedAt": "2016-04-27T16:21:41.2493688-04:00",
|
|
"Recipient": "recipient@example.com"
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "opened")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2016, 4, 27, 16, 21, 41, microsecond=249368,
|
|
tzinfo=get_fixed_timezone(-4*60)))
|
|
self.assertEqual(event.message_id, "f4830d10-9c35-4f0c-bca3-3d9b459821f8")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(event.user_agent, "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0")
|
|
self.assertEqual(event.tags, [])
|
|
self.assertEqual(event.metadata, {})
|
|
|
|
def test_click_event(self):
|
|
raw_event = {
|
|
"RecordType": "Click",
|
|
"ClickLocation": "HTML",
|
|
"Client": {
|
|
"Name": "Chrome 35.0.1916.153",
|
|
"Company": "Google",
|
|
"Family": "Chrome"
|
|
},
|
|
"OS": {
|
|
"Name": "OS X 10.7 Lion",
|
|
"Company": "Apple Computer, Inc.",
|
|
"Family": "OS X 10"
|
|
},
|
|
"Platform": "Desktop",
|
|
"UserAgent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_5) etc.",
|
|
"OriginalLink": "https://example.com/click/me",
|
|
"Geo": {
|
|
"CountryISOCode": "RS",
|
|
"Country": "Serbia",
|
|
"RegionISOCode": "VO",
|
|
"Region": "Autonomna Pokrajina Vojvodina",
|
|
"City": "Novi Sad",
|
|
"Zip": "21000",
|
|
"Coords": "45.2517,19.8369",
|
|
"IP": "8.8.8.8"
|
|
},
|
|
"MessageID": "f4830d10-9c35-4f0c-bca3-3d9b459821f8",
|
|
"ReceivedAt": "2017-10-25T15:21:11.9065619Z",
|
|
"Tag": "welcome-email",
|
|
"Recipient": "recipient@example.com"
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "clicked")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2017, 10, 25, 15, 21, 11, microsecond=906561,
|
|
tzinfo=utc))
|
|
self.assertEqual(event.message_id, "f4830d10-9c35-4f0c-bca3-3d9b459821f8")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(event.user_agent, "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_5) etc.")
|
|
self.assertEqual(event.click_url, "https://example.com/click/me")
|
|
self.assertEqual(event.tags, ["welcome-email"])
|
|
self.assertEqual(event.metadata, {})
|
|
|
|
def test_spam_event(self):
|
|
raw_event = {
|
|
"RecordType": "SpamComplaint",
|
|
"ID": 901542550,
|
|
"Type": "SpamComplaint",
|
|
"TypeCode": 512,
|
|
"Name": "Spam complaint",
|
|
"Tag": "Test",
|
|
"MessageID": "2706ee8a-737c-4285-b032-ccd317af53ed",
|
|
"ServerID": 1234,
|
|
"Description": "",
|
|
"Details": "Test spam complaint details",
|
|
"Email": "spam@example.com",
|
|
"From": "sender@example.com",
|
|
"BouncedAt": "2016-04-27T16:28:50.3963933-04:00",
|
|
"DumpAvailable": True,
|
|
"Inactive": True,
|
|
"CanActivate": False,
|
|
"Subject": "Postmark event test"
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "complained")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2016, 4, 27, 16, 28, 50, microsecond=396393,
|
|
tzinfo=get_fixed_timezone(-4*60)))
|
|
self.assertEqual(event.message_id, "2706ee8a-737c-4285-b032-ccd317af53ed")
|
|
self.assertEqual(event.event_id, "901542550")
|
|
self.assertEqual(event.recipient, "spam@example.com")
|
|
self.assertEqual(event.reject_reason, "spam")
|
|
self.assertEqual(event.description, "")
|
|
self.assertEqual(event.mta_response, "Test spam complaint details")
|
|
|
|
def test_misconfigured_inbound(self):
|
|
errmsg = "You seem to have set Postmark's *inbound* webhook to Anymail's Postmark *tracking* webhook URL."
|
|
with self.assertRaisesMessage(AnymailConfigurationError, errmsg):
|
|
self.client.post('/anymail/postmark/tracking/', content_type='application/json',
|
|
data=json.dumps({"FromFull": {"Email": "from@example.org"}}))
|
|
|
|
with self.assertRaisesMessage(AnymailConfigurationError, errmsg):
|
|
self.client.post('/anymail/postmark/tracking/', content_type='application/json',
|
|
data=json.dumps({"RecordType": "Inbound"}))
|
|
|
|
def test_unsubscribe(self):
|
|
raw_event = {
|
|
"RecordType": "SubscriptionChange",
|
|
"MessageID": "a4909a96-73d7-4c49-b148-a54522d3f7ac",
|
|
"ServerID": 23,
|
|
"MessageStream": "outbound",
|
|
"ChangedAt": "2022-06-05T17:17:32Z",
|
|
"Recipient": "john@example.com",
|
|
"Origin": "Recipient",
|
|
"SuppressSending": True,
|
|
"SuppressionReason": "ManualSuppression",
|
|
"Tag": "welcome-email",
|
|
"Metadata": {
|
|
"example": "value",
|
|
"example_2": "value"
|
|
}
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "unsubscribed")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2022, 6, 5, 17, 17, 32, tzinfo=utc))
|
|
self.assertEqual(event.message_id, "a4909a96-73d7-4c49-b148-a54522d3f7ac")
|
|
self.assertEqual(event.recipient, "john@example.com",)
|
|
self.assertEqual(event.reject_reason, "unsubscribed")
|
|
|
|
def test_resubscribe(self):
|
|
raw_event = {
|
|
"RecordType": "SubscriptionChange",
|
|
"MessageID": "a4909a96-73d7-4c49-b148-a54522d3f7ac",
|
|
"ServerID": 23,
|
|
"MessageStream": "outbound",
|
|
"ChangedAt": "2022-06-05T17:17:32Z",
|
|
"Recipient": "john@example.com",
|
|
"Origin": "Recipient",
|
|
"SuppressSending": False,
|
|
"SuppressionReason": None,
|
|
"Tag": "welcome-email",
|
|
"Metadata": {
|
|
"example": "value",
|
|
"example_2": "value"
|
|
}
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "subscribed")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2022, 6, 5, 17, 17, 32, tzinfo=utc))
|
|
self.assertEqual(event.message_id, "a4909a96-73d7-4c49-b148-a54522d3f7ac")
|
|
self.assertEqual(event.recipient, "john@example.com",)
|
|
self.assertEqual(event.reject_reason, None)
|
|
|
|
def test_subscription_change_bounce(self):
|
|
raw_event = {
|
|
"RecordType": "SubscriptionChange",
|
|
"MessageID": "b4cb783d-78ed-43f2-983b-63f55c712dc8",
|
|
"ServerID": 23,
|
|
"MessageStream": "outbound",
|
|
"ChangedAt": "2022-06-05T17:17:32Z",
|
|
"Recipient": "john@example.com",
|
|
"Origin": "Recipient",
|
|
"SuppressSending": True,
|
|
"SuppressionReason": "HardBounce",
|
|
"Tag": "my-tag",
|
|
"Metadata": {
|
|
"example": "value",
|
|
"example_2": "value"
|
|
}
|
|
}
|
|
response = self.client.post('/anymail/postmark/tracking/',
|
|
content_type='application/json', data=json.dumps(raw_event))
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=PostmarkTrackingWebhookView,
|
|
event=ANY, esp_name='Postmark')
|
|
event = kwargs['event']
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "bounced")
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
self.assertEqual(event.timestamp, datetime(2022, 6, 5, 17, 17, 32, tzinfo=utc))
|
|
self.assertEqual(event.message_id, "b4cb783d-78ed-43f2-983b-63f55c712dc8")
|
|
self.assertEqual(event.recipient, "john@example.com",)
|
|
self.assertEqual(event.reject_reason, "bounced")
|