pacnpal/django-anymail
1
0
Fork 0
mirror of https://github.com/pacnpal/django-anymail.git synced 2025-12-20 11:51:05 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
pixeebot/drip-2024-12-11-pixee-python/add-requests-timeouts
django-anymail/tests/test_postmark_webhooks.py
medmunds b4e22c63b3 Reformat code with automated tools 
Apply standardized code style
2023-02-06 15:05:24 -08:00

448 lines
16 KiB
Python
Raw Permalink Blame History

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")
Reference in New Issue View Git Blame Copy Permalink