www/membership/members/tests/test_newsletter.py

import os
import re
import unittest
from unittest.mock import patch

from django.conf import settings
from django.core import mail
from django.test import SimpleTestCase, TestCase, override_settings
from django.urls import reverse

from members import newsletter
from members.forms import RegistrationForm
from members.models import Account, Registration

ENABLED = {"LISTMONK_ENABLED": True, "LISTMONK_URL": "https://listmonk.example.org", "LISTMONK_API_USER": "membership", "LISTMONK_API_TOKEN": "test-token", "LISTMONK_LIST_NAME": "eom.dev"}
DN = "uid=alice,ou=people,dc=example,dc=org"


class FakeListmonk:
    """Just enough of listmonk's subscriber API, scoped like the real API user."""

    def __init__(self):
        self.lists = [{"id": 3, "name": "eom.dev"}]
        self.subscribers = {}
        self.calls = []

    def add(self, email, status="enabled", sub_status="confirmed"):
        sid = len(self.subscribers) + 1
        self.subscribers[sid] = {"id": sid, "email": email, "status": status, "lists": [{"id": 3, "subscription_status": sub_status}]}
        return sid

    def __call__(self, method, path, *, params=None, payload=None):
        self.calls.append((method, path, payload))
        if (method, path) == ("GET", "/api/lists"):
            return self.lists
        if (method, path) == ("GET", "/api/subscribers"):
            return {"results": [s for s in self.subscribers.values() if params["search"] in s["email"]]}
        if (method, path) == ("POST", "/api/subscribers"):
            if any(s["email"] == payload["email"] for s in self.subscribers.values()):
                raise newsletter.Conflict()
            self.add(payload["email"])
            return {}
        if (method, path) == ("PUT", "/api/subscribers/lists"):
            for sid in payload["ids"]:
                status = {"add": payload.get("status", "unconfirmed"), "unsubscribe": "unsubscribed"}[payload["action"]]
                self.subscribers[sid]["lists"] = [{"id": 3, "subscription_status": status}]
            return True
        if method == "PATCH":
            self.subscribers[int(path.rsplit("/", 1)[1])].update(payload)
            return {}
        raise AssertionError(f"unexpected call {method} {path}")


class NewsletterClientTests(SimpleTestCase):
    def setUp(self):
        newsletter._list_id = None
        self.fake = FakeListmonk()
        patcher = patch("members.newsletter._request", side_effect=self.fake)
        patcher.start()
        self.addCleanup(patcher.stop)

    def writes(self):
        return [c for c in self.fake.calls if c[0] != "GET"]

    def test_new_address_is_created_preconfirmed_on_the_list(self):
        self.assertEqual(newsletter.subscribe("alice@example.org", uid="alice"), newsletter.SUBSCRIBED)
        (method, path, payload), = self.writes()
        self.assertEqual((method, path), ("POST", "/api/subscribers"))
        self.assertEqual(payload["lists"], [3])
        self.assertTrue(payload["preconfirm_subscriptions"])
        self.assertEqual(payload["attribs"], {"source": "membership", "uid": "alice"})
        self.assertEqual(newsletter.status("alice@example.org"), newsletter.SUBSCRIBED)

    def test_unsubscribed_address_is_resubscribed_and_matches_exactly(self):
        self.fake.add("xalice@example.org")
        sid = self.fake.add("alice@example.org", sub_status="unsubscribed")
        self.assertEqual(newsletter.status("ALICE@example.org"), newsletter.UNSUBSCRIBED)
        newsletter.subscribe("alice@example.org", uid="alice")
        self.assertEqual(self.writes(), [("PUT", "/api/subscribers/lists", {"ids": [sid], "action": "add", "target_list_ids": [3], "status": "confirmed"})])

    def test_blocklisted_address_is_never_resubscribed(self):
        self.fake.add("alice@example.org", status="blocklisted", sub_status="unsubscribed")
        self.assertEqual(newsletter.status("alice@example.org"), newsletter.BLOCKED)
        self.assertEqual(newsletter.subscribe("alice@example.org", uid="alice"), newsletter.BLOCKED)
        self.assertEqual(self.writes(), [])

    def test_unsubscribe_and_change_email(self):
        sid = self.fake.add("alice@example.org")
        newsletter.unsubscribe("alice@example.org")
        self.assertEqual(newsletter.status("alice@example.org"), newsletter.UNSUBSCRIBED)
        newsletter.change_email("alice@example.org", "alice@new.example.org")
        self.assertEqual(self.fake.subscribers[sid]["email"], "alice@new.example.org")
        newsletter.unsubscribe("nobody@example.org")
        newsletter.change_email("nobody@example.org", "x@example.org")
        self.assertEqual(len(self.writes()), 2)

    def test_change_email_onto_an_existing_subscriber_merges(self):
        old = self.fake.add("alice@example.org")
        new = self.fake.add("alice@new.example.org", sub_status="unsubscribed")
        newsletter.change_email("alice@example.org", "alice@new.example.org")
        self.assertFalse(any(c[0] == "PATCH" for c in self.fake.calls))
        self.assertEqual(newsletter.status("alice@new.example.org"), newsletter.SUBSCRIBED)
        self.assertEqual(newsletter.status("alice@example.org"), newsletter.UNSUBSCRIBED)
        self.assertNotEqual(old, new)

    def test_missing_list_is_an_error(self):
        self.fake.lists = []
        with self.assertRaises(ValueError):
            newsletter.status("alice@example.org")


