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