monica_site/site/messaging/tests.py

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", 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))