import json from datetime import datetime, timezone from unittest.mock import ANY from django.test import override_settings, tag from anymail.signals import AnymailTrackingEvent from anymail.webhooks.sparkpost import SparkPostTrackingWebhookView from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase @tag("sparkpost") class SparkPostWebhookSecurityTestCase(WebhookBasicAuthTestCase): def call_webhook(self): return self.client.post( "/anymail/sparkpost/tracking/", content_type="application/json", data=json.dumps([]), ) # Actual tests are in WebhookBasicAuthTestCase @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=timezone.utc) ) self.assertEqual(event.esp_event, raw_events[0]) # normalized "message_id" is actually transmission_id: self.assertEqual(event.message_id, "65832150921904138") self.assertEqual(event.event_id, "92356927693813856") self.assertEqual(event.recipient, "recipient@example.com") # campaign_id (rcpt_tags not available at send): self.assertEqual(event.tags, ["Example Campaign Name"]) # includes transmissions.send metadata: self.assertEqual(event.metadata, {"customKey": "customValue"}) 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 :" " 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 : 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", ) @override_settings(ANYMAIL_SPARKPOST_TRACK_INITIAL_OPEN_AS_OPENED=True) def test_initial_open_event_as_opened(self): # Mapping SparkPost "initial_open" to Anymail normalized "opened" is opt-in # via a setting, for backwards compatibility and to avoid reporting duplicate # "opened" events when all SparkPost event types are enabled. raw_events = [ { "msys": { "track_event": { "type": "initial_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_initial_open_event_as_unknown(self): # By default, SparkPost "initial_open" is *not* mapped to Anymail "opened". raw_events = [ { "msys": { "track_event": { "type": "initial_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, "unknown") # Here's how to get the raw SparkPost event type: self.assertEqual( event.esp_event["msys"].get("track_event", {}).get("type"), "initial_open" ) # Note that other Anymail normalized event properties are still available: 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": "amp_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") def test_amp_events(self): raw_events = [ { "msys": { "track_event": { "type": "amp_open", } } }, { "msys": { "track_event": { "type": "amp_initial_open", } } }, { "msys": { "track_event": { "type": "amp_click", } } }, ] response = self.client.post( "/anymail/sparkpost/tracking/", content_type="application/json", data=json.dumps(raw_events), ) self.assertEqual(response.status_code, 200) self.assertEqual(self.tracking_handler.call_count, 3) events = [ kwargs["event"] for (args, kwargs) in self.tracking_handler.call_args_list ] self.assertEqual(events[0].event_type, "opened") # amp_initial_open is mapped to "unknown" by default: self.assertEqual(events[1].event_type, "unknown") self.assertEqual(events[2].event_type, "clicked")