pacnpal/django-anymail
1
0
Fork 0
mirror of https://github.com/pacnpal/django-anymail.git synced 2025-12-20 03:41:05 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
6cb1a8b4e94af86c8ce2cf6099a1f742e3732560
django-anymail/tests/test_brevo_webhooks.py
Luis Nell d05f448891 Brevo: support proxy open, complained, error events 
Add support for Brevo's new "Complained," "Error" and 
"Loaded by proxy" events in Brevo tracking webhook.

Closes #385.

---------

Co-authored-by: Mike Edmunds <medmunds@gmail.com>
2024-07-25 11:52:32 -07:00

532 lines
20 KiB
Python
Raw Blame History

import json
from datetime import datetime, timezone
from unittest.mock import ANY
from django.test import tag
from anymail.exceptions import AnymailConfigurationError
from anymail.signals import AnymailTrackingEvent
from anymail.webhooks.brevo import BrevoTrackingWebhookView
from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase
@tag("brevo")
class BrevoWebhookSecurityTestCase(WebhookBasicAuthTestCase):
def call_webhook(self):
return self.client.post(
"/anymail/brevo/tracking/",
content_type="application/json",
data=json.dumps({}),
)
# Actual tests are in WebhookBasicAuthTestCase
@tag("brevo")
class BrevoDeliveryTestCase(WebhookTestCase):
# Brevo's webhook payload data is documented at
# https://developers.brevo.com/docs/transactional-webhooks.
# The payloads below were obtained through live testing.
def test_sent_event(self):
raw_event = {
"event": "request",
"email": "recipient@example.com",
"id": 9999999, # this seems to be Brevo account id (not an event id)
"message-id": "<201803062010.27287306012@smtp-relay.mailin.fr>",
"subject": "Test subject",
# From a message sent at 2018-03-06 11:10:23-08:00
# (2018-03-06 19:10:23+00:00)...
"date": "2018-03-06 11:10:23", # tz from Brevo account's preferences
"ts": 1520331023, # 2018-03-06 10:10:23 -- what time zone is this?
"ts_event": 1520331023, # unclear if this ever differs from "ts"
"ts_epoch": 1520363423000, # 2018-03-06 19:10:23.000+00:00 -- UTC (msec)
"X-Mailin-custom": '{"meta": "data"}',
# "tag" is JSON-serialized tags array if `tags` param set on send,
# else single tag string if `X-Mailin-Tag` header set on send,
# else template name if sent using a template,
# else not present.
# "tags" is tags list if `tags` param set on send, else not present.
"tag": '["tag1","tag2"]',
"tags": ["tag1", "tag2"],
"template_id": 12,
"sending_ip": "333.33.33.33",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "queued")
self.assertEqual(event.esp_event, raw_event)
self.assertEqual(
event.timestamp,
datetime(2018, 3, 6, 19, 10, 23, microsecond=0, tzinfo=timezone.utc),
)
self.assertEqual(
event.message_id, "<201803062010.27287306012@smtp-relay.mailin.fr>"
)
# Brevo does not provide a unique event id:
self.assertIsNone(event.event_id)
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.metadata, {"meta": "data"})
self.assertEqual(event.tags, ["tag1", "tag2"])
def test_delivered_event(self):
raw_event = {
# For brevity, this and following tests omit some webhook data
# that was tested earlier, or that is not used by Anymail
"event": "delivered",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "delivered")
self.assertEqual(event.esp_event, raw_event)
self.assertEqual(
event.message_id, "<201803011158.9876543210@smtp-relay.mailin.fr>"
)
self.assertEqual(event.recipient, "recipient@example.com")
# empty dict when no X-Mailin-custom header given:
self.assertEqual(event.metadata, {})
self.assertEqual(event.tags, []) # empty list when no tags given
def test_hard_bounce(self):
raw_event = {
"event": "hard_bounce",
"email": "not-a-user@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
# the leading space in the reason is as received in actual testing:
"reason": " RecipientError: 550 5.5.0"
" Requested action not taken: mailbox unavailable.",
"tag": "header-tag",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.reject_reason, "bounced")
self.assertEqual(
event.mta_response,
" RecipientError: 550 5.5.0"
" Requested action not taken: mailbox unavailable.",
)
self.assertEqual(event.tags, ["header-tag"])
def test_soft_bounce_event(self):
raw_event = {
"event": "soft_bounce",
"email": "recipient@no-mx.example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
"reason": "undefined Unable to find MX of domain no-mx.example.com",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.reject_reason, "bounced")
# no human-readable description consistently available:
self.assertIsNone(event.description)
self.assertEqual(
event.mta_response,
"undefined Unable to find MX of domain no-mx.example.com",
)
def test_blocked(self):
raw_event = {
"event": "blocked",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
"reason": "blocked : due to blacklist user",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "rejected")
self.assertEqual(event.reject_reason, "blocked")
self.assertEqual(event.mta_response, "blocked : due to blacklist user")
def test_spam(self):
# "When a person who received your email reported that it is a spam."
# (haven't observed "spam" event in actual testing; payload below is a guess)
raw_event = {
"event": "spam",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "complained")
self.assertEqual(event.reject_reason, "spam")
def test_complaint(self):
# Sadly, this is not well documented in the official Brevo API documentation.
raw_event = {
"event": "complaint",
"email": "example@domain.com",
"id": "xxxxx",
"date": "2020-10-09 00:00:00",
"ts": 1604933619,
"message-id": "201798300811.5787683@relay.domain.com",
"ts_event": 1604933654,
"X-Mailin-custom": '{"meta": "data"}',
"tags": ["transac_messages"],
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "complained")
self.assertEqual(event.reject_reason, "spam")
def test_invalid_email(self):
# "If a ISP again indicated us that the email is not valid or if we discovered
# that the email is not valid." (unclear whether this error originates with the
# receiving MTA or with Brevo pre-send) (haven't observed "invalid_email"
# event in actual testing; payload below is a guess)
raw_event = {
"event": "invalid_email",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
"reason": "(guessing invalid_email includes a reason)",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.reject_reason, "invalid")
self.assertEqual(
event.mta_response, "(guessing invalid_email includes a reason)"
)
def test_error_email(self):
# Sadly, this is not well documented in the official Brevo API documentation.
raw_event = {
"event": "error",
"email": "example@domain.com",
"id": "xxxxx",
"date": "2020-10-09 00:00:00",
"ts": 1604933619,
"message-id": "201798300811.5787683@relay.domain.com",
"ts_event": 1604933654,
"subject": "My first Transactional",
"X-Mailin-custom": '{"meta": "data"}',
"template_id": 22,
"tags": ["transac_messages"],
"ts_epoch": 1604933623,
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "failed")
self.assertEqual(event.reject_reason, None)
def test_deferred_event(self):
# Note: the example below is an actual event capture (with 'example.com'
# substituted for the real receiving domain). It's pretty clearly a bounce, not
# a deferral. It looks like Brevo mis-categorizes this SMTP response code.
raw_event = {
"event": "deferred",
"email": "notauser@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
"reason": "550 RecipientError: 550 5.1.1 <notauser@example.com>: Recipient"
" address rejected: User unknown in virtual alias table",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "deferred")
# no human-readable description consistently available:
self.assertIsNone(event.description)
self.assertEqual(
event.mta_response,
"550 RecipientError: 550 5.1.1 <notauser@example.com>:"
" Recipient address rejected: User unknown in virtual alias table",
)
def test_opened_event(self):
# Brevo delivers 'unique_opened' only on the first open, and 'opened'
# only on the second or later tracking pixel views. (But they used to deliver
# both on the first open.)
raw_event = {
"event": "opened",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
self.assertIsNone(event.user_agent) # Brevo doesn't report user agent
def test_unique_opened_event(self):
# See note in test_opened_event above
raw_event = {
"event": "unique_opened",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
def test_proxy_open_event(self):
# Equivalent to "Loaded via Proxy" in the Brevo UI.
# This is sent when a tracking pixel is loaded via a 'privacy proxy server'.
# This technique is used by Apple Mail, for example, to protect user's IP
# addresses.
raw_event = {
"event": "proxy_open",
"email": "example@domain.com",
"id": 1,
"date": "2020-10-09 00:00:00",
"message-id": "201798300811.5787683@relay.domain.com",
"subject": "My first Transactional",
"tag": ["transactionalTag"],
"sending_ip": "xxx.xxx.xxx.xxx",
"s_epoch": 1534486682000,
"template_id": 1,
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
def test_unique_proxy_open_event(self):
# Sadly, undocumented in Brevo.
# Equivalent to "First Open but loaded via Proxy".
# This is sent when a tracking pixel is loaded via a 'privacy proxy server'.
# This technique is used by Apple Mail, for example, to protect user's IP
# addresses.
raw_event = {
"event": "unique_proxy_open",
"email": "example@domain.com",
"id": 1,
"date": "2020-10-09 00:00:00",
"message-id": "201798300811.5787683@relay.domain.com",
"subject": "My first Transactional",
"tag": ["transactionalTag"],
"sending_ip": "xxx.xxx.xxx.xxx",
"s_epoch": 1534486682000,
"template_id": 1,
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "opened")
def test_clicked_event(self):
raw_event = {
"event": "click",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
"link": "https://example.com/click/me",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "clicked")
self.assertEqual(event.click_url, "https://example.com/click/me")
self.assertIsNone(event.user_agent) # Brevo doesn't report user agent
def test_unsubscribe(self):
# "When a person unsubscribes from the email received."
# (haven't observed "unsubscribe" event in actual testing;
# payload below is a guess)
raw_event = {
"event": "unsubscribe",
"email": "recipient@example.com",
"ts_epoch": 1520363423000,
"message-id": "<201803011158.9876543210@smtp-relay.mailin.fr>",
}
response = self.client.post(
"/anymail/brevo/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=BrevoTrackingWebhookView,
event=ANY,
esp_name="Brevo",
)
event = kwargs["event"]
self.assertEqual(event.event_type, "unsubscribed")
def test_misconfigured_inbound(self):
errmsg = (
"You seem to have set Brevo's *inbound* webhook URL"
" to Anymail's Brevo *tracking* webhook URL."
)
with self.assertRaisesMessage(AnymailConfigurationError, errmsg):
self.client.post(
"/anymail/brevo/tracking/",
content_type="application/json",
data={"items": []},
)
Reference in New Issue View Git Blame Copy Permalink