mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-20 03:41:05 -05:00
Unisender Go: Fix status tracking webhook and tests.
- Fix signature checking to avoid false validation errors on webhook payloads including `/` (including all "clicked" and most "opened" events). And in general, avoid depending on specific details of Unisender Go's JSON serialization. (Fixes #398.) - Handle "use single event" webhook option (which has a different payload format). - Verify basic auth when Anymail's WEBHOOK_SECRET is used. (This is optional for Unisender Go, since payloads are signed, but it needs to be checked when enabled.) - Treat "soft_bounced" events as "deferred" rather than "bounced", since they will be retried later. - Update validation error to reference Project ID if the webhook is configured for a specific project. - Expose Unisender Go's delivery_status code and unsubscribe form comment as Anymail's normalized event.description. - Update webhook tests based on actual payloads and add several missing tests. - Update docs to clarify webhook use with Unisender Go projects.
This commit is contained in:
Mike Edmunds
@@ -1,177 +1,518 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import hashlib
|
||||
import uuid
|
||||
from datetime import timezone
|
||||
import json
|
||||
from copy import deepcopy
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import ANY
|
||||
|
||||
from django.test import RequestFactory, SimpleTestCase, override_settings, tag
|
||||
from django.test import override_settings, tag
|
||||
|
||||
from anymail.exceptions import AnymailWebhookValidationFailure
|
||||
from anymail.signals import EventType, RejectReason
|
||||
from anymail.exceptions import AnymailConfigurationError
|
||||
from anymail.signals import AnymailTrackingEvent
|
||||
from anymail.webhooks.unisender_go import UnisenderGoTrackingWebhookView
|
||||
|
||||
EVENT_TYPE = EventType.SENT
|
||||
EVENT_TIME = "2015-11-30 15:09:42"
|
||||
EVENT_DATETIME = datetime.datetime(2015, 11, 30, 15, 9, 42, tzinfo=timezone.utc)
|
||||
JOB_ID = "1a3Q2V-0000OZ-S0"
|
||||
DELIVERY_RESPONSE = "550 Spam rejected"
|
||||
UNISENDER_TEST_EMAIL = "recipient.email@example.com"
|
||||
TEST_API_KEY = "api_key"
|
||||
TEST_EMAIL_ID = str(uuid.uuid4())
|
||||
UNISENDER_TEST_DEFAULT_EXAMPLE = {
|
||||
"auth": TEST_API_KEY,
|
||||
"events_by_user": [
|
||||
{
|
||||
"user_id": 456,
|
||||
"project_id": "6432890213745872",
|
||||
"project_name": "MyProject",
|
||||
"events": [
|
||||
{
|
||||
"event_name": "transactional_email_status",
|
||||
"event_data": {
|
||||
"job_id": JOB_ID,
|
||||
"metadata": {"key1": "val1", "anymail_id": TEST_EMAIL_ID},
|
||||
"email": UNISENDER_TEST_EMAIL,
|
||||
"status": EVENT_TYPE,
|
||||
"event_time": EVENT_TIME,
|
||||
"url": "http://some.url.com",
|
||||
"delivery_info": {
|
||||
"delivery_status": "err_delivery_failed",
|
||||
"destination_response": DELIVERY_RESPONSE,
|
||||
"user_agent": (
|
||||
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36"
|
||||
),
|
||||
"ip": "111.111.111.111",
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
"event_name": "transactional_spam_block",
|
||||
"event_data": {
|
||||
"block_time": "YYYY-MM-DD HH:MM:SS",
|
||||
"block_type": "one_smtp",
|
||||
"domain": "domain_name",
|
||||
"SMTP_blocks_count": 8,
|
||||
"domain_status": "blocked",
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
EXAMPLE_WITHOUT_DELIVERY_INFO = {
|
||||
"auth": "",
|
||||
"events_by_user": [
|
||||
{
|
||||
"events": [
|
||||
{
|
||||
"event_name": "transactional_email_status",
|
||||
"event_data": {
|
||||
"job_id": JOB_ID,
|
||||
"metadata": {},
|
||||
"email": UNISENDER_TEST_EMAIL,
|
||||
"status": EVENT_TYPE,
|
||||
"event_time": EVENT_TIME,
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
}
|
||||
REQUEST_JSON = '{"auth":"api_key","key":"value"}'
|
||||
REQUEST_JSON_MD5 = "8c64386327f53722434f44021a7a0d40" # md5 hash of REQUEST_JSON
|
||||
REQUEST_DATA_AUTH = {"auth": REQUEST_JSON_MD5, "key": "value"}
|
||||
from .utils import test_file_content
|
||||
from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase
|
||||
|
||||
TEST_API_KEY = "TEST_API_KEY"
|
||||
|
||||
|
||||
def _request_json_to_dict_with_hashed_key(request_json: bytes) -> dict[str, str]:
|
||||
new_auth = hashlib.md5(request_json).hexdigest()
|
||||
return {"auth": new_auth, "key": "value"}
|
||||
def unisender_go_signed_payload(
|
||||
data: dict, api_key: str, json_options: dict | None = None
|
||||
) -> bytes:
|
||||
"""
|
||||
Return data serialized to JSON and signed with api_key, using Unisender Go's
|
||||
webhook signature in the top-level "auth" field:
|
||||
|
||||
"MD5 hash of the string body of the message, with the auth value replaced
|
||||
by the api_key of the user/project whose handler is being called."
|
||||
https://godocs.unisender.ru/web-api-ref#callback-format
|
||||
|
||||
Any json_options are passed to json.dumps as kwargs.
|
||||
|
||||
This modifies data to add the "auth" field.
|
||||
"""
|
||||
json_options = json_options or {"separators": (",", ":")}
|
||||
placeholder = "__PLACEHOLDER_FOR_SIGNATURE__"
|
||||
data["auth"] = placeholder
|
||||
serialized_data = json.dumps(data, **json_options)
|
||||
signature = hashlib.md5(
|
||||
serialized_data.replace(placeholder, api_key).encode()
|
||||
).hexdigest()
|
||||
signed_data = serialized_data.replace(placeholder, signature)
|
||||
data["auth"] = signature # make available to the caller
|
||||
return signed_data.encode()
|
||||
|
||||
|
||||
class UnisenderGoWebhookTestCase(WebhookTestCase):
|
||||
def client_post_signed(
|
||||
self,
|
||||
url: str,
|
||||
data: dict,
|
||||
api_key: str = TEST_API_KEY,
|
||||
json_options: dict | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
"""
|
||||
Return self.client.post(url, serialized json_data) signed with api_key
|
||||
using json_options.
|
||||
|
||||
Additional kwargs are passed to self.client.post()
|
||||
"""
|
||||
signed_data = unisender_go_signed_payload(data, api_key, json_options)
|
||||
return self.client.post(
|
||||
url, content_type="application/json", data=signed_data, **kwargs
|
||||
)
|
||||
|
||||
|
||||
@tag("unisender_go")
|
||||
class TestUnisenderGoWebhooks(SimpleTestCase):
|
||||
class UnisenderGoWebhookSettingsTestCase(UnisenderGoWebhookTestCase):
|
||||
def test_requires_api_key(self):
|
||||
with self.assertRaisesMessage(
|
||||
AnymailConfigurationError, "UNISENDER_GO_API_KEY"
|
||||
):
|
||||
self.client_post_signed("/anymail/unisender_go/tracking/", {})
|
||||
|
||||
@override_settings(ANYMAIL={"UNISENDER_GO_API_KEY": "SETTINGS_API_KEY"})
|
||||
def test_view_params_api_key_override(self):
|
||||
"""Webhook api_key can be provided as a view param"""
|
||||
view = UnisenderGoTrackingWebhookView.as_view(api_key="VIEW_API_KEY")
|
||||
view_instance = view.view_class(**view.view_initkwargs)
|
||||
self.assertEqual(view_instance.api_key_bytes, b"VIEW_API_KEY")
|
||||
|
||||
|
||||
@tag("unisender_go")
|
||||
@override_settings(
|
||||
# (Use expanded setting name because WebhookBasicAuthTestCase sets ANYMAIL={}.)
|
||||
ANYMAIL_UNISENDER_GO_API_KEY=TEST_API_KEY,
|
||||
)
|
||||
class UnisenderGoWebhookSecurityTestCase(
|
||||
UnisenderGoWebhookTestCase, WebhookBasicAuthTestCase
|
||||
):
|
||||
should_warn_if_no_auth = False # because we check webhook signature
|
||||
|
||||
payload = {
|
||||
# "auth" is added by self.client_post_signed()
|
||||
"events_by_user": [
|
||||
{
|
||||
"user_id": 123456,
|
||||
"events": [
|
||||
{
|
||||
"event_name": "transactional_email_status",
|
||||
"event_data": {
|
||||
"job_id": "1sn15Z-0007Le-GtVN",
|
||||
"email": "test@example.com",
|
||||
"status": "sent",
|
||||
"metadata": {"can_be_unicode": "Метаданные"},
|
||||
"event_time": "2024-09-07 19:27:17",
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
def call_webhook(self):
|
||||
return self.client_post_signed("/anymail/unisender_go/tracking/", self.payload)
|
||||
|
||||
# Additional tests are in WebhookBasicAuthTestCase
|
||||
|
||||
def test_verifies_correct_signature(self):
|
||||
response = self.client_post_signed(
|
||||
"/anymail/unisender_go/tracking/", self.payload
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_rejects_bad_signature(self):
|
||||
# This also verifies that the error log references the correct setting to check.
|
||||
with self.assertLogs() as logs:
|
||||
response = self.client_post_signed(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
self.payload,
|
||||
api_key="OTHER_API_KEY",
|
||||
)
|
||||
# SuspiciousOperation causes 400 response (even in test client):
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn("check Anymail UNISENDER_GO_API_KEY", logs.output[0])
|
||||
self.assertNotIn("Project ID", logs.output[0])
|
||||
|
||||
def test_rejects_missing_signature(self):
|
||||
payload = deepcopy(self.payload)
|
||||
del payload["auth"]
|
||||
# Post directly (without signing to add auth):
|
||||
response = self.client.post(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
content_type="application/json",
|
||||
data=json.dumps(payload),
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_rejects_problem_signatures(self):
|
||||
# Make sure our `body.replace(auth, key)` approach can't be confused
|
||||
# by invalid payloads.
|
||||
for bad_auth in ["", " ", ":", "{", '"', 0, None, [], {}]:
|
||||
with self.subTest(auth=bad_auth):
|
||||
payload = deepcopy(self.payload)
|
||||
payload["auth"] = bad_auth
|
||||
response = self.client.post(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
content_type="application/json",
|
||||
data=json.dumps(payload),
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_error_includes_project_id(self):
|
||||
# If the webhook has a selected project, mention
|
||||
# its id in the validation error to assist in debugging.
|
||||
payload = deepcopy(self.payload)
|
||||
payload["events_by_user"][0].update(
|
||||
{"project_id": 999999, "project_name": "Test project"}
|
||||
)
|
||||
with self.assertLogs() as logs:
|
||||
response = self.client_post_signed(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
payload,
|
||||
api_key="OTHER_API_KEY",
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn(
|
||||
"check Anymail UNISENDER_GO_API_KEY setting is for Project ID 999999",
|
||||
logs.output[0],
|
||||
)
|
||||
|
||||
def test_error_includes_project_id_single_event(self):
|
||||
# Selected project works with "single event" option.
|
||||
payload = {
|
||||
"user_id": 123456,
|
||||
"project_id": 999999,
|
||||
"project_name": "Test project",
|
||||
"event_name": "transactional_email_status",
|
||||
"job_id": "1sn15Z-0007Le-GtVN",
|
||||
"email": "test@example.com",
|
||||
"status": "sent",
|
||||
"event_time": "2024-09-07 19:27:17",
|
||||
}
|
||||
with self.assertLogs() as logs:
|
||||
response = self.client_post_signed(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
payload,
|
||||
api_key="OTHER_API_KEY",
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn(
|
||||
"check Anymail UNISENDER_GO_API_KEY setting is for Project ID 999999",
|
||||
logs.output[0],
|
||||
)
|
||||
|
||||
def test_insensitive_to_json_serialization_options(self):
|
||||
# Our webhook signature verification must not depend on the exact
|
||||
# details of how Unisender Go serializes the JSON payload.
|
||||
for json_options in [
|
||||
{"separators": None},
|
||||
{"ensure_ascii": False},
|
||||
{"indent": 4},
|
||||
{"sort_keys": True},
|
||||
]:
|
||||
with self.subTest(options=json_options):
|
||||
response = self.client_post_signed(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
self.payload,
|
||||
json_options=json_options,
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@override_settings(
|
||||
# noqa: secret-scanning: this API key has been disabled
|
||||
ANYMAIL={"UNISENDER_GO_API_KEY": "6mjstx9gwi7qj8eni6m77hfiiw6aifmss154y4ze"}
|
||||
)
|
||||
def test_actual_signed_payload(self):
|
||||
# Test our signature verification using an actual payload and API key.
|
||||
payload = test_file_content("unisender-go-tracking-test-payload.json.raw")
|
||||
# (If an editor or pre-commit forces a trailing newline, the test breaks.)
|
||||
assert payload[-1] != b"\n", "Test payload must not have end-of-file newline"
|
||||
response = self.client.post(
|
||||
"/anymail/unisender_go/tracking/",
|
||||
content_type="application/json",
|
||||
data=payload,
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
|
||||
@tag("unisender_go")
|
||||
@override_settings(ANYMAIL={"UNISENDER_GO_API_KEY": TEST_API_KEY})
|
||||
class UnisenderGoTestCase(UnisenderGoWebhookTestCase):
|
||||
# Most of these tests use Unisender Go's "single event" option for brevity.
|
||||
# Anymail also supports (and recommends) multiple event webhook option;
|
||||
# tests for that are toward the end.
|
||||
|
||||
def test_sent_event(self):
|
||||
request = RequestFactory().post(
|
||||
path="/",
|
||||
data=UNISENDER_TEST_DEFAULT_EXAMPLE,
|
||||
content_type="application/json",
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"user_id": 111111,
|
||||
"project_id": 999999,
|
||||
"project_name": "Testing",
|
||||
"job_id": "1smi9f-00057m-86zr",
|
||||
"metadata": {
|
||||
"anymail_id": "00001111-2222-3333-4444-555566667777",
|
||||
"cohort": "group a121",
|
||||
},
|
||||
"email": "recipient@example.com",
|
||||
"status": "sent",
|
||||
"event_time": "2024-09-06 23:14:19",
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
view = UnisenderGoTrackingWebhookView()
|
||||
|
||||
events = view.parse_events(request)
|
||||
event = events[0]
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
self.assertEqual(event.event_type, EVENT_TYPE)
|
||||
self.assertEqual(event.timestamp, EVENT_DATETIME)
|
||||
event = kwargs["event"]
|
||||
self.assertIsInstance(event, AnymailTrackingEvent)
|
||||
self.assertEqual(event.event_type, "sent")
|
||||
self.assertEqual(
|
||||
event.timestamp,
|
||||
datetime(2024, 9, 6, 23, 14, 19, tzinfo=timezone.utc),
|
||||
)
|
||||
# event.message_id matches the message.anymail_status.message_id
|
||||
# from when the message was sent. It comes from metadata.anymail_id
|
||||
# if present (added by UNISENDER_GO_GENERATE_MESSAGE_ID).
|
||||
self.assertEqual(event.message_id, "00001111-2222-3333-4444-555566667777")
|
||||
# Unisender Go does not include a useful event_id
|
||||
self.assertIsNone(event.event_id)
|
||||
self.assertEqual(event.recipient, UNISENDER_TEST_EMAIL)
|
||||
self.assertEqual(event.reject_reason, RejectReason.OTHER)
|
||||
self.assertEqual(event.mta_response, DELIVERY_RESPONSE)
|
||||
self.assertDictEqual(event.metadata, {"key1": "val1"})
|
||||
self.assertEqual(event.recipient, "recipient@example.com")
|
||||
# Although Unisender Go's email-send docs claim tags are sent to webhooks,
|
||||
# its webhook docs don't show tags (and they aren't actually sent 9/2024).
|
||||
# self.assertEqual(event.tags, ["tag1", "Tag 2"])
|
||||
# Our added "anymail_id" should be removed from metadata.
|
||||
self.assertEqual(event.metadata, {"cohort": "group a121"})
|
||||
self.assertEqual(event.esp_event, raw_event)
|
||||
|
||||
def test_without_delivery_info(self):
|
||||
request = RequestFactory().post(
|
||||
path="/",
|
||||
data=EXAMPLE_WITHOUT_DELIVERY_INFO,
|
||||
content_type="application/json",
|
||||
def test_delivered_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"user_id": 111111,
|
||||
"job_id": "1smi9f-00057m-86zr",
|
||||
"email": "recipient@example.com",
|
||||
"status": "delivered",
|
||||
"event_time": "2024-09-06 23:14:24",
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
view = UnisenderGoTrackingWebhookView()
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "delivered")
|
||||
# If UNISENDER_GO_GENERATE_MESSAGE_ID not enabled, Unisender Go's
|
||||
# job_id becomes Anymail's message_id.
|
||||
self.assertEqual(event.message_id, "1smi9f-00057m-86zr")
|
||||
self.assertEqual(event.recipient, "recipient@example.com")
|
||||
|
||||
events = view.parse_events(request)
|
||||
|
||||
self.assertEqual(len(events), 1)
|
||||
# Without metadata["anymail_id"], message_id uses the job_id.
|
||||
# (This covers messages sent with "UNISENDER_GO_GENERATE_MESSAGE_ID": False.)
|
||||
self.assertEqual(events[0].message_id, JOB_ID)
|
||||
|
||||
@override_settings(ANYMAIL_UNISENDER_GO_API_KEY=TEST_API_KEY)
|
||||
def test_check_authorization(self):
|
||||
"""Asserts that nothing is failing"""
|
||||
request_data = _request_json_to_dict_with_hashed_key(
|
||||
b'{"auth":"api_key","key":"value"}',
|
||||
def test_hard_bounced_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"status": "hard_bounced",
|
||||
"email": "bounce@example.com",
|
||||
"delivery_info": {
|
||||
"delivery_status": "err_user_unknown",
|
||||
"destination_response": "555 5.7.1 User unknown 'bounce@example.com'.",
|
||||
},
|
||||
"event_time": "2024-09-06 23:22:40",
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
request = RequestFactory().post(
|
||||
path="/", data=request_data, content_type="application/json"
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "bounced")
|
||||
self.assertEqual(event.recipient, "bounce@example.com")
|
||||
self.assertEqual(event.reject_reason, "bounced")
|
||||
self.assertEqual(event.description, "err_user_unknown")
|
||||
self.assertEqual(
|
||||
event.mta_response, "555 5.7.1 User unknown 'bounce@example.com'."
|
||||
)
|
||||
view = UnisenderGoTrackingWebhookView()
|
||||
|
||||
view.validate_request(request)
|
||||
|
||||
@override_settings(ANYMAIL_UNISENDER_GO_API_KEY=TEST_API_KEY)
|
||||
def test_check_authorization__fail__ordinar_quoters(self):
|
||||
request_json = b"{'auth':'api_key','key':'value'}"
|
||||
request_data = _request_json_to_dict_with_hashed_key(request_json)
|
||||
request = RequestFactory().post(
|
||||
path="/", data=request_data, content_type="application/json"
|
||||
def test_soft_bounced_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"status": "soft_bounced",
|
||||
"email": "full@example.com",
|
||||
"delivery_info": {
|
||||
"delivery_status": "err_mailbox_full",
|
||||
"destination_response": "554 5.2.2 Mailbox full 'full@example.com'.",
|
||||
},
|
||||
"event_time": "2024-09-06 23:22:40",
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
view = UnisenderGoTrackingWebhookView()
|
||||
|
||||
with self.assertRaises(AnymailWebhookValidationFailure):
|
||||
view.validate_request(request)
|
||||
|
||||
@override_settings(ANYMAIL_UNISENDER_GO_API_KEY=TEST_API_KEY)
|
||||
def test_check_authorization__fail__spaces_after_semicolon(self):
|
||||
request_json = b'{"auth": "api_key","key": "value"}'
|
||||
request_data = _request_json_to_dict_with_hashed_key(request_json)
|
||||
request = RequestFactory().post(
|
||||
path="/", data=request_data, content_type="application/json"
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "deferred")
|
||||
self.assertEqual(event.recipient, "full@example.com")
|
||||
self.assertEqual(event.reject_reason, "bounced")
|
||||
self.assertEqual(event.description, "err_mailbox_full")
|
||||
self.assertEqual(
|
||||
event.mta_response, "554 5.2.2 Mailbox full 'full@example.com'."
|
||||
)
|
||||
view = UnisenderGoTrackingWebhookView()
|
||||
|
||||
with self.assertRaises(AnymailWebhookValidationFailure):
|
||||
view.validate_request(request)
|
||||
|
||||
@override_settings(ANYMAIL_UNISENDER_GO_API_KEY=TEST_API_KEY)
|
||||
def test_check_authorization__fail__spaces_after_comma(self):
|
||||
request_json = b'{"auth":"api_key", "key":"value"}'
|
||||
request_data = _request_json_to_dict_with_hashed_key(request_json)
|
||||
request = RequestFactory().post(
|
||||
path="/", data=request_data, content_type="application/json"
|
||||
def test_spam_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"status": "spam",
|
||||
"email": "to@example.com",
|
||||
"delivery_info": {
|
||||
"delivery_status": "err_spam_rejected",
|
||||
"destination_response": "550 Spam rejected",
|
||||
},
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
view = UnisenderGoTrackingWebhookView()
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "complained")
|
||||
self.assertEqual(event.recipient, "to@example.com")
|
||||
self.assertEqual(event.description, "err_spam_rejected")
|
||||
self.assertEqual(event.mta_response, "550 Spam rejected")
|
||||
|
||||
with self.assertRaises(AnymailWebhookValidationFailure):
|
||||
view.validate_request(request)
|
||||
def test_unsubscribed_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"status": "unsubscribed",
|
||||
"email": "to@example.com",
|
||||
"comment": "From unsubscribe page 'comment' field",
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "unsubscribed")
|
||||
self.assertEqual(event.recipient, "to@example.com")
|
||||
self.assertEqual(event.description, "From unsubscribe page 'comment' field")
|
||||
|
||||
def test_opened_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"status": "opened",
|
||||
"email": "to@example.com",
|
||||
"delivery_info": {
|
||||
"user_agent": "... via ggpht.com GoogleImageProxy",
|
||||
"ip": "10.10.1.333",
|
||||
},
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "opened")
|
||||
self.assertEqual(event.recipient, "to@example.com")
|
||||
self.assertEqual(event.user_agent, "... via ggpht.com GoogleImageProxy")
|
||||
|
||||
def test_clicked_event(self):
|
||||
raw_event = {
|
||||
"event_name": "transactional_email_status",
|
||||
"status": "clicked",
|
||||
"email": "to@example.com",
|
||||
"url": "https://example.com",
|
||||
"delivery_info": {
|
||||
"user_agent": "Mozilla/5.0 AppleWebKit/537.36 ...",
|
||||
"ip": "192.168.1.333",
|
||||
},
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
kwargs = self.assert_handler_called_once_with(
|
||||
self.tracking_handler,
|
||||
sender=UnisenderGoTrackingWebhookView,
|
||||
event=ANY,
|
||||
esp_name="Unisender Go",
|
||||
)
|
||||
event = kwargs["event"]
|
||||
self.assertEqual(event.event_type, "clicked")
|
||||
self.assertEqual(event.recipient, "to@example.com")
|
||||
self.assertEqual(event.click_url, "https://example.com")
|
||||
self.assertEqual(event.user_agent, "Mozilla/5.0 AppleWebKit/537.36 ...")
|
||||
|
||||
def test_multiple_event_option(self):
|
||||
# Payload format is different when "Use single event" not checked.
|
||||
raw_event = {
|
||||
"events_by_user": [
|
||||
{
|
||||
"user_id": 111111,
|
||||
"events": [
|
||||
{
|
||||
"event_name": "transactional_email_status",
|
||||
"event_data": {
|
||||
"job_id": "1sn15Z-0007Le-GtVN",
|
||||
"email": "to@example.com",
|
||||
"status": "sent",
|
||||
"event_time": "2024-09-07 19:27:17",
|
||||
},
|
||||
},
|
||||
{
|
||||
"event_name": "transactional_email_status",
|
||||
"event_data": {
|
||||
"job_id": "1sn15Z-0007Le-GtVN",
|
||||
"email": "cc@example.com",
|
||||
"status": "delivered",
|
||||
"event_time": "2024-09-07 19:27:17",
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
response = self.client_post_signed("/anymail/unisender_go/tracking/", raw_event)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(self.tracking_handler.call_count, 2)
|
||||
events = [
|
||||
kwargs["event"] for (args, kwargs) in self.tracking_handler.call_args_list
|
||||
]
|
||||
self.assertEqual(events[0].event_type, "sent")
|
||||
self.assertEqual(events[0].recipient, "to@example.com")
|
||||
self.assertEqual(events[1].event_type, "delivered")
|
||||
self.assertEqual(events[1].recipient, "cc@example.com")
|
||||
|
||||
# esp_event is event_data for each event
|
||||
self.assertEqual(
|
||||
events[0].esp_event,
|
||||
raw_event["events_by_user"][0]["events"][0]["event_data"],
|
||||
)
|
||||
self.assertEqual(
|
||||
events[1].esp_event,
|
||||
raw_event["events_by_user"][0]["events"][1]["event_data"],
|
||||
)
|
||||
|
||||
def test_webhook_setup_verification(self):
|
||||
# Unisender Go verifies webhook at setup time by calling GET.
|
||||
response = self.client.get("/anymail/unisender_go/tracking/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
Reference in New Issue
Block a user