✅ [#2157] Add tests for newsletter subscription form

src/open_inwoner/accounts/tests/test_profile_views.py

@@ -6,7 +6,7 @@
 from django.template.defaultfilters import date as django_date
 from django.test import override_settings
 from django.urls import reverse
-from django.utils.translation import gettext_lazy as _
+from django.utils.translation import gettext as _
 
 import requests_mock
 from cms import api
@@ -17,10 +17,13 @@
 from open_inwoner.accounts.choices import StatusChoices
 from open_inwoner.cms.profile.cms_appconfig import ProfileConfig
 from open_inwoner.haalcentraal.tests.mixins import HaalCentraalMixin
+from open_inwoner.laposta.models import LapostaConfig
+from open_inwoner.laposta.tests.factories import LapostaListFactory, SubscriptionFactory
 from open_inwoner.openklant.models import OpenKlantConfig
 from open_inwoner.pdc.tests.factories import CategoryFactory
 from open_inwoner.plans.tests.factories import PlanFactory
 from open_inwoner.utils.logentry import LOG_ACTIONS
+from open_inwoner.utils.test import ClearCachesMixin
 from open_inwoner.utils.tests.helpers import AssertTimelineLogMixin, create_image_bytes
 
 from ...cms.profile.cms_apps import ProfileApphook
@@ -1048,3 +1051,63 @@ def test_collaborate_notifications_display(self):
         form = response.forms["change-notifications"]
 
         self.assertIn("plans_notifications", form.fields)
+
+
+@requests_mock.Mocker()
+@override_settings(
+    ROOT_URLCONF="open_inwoner.cms.tests.urls", MIDDLEWARE=PATCHED_MIDDLEWARE
+)
+class NewsletterSubscriptionTests(ClearCachesMixin, WebTest):
+    def setUp(self):
+        super().setUp()
+
+        self.profile_url = reverse("profile:detail")
+        self.user = DigidUserFactory()
+
+        self.config = LapostaConfig.get_solo()
+        self.config.api_root = "https://laposta.local/api/v2/"
+        self.config.basic_auth_username = "username"
+        self.config.basic_auth_password = "password"
+        self.config.save()
+
+    def test_do_not_render_form_if_config_is_missing(self, m):
+        self.config.api_root = ""
+        self.config.save()
+
+        response = self.app.get(self.profile_url, user=self.user)
+
+        self.assertNotIn(_("Nieuwsbrieven"), response.text)
+        self.assertNotIn("newsletter-form", response.forms)
+
+    def test_do_not_render_form_if_no_newsletters_are_found(self, m):
+        m.get("https://laposta.local/api/v2/list", json=[])
+
+        response = self.app.get(self.profile_url, user=self.user)
+
+        self.assertNotIn(_("Nieuwsbrieven"), response.text)
+        self.assertNotIn("newsletter-form", response.forms)
+
+    def test_render_form_if_newsletters_are_found(self, m):
+        list1, list2 = LapostaListFactory.build(
+            name="Nieuwsbrief1", remarks="foo"
+        ), LapostaListFactory.build(name="Nieuwsbrief2", remarks="bar")
+
+        SubscriptionFactory.create(list_id=list1.list_id, user=self.user)
+
+        m.get(
+            "https://laposta.local/api/v2/list",
+            json={"data": [{"list": list1.dict()}, {"list": list2.dict()}]},
+        )
+
+        response = self.app.get(self.profile_url, user=self.user)
+
+        self.assertIn(_("Nieuwsbrieven"), response.text)
+        self.assertIn("newsletter-form", response.forms)
+
+        form = response.forms["newsletter-form"]
+
+        # First checkbox should be checked, because the user is already subscribed
+        self.assertTrue(form.fields["newsletters"][0].checked)
+        self.assertFalse(form.fields["newsletters"][1].checked)
+        self.assertIn("Nieuwsbrief1: foo", response.text)
+        self.assertIn("Nieuwsbrief2: bar", response.text)

src/open_inwoner/laposta/api_models.py

@@ -1,8 +1,11 @@
 from datetime import datetime
+from ipaddress import IPv4Address
 from typing import Any, Literal
 
 from pydantic import BaseModel, EmailStr, HttpUrl, IPvAnyAddress, NonNegativeInt
 
+from open_inwoner.utils.time import convert_datetime_to_iso_8601_with_z_suffix
+
 
 class Members(BaseModel):
     active: NonNegativeInt
@@ -23,6 +26,20 @@ class LapostaList(BaseModel):
     account_id: str
     members: Members
 
+    # Let `.dict()` handle JSON encoding
+    # Source: https://github.com/pydantic/pydantic/issues/1409#issuecomment-1381655025
+    def _iter(self, **kwargs):
+        for key, value in super()._iter(**kwargs):
+            yield key, self.__config__.json_encoders.get(type(value), lambda v: v)(
+                value
+            )
+
+    class Config:
+        json_encoders = {
+            # custom output conversion for datetime
+            datetime: convert_datetime_to_iso_8601_with_z_suffix
+        }
+
 
 class UserData(BaseModel):
     ip: IPvAnyAddress
@@ -36,8 +53,23 @@ class Member(BaseModel):
     member_id: str
     list_id: str
     email: EmailStr
-    state: str
-    signup_date: datetime
+    state: str | None = None
+    signup_date: datetime | None = None
     ip: IPvAnyAddress
-    source_url: HttpUrl | Literal[""] | None
-    custom_fields: dict | None
+    source_url: HttpUrl | Literal[""] | None = None
+    custom_fields: dict | None = None
+
+    # Let `.dict()` handle JSON encoding
+    # Source: https://github.com/pydantic/pydantic/issues/1409#issuecomment-1381655025
+    def _iter(self, **kwargs):
+        for key, value in super()._iter(**kwargs):
+            yield key, self.__config__.json_encoders.get(type(value), lambda v: v)(
+                value
+            )
+
+    class Config:
+        json_encoders = {
+            # custom output conversion for datetime
+            datetime: convert_datetime_to_iso_8601_with_z_suffix,
+            IPv4Address: str,
+        }

src/open_inwoner/laposta/tests/factories.py

@@ -0,0 +1,22 @@
+import factory
+from polyfactory.factories.pydantic_factory import ModelFactory
+
+from open_inwoner.accounts.tests.factories import UserFactory
+
+from ..api_models import LapostaList, Member
+from ..models import Subscription
+
+
+class LapostaListFactory(ModelFactory[LapostaList]):
+    __model__ = LapostaList
+
+
+class MemberFactory(ModelFactory[Member]):
+    __model__ = Member
+
+
+class SubscriptionFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = Subscription
+
+    user = factory.SubFactory(UserFactory)

src/open_inwoner/laposta/tests/test_forms.py

@@ -0,0 +1,192 @@
+from urllib.parse import parse_qs
+
+from django.test import RequestFactory, TestCase
+
+import requests_mock
+
+from open_inwoner.accounts.tests.factories import DigidUserFactory
+from open_inwoner.utils.test import ClearCachesMixin
+
+from ..forms import NewsletterSubscriptionForm
+from ..models import LapostaConfig, Subscription
+from .factories import LapostaListFactory, MemberFactory, SubscriptionFactory
+
+LAPOSTA_API_ROOT = "https://laposta.local/api/v2/"
+
+
+@requests_mock.Mocker()
+class NewsletterSubscriptionFormTestCase(ClearCachesMixin, TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self.user = DigidUserFactory()
+
+        self.request = RequestFactory().get("/")
+        self.request.user = self.user
+
+        self.config = LapostaConfig.get_solo()
+        self.config.api_root = LAPOSTA_API_ROOT
+        self.config.basic_auth_username = "username"
+        self.config.basic_auth_password = "password"
+        self.config.save()
+
+        self.list1 = LapostaListFactory.build(
+            list_id="123", name="Nieuwsbrief1", remarks="foo"
+        )
+        self.list2 = LapostaListFactory.build(
+            list_id="456", name="Nieuwsbrief2", remarks="bar"
+        )
+        self.list3 = LapostaListFactory.build(
+            list_id="789", name="Nieuwsbrief3", remarks="baz"
+        )
+
+    def setUpMocks(self, m):
+        m.get(
+            f"{LAPOSTA_API_ROOT}list",
+            json={
+                "data": [
+                    {"list": self.list1.dict()},
+                    {"list": self.list2.dict()},
+                    {"list": self.list3.dict()},
+                ]
+            },
+        )
+
+    def test_save_form(self, m):
+        """
+        Verify that the form can create and delete subscriptions
+        """
+        self.setUpMocks(m)
+
+        SubscriptionFactory.create(list_id="123", member_id="member123", user=self.user)
+        SubscriptionFactory.create(list_id="456", member_id="member456", user=self.user)
+
+        form = NewsletterSubscriptionForm(data={}, user=self.user)
+
+        # User already has a subscription for the first two newsletters
+        self.assertEqual(form["newsletters"].initial, ["123", "456"])
+
+        form = NewsletterSubscriptionForm(
+            data={"newsletters": ["456", "789"]}, user=self.user
+        )
+
+        self.assertTrue(form.is_valid())
+
+        post_matcher = m.post(
+            f"{LAPOSTA_API_ROOT}member",
+            json={
+                "member": MemberFactory.build(
+                    list_id="789",
+                    member_id="member789",
+                    email=self.user.email,
+                    custom_fields=None,
+                ).dict()
+            },
+        )
+        delete_matcher = m.delete(f"{LAPOSTA_API_ROOT}member/member123?list_id=123")
+
+        form.save(self.request)
+
+        self.assertEqual(
+            len(post_matcher.request_history),
+            1,
+            "Subscribe to list if present in the form data (and no subscription exists yet)",
+        )
+
+        [post_request] = post_matcher.request_history
+
+        self.assertEqual(
+            parse_qs(post_request.body),
+            {"list_id": ["789"], "ip": ["127.0.0.1"], "email": [self.user.email]},
+        )
+
+        # Because list_id 123 was present in the
+        self.assertEqual(
+            len(delete_matcher.request_history),
+            1,
+            "Unsubscribe from list if not present in the form data",
+        )
+
+        subscriptions = Subscription.objects.filter(user=self.user)
+
+        self.assertEqual(subscriptions.count(), 2)
+
+        subscription1, subscription2 = subscriptions
+
+        self.assertEqual(subscription1.list_id, "456")
+        self.assertEqual(subscription1.member_id, "member456")
+        self.assertEqual(subscription2.list_id, "789")
+        self.assertEqual(subscription2.member_id, "member789")
+
+    def test_save_form_create_duplicate_subscription(self, m):
+        """
+        Verify that the client properly handles the scenario where the user is a member
+        of the list in the API, but no Subscription exists locally and tries to create one
+        """
+        self.setUpMocks(m)
+
+        form = NewsletterSubscriptionForm(data={"newsletters": ["789"]}, user=self.user)
+
+        self.assertTrue(form.is_valid())
+
+        post_matcher = m.post(
+            f"{LAPOSTA_API_ROOT}member",
+            json={
+                "error": {
+                    "type": "invalid_input",
+                    "message": "Email address exists",
+                    "code": 204,
+                    "parameter": "email",
+                    "id": "pqfozv6xqu",
+                    "member_id": "member789",
+                }
+            },
+            status_code=400,
+        )
+
+        form.save(self.request)
+
+        self.assertEqual(len(post_matcher.request_history), 1)
+
+        subscriptions = Subscription.objects.filter(user=self.user)
+
+        self.assertEqual(subscriptions.count(), 1)
+
+        subscription = subscriptions.get()
+
+        # Subscription should be created locally, based on data returned by API
+        self.assertEqual(subscription.list_id, "789")
+        self.assertEqual(subscription.member_id, "member789")
+
+    def test_save_form_delete_non_existent_subscription(self, m):
+        """
+        Verify that the client properly handles the scenario where the user has a
+        Subscription to a list locally and tries to delete it, but that relationship
+        does not exist in the API
+        """
+        self.setUpMocks(m)
+
+        SubscriptionFactory.create(list_id="789", member_id="member789", user=self.user)
+
+        form = NewsletterSubscriptionForm(data={"newsletters": []}, user=self.user)
+
+        self.assertTrue(form.is_valid())
+
+        delete_matcher = m.delete(
+            f"{LAPOSTA_API_ROOT}member/member789?list_id=789",
+            json={
+                "error": {
+                    "type": "invalid_input",
+                    "message": "Unknown member",
+                    "code": 203,
+                    "parameter": "member_id",
+                }
+            },
+            status_code=400,
+        )
+
+        form.save(self.request)
+
+        self.assertEqual(len(delete_matcher.request_history), 1)
+
+        self.assertFalse(Subscription.objects.filter(user=self.user).exists())

src/open_inwoner/utils/time.py

@@ -38,3 +38,7 @@ def has_new_elements(
     otherwise
     """
     return any(is_new(elem, attribute_name, delta) for elem in collection)
+
+
+def convert_datetime_to_iso_8601_with_z_suffix(dt: dt.datetime) -> str:
+    return dt.strftime("%Y-%m-%dT%H:%M:%SZ")