import json from datetime import datetime import hashlib import hmac from django.core.exceptions import ImproperlyConfigured from django.test import override_settings from django.utils.timezone import utc from mock import ANY from anymail.exceptions import AnymailConfigurationError from anymail.signals import AnymailTrackingEvent from anymail.webhooks.mailgun import MailgunTrackingWebhookView from .webhook_cases import WebhookTestCase, WebhookBasicAuthTestsMixin TEST_API_KEY = 'TEST_API_KEY' def mailgun_signature(timestamp, token, api_key): """Generates a Mailgun webhook signature""" # https://documentation.mailgun.com/en/latest/user_manual.html#securing-webhooks return hmac.new( key=api_key.encode('ascii'), msg='{timestamp}{token}'.format(timestamp=timestamp, token=token).encode('ascii'), digestmod=hashlib.sha256).hexdigest() def mailgun_sign_payload(data, api_key=TEST_API_KEY): """Add or complete Mailgun webhook signature block in data dict""" # Modifies the dict in place event_data = data.get('event-data', {}) signature = data.setdefault('signature', {}) token = signature.setdefault('token', '1234567890abcdef1234567890abcdef') timestamp = signature.setdefault('timestamp', str(int(float(event_data.get('timestamp', '1234567890.123'))))) signature['signature'] = mailgun_signature(timestamp, token, api_key=api_key) return data def mailgun_sign_legacy_payload(data, api_key=TEST_API_KEY): """Add a Mailgun webhook signature to data dict""" # Modifies the dict in place data.setdefault('timestamp', '1234567890') data.setdefault('token', '1234567890abcdef1234567890abcdef') data['signature'] = mailgun_signature(data['timestamp'], data['token'], api_key=api_key) return data def querydict_to_postdict(qd): """Converts a Django QueryDict to a TestClient.post(data)-style dict Single-value fields appear as normal Multi-value fields appear as a list (differs from QueryDict.dict) """ return { key: values if len(values) > 1 else values[0] for key, values in qd.lists() } class MailgunWebhookSettingsTestCase(WebhookTestCase): def test_requires_api_key(self): with self.assertRaises(ImproperlyConfigured): self.client.post('/anymail/mailgun/tracking/', content_type="application/json", data=json.dumps(mailgun_sign_payload({'event-data': {'event': 'delivered'}}))) @override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY) class MailgunWebhookSecurityTestCase(WebhookTestCase, WebhookBasicAuthTestsMixin): should_warn_if_no_auth = False # because we check webhook signature def call_webhook(self): return self.client.post('/anymail/mailgun/tracking/', content_type="application/json", data=json.dumps(mailgun_sign_payload({'event-data': {'event': 'delivered'}}))) # Additional tests are in WebhookBasicAuthTestsMixin def test_verifies_correct_signature(self): response = self.client.post('/anymail/mailgun/tracking/', content_type="application/json", data=json.dumps(mailgun_sign_payload({'event-data': {'event': 'delivered'}}))) self.assertEqual(response.status_code, 200) def test_verifies_missing_signature(self): response = self.client.post('/anymail/mailgun/tracking/', content_type="application/json", data=json.dumps({'event-data': {'event': 'delivered'}})) self.assertEqual(response.status_code, 400) def test_verifies_bad_signature(self): data = mailgun_sign_payload({'event-data': {'event': 'delivered'}}, api_key="wrong API key") response = self.client.post('/anymail/mailgun/tracking/', content_type="application/json", data=json.dumps(data)) self.assertEqual(response.status_code, 400) @override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY) class MailgunTestCase(WebhookTestCase): # Tests for Mailgun's new webhooks (announced 2018-06-29) def test_delivered_event(self): # This is an actual, complete (sanitized) "delivered" event as received from Mailgun. # (For brevity, later tests omit several payload fields that aren't used by Anymail.) raw_event = mailgun_sign_payload({ "signature": { "timestamp": "1534108637", "token": "651869375b9df3c98fc15c4889b102119add1235c38fc92824", "signature": "...", }, "event-data": { "tags": [], "timestamp": 1534108637.153125, "storage": { "url": "https://sw.api.mailgun.net/v3/domains/example.org/messages/eyJwI...", "key": "eyJwI...", }, "recipient-domain": "example.com", "id": "hTWCTD81RtiDN-...", "campaigns": [], "user-variables": {}, "flags": { "is-routed": False, "is-authenticated": True, "is-system-test": False, "is-test-mode": False, }, "log-level": "info", "envelope": { "sending-ip": "333.123.123.200", "sender": "test@example.org", "transport": "smtp", "targets": "recipient@example.com", }, "message": { "headers": { "to": "recipient@example.com", "message-id": "20180812211713.1.DF5966851B4BAA99@example.org", "from": "test@example.org", "subject": "Testing", }, "attachments": [], "size": 809, }, "recipient": "recipient@example.com", "event": "delivered", "delivery-status": { "tls": True, "mx-host": "smtp-in.example.com", "attempt-no": 1, "description": "", "session-seconds": 3.5700838565826416, "utf8": True, "code": 250, "message": "OK", "certificate-verified": True, }, }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "delivered") self.assertEqual(event.timestamp, datetime(2018, 8, 12, 21, 17, 17, microsecond=153125, tzinfo=utc)) self.assertEqual(event.message_id, "<20180812211713.1.DF5966851B4BAA99@example.org>") # Note that Anymail uses the "token" as its normalized event_id: self.assertEqual(event.event_id, "651869375b9df3c98fc15c4889b102119add1235c38fc92824") # ... if you want the Mailgun "event id", that's available through the raw esp_event: self.assertEqual(event.esp_event["event-data"]["id"], "hTWCTD81RtiDN-...") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual(event.esp_event, raw_event) self.assertEqual(event.tags, []) self.assertEqual(event.metadata, {}) def test_failed_permanent_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "failed", "severity": "permanent", "reason": "bounce", "recipient": "invalid@example.com", "timestamp": 1534110422.389832, "log-level": "error", "message": { "headers": { "to": "invalid@example.com", "message-id": "20180812214658.1.0DF563D0B3597700@example.org", "from": "Test Sender ", }, }, "delivery-status": { "tls": True, "mx-host": "aspmx.l.example.org", "attempt-no": 1, "description": "", "session-seconds": 2.952177047729492, "utf8": True, "code": 550, "message": "5.1.1 The email account that you tried to reach does not exist. Please try\n" "5.1.1 double-checking the recipient's email address for typos", "certificate-verified": True } }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "bounced") self.assertEqual(event.recipient, "invalid@example.com") self.assertEqual(event.reject_reason, "bounced") self.assertEqual(event.description, "") self.assertEqual(event.mta_response, "5.1.1 The email account that you tried to reach does not exist. Please try\n" "5.1.1 double-checking the recipient's email address for typos") def test_failed_temporary_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "failed", "severity": "temporary", "reason": "generic", "timestamp": 1534111899.659519, "log-level": "warn", "message": { "headers": { "to": "undeliverable@nomx.example.com", "message-id": "20180812214638.1.4A7D468E9BC18C5D@example.org", "from": "Test Sender ", "subject": "Testing" }, }, "recipient": "undeliverable@nomx.example.com", "delivery-status": { "attempt-no": 3, "description": "No MX for nomx.example.com", "session-seconds": 0.0, "retry-seconds": 1800, "code": 498, "message": "No MX for nomx.example.com" } }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "deferred") self.assertEqual(event.recipient, "undeliverable@nomx.example.com") self.assertEqual(event.reject_reason, "other") self.assertEqual(event.description, "No MX for nomx.example.com") self.assertEqual(event.mta_response, "No MX for nomx.example.com") def test_failed_greylisted_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "failed", "severity": "temporary", "reason": "greylisted", "timestamp": 1534111899.659519, "log-level": "warn", "message": { "headers": { "to": "undeliverable@nomx.example.com", "message-id": "20180812214638.1.4A7D468E9BC18C5D@example.org", "from": "Test Sender ", "subject": "Testing" }, }, "recipient": "undeliverable@mx.example.com", "delivery-status": { "mx-host": "mx.example.com", "attempt-no": 1, "description": "Recipient address rejected: Greylisted", "session-seconds": 0.0, "retry-seconds": 300, "code": 450, "message": "Recipient address rejected: Greylisted" } }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "deferred") self.assertEqual(event.recipient, "undeliverable@mx.example.com") self.assertEqual(event.reject_reason, "other") self.assertEqual(event.description, "Recipient address rejected: Greylisted") self.assertEqual(event.mta_response, "Recipient address rejected: Greylisted") def test_rejected_event(self): # (The "rejected" event is documented and appears in Mailgun dashboard logs, # but it doesn't appear to be delivered through webhooks as of 8/2018.) # Note that this payload lacks the recipient field present in all other events. raw_event = mailgun_sign_payload({ "event-data": { "event": "rejected", "timestamp": 1529704976.104692, "log-level": "warn", "reject": { "reason": "Sandbox subdomains are for test purposes only.", "description": "", }, "message": { "headers": { "to": "Recipient Name ", "message-id": "20180622220256.1.B31A451A2E5422BB@sandbox55887.mailgun.org", "from": "test@sandbox55887.mailgun.org", "subject": "Test Subject" }, }, }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "rejected") self.assertEqual(event.reject_reason, "other") self.assertEqual(event.description, "Sandbox subdomains are for test purposes only.") self.assertEqual(event.recipient, "recipient@example.org") def test_complained_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "complained", "id": "ncV2XwymRUKbPek_MIM-Gw", "timestamp": 1377214260.049634, "log-level": "warn", "recipient": "recipient@example.com", "message": { "headers": { "to": "foo@recipient.com", "message-id": "20130718032413.263EE2E0926@example.org", "from": "Sender Name ", "subject": "We are not spammer", }, }, }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "complained") self.assertEqual(event.recipient, "recipient@example.com") def test_unsubscribed_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "unsubscribed", "id": "W3X4JOhFT-OZidZGKKr9iA", "timestamp": 1377213791.421473, "log-level": "info", "recipient": "recipient@example.com", "message": { "headers": { "message-id": "20130822232216.13966.79700@samples.mailgun.org" } }, }, }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "unsubscribed") self.assertEqual(event.recipient, "recipient@example.com") def test_opened_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "opened", "timestamp": 1534109600.089676, "recipient": "recipient@example.com", "tags": ["welcome", "variation-A"], "user-variables": { "cohort": "2018-08-B", "user_id": "123456" }, "message": { # Mailgun *only* includes the message-id header for opened, clicked events... "headers": { "message-id": "20180812213139.1.BC6694A917BB7E6A@example.org" } }, "geolocation": { "country": "US", "region": "CA", "city": "San Francisco" }, "ip": "888.222.444.111", "client-info": { "client-type": "browser", "client-os": "OS X", "device-type": "desktop", "client-name": "Chrome", "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6)..." }, } }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "opened") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual(event.tags, ["welcome", "variation-A"]) self.assertEqual(event.metadata, {"cohort": "2018-08-B", "user_id": "123456"}) def test_clicked_event(self): raw_event = mailgun_sign_payload({ "event-data": { "event": "clicked", "timestamp": 1534109600.089676, "recipient": "recipient@example.com", "url": "https://example.com/test" } }) response = self.client.post('/anymail/mailgun/tracking/', data=json.dumps(raw_event), content_type='application/json') self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "clicked") self.assertEqual(event.click_url, "https://example.com/test") @override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY) class MailgunLegacyTestCase(WebhookTestCase): # Tests for Mailgun's "legacy" webhooks # (which were the only webhooks available prior to Anymail 4.0) def test_delivered_event(self): raw_event = mailgun_sign_legacy_payload({ 'domain': 'example.com', 'message-headers': json.dumps([ ["Sender", "from=example.com"], ["Date", "Thu, 21 Apr 2016 17:55:29 +0000"], ["X-Mailgun-Sid", "WyIxZmY4ZSIsICJtZWRtdW5kc0BnbWFpbC5jb20iLCAiZjFjNzgyIl0="], ["Received", "by luna.mailgun.net with HTTP; Thu, 21 Apr 2016 17:55:29 +0000"], ["Message-Id", "<20160421175529.19495.89030.B3AE3728@example.com>"], ["To", "recipient@example.com"], ["From", "from@example.com"], ["Subject", "Webhook testing"], ["Mime-Version", "1.0"], ["Content-Type", ["multipart/alternative", {"boundary": "74fb561763da440d8e6a034054974251"}]] ]), 'X-Mailgun-Sid': 'WyIxZmY4ZSIsICJtZWRtdW5kc0BnbWFpbC5jb20iLCAiZjFjNzgyIl0=', 'token': '06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0', 'timestamp': '1461261330', 'Message-Id': '<20160421175529.19495.89030.B3AE3728@example.com>', 'recipient': 'recipient@example.com', 'event': 'delivered', }) response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "delivered") self.assertEqual(event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=utc)) self.assertEqual(event.message_id, "<20160421175529.19495.89030.B3AE3728@example.com>") self.assertEqual(event.event_id, "06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual(querydict_to_postdict(event.esp_event), raw_event) self.assertEqual(event.tags, []) self.assertEqual(event.metadata, {}) def test_dropped_bounce(self): raw_event = mailgun_sign_legacy_payload({ 'code': '605', 'domain': 'example.com', 'description': 'Not delivering to previously bounced address', 'attachment-count': '1', 'Message-Id': '<20160421180324.70521.79375.96884DDB@example.com>', 'reason': 'hardfail', 'event': 'dropped', 'message-headers': json.dumps([ ["X-Mailgun-Sid", "WyI3Y2VjMyIsICJib3VuY2VAZXhhbXBsZS5jb20iLCAiZjFjNzgyIl0="], ["Received", "by luna.mailgun.net with HTTP; Thu, 21 Apr 2016 18:03:24 +0000"], ["Message-Id", "<20160421180324.70521.79375.96884DDB@example.com>"], ["To", "bounce@example.com"], ["From", "from@example.com"], ["Subject", "Webhook testing"], ["Mime-Version", "1.0"], ["Content-Type", ["multipart/alternative", {"boundary": "a5b51388a4e3455d8feb8510bb8c9fa2"}]] ]), 'recipient': 'bounce@example.com', 'timestamp': '1461261330', 'X-Mailgun-Sid': 'WyI3Y2VjMyIsICJib3VuY2VAZXhhbXBsZS5jb20iLCAiZjFjNzgyIl0=', 'token': 'a3fe1fa1640349ac552b84ddde373014b4c41645830c8dd3fc', }) response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "rejected") self.assertEqual(event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=utc)) self.assertEqual(event.message_id, "<20160421180324.70521.79375.96884DDB@example.com>") self.assertEqual(event.event_id, "a3fe1fa1640349ac552b84ddde373014b4c41645830c8dd3fc") self.assertEqual(event.recipient, "bounce@example.com") self.assertEqual(event.reject_reason, "bounced") self.assertEqual(event.description, 'Not delivering to previously bounced address') self.assertEqual(querydict_to_postdict(event.esp_event), raw_event) def test_dropped_spam(self): raw_event = mailgun_sign_legacy_payload({ 'code': '607', 'description': 'Not delivering to a user who marked your messages as spam', 'reason': 'hardfail', 'event': 'dropped', 'recipient': 'complaint@example.com', # (omitting some fields that aren't relevant to the test) }) response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "rejected") self.assertEqual(event.reject_reason, "spam") self.assertEqual(event.description, 'Not delivering to a user who marked your messages as spam') def test_dropped_timed_out(self): raw_event = mailgun_sign_legacy_payload({ 'code': '499', 'description': 'Unable to connect to MX servers: [example.com]', 'reason': 'old', 'event': 'dropped', 'recipient': 'complaint@example.com', # (omitting some fields that aren't relevant to the test) }) response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "rejected") self.assertEqual(event.reject_reason, "timed_out") self.assertEqual(event.description, 'Unable to connect to MX servers: [example.com]') def test_invalid_mailbox(self): raw_event = mailgun_sign_legacy_payload({ 'code': '550', 'error': "550 5.1.1 The email account that you tried to reach does not exist. Please try " " 5.1.1 double-checking the recipient's email address for typos or " " 5.1.1 unnecessary spaces.", 'event': 'bounced', 'recipient': 'noreply@example.com', # (omitting some fields that aren't relevant to the test) }) response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "bounced") self.assertEqual(event.reject_reason, "bounced") self.assertIn("The email account that you tried to reach does not exist", event.mta_response) def test_alt_smtp_code(self): # In some cases, Mailgun uses RFC-3463 extended SMTP status codes (x.y.z, rather than nnn). # See issue #62. raw_event = mailgun_sign_legacy_payload({ 'code': '5.1.1', 'error': 'smtp;550 5.1.1 RESOLVER.ADR.RecipNotFound; not found', 'event': 'bounced', 'recipient': 'noreply@example.com', # (omitting some fields that aren't relevant to the test) }) response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView, event=ANY, esp_name='Mailgun') event = kwargs['event'] self.assertEqual(event.event_type, "bounced") self.assertEqual(event.reject_reason, "bounced") self.assertIn("RecipNotFound", event.mta_response) def test_metadata_message_headers(self): # Metadata fields are interspersed with other data, but also in message-headers # for delivered, bounced and dropped events raw_event = mailgun_sign_legacy_payload({ 'event': 'delivered', 'message-headers': json.dumps([ ["X-Mailgun-Variables", "{\"custom1\": \"value1\", \"custom2\": \"{\\\"key\\\":\\\"value\\\"}\"}"], ]), 'custom1': 'value1', 'custom2': '{"key":"value"}', # you can store JSON, but you'll need to unpack it yourself }) self.client.post('/anymail/mailgun/tracking/', data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs['event'] self.assertEqual(event.metadata, {"custom1": "value1", "custom2": '{"key":"value"}'}) def test_metadata_post_fields(self): # Metadata fields are only interspersed with other event params # for opened, clicked, unsubscribed events raw_event = mailgun_sign_legacy_payload({ 'event': 'clicked', 'custom1': 'value1', 'custom2': '{"key":"value"}', # you can store JSON, but you'll need to unpack it yourself }) self.client.post('/anymail/mailgun/tracking/', data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs['event'] self.assertEqual(event.metadata, {"custom1": "value1", "custom2": '{"key":"value"}'}) def test_metadata_key_conflicts(self): # If you happen to name metadata (user-variable) keys the same as Mailgun # event properties, Mailgun will include both in the webhook post. # Make sure we don't confuse them. metadata = { "event": "metadata-event", "recipient": "metadata-recipient", "signature": "metadata-signature", "timestamp": "metadata-timestamp", "token": "metadata-token", "ordinary field": "ordinary metadata value", } raw_event = mailgun_sign_legacy_payload({ 'event': 'clicked', 'recipient': 'actual-recipient@example.com', 'token': 'actual-event-token', 'timestamp': '1461261330', 'url': 'http://clicked.example.com/actual/event/param', 'h': "an (undocumented) Mailgun event param", 'tag': ["actual-tag-1", "actual-tag-2"], }) # Simulate how Mailgun merges user-variables fields into event: for key in metadata.keys(): if key in raw_event: if key in {'signature', 'timestamp', 'token'}: # For these fields, Mailgun's value appears after the metadata value raw_event[key] = [metadata[key], raw_event[key]] elif key == 'message-headers': pass # Mailgun won't merge this field into the event else: # For all other fields, the defined event value comes first raw_event[key] = [raw_event[key], metadata[key]] else: raw_event[key] = metadata[key] response = self.client.post('/anymail/mailgun/tracking/', data=raw_event) self.assertEqual(response.status_code, 200) # if this fails, signature checking is using metadata values kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs['event'] self.assertEqual(event.event_type, "clicked") self.assertEqual(event.recipient, "actual-recipient@example.com") self.assertEqual(event.timestamp.isoformat(), "2016-04-21T17:55:30+00:00") self.assertEqual(event.event_id, "actual-event-token") self.assertEqual(event.tags, ["actual-tag-1", "actual-tag-2"]) self.assertEqual(event.metadata, metadata) def test_tags(self): # Most events include multiple 'tag' fields for message's tags raw_event = mailgun_sign_legacy_payload({ 'tag': ['tag1', 'tag2'], # Django TestClient encodes list as multiple field values 'event': 'opened', }) self.client.post('/anymail/mailgun/tracking/', data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs['event'] self.assertEqual(event.tags, ["tag1", "tag2"]) def test_x_tags(self): # Delivery events don't include 'tag', but do include 'X-Mailgun-Tag' fields raw_event = mailgun_sign_legacy_payload({ 'X-Mailgun-Tag': ['tag1', 'tag2'], 'event': 'delivered', }) self.client.post('/anymail/mailgun/tracking/', data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs['event'] self.assertEqual(event.tags, ["tag1", "tag2"]) def test_misconfigured_inbound(self): raw_event = mailgun_sign_legacy_payload({ 'recipient': 'test@inbound.example.com', 'sender': 'envelope-from@example.org', 'message-headers': '[]', 'body-plain': 'Test body plain', 'body-html': '
Test body html
', }) errmsg = "You seem to have set Mailgun's *inbound* route to Anymail's Mailgun *tracking* webhook URL." with self.assertRaisesMessage(AnymailConfigurationError, errmsg): self.client.post('/anymail/mailgun/tracking/', data=raw_event)