print_forge/site/email_sms/tests.py

from django.contrib.auth import get_user_model
from django.test import Client, TestCase, override_settings
from django.urls import reverse

from contacts.models import Channel, ConsentRecord, Contact, Suppression
from email_sms.models import Campaign, Message, MessageTemplate, ProviderEvent
from email_sms.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("email_sms: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", is_staff=True
        )
        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", is_staff=True
        )
        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("email_sms: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("email_sms: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("email_sms: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", is_staff=True
        )
        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(
            "email_sms: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(
            "email_sms: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(
            "email_sms: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("#00aeef", 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("email_sms:email_webhook")
        response = self.client.post(
            url,
            data={
                "event": "delivered",
                "rcpt": "pat@example.com",
                "email_id": "smtp2go-abc-123",
                "X-Monica-Message-Id": str(self.message.pk),
            },
        )
        self.assertEqual(response.status_code, 200)
        self.message.refresh_from_db()
        self.assertEqual(self.message.status, Message.Status.DELIVERED)
        self.assertEqual(self.message.provider_message_id, "smtp2go-abc-123")
        self.assertTrue(
            ProviderEvent.objects.filter(
                message=self.message, event_type="delivered"
            ).exists()
        )

    def test_open_records_event_sets_opened(self):
        self.message.status = Message.Status.DELIVERED
        self.message.save(update_fields=["status"])
        url = reverse("email_sms:email_webhook")
        self.client.post(
            url,
            data={
                "event": "open",
                "rcpt": "pat@example.com",
                "X-Monica-Message-Id": str(self.message.pk),
            },
        )
        self.message.refresh_from_db()
        self.assertEqual(self.message.status, Message.Status.OPENED)
        self.assertTrue(
            ProviderEvent.objects.filter(
                message=self.message, event_type="open"
            ).exists()
        )

    def test_click_sets_clicked_from_sent(self):
        url = reverse("email_sms:email_webhook")
        self.client.post(
            url,
            data={
                "event": "click",
                "rcpt": "pat@example.com",
                "X-Monica-Message-Id": str(self.message.pk),
            },
        )
        self.message.refresh_from_db()
        self.assertEqual(self.message.status, Message.Status.CLICKED)
        self.assertTrue(
            ProviderEvent.objects.filter(
                message=self.message, event_type="click"
            ).exists()
        )

    def test_open_does_not_downgrade_clicked(self):
        self.message.status = Message.Status.CLICKED
        self.message.save(update_fields=["status"])
        url = reverse("email_sms:email_webhook")
        self.client.post(
            url,
            data={
                "event": "open",
                "rcpt": "pat@example.com",
                "X-Monica-Message-Id": str(self.message.pk),
            },
        )
        self.message.refresh_from_db()
        self.assertEqual(self.message.status, Message.Status.CLICKED)

    def test_hard_bounce_suppresses_contact(self):
        url = reverse("email_sms: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("email_sms: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", is_staff=True)
        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(
            "email_sms:campaign_status_json", kwargs={"pk": self.campaign.pk}
        )
        response = self.client.get(url)
        self.assertEqual(response.status_code, 200)
        data = response.json()
        self.assertEqual(data["stats"]["opens"], 1)
        self.assertIn("chart_bars", data["stats"])
        labels = [bar["label"] for bar in data["stats"]["chart_bars"]]
        self.assertEqual(labels, ["Sent", "Delivered", "Opens", "Clicks", "Failed"])

    def test_bearer_authorization_header(self):
        import json

        url = reverse("email_sms: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("email_sms: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("email_sms: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("email_sms: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("email_sms: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_sms_short_event_name_delivered(self):
        """API sms_events use short names; body may omit sms_ prefix."""
        url = reverse("email_sms:sms_webhook")
        import json

        response = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "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_ui_label_and_form_encoded(self):
        url = reverse("email_sms:sms_webhook")
        self.message.status = Message.Status.QUEUED
        self.message.save(update_fields=["status"])
        response = self.client.post(
            url,
            data={
                "event": "Submitted",
                "message_id": "sms-provider-99",
                "destination_number": "5550142291",
            },
        )
        self.assertEqual(response.status_code, 200)
        self.message.refresh_from_db()
        self.assertEqual(self.message.status, Message.Status.SENT)
        self.assertTrue(
            ProviderEvent.objects.filter(
                message=self.message, event_type="sms_submitted"
            ).exists()
        )

    def test_sms_rejected_marks_failed(self):
        url = reverse("email_sms:sms_webhook")
        import json

        response = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "sms_rejected",
                    "message_id": "sms-provider-99",
                    "destination_number": "5550142291",
                    "status_code": "blocked",
                }
            ),
            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_sms_webhook_id_is_not_used_as_message_id(self):
        """Docs: `id` is the webhook notification id, not the SMS id."""
        url = reverse("email_sms:sms_webhook")
        import json

        response = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "sms_delivered",
                    "id": "sms-provider-99",
                    "destination_number": "19999999999",
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 200)
        self.message.refresh_from_db()
        self.assertEqual(self.message.status, Message.Status.SENT)
        self.assertTrue(
            ProviderEvent.objects.filter(
                message=None, event_type="sms_delivered"
            ).exists()
        )

    def test_sms_opt_out_unmatched_suppresses_by_phone(self):
        url = reverse("email_sms:sms_webhook")
        import json

        self.message.delete()
        response = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "opt-out",
                    "message_id": "unknown",
                    "destination_number": "+15550142291",
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 200)
        self.assertFalse(contact_may_receive(self.contact, Channel.SMS))

    def test_inbound_stop_still_works(self):
        url = reverse("email_sms: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_inbound_unsub_keyword(self):
        url = reverse("email_sms:sms_webhook")
        response = self.client.post(
            url, {"from": "5550142291", "text": "UNSUB"}
        )
        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("email_sms: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 UnifiedSmtp2goWebhookTests(TestCase):
    def setUp(self):
        self.client = Client()
        self.email_contact = Contact.objects.create(
            email="pat@example.com",
            first_name="Pat",
        )
        set_channel_consent(
            self.email_contact, Channel.EMAIL, opted_in=True, reason="test"
        )
        self.email_campaign = create_campaign_draft(
            name="Unified email",
            audience=Campaign.Audience.EMAIL_OPT_IN,
            subject="Hello",
            body="Body",
        )
        self.email_message = self.email_campaign.messages.get()
        self.email_message.status = Message.Status.SENT
        self.email_message.save(update_fields=["status"])

        self.sms_contact = Contact.objects.create(
            email="sms.pat@example.com",
            phone="+15550142291",
            first_name="Pat",
        )
        set_channel_consent(
            self.sms_contact, Channel.SMS, opted_in=True, reason="test"
        )
        self.sms_campaign = create_campaign_draft(
            name="Unified SMS",
            audience=Campaign.Audience.SMS_OPT_IN,
            subject="",
            body="Hi",
        )
        self.sms_message = self.sms_campaign.messages.get()
        self.sms_message.status = Message.Status.SENT
        self.sms_message.provider_message_id = "sms-unified-1"
        self.sms_message.save()

    def test_classify_helpers(self):
        from email_sms.webhooks import classify_smtp2go_payload

        self.assertEqual(
            classify_smtp2go_payload({"from": "555", "text": "STOP"}),
            "sms_inbound",
        )
        self.assertEqual(
            classify_smtp2go_payload(
                {"event": "sms_delivered", "message_id": "x"}
            ),
            "sms",
        )
        self.assertEqual(
            classify_smtp2go_payload(
                {
                    "event": "delivered",
                    "destination_number": "5550142291",
                    "message_id": "x",
                }
            ),
            "sms",
        )
        self.assertEqual(
            classify_smtp2go_payload(
                {
                    "event": "delivered",
                    "rcpt": "pat@example.com",
                    "email_id": "e1",
                }
            ),
            "email",
        )
        self.assertEqual(
            classify_smtp2go_payload({"event": "click", "rcpt": "a@b.c"}),
            "email",
        )

    def test_unified_url_routes_email_and_sms(self):
        import json

        url = reverse("email_sms:smtp2go_webhook")
        email_resp = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "click",
                    "rcpt": "pat@example.com",
                    "X-Monica-Message-Id": str(self.email_message.pk),
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(email_resp.status_code, 200)
        self.assertEqual(email_resp.json()["channel"], "email")
        self.email_message.refresh_from_db()
        self.assertEqual(self.email_message.status, Message.Status.CLICKED)

        sms_resp = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "sms_delivered",
                    "message_id": "sms-unified-1",
                    "destination_number": "5550142291",
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(sms_resp.status_code, 200)
        self.assertEqual(sms_resp.json()["channel"], "sms")
        self.sms_message.refresh_from_db()
        self.assertEqual(self.sms_message.status, Message.Status.DELIVERED)

    def test_legacy_email_path_still_accepts_sms(self):
        """Legacy /email/ alias is the unified handler."""
        import json

        url = reverse("email_sms:email_webhook")
        response = self.client.post(
            url,
            data=json.dumps(
                {
                    "event": "delivered",
                    "message_id": "sms-unified-1",
                    "destination_number": "5550142291",
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["channel"], "sms")
        self.sms_message.refresh_from_db()
        self.assertEqual(self.sms_message.status, Message.Status.DELIVERED)


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 email_sms.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 StoredFileUploadTests(TestCase):
    def setUp(self):
        User = get_user_model()
        self.user = User.objects.create_user(
            username="uploader", password="test-pass-123", is_staff=True
        )
        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 core.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("email_sms: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("core: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 email_sms.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 email_sms.providers.email.smtp2go import send_email
        from email_sms.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 email_sms.services import create_campaign_draft, set_channel_consent

        User = get_user_model()
        self.user = User.objects.create_user(
            username="recip", password="test-pass-123", is_staff=True
        )
        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(
            "email_sms: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(
            "email_sms: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)


@override_settings(SHORTENER_BASE_URL="", SHORTENER_API_TOKEN="")
class Smtp2goSmsSendTests(TestCase):
    def setUp(self):
        self.contact = Contact.objects.create(
            phone="330-402-2675",
            first_name="Rufus",
            last_name="Firefly",
        )
        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,
            body="Hello {{first_name}}",
        )
        self.message = self.campaign.messages.get(contact=self.contact)

    def test_format_destination_us_local(self):
        from email_sms.providers.sms.smtp2go import _format_destination

        self.assertEqual(_format_destination("330-402-2675"), "+13304022675")
        self.assertEqual(_format_destination("+1 (330) 402-2675"), "+13304022675")
        self.assertEqual(_format_destination("13304022675"), "+13304022675")

    def test_send_sms_uses_destination_and_content(self):
        from unittest.mock import MagicMock, patch

        from email_sms.providers.sms.smtp2go import send_sms

        resp = MagicMock()
        resp.ok = True
        resp.content = b'{"request_id":"req-1","data":{"messages":[{"message_id":"sms-abc","destination":"+13304022675","status":"queued"}],"total_sent":1}}'
        resp.json.return_value = {
            "request_id": "req-1",
            "data": {
                "messages": [
                    {
                        "message_id": "sms-abc",
                        "destination": "+13304022675",
                        "status": "queued",
                    }
                ],
                "total_sent": 1,
            },
        }

        with self.settings(SMTP2GO_SMS_API_KEY="api-test-key"):
            with patch(
                "email_sms.providers.sms.smtp2go.requests.post",
                return_value=resp,
            ) as post:
                provider_id = send_sms(self.message)

        self.assertEqual(provider_id, "sms-abc")
        payload = post.call_args.kwargs["json"]
        self.assertEqual(payload["api_key"], "api-test-key")
        self.assertEqual(payload["destination"], ["+13304022675"])
        self.assertTrue(payload["content"].startswith("Hello Rufus"))
        self.assertIn("utm_medium=sms", payload["content"])
        self.assertNotIn("to", payload)
        self.assertNotIn("text", payload)

    def test_send_sms_includes_api_error_body(self):
        from unittest.mock import MagicMock, patch

        import requests

        from email_sms.providers.sms.smtp2go import send_sms

        resp = MagicMock()
        resp.ok = False
        resp.status_code = 400
        resp.url = "https://api.smtp2go.com/v3/sms/send"
        resp.reason = "Bad Request"
        resp.text = '{"data":{"error":"Missing required field","error_code":"E_ApiResponseCodes.INVALID_REQUEST"}}'
        resp.json.return_value = {
            "data": {
                "error": "Missing required field",
                "error_code": "E_ApiResponseCodes.INVALID_REQUEST",
            }
        }

        with self.settings(SMTP2GO_SMS_API_KEY="api-test-key"):
            with patch(
                "email_sms.providers.sms.smtp2go.requests.post",
                return_value=resp,
            ):
                with self.assertRaises(requests.HTTPError) as ctx:
                    send_sms(self.message)

        self.assertIn("Missing required field", str(ctx.exception))
        self.assertIn("INVALID_REQUEST", str(ctx.exception))


class CampaignUtmLinkTests(TestCase):
    def setUp(self):
        User = get_user_model()
        self.user = User.objects.create_user(
            username="utm-composer", password="test-pass-123", is_staff=True
        )
        self.client = Client()
        self.client.login(username="utm-composer", password="test-pass-123")
        self.contact = Contact.objects.create(
            email="pat@example.com",
            phone="5550100199",
            first_name="Pat",
        )
        set_channel_consent(
            self.contact, Channel.EMAIL, opted_in=True, reason="test"
        )
        set_channel_consent(
            self.contact, Channel.SMS, opted_in=True, reason="test"
        )

    @override_settings(
        PUBLIC_SITE_URL="https://acmehvac.com",
        SITE_NAME="Acme HVAC",
        UTM_SOURCE="",
        SHORTENER_BASE_URL="",
        SHORTENER_API_TOKEN="",
    )
    def test_email_draft_inserts_html_link(self):
        campaign = create_campaign_draft(
            name="Spring seller tips",
            audience=Campaign.Audience.EMAIL_OPT_IN,
            subject="Hello",
            body="<p>Hi there</p>",
            created_by=self.user,
        )
        self.assertIn('data-campaign-utm="1"', campaign.body_override)
        self.assertIn("utm_source=acme-hvac", campaign.body_override)
        self.assertIn("utm_medium=email", campaign.body_override)
        self.assertIn("utm_campaign=spring-seller-tips", campaign.body_override)
        self.assertIn("Hi there", campaign.body_override)

    @override_settings(
        PUBLIC_SITE_URL="https://acmehvac.com",
        SITE_NAME="Acme HVAC",
        UTM_SOURCE="",
        SHORTENER_BASE_URL="",
        SHORTENER_API_TOKEN="",
    )
    def test_sms_draft_inserts_plain_url(self):
        campaign = create_campaign_draft(
            name="Open house",
            audience=Campaign.Audience.SMS_OPT_IN,
            body="See you Saturday",
            created_by=self.user,
        )
        self.assertIn("See you Saturday", campaign.body_override)
        self.assertIn(
            "https://acmehvac.com/?utm_source=acme-hvac&utm_medium=sms"
            "&utm_campaign=open-house",
            campaign.body_override,
        )
        self.assertNotIn("<a ", campaign.body_override)

    def test_composer_shows_tracked_link_panel(self):
        url = reverse("email_sms:campaign_list")
        response = self.client.get(url)
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Tracked site link")
        self.assertContains(response, "campaign-utm.js")
        self.assertContains(response, "qrcode.min.js")
        self.assertContains(response, "data-utm-qr")
        self.assertContains(response, "utm_campaign")
        self.assertContains(response, reverse("campaign_short_link"))

    @override_settings(
        PUBLIC_SITE_URL="https://acmehvac.com",
        SITE_NAME="Acme HVAC",
        UTM_SOURCE="",
        SHORTENER_BASE_URL="",
        SHORTENER_API_TOKEN="",
    )
    def test_detail_shows_tracked_url(self):
        campaign = create_campaign_draft(
            name="Spring seller tips",
            audience=Campaign.Audience.EMAIL_OPT_IN,
            subject="Hello",
            body="Hi",
            created_by=self.user,
        )
        url = reverse("email_sms:campaign_detail", kwargs={"pk": campaign.pk})
        response = self.client.get(url)
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "utm_campaign=spring-seller-tips")
        self.assertContains(response, "utm_medium=email")
        self.assertContains(response, "qrcode.min.js")
        self.assertContains(response, "data-utm-qr")

    @override_settings(
        PUBLIC_SITE_URL="https://acmehvac.com",
        SITE_NAME="Acme HVAC",
        UTM_SOURCE="",
        SHORTENER_BASE_URL="https://shortener.aimloperations.com",
        SHORTENER_API_TOKEN="acme:test-token",
    )
    def test_shortener_mints_into_sms_body(self):
        from unittest.mock import MagicMock, patch

        mock = MagicMock()
        mock.status_code = 201
        mock.json.return_value = {
            "code": "a3k9xm",
            "short_url": "https://piha.li/a3k9xm",
            "target_url": (
                "https://acmehvac.com/?utm_source=acme-hvac"
                "&utm_medium=sms&utm_campaign=open-house"
            ),
        }
        with patch("core.shortener.requests.post", return_value=mock) as post:
            campaign = create_campaign_draft(
                name="Open house",
                audience=Campaign.Audience.SMS_OPT_IN,
                body="See you Saturday",
                created_by=self.user,
            )
        self.assertIn("https://piha.li/a3k9xm", campaign.body_override)
        self.assertIn("See you Saturday", campaign.body_override)
        self.assertNotIn("utm_medium=sms", campaign.body_override)
        post.assert_called_once()
        headers = post.call_args.kwargs["headers"]
        self.assertEqual(headers["Authorization"], "Bearer acme:test-token")
        payload = post.call_args.kwargs["json"]
        self.assertIn("utm_medium=sms", payload["target_url"])
        self.assertIn("utm_campaign=open-house", payload["target_url"])
        self.assertEqual(payload["external_ref"], str(campaign.pk))
        self.assertEqual(campaign.messages.get().body_snapshot, campaign.body_override)

    @override_settings(
        PUBLIC_SITE_URL="https://acmehvac.com",
        SITE_NAME="Acme HVAC",
        UTM_SOURCE="",
        SHORTENER_BASE_URL="https://shortener.aimloperations.com",
        SHORTENER_API_TOKEN="acme:test-token",
    )
    def test_shortener_failure_falls_back_to_long_url(self):
        from unittest.mock import MagicMock, patch

        mock = MagicMock()
        mock.status_code = 401
        mock.text = '{"detail":"Unauthorized"}'
        with patch("core.shortener.requests.post", return_value=mock):
            campaign = create_campaign_draft(
                name="Open house",
                audience=Campaign.Audience.SMS_OPT_IN,
                body="See you Saturday",
                created_by=self.user,
            )
        self.assertIn("utm_medium=sms", campaign.body_override)
        self.assertIn("utm_campaign=open-house", campaign.body_override)
        self.assertNotIn("piha.li", campaign.body_override)