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
dc0a46a815d062d52660b9237627b22f89093bce
django-anymail/tests/test_mailjet_webhooks.py
Mike Edmunds adc92f037b Tests: tox and mock cleanup 
* Remove mock dependency
  (mock is now part of standard unittest package)
* Cleanup tox dependency installation logic
2021-06-13 12:13:31 -07:00

259 lines
12 KiB
Python
Raw Blame History

import json
from datetime import datetime
from unittest.mock import ANY
from django.test import tag
from django.utils.timezone import utc
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=utc))
self.assertEqual(event.esp_event, raw_events[0])
self.assertEqual(event.mta_response, "sent (250 2.0.0 OK 1498093527 a67bc12345def.22 - gsmtp)")
self.assertEqual(event.message_id, "12345678901234567") # converted to str (matching backend status)
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=utc))
self.assertEqual(event.esp_event, raw_event)
self.assertEqual(event.mta_response, "sent (250 2.0.0 OK 1498093527 a67bc12345def.22 - gsmtp)")
self.assertEqual(event.message_id, "12345678901234567") # converted to str (matching backend status)
self.assertEqual(event.recipient, "recipient@example.com")
Reference in New Issue View Git Blame Copy Permalink