mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-19 19:31:06 -05:00
448 lines
16 KiB
Python
448 lines
16 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import ANY
|
|
|
|
from django.test import tag
|
|
from django.utils.timezone import get_fixed_timezone
|
|
|
|
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=timezone.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=timezone.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=timezone.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=timezone.utc)
|
|
)
|
|
self.assertEqual(event.message_id, "b4cb783d-78ed-43f2-983b-63f55c712dc8")
|
|
self.assertEqual(
|
|
event.recipient,
|
|
"john@example.com",
|
|
)
|
|
self.assertEqual(event.reject_reason, "bounced")
|