Add Django site, Docker packaging, and beta/prod Gitea deploys.
Unignore site/ (was blocked by mkdocs /site rule), add compose/Docker/uv tooling, and split deploys so push to main goes to beta while prod stays manual.
This commit is contained in:
@@ -0,0 +1,629 @@
|
||||
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]"))
|
||||
# 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])
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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"):
|
||||
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)
|
||||
Reference in New Issue
Block a user