westfarn
d830f07757
Fix campaign stuck on sending under async queue, and stop UUID ValidationError when SMTP2GO tests send "Headers Unavailable".
760 lines
28 KiB
Python
760 lines
28 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 (
|
|
contact_may_receive,
|
|
create_campaign_draft,
|
|
make_unsubscribe_token,
|
|
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,
|
|
{
|
|
"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")
|
|
|
|
|
|
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_keeps_delivered(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.DELIVERED)
|
|
self.assertTrue(
|
|
ProviderEvent.objects.filter(
|
|
message=self.message, event_type="open"
|
|
).exists()
|
|
)
|
|
|
|
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)
|
|
|
|
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))
|