westfarn
f7f3174e1c
SMTP2GO open/click webhooks now advance Message to opened/clicked, and the campaign engagement chart uses live stats instead of placeholder bars.
1079 lines
39 KiB
Python
1079 lines
39 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.test import Client, TestCase
|
|
from django.urls import reverse
|
|
|
|
from contacts.models import Channel, ConsentRecord, Contact, Suppression
|
|
from messaging.models import Campaign, Message, MessageTemplate, ProviderEvent
|
|
from messaging.services import (
|
|
channel_preferences,
|
|
contact_may_receive,
|
|
create_campaign_draft,
|
|
make_unsubscribe_token,
|
|
opted_in_contacts,
|
|
set_channel_consent,
|
|
)
|
|
|
|
|
|
class PreferenceCenterTests(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
self.contact = Contact.objects.create(
|
|
email="jordan.lee@example.com",
|
|
phone="5550184420",
|
|
first_name="Jordan",
|
|
last_name="Lee",
|
|
)
|
|
for channel in Channel:
|
|
set_channel_consent(
|
|
self.contact, channel.value, opted_in=True, reason="test_seed"
|
|
)
|
|
self.token = make_unsubscribe_token(str(self.contact.pk), Channel.EMAIL)
|
|
|
|
def test_preferences_get_shows_form(self):
|
|
url = reverse("public:unsubscribe", kwargs={"token": self.token})
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Email marketing")
|
|
self.assertContains(response, "SMS updates")
|
|
self.assertContains(response, "Postcard mailings")
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.EMAIL))
|
|
|
|
def test_invalid_token(self):
|
|
url = reverse("public:unsubscribe", kwargs={"token": "not-a-valid-token"})
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "invalid or has expired")
|
|
|
|
def test_one_click_query_opts_out_email_only(self):
|
|
url = reverse("public:unsubscribe", kwargs={"token": self.token})
|
|
response = self.client.get(f"{url}?one_click=1")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.contact.refresh_from_db()
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.EMAIL))
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.SMS))
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.POSTCARD))
|
|
|
|
def test_one_click_endpoint(self):
|
|
url = reverse(
|
|
"public:unsubscribe_one_click", kwargs={"token": self.token}
|
|
)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.EMAIL))
|
|
|
|
def test_one_click_post_rfc8058(self):
|
|
url = reverse(
|
|
"public:unsubscribe_one_click", kwargs={"token": self.token}
|
|
)
|
|
response = self.client.post(
|
|
url, {"List-Unsubscribe": "One-Click"}
|
|
)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.EMAIL))
|
|
|
|
def test_save_preferences_partial_opt_out(self):
|
|
url = reverse("public:unsubscribe", kwargs={"token": self.token})
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"action": "save",
|
|
"consent_email": "1",
|
|
# SMS unchecked
|
|
"consent_postcard": "1",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.EMAIL))
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.SMS))
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.POSTCARD))
|
|
sms_sup = Suppression.objects.get(
|
|
contact=self.contact, channel=Channel.SMS
|
|
)
|
|
self.assertTrue(sms_sup.active)
|
|
|
|
def test_unsubscribe_all(self):
|
|
url = reverse("public:unsubscribe", kwargs={"token": self.token})
|
|
response = self.client.post(url, {"action": "unsubscribe_all"})
|
|
self.assertEqual(response.status_code, 302)
|
|
for channel in Channel:
|
|
self.assertFalse(contact_may_receive(self.contact, channel.value))
|
|
|
|
|
|
class SmsStopWebhookTests(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
self.contact = Contact.objects.create(
|
|
email="avery@example.com",
|
|
phone="5550142291",
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.SMS, opted_in=True, reason="test"
|
|
)
|
|
|
|
def test_stop_webhook(self):
|
|
url = reverse("messaging:sms_webhook")
|
|
response = self.client.post(
|
|
url, {"from": "5550142291", "text": "STOP"}
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.SMS))
|
|
|
|
|
|
class PortalConsentToggleTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="monica", password="test-pass-123"
|
|
)
|
|
self.client = Client()
|
|
self.client.login(username="monica", password="test-pass-123")
|
|
self.contact = Contact.objects.create(
|
|
email="sam@example.com",
|
|
first_name="Sam",
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.EMAIL, opted_in=True, reason="test"
|
|
)
|
|
|
|
def test_portal_can_toggle_consent(self):
|
|
url = reverse("contacts:detail", kwargs={"pk": self.contact.pk})
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"first_name": "Sam",
|
|
"last_name": "",
|
|
"email": "sam@example.com",
|
|
"phone": "",
|
|
"notes": "updated",
|
|
"consent_sms": "1",
|
|
"consent_postcard": "1",
|
|
# email unchecked → opt out
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.EMAIL))
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.SMS))
|
|
self.assertTrue(contact_may_receive(self.contact, Channel.POSTCARD))
|
|
email_consent = ConsentRecord.objects.get(
|
|
contact=self.contact, channel=Channel.EMAIL
|
|
)
|
|
self.assertEqual(email_consent.reason, "portal_manual")
|
|
|
|
def test_portal_can_update_profile(self):
|
|
url = reverse("contacts:detail", kwargs={"pk": self.contact.pk})
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"first_name": "Samantha",
|
|
"last_name": "Lee",
|
|
"email": "sam.lee@example.com",
|
|
"phone": "6305550100",
|
|
"address_line1": "10 Main St",
|
|
"address_city": "Naperville",
|
|
"address_state": "IL",
|
|
"address_zip": "60540",
|
|
"notes": "VIP",
|
|
"consent_email": "1",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.contact.refresh_from_db()
|
|
self.assertEqual(self.contact.first_name, "Samantha")
|
|
self.assertEqual(self.contact.last_name, "Lee")
|
|
self.assertEqual(self.contact.email, "sam.lee@example.com")
|
|
self.assertEqual(self.contact.phone, "6305550100")
|
|
self.assertEqual(self.contact.postal_address.get("line1"), "10 Main St")
|
|
self.assertEqual(self.contact.notes, "VIP")
|
|
|
|
|
|
class CampaignDraftSaveTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="composer", password="test-pass-123"
|
|
)
|
|
self.client = Client()
|
|
self.client.login(username="composer", password="test-pass-123")
|
|
self.contact = Contact.objects.create(
|
|
email="pat@example.com",
|
|
first_name="Pat",
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.EMAIL, opted_in=True, reason="test"
|
|
)
|
|
|
|
def test_save_draft_creates_campaign_and_messages(self):
|
|
url = reverse("messaging:campaign_list")
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"name": "Spring tips",
|
|
"subject": "Hello sellers",
|
|
"body": "Hi {first_name}",
|
|
"audience": Campaign.Audience.EMAIL_OPT_IN,
|
|
"scheduled_for": "2026-08-10T09:30",
|
|
},
|
|
)
|
|
campaign = Campaign.objects.get(name="Spring tips")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(
|
|
response.url,
|
|
reverse("messaging:campaign_detail", kwargs={"pk": campaign.pk}),
|
|
)
|
|
self.assertEqual(campaign.status, Campaign.Status.DRAFT)
|
|
self.assertEqual(campaign.channel, Channel.EMAIL)
|
|
self.assertEqual(campaign.audience, Campaign.Audience.EMAIL_OPT_IN)
|
|
self.assertIsNotNone(campaign.scheduled_for)
|
|
self.assertEqual(campaign.messages.count(), 1)
|
|
msg = campaign.messages.get()
|
|
self.assertEqual(msg.contact, self.contact)
|
|
self.assertEqual(msg.status, Message.Status.DRAFT)
|
|
|
|
def test_save_draft_requires_subject_for_email(self):
|
|
url = reverse("messaging:campaign_list")
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"name": "No subject",
|
|
"subject": "",
|
|
"body": "Body only",
|
|
"audience": Campaign.Audience.EMAIL_OPT_IN,
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Subject is required")
|
|
self.assertFalse(Campaign.objects.filter(name="No subject").exists())
|
|
|
|
def test_create_campaign_draft_helper(self):
|
|
campaign = create_campaign_draft(
|
|
name="Helper draft",
|
|
audience=Campaign.Audience.EMAIL_OPT_IN,
|
|
subject="Subj",
|
|
body="Body",
|
|
created_by=self.user,
|
|
)
|
|
self.assertEqual(campaign.messages.count(), 1)
|
|
self.assertEqual(campaign.created_by, self.user)
|
|
|
|
|
|
class CampaignSendTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="sender", password="test-pass-123", email="sender@example.com"
|
|
)
|
|
self.client = Client()
|
|
self.client.login(username="sender", password="test-pass-123")
|
|
self.contact = Contact.objects.create(
|
|
email="pat@example.com",
|
|
first_name="Pat",
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.EMAIL, opted_in=True, reason="test"
|
|
)
|
|
self.campaign = create_campaign_draft(
|
|
name="Send me",
|
|
audience=Campaign.Audience.EMAIL_OPT_IN,
|
|
subject="Hello",
|
|
body="Body text",
|
|
created_by=self.user,
|
|
)
|
|
|
|
def test_detail_shows_send_controls(self):
|
|
url = reverse(
|
|
"messaging:campaign_detail", kwargs={"pk": self.campaign.pk}
|
|
)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, "Send test email")
|
|
self.assertContains(response, "Send now to recipients")
|
|
|
|
def test_test_send_uses_locmem(self):
|
|
from django.core import mail
|
|
|
|
url = reverse(
|
|
"messaging:campaign_test_send", kwargs={"pk": self.campaign.pk}
|
|
)
|
|
with self.settings(
|
|
EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
|
|
DEFAULT_FROM_EMAIL="noreply@example.com",
|
|
):
|
|
response = self.client.post(url, {"test_email": "me@example.com"})
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
self.assertEqual(mail.outbox[0].to, ["me@example.com"])
|
|
self.assertTrue(mail.outbox[0].subject.startswith("[TEST]"))
|
|
self.assertTrue(mail.outbox[0].alternatives)
|
|
self.assertEqual(mail.outbox[0].alternatives[0][1], "text/html")
|
|
# Recipients untouched
|
|
self.assertEqual(
|
|
self.campaign.messages.filter(status=Message.Status.DRAFT).count(), 1
|
|
)
|
|
|
|
def test_send_now_marks_messages_sent(self):
|
|
from django.core import mail
|
|
|
|
url = reverse(
|
|
"messaging:campaign_send", kwargs={"pk": self.campaign.pk}
|
|
)
|
|
with self.settings(
|
|
EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
|
|
DEFAULT_FROM_EMAIL="noreply@example.com",
|
|
PUBLIC_SITE_URL="http://testserver",
|
|
):
|
|
response = self.client.post(url)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.campaign.refresh_from_db()
|
|
self.assertEqual(self.campaign.status, Campaign.Status.COMPLETED)
|
|
self.assertEqual(
|
|
self.campaign.messages.filter(status=Message.Status.SENT).count(), 1
|
|
)
|
|
# Recipient campaign email + one realtor completion summary.
|
|
self.assertEqual(len(mail.outbox), 2)
|
|
self.assertIsNotNone(self.campaign.notify_sent_at)
|
|
summary = mail.outbox[1]
|
|
self.assertIn("Campaign sent:", summary.subject)
|
|
self.assertEqual(summary.to, [self.user.email])
|
|
self.assertTrue(mail.outbox[0].alternatives)
|
|
self.assertEqual(mail.outbox[0].alternatives[0][1], "text/html")
|
|
self.assertIn("#00626c", mail.outbox[0].alternatives[0][0])
|
|
self.assertTrue(summary.alternatives)
|
|
|
|
|
|
class Smtp2goEmailWebhookTests(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
self.contact = Contact.objects.create(
|
|
email="pat@example.com",
|
|
first_name="Pat",
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.EMAIL, opted_in=True, reason="test"
|
|
)
|
|
self.campaign = create_campaign_draft(
|
|
name="Webhook campaign",
|
|
audience=Campaign.Audience.EMAIL_OPT_IN,
|
|
subject="Hello",
|
|
body="Body",
|
|
)
|
|
self.message = self.campaign.messages.get()
|
|
self.message.status = Message.Status.SENT
|
|
self.message.provider_message_id = f"smtp-{self.message.pk}"
|
|
self.message.save()
|
|
|
|
def test_delivered_updates_status_via_monica_header(self):
|
|
url = reverse("messaging:email_webhook")
|
|
response = self.client.post(
|
|
url,
|
|
data={
|
|
"event": "delivered",
|
|
"rcpt": "pat@example.com",
|
|
"email_id": "smtp2go-abc-123",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.DELIVERED)
|
|
self.assertEqual(self.message.provider_message_id, "smtp2go-abc-123")
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, event_type="delivered"
|
|
).exists()
|
|
)
|
|
|
|
def test_open_records_event_sets_opened(self):
|
|
self.message.status = Message.Status.DELIVERED
|
|
self.message.save(update_fields=["status"])
|
|
url = reverse("messaging:email_webhook")
|
|
self.client.post(
|
|
url,
|
|
data={
|
|
"event": "open",
|
|
"rcpt": "pat@example.com",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.OPENED)
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, event_type="open"
|
|
).exists()
|
|
)
|
|
|
|
def test_click_sets_clicked_from_sent(self):
|
|
url = reverse("messaging:email_webhook")
|
|
self.client.post(
|
|
url,
|
|
data={
|
|
"event": "click",
|
|
"rcpt": "pat@example.com",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.CLICKED)
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, event_type="click"
|
|
).exists()
|
|
)
|
|
|
|
def test_open_does_not_downgrade_clicked(self):
|
|
self.message.status = Message.Status.CLICKED
|
|
self.message.save(update_fields=["status"])
|
|
url = reverse("messaging:email_webhook")
|
|
self.client.post(
|
|
url,
|
|
data={
|
|
"event": "open",
|
|
"rcpt": "pat@example.com",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.CLICKED)
|
|
|
|
def test_hard_bounce_suppresses_contact(self):
|
|
url = reverse("messaging:email_webhook")
|
|
self.client.post(
|
|
url,
|
|
data={
|
|
"event": "bounce",
|
|
"bounce": "hard",
|
|
"rcpt": "pat@example.com",
|
|
"message": "550 user unknown",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.BOUNCED)
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.EMAIL))
|
|
|
|
def test_webhook_secret_required_when_configured(self):
|
|
url = reverse("messaging:email_webhook")
|
|
with self.settings(SMTP2GO_WEBHOOK_SECRET="s3cret"):
|
|
denied = self.client.post(url, data={"event": "delivered"})
|
|
self.assertEqual(denied.status_code, 403)
|
|
ok = self.client.post(
|
|
f"{url}?token=s3cret",
|
|
data={
|
|
"event": "delivered",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
|
|
def test_status_json_includes_opens(self):
|
|
User = get_user_model()
|
|
user = User.objects.create_user(username="viewer", password="test-pass-123")
|
|
self.client.login(username="viewer", password="test-pass-123")
|
|
ProviderEvent.objects.create(
|
|
message=self.message,
|
|
provider="smtp2go_email",
|
|
event_type="open",
|
|
payload={"event": "open"},
|
|
)
|
|
url = reverse(
|
|
"messaging:campaign_status_json", kwargs={"pk": self.campaign.pk}
|
|
)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data["stats"]["opens"], 1)
|
|
self.assertIn("chart_bars", data["stats"])
|
|
labels = [bar["label"] for bar in data["stats"]["chart_bars"]]
|
|
self.assertEqual(labels, ["Sent", "Delivered", "Opens", "Clicks", "Failed"])
|
|
|
|
def test_bearer_authorization_header(self):
|
|
import json
|
|
|
|
url = reverse("messaging:email_webhook")
|
|
with self.settings(SMTP2GO_WEBHOOK_SECRET="s3cret"):
|
|
denied = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"event": "delivered",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer wrong",
|
|
)
|
|
self.assertEqual(denied.status_code, 403)
|
|
ok = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"event": "delivered",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
"email_id": "e-1",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer s3cret",
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
|
|
def test_invalid_monica_header_does_not_500(self):
|
|
"""SMTP2GO tests often send a non-UUID sample custom header."""
|
|
import json
|
|
|
|
url = reverse("messaging:email_webhook")
|
|
response = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"event": "delivered",
|
|
"rcpt": "nobody@example.com",
|
|
"X-Monica-Message-Id": "Headers Unavailable",
|
|
"email_id": "smtp2go-test-id",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertTrue(data["ok"])
|
|
self.assertFalse(data["matched"])
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(event_type="delivered").exists()
|
|
)
|
|
|
|
def test_ui_event_label_aliases(self):
|
|
url = reverse("messaging:email_webhook")
|
|
response = self.client.post(
|
|
url,
|
|
data={
|
|
"event": "bounced",
|
|
"bounce": "soft",
|
|
"X-Monica-Message-Id": str(self.message.pk),
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.BOUNCED)
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, event_type="bounce"
|
|
).exists()
|
|
)
|
|
|
|
|
|
class Smtp2goSmsWebhookTests(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
self.contact = Contact.objects.create(
|
|
email="pat@example.com",
|
|
phone="+15550142291",
|
|
first_name="Pat",
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.SMS, opted_in=True, reason="test"
|
|
)
|
|
self.campaign = create_campaign_draft(
|
|
name="SMS blast",
|
|
audience=Campaign.Audience.SMS_OPT_IN,
|
|
subject="",
|
|
body="Hi there",
|
|
)
|
|
self.message = self.campaign.messages.get()
|
|
self.message.status = Message.Status.SENT
|
|
self.message.provider_message_id = "sms-provider-99"
|
|
self.message.save()
|
|
|
|
def test_sms_delivered_by_message_id(self):
|
|
url = reverse("messaging:sms_webhook")
|
|
import json
|
|
|
|
response = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"event": "sms_delivered",
|
|
"message_id": "sms-provider-99",
|
|
"destination_number": "5550142291",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.DELIVERED)
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, event_type="sms_delivered"
|
|
).exists()
|
|
)
|
|
|
|
def test_sms_failed_by_phone_fallback(self):
|
|
url = reverse("messaging:sms_webhook")
|
|
import json
|
|
|
|
self.message.provider_message_id = "other-id"
|
|
self.message.save(update_fields=["provider_message_id"])
|
|
response = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"event": "sms_failed",
|
|
"message_id": "unknown",
|
|
"destination_number": "+1 (555) 014-2291",
|
|
"status_code": "undeliverable",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.FAILED)
|
|
|
|
def test_inbound_stop_still_works(self):
|
|
url = reverse("messaging:sms_webhook")
|
|
response = self.client.post(
|
|
url, {"from": "5550142291", "text": "STOP"}
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertFalse(contact_may_receive(self.contact, Channel.SMS))
|
|
|
|
def test_sms_webhook_requires_bearer_when_secret_set(self):
|
|
url = reverse("messaging:sms_webhook")
|
|
import json
|
|
|
|
with self.settings(SMTP2GO_WEBHOOK_SECRET="s3cret"):
|
|
denied = self.client.post(
|
|
url,
|
|
data=json.dumps({"event": "sms_delivered", "message_id": "x"}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(denied.status_code, 403)
|
|
ok = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"event": "sms_delivered",
|
|
"message_id": "sms-provider-99",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer s3cret",
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
|
|
|
|
class PcmPostcardWebhookTests(TestCase):
|
|
def setUp(self):
|
|
self.client = Client()
|
|
self.contact = Contact.objects.create(
|
|
email="mail@example.com",
|
|
first_name="Pat",
|
|
last_name="Lee",
|
|
postal_address=Contact.make_postal_address(
|
|
line1="123 Main St",
|
|
city="Naperville",
|
|
state="IL",
|
|
zip_code="60540",
|
|
),
|
|
)
|
|
set_channel_consent(
|
|
self.contact, Channel.POSTCARD, opted_in=True, reason="test"
|
|
)
|
|
template = MessageTemplate.objects.create(
|
|
name="PCM test design",
|
|
channel=Channel.POSTCARD,
|
|
body="Postcard",
|
|
postcard_front={"design_id": 99, "size": "46", "provider": "pcm"},
|
|
)
|
|
self.campaign = create_campaign_draft(
|
|
name="March mailer",
|
|
audience=Campaign.Audience.POSTCARD_OPT_IN,
|
|
body="Postcard mailing",
|
|
template=template,
|
|
)
|
|
self.message = self.campaign.messages.get()
|
|
self.message.status = Message.Status.SENT
|
|
self.message.provider = "pcm"
|
|
self.message.provider_message_id = "order-555"
|
|
self.message.save()
|
|
|
|
def test_delivered_by_ext_ref(self):
|
|
import json
|
|
|
|
url = reverse("messaging:postcard_webhook")
|
|
response = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"status": "Delivered",
|
|
"extRefNbr": str(self.message.pk),
|
|
"orderID": 555,
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.message.refresh_from_db()
|
|
self.assertEqual(self.message.status, Message.Status.DELIVERED)
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, provider="pcm", event_type="Delivered"
|
|
).exists()
|
|
)
|
|
|
|
def test_requires_bearer_when_secret_set(self):
|
|
import json
|
|
|
|
url = reverse("messaging:postcard_webhook")
|
|
with self.settings(PCM_WEBHOOK_SECRET="pcm-secret", PCM_WEBHOOK_SECRETS=""):
|
|
denied = self.client.post(
|
|
url,
|
|
data=json.dumps({"status": "Delivered", "orderID": 555}),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(denied.status_code, 403)
|
|
ok = self.client.post(
|
|
url,
|
|
data=json.dumps(
|
|
{
|
|
"status": "Delivered",
|
|
"orderID": "order-555",
|
|
}
|
|
),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer pcm-secret",
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
|
|
def test_accepts_any_secret_from_list(self):
|
|
import json
|
|
|
|
url = reverse("messaging:postcard_webhook")
|
|
with self.settings(
|
|
PCM_WEBHOOK_SECRET="",
|
|
PCM_WEBHOOK_SECRETS="sec-a,sec-b",
|
|
):
|
|
denied = self.client.post(
|
|
url,
|
|
data=json.dumps({"status": "Delivered", "orderID": "order-555"}),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer wrong",
|
|
)
|
|
self.assertEqual(denied.status_code, 403)
|
|
ok = self.client.post(
|
|
url,
|
|
data=json.dumps({"status": "Delivered", "orderID": "order-555"}),
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer sec-b",
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
|
|
|
|
class PcmAuthTests(TestCase):
|
|
def setUp(self):
|
|
from messaging.providers.postcard import pcm as pcm_mod
|
|
|
|
pcm_mod.clear_token_cache()
|
|
|
|
def tearDown(self):
|
|
from messaging.providers.postcard import pcm as pcm_mod
|
|
|
|
pcm_mod.clear_token_cache()
|
|
|
|
def test_login_required_before_design_list(self):
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from messaging.providers.postcard import pcm as pcm_mod
|
|
|
|
login_resp = MagicMock()
|
|
login_resp.status_code = 200
|
|
login_resp.content = b'{"token":"session-tok","expires":"2099-01-01T00:00:00.000Z"}'
|
|
login_resp.json.return_value = {
|
|
"token": "session-tok",
|
|
"expires": "2099-01-01T00:00:00.000Z",
|
|
}
|
|
|
|
design_resp = MagicMock()
|
|
design_resp.status_code = 200
|
|
design_resp.content = b'{"results":[]}'
|
|
design_resp.json.return_value = {"results": []}
|
|
|
|
with self.settings(PCM_API_KEY="key", PCM_API_SECRET="secret"):
|
|
with patch("messaging.providers.postcard.pcm.requests.post") as post:
|
|
with patch(
|
|
"messaging.providers.postcard.pcm.requests.request"
|
|
) as request:
|
|
post.return_value = login_resp
|
|
request.return_value = design_resp
|
|
designs = pcm_mod.list_designs()
|
|
|
|
self.assertEqual(designs, [])
|
|
post.assert_called_once()
|
|
self.assertIn("/auth/login", post.call_args.args[0])
|
|
self.assertEqual(
|
|
post.call_args.kwargs["json"],
|
|
{"apiKey": "key", "apiSecret": "secret"},
|
|
)
|
|
auth = request.call_args.kwargs["headers"]["Authorization"]
|
|
self.assertEqual(auth, "Bearer session-tok")
|
|
|
|
def test_missing_secret_raises(self):
|
|
from messaging.providers.postcard import pcm as pcm_mod
|
|
|
|
with self.settings(PCM_API_KEY="key", PCM_API_SECRET=""):
|
|
with self.assertRaises(pcm_mod.PcmApiError) as ctx:
|
|
pcm_mod.login(force=True)
|
|
self.assertIn("PCM_API_SECRET", str(ctx.exception))
|
|
|
|
|
|
class PostcardAddressDefaultConsentTests(TestCase):
|
|
def test_address_without_consent_is_postcard_eligible(self):
|
|
contact = Contact.objects.create(
|
|
email="addr@example.com",
|
|
first_name="Ann",
|
|
postal_address=Contact.make_postal_address(
|
|
line1="1 Oak St",
|
|
city="Naperville",
|
|
state="IL",
|
|
zip_code="60540",
|
|
),
|
|
)
|
|
self.assertTrue(contact_may_receive(contact, Channel.POSTCARD))
|
|
prefs = channel_preferences(contact)
|
|
self.assertTrue(prefs[Channel.POSTCARD])
|
|
self.assertEqual(opted_in_contacts(Channel.POSTCARD).count(), 1)
|
|
|
|
def test_explicit_postcard_opt_out_respected(self):
|
|
contact = Contact.objects.create(
|
|
email="out@example.com",
|
|
postal_address=Contact.make_postal_address(line1="2 Oak St"),
|
|
)
|
|
set_channel_consent(
|
|
contact, Channel.POSTCARD, opted_in=False, reason="opt_out"
|
|
)
|
|
self.assertFalse(contact_may_receive(contact, Channel.POSTCARD))
|
|
self.assertEqual(opted_in_contacts(Channel.POSTCARD).count(), 0)
|
|
|
|
|
|
class CampaignHtmlBodyTests(TestCase):
|
|
def test_html_body_preserved_in_email(self):
|
|
from django.core import mail
|
|
|
|
User = get_user_model()
|
|
user = User.objects.create_user(
|
|
username="htmlsender", password="test-pass-123", email="s@example.com"
|
|
)
|
|
contact = Contact.objects.create(email="pat@example.com", first_name="Pat")
|
|
set_channel_consent(contact, Channel.EMAIL, opted_in=True, reason="test")
|
|
campaign = create_campaign_draft(
|
|
name="HTML tip",
|
|
audience=Campaign.Audience.EMAIL_OPT_IN,
|
|
subject="Styled",
|
|
body='<p><strong>Hello</strong> <img src="https://example.com/a.png"></p>',
|
|
created_by=user,
|
|
)
|
|
from messaging.providers.email.smtp2go import send_email
|
|
|
|
send_email(campaign.messages.get())
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
html = mail.outbox[0].alternatives[0][0]
|
|
self.assertIn("<strong>Hello</strong>", html)
|
|
self.assertIn('src="https://example.com/a.png"', html)
|
|
|
|
|
|
class PostcardDesignPickTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="pcm-pick", password="test-pass-123"
|
|
)
|
|
self.client = Client()
|
|
self.client.login(username="pcm-pick", password="test-pass-123")
|
|
self.contact = Contact.objects.create(
|
|
email="mail@example.com",
|
|
postal_address=Contact.make_postal_address(line1="9 Main"),
|
|
)
|
|
|
|
def test_compose_can_pick_pcm_design_id(self):
|
|
from unittest.mock import patch
|
|
|
|
with patch(
|
|
"messaging.views._fetch_pcm_designs",
|
|
return_value=(
|
|
[{"design_id": "42", "name": "Spring card", "size": "46"}],
|
|
"",
|
|
),
|
|
):
|
|
url = reverse("messaging:campaign_list")
|
|
response = self.client.post(
|
|
url,
|
|
{
|
|
"name": "Mailer",
|
|
"subject": "",
|
|
"body": "",
|
|
"audience": Campaign.Audience.POSTCARD_OPT_IN,
|
|
"template_id": "d:42",
|
|
},
|
|
)
|
|
campaign = Campaign.objects.get(name="Mailer")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(campaign.channel, Channel.POSTCARD)
|
|
self.assertIsNotNone(campaign.template_id)
|
|
self.assertEqual(
|
|
campaign.template.postcard_front.get("design_id"), 42
|
|
)
|
|
self.assertEqual(campaign.messages.count(), 1)
|
|
|
|
|
|
class StoredFileUploadTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="uploader", password="test-pass-123"
|
|
)
|
|
self.client = Client()
|
|
self.client.login(username="uploader", password="test-pass-123")
|
|
|
|
def test_upload_stores_bytes_in_database(self):
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
from messaging.models import StoredFile
|
|
|
|
png = (
|
|
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01"
|
|
b"\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00"
|
|
b"\x00\x0cIDATx\x9cc\xf8\x0f\x00\x00\x01\x01\x00\x05\x18"
|
|
b"\xd8N\x00\x00\x00\x00IEND\xaeB`\x82"
|
|
)
|
|
upload = SimpleUploadedFile("dot.png", png, content_type="image/png")
|
|
response = self.client.post(
|
|
reverse("messaging:campaign_image_upload"),
|
|
{"image": upload},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertIn("url", payload)
|
|
stored = StoredFile.objects.get()
|
|
self.assertEqual(bytes(stored.data), png)
|
|
self.assertEqual(stored.content_type, "image/png")
|
|
self.assertEqual(stored.uploaded_by, self.user)
|
|
|
|
fetch = self.client.get(
|
|
reverse("messaging:stored_file", kwargs={"pk": stored.pk})
|
|
)
|
|
self.assertEqual(fetch.status_code, 200)
|
|
self.assertEqual(fetch["Content-Type"], "image/png")
|
|
self.assertEqual(b"".join(fetch.streaming_content), png)
|
|
|
|
|
|
|
|
class MergeTagTests(TestCase):
|
|
def test_render_merge_tags_double_and_single_braces(self):
|
|
from messaging.services import render_merge_tags
|
|
|
|
contact = Contact.objects.create(
|
|
email="pat@example.com",
|
|
first_name="Pat",
|
|
last_name="Lee",
|
|
phone="6305551212",
|
|
)
|
|
text = "Hi {{first_name}} {{last_name}} / {first_name} {email}"
|
|
out = render_merge_tags(text, contact)
|
|
self.assertEqual(out, "Hi Pat Lee / Pat pat@example.com")
|
|
|
|
def test_email_send_substitutes_merge_tags(self):
|
|
from django.core import mail
|
|
|
|
from messaging.providers.email.smtp2go import send_email
|
|
from messaging.services import create_campaign_draft, set_channel_consent
|
|
|
|
User = get_user_model()
|
|
user = User.objects.create_user(username="merge", password="x")
|
|
contact = Contact.objects.create(
|
|
email="pat@example.com", first_name="Pat", last_name="Lee"
|
|
)
|
|
set_channel_consent(contact, Channel.EMAIL, opted_in=True, reason="test")
|
|
campaign = create_campaign_draft(
|
|
name="Merge",
|
|
audience=Campaign.Audience.EMAIL_OPT_IN,
|
|
subject="Hello {{first_name}}",
|
|
body="<p>Dear {{first_name}} {{last_name}}</p>",
|
|
created_by=user,
|
|
)
|
|
send_email(campaign.messages.get())
|
|
self.assertEqual(mail.outbox[0].subject, "Hello Pat")
|
|
html = mail.outbox[0].alternatives[0][0]
|
|
self.assertIn("Dear Pat Lee", html)
|
|
self.assertNotIn("{{first_name}}", html)
|
|
|
|
|
|
class CampaignRecipientTableTests(TestCase):
|
|
def setUp(self):
|
|
from messaging.services import create_campaign_draft, set_channel_consent
|
|
|
|
User = get_user_model()
|
|
self.user = User.objects.create_user(
|
|
username="recip", password="test-pass-123"
|
|
)
|
|
self.client = Client()
|
|
self.client.login(username="recip", password="test-pass-123")
|
|
self.contacts = []
|
|
for i in range(3):
|
|
c = Contact.objects.create(
|
|
email=f"c{i}@example.com", first_name=f"Name{i}"
|
|
)
|
|
set_channel_consent(c, Channel.EMAIL, opted_in=True, reason="test")
|
|
self.contacts.append(c)
|
|
self.campaign = create_campaign_draft(
|
|
name="Recipients",
|
|
audience=Campaign.Audience.EMAIL_OPT_IN,
|
|
subject="Hi",
|
|
body="Body",
|
|
created_by=self.user,
|
|
)
|
|
|
|
def test_detail_does_not_flash_message_objects(self):
|
|
url = reverse(
|
|
"messaging:campaign_detail", kwargs={"pk": self.campaign.pk}
|
|
)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertNotContains(response, "email →")
|
|
self.assertContains(response, "Recipients")
|
|
self.assertContains(response, "Remove")
|
|
|
|
def test_remove_recipient(self):
|
|
msg = self.campaign.messages.first()
|
|
url = reverse(
|
|
"messaging:campaign_message_remove",
|
|
kwargs={"pk": self.campaign.pk, "message_id": msg.pk},
|
|
)
|
|
response = self.client.post(url)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(self.campaign.messages.count(), 2)
|
|
|
|
def test_postcard_shows_pcm_events_title(self):
|
|
from messaging.services import create_campaign_draft
|
|
|
|
Contact.objects.create(
|
|
email="pc@example.com",
|
|
postal_address=Contact.make_postal_address(line1="1 Main"),
|
|
)
|
|
campaign = create_campaign_draft(
|
|
name="Cards",
|
|
audience=Campaign.Audience.POSTCARD_OPT_IN,
|
|
body="Postcard mailing",
|
|
created_by=self.user,
|
|
template=MessageTemplate.objects.create(
|
|
name="Design",
|
|
channel=Channel.POSTCARD,
|
|
body="",
|
|
postcard_front={"design_id": "99"},
|
|
),
|
|
)
|
|
url = reverse("messaging:campaign_detail", kwargs={"pk": campaign.pk})
|
|
response = self.client.get(url)
|
|
self.assertContains(response, "Recent PCM Integrations events")
|
|
self.assertNotContains(response, "Recent SMTP2GO events")
|