class RegistrationFormTests(SimpleTestCase):
    def test_checkbox_only_when_configured_and_unticked(self):
        with override_settings(LISTMONK_ENABLED=False):
            self.assertNotIn("newsletter", RegistrationForm().fields)
        with override_settings(**ENABLED):
            form = RegistrationForm()
            self.assertIn("newsletter", form.fields)
            self.assertNotIn("checked", str(form["newsletter"]))
            form = RegistrationForm({"uid": "alice", "email": "alice@example.org"})
            self.assertTrue(form.is_valid())
            self.assertFalse(form.cleaned_data["newsletter"])


@override_settings(**ENABLED)
@patch("members.views.directory.create_account", return_value={"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"})
class SignupTests(TestCase):
    password = "correct-horse-purple-planet-943!"

    def sign_up(self, **extra):
        self.client.post("/members/register/", {"uid": "alice", "email": "alice@example.org", **extra})
        token = re.search(r"/verify/([^/]+)/", mail.outbox[-1].body).group(1)
        self.client.get(f"/members/verify/{token}/")
        return self.client.post("/members/register/finish/", {"password": self.password, "confirm_password": self.password})

    @patch("members.views.newsletter.subscribe", return_value=newsletter.SUBSCRIBED)
    def test_opt_in_subscribes_after_the_account_exists(self, subscribe, provision):
        subscribe.side_effect = lambda *a, **k: self.assertTrue(Account.objects.exists()) or newsletter.SUBSCRIBED
        response = self.sign_up(newsletter="on")
        self.assertContains(response, "subscribed to premiere emails")
        subscribe.assert_called_once_with("alice@example.org", uid="alice")

    @patch("members.views.newsletter.subscribe")
    def test_no_opt_in_no_subscription(self, subscribe, provision):
        self.assertContains(self.sign_up(), "Account created")
        subscribe.assert_not_called()

    @patch("members.views.newsletter.subscribe", side_effect=ConnectionError("down"))
    def test_newsletter_outage_never_blocks_the_account(self, subscribe, provision):
        response = self.sign_up(newsletter="on")
        self.assertContains(response, "Account created")
        self.assertContains(response, "try again from your profile")
        self.assertEqual(Account.objects.count(), 1)
        self.assertFalse(Registration.objects.exists())


@override_settings(**ENABLED)
class PreferenceTests(TestCase):
    def setUp(self):
        self.account = Account.objects.create(uid="alice", email="alice@example.org", ldap_dn=DN, ldap_uuid="uuid-alice")
        session = self.client.session
        session["account_id"] = str(self.account.pk)
        session.save()
        self.url = reverse("members:newsletter")

    @patch("members.views.newsletter.subscribe", return_value=newsletter.SUBSCRIBED)
    @patch("members.views.newsletter.unsubscribe", return_value=newsletter.UNSUBSCRIBED)
    def test_toggle_uses_the_account_address(self, unsubscribe, subscribe):
        self.assertRedirects(self.client.post(self.url, {"action": "subscribe", "email": "evil@example.org"}), reverse("members:profile") + "?newsletter=saved#newsletter", fetch_redirect_response=False)
        subscribe.assert_called_once_with("alice@example.org", uid="alice")
        self.client.post(self.url, {"action": "unsubscribe"})
        unsubscribe.assert_called_once_with("alice@example.org")
        self.assertEqual(self.client.post(self.url, {"action": "delete"}).status_code, 400)
        self.assertEqual(self.client.get(self.url).status_code, 405)

    @patch("members.views.newsletter.subscribe", side_effect=ConnectionError("down"))
    def test_outage_reports_error(self, subscribe):
        self.assertRedirects(self.client.post(self.url, {"action": "subscribe"}), reverse("members:profile") + "?newsletter=error#newsletter", fetch_redirect_response=False)

    @patch("members.views.newsletter.subscribe")
    def test_requires_sign_in_and_configuration(self, subscribe):
        self.client.session.flush()
        self.client.cookies.clear()
        self.assertRedirects(self.client.post(self.url, {"action": "subscribe"}), reverse("members:home"), fetch_redirect_response=False)
        with override_settings(LISTMONK_ENABLED=False):
            session = self.client.session
            session["account_id"] = str(self.account.pk)
            session.save()
            self.assertEqual(self.client.post(self.url, {"action": "subscribe"}).status_code, 400)
        subscribe.assert_not_called()


@unittest.skipUnless(os.environ.get("LISTMONK_TEST_URL"), "Set LISTMONK_TEST_URL/USER/TOKEN to test against a disposable listmonk.")
class LiveListmonkTests(SimpleTestCase):
    """Against a throwaway listmonk configured by ansible-role-listmonk."""

    def test_round_trip(self):
        with override_settings(LISTMONK_URL=os.environ["LISTMONK_TEST_URL"], LISTMONK_API_USER=os.environ["LISTMONK_TEST_USER"], LISTMONK_API_TOKEN=os.environ["LISTMONK_TEST_TOKEN"], LISTMONK_LIST_NAME="eom.dev"):
            newsletter._list_id = None
            email = "live-test@example.org"
            self.assertEqual(newsletter.subscribe(email, uid="livetest"), newsletter.SUBSCRIBED)
            self.assertEqual(newsletter.status(email), newsletter.SUBSCRIBED)
            self.assertEqual(newsletter.unsubscribe(email), newsletter.UNSUBSCRIBED)
            self.assertEqual(newsletter.status(email), newsletter.UNSUBSCRIBED)
            self.assertEqual(newsletter.subscribe(email, uid="livetest"), newsletter.SUBSCRIBED)
            self.assertEqual(newsletter.status(email), newsletter.SUBSCRIBED)
            newsletter.change_email(email, "live-test-2@example.org")
            self.assertEqual(newsletter.status("live-test-2@example.org"), newsletter.SUBSCRIBED)
            self.assertEqual(newsletter.status(email), newsletter.UNSUBSCRIBED)
            newsletter.unsubscribe("live-test-2@example.org")