pacnpal/django-anymail
1
0
Fork 0
mirror of https://github.com/pacnpal/django-anymail.git synced 2025-12-19 19:31:06 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
django-anymail/tests/test_mailjet_webhooks.py
medmunds b4e22c63b3 Reformat code with automated tools 
Apply standardized code style
2023-02-06 15:05:24 -08:00

359 lines
13 KiB
Python
Raw Permalink Blame History

import json
from datetime import datetime, timezone
from unittest.mock import ANY
from django.test import tag
from anymail.signals import AnymailTrackingEvent
from anymail.webhooks.mailjet import MailjetTrackingWebhookView
from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase
@tag("mailjet")
class MailjetWebhookSecurityTestCase(WebhookBasicAuthTestCase):
def call_webhook(self):
return self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps([]),
)
# Actual tests are in WebhookBasicAuthTestCase
@tag("mailjet")
class MailjetDeliveryTestCase(WebhookTestCase):
def test_sent_event(self):
# Mailjet's "sent" event indicates receiving MTA has accepted message;
# Anymail calls this "delivered"
raw_events = [
{
"event": "sent",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "recipient@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "tag1",
"mj_message_id": "12345678901234567",
"smtp_reply": "sent (250 2.0.0 OK 1498093527 a67bc12345def.22 - gsmtp)",
"Payload": '{"meta1": "simple string", "meta2": 2}',
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "delivered")
self.assertEqual(
event.timestamp, datetime(2017, 6, 22, 1, 5, 27, tzinfo=timezone.utc)
)
self.assertEqual(event.esp_event, raw_events[0])
self.assertEqual(
event.mta_response,
"sent (250 2.0.0 OK 1498093527 a67bc12345def.22 - gsmtp)",
)
# message_id converted to str (matching backend status):
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.tags, ["tag1"])
self.assertEqual(event.metadata, {"meta1": "simple string", "meta2": 2})
def test_open_event(self):
raw_events = [
{
"event": "open",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "recipient@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"ip": "192.168.100.100",
"geo": "US",
"agent": "Mozilla/5.0 (via ggpht.com GoogleImageProxy)",
"Payload": "",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(
event.user_agent, "Mozilla/5.0 (via ggpht.com GoogleImageProxy)"
)
self.assertEqual(event.tags, [])
self.assertEqual(event.metadata, {})
def test_click_event(self):
raw_events = [
{
"event": "open",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "recipient@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"url": "http://example.com",
"ip": "192.168.100.100",
"geo": "US",
"agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_5)"
" Chrome/58.0.3029.110",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(
event.user_agent,
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_5) Chrome/58.0.3029.110",
)
self.assertEqual(event.click_url, "http://example.com")
self.assertEqual(event.tags, [])
self.assertEqual(event.metadata, {})
def test_bounce_event(self):
raw_events = [
{
"event": "bounce",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "invalid@invalid",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"blocked": True,
"hard_bounce": True,
"error_related_to": "domain",
"error": "invalid domain",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "invalid@invalid")
self.assertEqual(event.reject_reason, "bounced")
self.assertEqual(event.mta_response, None)
def test_blocked_event(self):
raw_events = [
{
"event": "blocked",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "bad@example.com",
"mj_campaign_id": 0,
"mj_contact_id": 9876543210,
"customcampaign": "",
"error_related_to": "domain",
"error": "typofix",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "rejected")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "bad@example.com")
self.assertEqual(event.reject_reason, "invalid")
self.assertEqual(event.mta_response, None)
def test_spam_event(self):
raw_events = [
{
"event": "spam",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "spam@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"source": "greylisted",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "complained")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "spam@example.com")
def test_unsub_event(self):
raw_events = [
{
"event": "unsub",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "recipient@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"mj_list_id": 0,
"ip": "127.0.0.4",
"geo": "",
"agent": "List-Unsubscribe",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "unsubscribed")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "recipient@example.com")
def test_bounced_greylist_event(self):
# greylist "bounce" should be reported as "deferred" (will be retried later)
raw_events = [
{
"event": "bounce",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "protected@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"blocked": True,
"hard_bounce": False,
"error_related_to": "domain",
"error": "greylisted",
}
]
response = self.client.post(
"/anymail/mailjet/tracking/",
content_type="application/json",
data=json.dumps(raw_events),
)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(
self.tracking_handler,
sender=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "deferred")
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "protected@example.com")
self.assertEqual(event.reject_reason, "other")
def test_non_grouped_event(self):
# If you don't enable "group events" on a webhook, Mailjet sends a single bare
# event (not a list of one event, despite what the docs say).
raw_event = {
"event": "sent",
"time": 1498093527,
"MessageID": 12345678901234567,
"email": "recipient@example.com",
"mj_campaign_id": 1234567890,
"mj_contact_id": 9876543210,
"customcampaign": "",
"mj_message_id": "12345678901234567",
"smtp_reply": "sent (250 2.0.0 OK 1498093527 a67bc12345def.22 - gsmtp)",
"Payload": "",
}
response = self.client.post(
"/anymail/mailjet/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=MailjetTrackingWebhookView,
event=ANY,
esp_name="Mailjet",
)
event = kwargs["event"]
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "delivered")
self.assertEqual(
event.timestamp, datetime(2017, 6, 22, 1, 5, 27, tzinfo=timezone.utc)
)
self.assertEqual(event.esp_event, raw_event)
self.assertEqual(
event.mta_response,
"sent (250 2.0.0 OK 1498093527 a67bc12345def.22 - gsmtp)",
)
# message_id converted to str (matching backend status)
self.assertEqual(event.message_id, "12345678901234567")
self.assertEqual(event.recipient, "recipient@example.com")
Reference in New Issue View Git Blame Copy Permalink