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
c80310848179cf4bf662c04a7f27e84cb319e4d5
django-anymail/tests/test_sparkpost_webhooks.py
Mike Edmunds 978996d7b8 Test without optional packages 
Tox:
* Add Tox factor for extras (all, none, individual ESP).
  For now, only break out ESPs that have specific extra
  dependencies (amazon_ses, sparkpost).
* Install most package dependencies (including extras)
  through the package itself.
* Use new runtests.py environment vars to limit test tags
  when Tox isn't installing all extras.

Travis:
* Rework matrix to request specific TOXENVs directly;
  drop tox-travis.

Test runner (runtests.py):
* Centralize RUN_LIVE_TESTS logic in runtests.py
* Add ANYMAIL_ONLY_TEST and ANYMAIL_SKIP_TESTS env vars
  (comma-separated lists of tags)

Test implementations:
* Tag all ESP-specific tests with ESP
* Tag live tests with "live"
* Don't import ESP-specific packages at test module level. 
  (Test discovery imports test modules before tag-based filtering.)

Closes #104
2019-02-09 15:04:08 -08:00

295 lines
15 KiB
Python
Raw Blame History

import json
from datetime import datetime
from django.test import tag
from django.utils.timezone import utc
from mock import ANY
from anymail.signals import AnymailTrackingEvent
from anymail.webhooks.sparkpost import SparkPostTrackingWebhookView
from .webhook_cases import WebhookBasicAuthTestsMixin, WebhookTestCase
@tag('sparkpost')
class SparkPostWebhookSecurityTestCase(WebhookTestCase, WebhookBasicAuthTestsMixin):
def call_webhook(self):
return self.client.post('/anymail/sparkpost/tracking/',
content_type='application/json', data=json.dumps([]))
# Actual tests are in WebhookBasicAuthTestsMixin
@tag('sparkpost')
class SparkPostDeliveryTestCase(WebhookTestCase):
def test_ping_event(self):
raw_events = [{'msys': {}}]
response = self.client.post('/anymail/sparkpost/tracking/',
content_type='application/json', data=json.dumps(raw_events))
self.assertEqual(response.status_code, 200)
self.assertFalse(self.tracking_handler.called) # no real events
def test_injection_event(self):
# Full event from SparkPost sample events API. (Later tests omit unused event fields.)
raw_events = [{"msys": {"message_event": {
"type": "injection",
"campaign_id": "Example Campaign Name",
"customer_id": "1",
"event_id": "92356927693813856",
"friendly_from": "sender@example.com",
"ip_pool": "Example-Ip-Pool",
"message_id": "000443ee14578172be22",
"msg_from": "sender@example.com",
"msg_size": "1337",
"rcpt_meta": {"customKey": "customValue"},
"rcpt_tags": ["male", "US"],
"rcpt_to": "recipient@example.com",
"raw_rcpt_to": "recipient@example.com",
"rcpt_type": "cc",
"routing_domain": "example.com",
"sending_ip": "127.0.0.1",
"sms_coding": "ASCII",
"sms_dst": "7876712656",
"sms_dst_npi": "E164",
"sms_dst_ton": "International",
"sms_segments": 5,
"sms_src": "1234",
"sms_src_npi": "E164",
"sms_src_ton": "Unknown",
"sms_text": "lol",
"subaccount_id": "101",
"subject": "Summer deals are here!",
"template_id": "templ-1234",
"template_version": "1",
"timestamp": "1454442600",
"transmission_id": "65832150921904138"
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "queued")
self.assertEqual(event.timestamp, datetime(2016, 2, 2, 19, 50, 00, tzinfo=utc))
self.assertEqual(event.esp_event, raw_events[0])
self.assertEqual(event.message_id, "65832150921904138") # actually transmission_id
self.assertEqual(event.event_id, "92356927693813856")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.tags, ["Example Campaign Name"]) # campaign_id (rcpt_tags not available at send)
self.assertEqual(event.metadata, {"customKey": "customValue"}) # includes transmissions.send metadata
def test_delivery_event(self):
raw_events = [{"msys": {"message_event": {
"type": "delivery",
"event_id": "92356927693813856",
"rcpt_to": "recipient@example.com",
"raw_rcpt_to": "Recipient@example.com",
"rcpt_meta": {},
"timestamp": "1454442600",
"transmission_id": "65832150921904138"
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "delivered")
self.assertEqual(event.recipient, "Recipient@example.com")
self.assertEqual(event.tags, [])
self.assertEqual(event.metadata, {})
def test_bounce_event(self):
raw_events = [{
"msys": {"message_event": {
"type": "bounce",
"bounce_class": "10",
"customer_id": "00000",
"error_code": "550",
"event_id": "84345317653491230",
"message_id": "0004e3724f57753a3561",
"raw_rcpt_to": "bounce@example.com",
"raw_reason": "550 5.1.1 <bounce@example.com>: Recipient address rejected: User unknown",
"rcpt_to": "bounce@example.com",
"reason": "550 5.1.1 ...@... Recipient address rejected: ...",
"timestamp": "1464824548",
"transmission_id": "84345317650824116",
}},
"cust": {"id": "00000"} # Included in real (non-example) event data
}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.esp_event, raw_events[0])
self.assertEqual(event.message_id, "84345317650824116") # transmission_id
self.assertEqual(event.event_id, "84345317653491230")
self.assertEqual(event.recipient, "bounce@example.com")
self.assertEqual(event.reject_reason, "invalid")
self.assertEqual(event.mta_response,
"550 5.1.1 <bounce@example.com>: Recipient address rejected: User unknown")
def test_delay_event(self):
raw_events = [{"msys": {"message_event": {
"type": "delay",
"bounce_class": "21",
"error_code": "454",
"event_id": "84345317653675522",
"message_id": "0004e3724f57753a3861",
"num_retries": "1",
"queue_time": "1200161",
"raw_rcpt_to": "recipient@nomx.example.com",
"raw_reason": "454 4.4.4 [internal] no MX or A for domain",
"rcpt_to": "recipient@nomx.example.com",
"reason": "454 4.4.4 [internal] no MX or A for domain",
"timestamp": "1464825748",
"transmission_id": "84345317650824116",
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "deferred")
self.assertEqual(event.esp_event, raw_events[0])
self.assertEqual(event.recipient, "recipient@nomx.example.com")
self.assertEqual(event.mta_response, "454 4.4.4 [internal] no MX or A for domain")
def test_unsubscribe_event(self):
raw_events = [{"msys": {"unsubscribe_event": {
"type": "list_unsubscribe",
"event_id": "66331590532986193",
"message_id": "0004278150574660124d",
"raw_rcpt_to": "recipient@example.com",
"rcpt_to": "recipient@example.com",
"timestamp": "1464894280",
"transmission_id": "84345993965073285",
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "unsubscribed")
self.assertEqual(event.recipient, "recipient@example.com")
def test_generation_rejection_event(self):
# This is what you get if you try to send to a suppressed address
raw_events = [{"msys": {"gen_event": {
"type": "generation_rejection",
"error_code": "554",
"event_id": "102360394390563734",
"message_id": "0005c29950577c61695d",
"raw_rcpt_to": "suppressed@example.com",
"raw_reason": "554 5.7.1 recipient address suppressed due to customer policy",
"rcpt_to": "suppressed@example.com",
"reason": "554 5.7.1 recipient address suppressed due to customer policy",
"timestamp": "1464900034",
"transmission_id": "102360394387646691",
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "rejected")
self.assertEqual(event.recipient, "suppressed@example.com")
self.assertEqual(event.mta_response, "554 5.7.1 recipient address suppressed due to customer policy")
def test_generation_failure_event(self):
# This is what you get from a template rendering failure
raw_events = [{"msys": {"message_event": {
"type": "generation_failure",
"error_code": "554",
"event_id": "139013368081587254",
"raw_rcpt_to": "recipient@example.com",
"raw_reason": "554 5.3.3 [internal] Error while rendering part html: ...",
"rcpt_subs": {"name": "Alice", "order_no": "12345"},
"rcpt_to": "recipient@example.com",
"reason": "554 5.3.3 [internal] Error while rendering part html: ...",
"tdate": "2018-10-11T23:24:45.000Z",
"template_id": "test-template",
"template_version": "3",
"transmission_id": "139013368081177607",
"timestamp": "2018-10-11T23:24:45.000+00:00"
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "failed")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.mta_response, "554 5.3.3 [internal] Error while rendering part html: ...")
def test_bounce_challenge_response(self):
# Test for changing initial event_type based on bounce_class
raw_events = [{"msys": {"message_event": {
"type": "bounce",
"bounce_class": "60",
"raw_rcpt_to": "vacationing@example.com",
"rcpt_to": "vacationing@example.com",
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "autoresponded")
self.assertEqual(event.reject_reason, "other")
self.assertEqual(event.recipient, "vacationing@example.com")
def test_open_event(self):
raw_events = [{"msys": {"track_event": {
"type": "open",
"raw_rcpt_to": "recipient@example.com",
"user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_3) AppleWebKit/537.36",
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "opened")
self.assertEqual(event.user_agent, "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_3) AppleWebKit/537.36")
def test_click_event(self):
raw_events = [{"msys": {"track_event": {
"type": "click",
"raw_rcpt_to": "recipient@example.com",
"target_link_name": "Example Link Name",
"target_link_url": "http://example.com",
"user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_3) AppleWebKit/537.36",
}}}]
response = self.client.post('/anymail/sparkpost/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=SparkPostTrackingWebhookView,
event=ANY, esp_name='SparkPost')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "clicked")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.user_agent, "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_3) AppleWebKit/537.36")
self.assertEqual(event.click_url, "http://example.com")
Reference in New Issue View Git Blame Copy Permalink