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 (
    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_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))


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