web_django_template/site/directmail/tests.py

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

from contacts.models import Channel, ConsentRecord, Contact, Suppression
from directmail.models import Campaign, Message, MessageTemplate, ProviderEvent
from contacts.consent import channel_preferences
from directmail.services import (
    contact_may_receive,
    create_campaign_draft,
    opted_in_contacts,
    set_channel_consent,
)


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("directmail: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("directmail: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("directmail: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 directmail.providers.postcard import pcm as pcm_mod

        pcm_mod.clear_token_cache()

    def tearDown(self):
        from directmail.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 directmail.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("directmail.providers.postcard.pcm.requests.post") as post:
                with patch(
                    "directmail.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 directmail.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))

    def test_return_address_requires_street_fields(self):
        from directmail.providers.postcard import pcm as pcm_mod

        with self.settings(
            SITE_NAME="Monica Dhillon",
            PCM_RETURN_ADDRESS="",
            PCM_RETURN_LINE1="",
            PCM_RETURN_CITY="",
            PCM_RETURN_STATE="",
            PCM_RETURN_ZIP="",
        ):
            with self.assertRaises(pcm_mod.PcmApiError) as ctx:
                pcm_mod.return_address_from_settings()
        self.assertIn("PCM return address incomplete", str(ctx.exception))

    def test_return_address_from_json(self):
        from directmail.providers.postcard import pcm as pcm_mod

        with self.settings(
            PCM_RETURN_ADDRESS=(
                '{"firstName":"Mo","lastName":"D","address":"1 Main",'
                '"city":"Naperville","state":"IL","zipCode":"60540"}'
            ),
        ):
            addr = pcm_mod.return_address_from_settings()
        self.assertEqual(addr["address"], "1 Main")
        self.assertEqual(addr["zipCode"], "60540")


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 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(
            "directmail.views._fetch_pcm_designs",
            return_value=(
                [{"design_id": "42", "name": "Spring card", "size": "46"}],
                "",
            ),
        ):
            url = reverse("directmail: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 PostcardUtmPanelTests(TestCase):
    def setUp(self):
        User = get_user_model()
        self.user = User.objects.create_user(
            username="utm-mail", password="test-pass-123"
        )
        self.client = Client()
        self.client.login(username="utm-mail", password="test-pass-123")
        self.contact = Contact.objects.create(
            email="mail@example.com",
            first_name="Pat",
            postal_address=Contact.make_postal_address(line1="9 Main"),
        )

    def test_composer_shows_tracked_link_panel(self):
        with patch(
            "directmail.views._fetch_pcm_designs",
            return_value=([], ""),
        ):
            url = reverse("directmail: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, reverse("campaign_short_link"))


class CarrierRoutePostcardTests(TestCase):
    def setUp(self):
        User = get_user_model()
        self.user = User.objects.create_user(
            username="routes", password="test-pass-123"
        )
        self.client = Client()
        self.client.login(username="routes", password="test-pass-123")
        self.contact = Contact.objects.create(
            email="mail@example.com",
            postal_address=Contact.make_postal_address(line1="9 Main"),
        )

    def test_pick_unit_price_uses_break_qty_and_fallback(self):
        from directmail.providers.postcard import pcm as pcm_mod

        price, source = pcm_mod.pick_unit_price(
            [
                {"mailClass": "Standard", "price": 0.8, "breakQty": 1},
                {"mailClass": "Standard", "price": 0.5, "breakQty": 200},
            ],
            mail_class="Standard",
            quantity=250,
        )
        self.assertEqual(price, 0.5)
        self.assertEqual(source, "pcm")
        with self.settings(PCM_POSTCARD_PRICE_STANDARD="0.42"):
            price, source = pcm_mod.pick_unit_price(
                [], mail_class="Standard", quantity=10
            )
        self.assertEqual(price, 0.42)
        self.assertEqual(source, "fallback")

    def test_quote_search_returns_route_costs(self):
        payload = {
            "listCountID": 11,
            "recordCount": 300,
            "breakdown": [
                {
                    "breakdownType": "ZipCRRT",
                    "data": [
                        {"code": "60540:C001", "text": "NAPERVILLE", "total": 120},
                        {"code": "60540:C002", "text": "NAPERVILLE", "total": 180},
                    ],
                }
            ],
        }
        with self.settings(PCM_POSTCARD_PRICE_STANDARD="0.50"):
            with patch(
                "directmail.providers.postcard.pcm.design_pricing", return_value=[]
            ):
                with patch(
                    "directmail.providers.postcard.pcm.list_count_zipcode",
                    return_value=payload,
                ):
                    response = self.client.post(
                        reverse("directmail:postcard_quote"),
                        data='{"action":"search","zip":"60540","size":"68"}',
                        content_type="application/json",
                    )
        self.assertEqual(response.status_code, 200)
        data = response.json()
        self.assertEqual(len(data["routes"]), 2)
        self.assertEqual(data["routes"][0]["cost"], 60.0)
        self.assertEqual(data["mail_class"], "Standard")

    def test_quote_select_and_save_draft(self):
        import json

        select_payload = {
            "listCountID": 99,
            "recordCount": 300,
            "breakdown": [
                {
                    "breakdownType": "ZipCRRT",
                    "data": [
                        {"code": "60540:C001", "text": "A", "total": 120},
                        {"code": "60540:C002", "text": "B", "total": 180},
                    ],
                }
            ],
        }
        with self.settings(PCM_POSTCARD_PRICE_STANDARD="0.50"):
            with patch(
                "directmail.providers.postcard.pcm.design_pricing", return_value=[]
            ):
                with patch(
                    "directmail.providers.postcard.pcm.list_count_carrier_route",
                    return_value=select_payload,
                ):
                    quoted = self.client.post(
                        reverse("directmail:postcard_quote"),
                        data=(
                            '{"action":"select","size":"68",'
                            '"routes":["60540:C001","60540:C002"]}'
                        ),
                        content_type="application/json",
                    )
        self.assertEqual(quoted.status_code, 200)
        quote = quoted.json()
        self.assertEqual(quote["list_count_id"], 99)
        self.assertEqual(quote["total"], 150.0)

        targeting = {
            "selected_codes": quote["selected_codes"],
            "list_count_id": quote["list_count_id"],
            "record_count": quote["record_count"],
            "mail_class": quote["mail_class"],
            "estimated": quote["estimated"],
            "route_rows": quote["routes"],
        }
        with patch(
            "directmail.views._fetch_pcm_designs",
            return_value=(
                [{"design_id": "42", "name": "Spring card", "size": "68"}],
                "",
            ),
        ):
            response = self.client.post(
                reverse("directmail:campaign_list"),
                {
                    "name": "Farm drop",
                    "audience": Campaign.Audience.CARRIER_ROUTE,
                    "template_id": "d:42",
                    "route_targeting": json.dumps(targeting),
                },
            )
        self.assertEqual(response.status_code, 302)
        campaign = Campaign.objects.get(name="Farm drop")
        self.assertEqual(campaign.audience, Campaign.Audience.CARRIER_ROUTE)
        self.assertEqual(campaign.route_targeting["list_count_id"], 99)
        self.assertEqual(campaign.messages.count(), 1)
        self.assertIsNone(campaign.messages.get().contact_id)

    def test_send_uses_list_count_endpoint(self):
        from directmail.providers.postcard import pcm as pcm_mod
        from directmail.services import enqueue_campaign_send

        template = MessageTemplate.objects.create(
            name="Route design",
            channel=Channel.POSTCARD,
            body="Postcard",
            postcard_front={"design_id": 42, "size": "68", "provider": "pcm"},
        )
        campaign = create_campaign_draft(
            name="Routes",
            audience=Campaign.Audience.CARRIER_ROUTE,
            body="note",
            created_by=self.user,
            template=template,
            route_targeting={
                "selected_codes": ["60540:C001"],
                "list_count_id": 77,
                "record_count": 50,
                "mail_class": "Standard",
            },
        )
        pcm_mod.clear_token_cache()
        with self.settings(
            PCM_API_KEY="key",
            PCM_API_SECRET="secret",
            PCM_RETURN_ADDRESS=(
                '{"firstName":"Mo","lastName":"D","address":"1 Main",'
                '"city":"Naperville","state":"IL","zipCode":"60540"}'
            ),
        ):
            with patch("directmail.providers.postcard.pcm.login", return_value="tok"):
                with patch(
                    "directmail.providers.postcard.pcm.pcm_request",
                    return_value={"orderID": 555},
                ) as req:
                    enqueue_campaign_send(campaign)
        path = req.call_args.args[1]
        self.assertEqual(path, "/order/postcard/with-list-count")
        body = req.call_args.kwargs["json_body"]
        self.assertEqual(body["listCountID"], 77)
        self.assertEqual(body["recordCount"], 50)
        campaign.refresh_from_db()
        self.assertEqual(campaign.messages.get().provider_message_id, "555")

    def test_mailing_list_postcard_still_creates_contacts(self):
        with patch(
            "directmail.views._fetch_pcm_designs",
            return_value=(
                [{"design_id": "42", "name": "Spring card", "size": "46"}],
                "",
            ),
        ):
            response = self.client.post(
                reverse("directmail:campaign_list"),
                {
                    "name": "List mailer",
                    "audience": Campaign.Audience.POSTCARD_OPT_IN,
                    "template_id": "d:42",
                },
            )
        self.assertEqual(response.status_code, 302)
        campaign = Campaign.objects.get(name="List mailer")
        self.assertEqual(campaign.messages.count(), 1)
        self.assertEqual(campaign.messages.get().contact_id, self.contact.pk)

    def test_postcard_does_not_inject_into_body(self):
        campaign = create_campaign_draft(
            name="March mailer",
            audience=Campaign.Audience.POSTCARD_OPT_IN,
            body="Internal note only",
            created_by=self.user,
        )
        self.assertEqual(campaign.body_override, "Internal note only")
        self.assertNotIn("utm_source", 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_detail_shows_tracked_url(self):
        campaign = create_campaign_draft(
            name="March mailer",
            audience=Campaign.Audience.POSTCARD_OPT_IN,
            body="Internal note only",
            created_by=self.user,
        )
        url = reverse("directmail:campaign_detail", kwargs={"pk": campaign.pk})
        response = self.client.get(url)
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "utm_campaign=march-mailer")
        self.assertContains(response, "utm_medium=postcard")
        self.assertContains(response, "qrcode.min.js")
        self.assertNotContains(response, "Internal note onlyutm")