🗃️ [#2041] Added hard database constraints on User's identifier/login_type fields

src/open_inwoner/accounts/migrations/0076_user_unique_bsn_when_login_digid_and_more.py

@@ -0,0 +1,76 @@
+# Generated by Django 4.2.11 on 2024-04-15 13:28
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("accounts", "0075_user_verified_email"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("login_type", "digid")),
+                fields=("bsn",),
+                name="unique_bsn_when_login_digid",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(
+                    ("login_type", "eherkenning"), models.Q(("rsin", ""), _negated=True)
+                ),
+                fields=("rsin",),
+                name="unique_rsin_when_login_eherkenning",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(
+                    ("login_type", "eherkenning"), models.Q(("kvk", ""), _negated=True)
+                ),
+                fields=("kvk",),
+                name="unique_kvk_when_login_eherkenning",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("login_type", "oidc")),
+                fields=("oidc_id",),
+                name="unique_oidc_id_when_login_oidc",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.CheckConstraint(
+                check=models.Q(("bsn", ""), ("login_type", "digid"), _connector="OR"),
+                name="check_bsn_only_set_when_login_digid",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.CheckConstraint(
+                check=models.Q(
+                    ("oidc_id", ""), ("login_type", "oidc"), _connector="OR"
+                ),
+                name="check_oidc_id_only_set_when_login_oidc",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="user",
+            constraint=models.CheckConstraint(
+                check=models.Q(
+                    models.Q(("kvk", ""), ("rsin", "")),
+                    ("login_type", "eherkenning"),
+                    _connector="OR",
+                ),
+                name="check_kvk_or_rsin_only_set_when_login_eherkenning",
+            ),
+        ),
+    ]

src/open_inwoner/accounts/models.py

@@ -7,7 +7,7 @@
 from django.contrib.contenttypes.fields import GenericRelation
 from django.core.exceptions import ValidationError
 from django.db import models
-from django.db.models import Q, UniqueConstraint
+from django.db.models import CheckConstraint, Q, UniqueConstraint
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.crypto import get_random_string
@@ -132,8 +132,6 @@ def has_verified_email(self):
     date_joined = models.DateTimeField(
         verbose_name=_("Date joined"), default=timezone.now
     )
-    # TODO shouldn't rsin & bsn be unique? (possibly fixed in model constraints)
-    # TODO fix rsin & bsn to not be both null AND blank (!)
     rsin = models.CharField(
         verbose_name=_("Rsin"), max_length=9, blank=True, default=""
     )
@@ -255,31 +253,43 @@ class Meta:
                 & ~Q(login_type=LoginTypeChoices.eherkenning),
                 name="unique_email_when_not_digid_or_eherkenning",
             ),
-            # UniqueConstraint(
-            #     fields=["bsn"],
-            #     condition=Q(login_type=LoginTypeChoices.digid)
-            #     # not quite sure if we want this (bsn shouldn't be null AND blank)
-            #     & ~(Q(bsn="") | Q(bsn__isnull=True)),
-            #     name="unique_bsn_when_digid",
-            # ),
-            # UniqueConstraint(
-            #     fields=["rsin"],
-            #     condition=Q(login_type=LoginTypeChoices.eherkenning)
-            #     # not quite sure if we want this (rsin shouldn't be null AND blank)
-            #     & ~(Q(rsin="") | Q(rsin__isnull=True)),
-            #     name="unique_rsin_when_eherkenning",
-            # ),
-            # UniqueConstraint(
-            #     fields=["oidc_id"],
-            #     condition=Q(login_type=LoginTypeChoices.oidc),
-            #     name="unique_bsn_when_digid",
-            # ),
-            # CheckConstraint(
-            #     # maybe this is not correct?
-            #     check=(Q(bsn="") | Q(bsn__isnull=True))
-            #     | ~Q(login_type=LoginTypeChoices.digid),
-            #     name="check_digid_bsn_required_when_digid",
-            # ),
+            UniqueConstraint(
+                fields=["bsn"],
+                condition=Q(login_type=LoginTypeChoices.digid),
+                name="unique_bsn_when_login_digid",
+            ),
+            UniqueConstraint(
+                fields=["rsin"],
+                condition=Q(login_type=LoginTypeChoices.eherkenning) & ~Q(rsin=""),
+                name="unique_rsin_when_login_eherkenning",
+            ),
+            UniqueConstraint(
+                fields=["kvk"],
+                condition=Q(login_type=LoginTypeChoices.eherkenning) & ~Q(kvk=""),
+                name="unique_kvk_when_login_eherkenning",
+            ),
+            UniqueConstraint(
+                fields=["oidc_id"],
+                condition=Q(login_type=LoginTypeChoices.oidc),
+                name="unique_oidc_id_when_login_oidc",
+            ),
+            # --- --- ---
+            CheckConstraint(
+                check=Q(bsn="") | Q(login_type=LoginTypeChoices.digid),
+                name="check_bsn_only_set_when_login_digid",
+            ),
+            CheckConstraint(
+                check=Q(oidc_id="") | Q(login_type=LoginTypeChoices.oidc),
+                name="check_oidc_id_only_set_when_login_oidc",
+            ),
+            CheckConstraint(
+                check=(Q(kvk="") & Q(rsin=""))
+                | Q(login_type=LoginTypeChoices.eherkenning),
+                name="check_kvk_or_rsin_only_set_when_login_eherkenning",
+            ),
+            # NOTE: we do not need a constraint that enforces exclusivity between
+            # `KVK` and `RSIN`, because companies have both of these attributes and we
+            # need both of them to fetch cases for these companies
         ]
 
     def __init__(self, *args, **kwargs):

src/open_inwoner/accounts/tests/test_auth.py

@@ -1134,7 +1134,7 @@ def test_eherkenning_user_success(self, mock_kvk, mock_retrieve_rsin_with_kvk):
         test_user = eHerkenningUserFactory.create(
             email="test@localhost",
             kvk="64819772",
-            rsin="123456789",
+            rsin="123456782",
         )
 
         url = reverse("eherkenning-mock:password")

src/open_inwoner/accounts/tests/test_contact_views.py

@@ -173,7 +173,7 @@ def test_existing_user_contact(self):
         self.assertEqual(existing_user, pending_invitation)
 
     def test_existing_user_contact_with_case_sensitive_email(self):
-        existing_user = UserFactory(email="[email protected]", bsn="111111111")
+        existing_user = DigidUserFactory(email="[email protected]", bsn="111111111")
         response = self.app.get(self.create_url, user=self.user)
         form = response.forms["contact-form"]
 

src/open_inwoner/accounts/tests/test_oidc_views.py

@@ -1055,7 +1055,7 @@ def test_new_user_is_created_when_new_kvk(
         # set up a user with a non existing email address
         mock_get_userinfo.return_value = {"sub": "00000000"}
         eHerkenningUserFactory.create(
-            kvk="12345678", rsin="123456789", email="[email protected]"
+            kvk="12345678", rsin="123456782", email="[email protected]"
         )
         session = self.client.session
         session["oidc_states"] = {"mock": {"nonce": "nonce"}}

src/open_inwoner/accounts/tests/test_profile_views.py

@@ -213,6 +213,7 @@ def test_info_eherkenning_user(self):
             housenumber="42",
             postcode="1234 XY",
             city="The good place",
+            kvk="87654321",
         )
         response = self.app.get(self.url, user=user)
 
@@ -511,7 +512,6 @@ def test_modify_phone_and_email_updates_klant_api(self, m):
         response = self.app.get(self.url, user=data.user)
 
         # reset noise from signals
-        m.reset_mock()
         self.clearTimelineLogs()
 
         form = response.forms["profile-edit"]
@@ -536,52 +536,76 @@ def test_modify_phone_and_email_updates_klant_api(self, m):
         )
 
     @requests_mock.Mocker()
-    def test_eherkenning_user_updates_klant_api(self, m):
+    def test_eherkenning_user_updates_klant_api_with_rsin(self, m):
         MockAPIReadPatchData.setUpServices()
+        data = MockAPIReadPatchData()
+
+        matchers = data.install_mocks_eherkenning(m, use_rsin=True).matchers
+
+        config = OpenKlantConfig.get_solo()
+        config.use_rsin_for_innNnpId_query_parameter = True
+        config.save()
+
+        response = self.app.get(self.url, user=data.eherkenning_user)
+
+        # reset noise from signals
+        self.clearTimelineLogs()
 
-        for use_rsin_for_innNnpId_query_parameter in [True, False]:
-            with self.subTest(
-                use_rsin_for_innNnpId_query_parameter=use_rsin_for_innNnpId_query_parameter
-            ):
-                # NOTE Explicitly creating a new Mocker object here, because for some reason
-                # `m` is overridden somewhere, which causes issues when `MockAPIReadPatchData.install_mocks`
-                # is run for the second time
-                with requests_mock.Mocker() as m:
-                    data = MockAPIReadPatchData().install_mocks_eherkenning(
-                        m, use_rsin=use_rsin_for_innNnpId_query_parameter
-                    )
-
-                    config = OpenKlantConfig.get_solo()
-                    config.use_rsin_for_innNnpId_query_parameter = (
-                        use_rsin_for_innNnpId_query_parameter
-                    )
-                    config.save()
-
-                    response = self.app.get(self.url, user=data.eherkenning_user)
-
-                    # reset noise from signals
-                    m.reset_mock()
-                    self.clearTimelineLogs()
-
-                    form = response.forms["profile-edit"]
-                    form["email"] = "[email protected]"
-                    form["phonenumber"] = "0612345678"
-                    form.submit()
-
-                    # user data tested in other cases
-                    self.assertTrue(data.matchers[0].called)
-                    klant_patch_data = data.matchers[1].request_history[0].json()
-                    self.assertEqual(
-                        klant_patch_data,
-                        {
-                            "emailadres": "[email protected]",
-                            "telefoonnummer": "0612345678",
-                        },
-                    )
-                    self.assertTimelineLog("retrieved klant for user")
-                    self.assertTimelineLog(
-                        "patched klant from user profile edit with fields: emailadres, telefoonnummer"
-                    )
+        form = response.forms["profile-edit"]
+        form["email"] = "[email protected]"
+        form["phonenumber"] = "0612345678"
+        form.submit()
+
+        # user data tested in other cases
+        self.assertTrue(matchers[0].called)
+        klant_patch_data = matchers[1].request_history[0].json()
+        self.assertEqual(
+            klant_patch_data,
+            {
+                "emailadres": "[email protected]",
+                "telefoonnummer": "0612345678",
+            },
+        )
+        self.assertTimelineLog("retrieved klant for user")
+        self.assertTimelineLog(
+            "patched klant from user profile edit with fields: emailadres, telefoonnummer"
+        )
+
+    @requests_mock.Mocker()
+    def test_eherkenning_user_updates_klant_api_with_kvk(self, m):
+        MockAPIReadPatchData.setUpServices()
+        data = MockAPIReadPatchData()
+
+        matchers = data.install_mocks_eherkenning(m, use_rsin=False).matchers
+
+        config = OpenKlantConfig.get_solo()
+        config.use_rsin_for_innNnpId_query_parameter = False
+        config.save()
+
+        response = self.app.get(self.url, user=data.eherkenning_user)
+
+        # reset noise from signals
+        self.clearTimelineLogs()
+
+        form = response.forms["profile-edit"]
+        form["email"] = "[email protected]"
+        form["phonenumber"] = "0612345678"
+        form.submit()
+
+        # user data tested in other cases
+        self.assertTrue(matchers[0].called)
+        klant_patch_data = matchers[1].request_history[0].json()
+        self.assertEqual(
+            klant_patch_data,
+            {
+                "emailadres": "[email protected]",
+                "telefoonnummer": "0612345678",
+            },
+        )
+        self.assertTimelineLog("retrieved klant for user")
+        self.assertTimelineLog(
+            "patched klant from user profile edit with fields: emailadres, telefoonnummer"
+        )
 
     @requests_mock.Mocker()
     def test_modify_phone_updates_klant_api_but_skips_unchanged(self, m):
@@ -610,7 +634,7 @@ def test_modify_phone_updates_klant_api_but_skip_unchanged_email(self, m):
         response = self.app.get(self.url, user=data.user)
 
         # reset noise from signals
-        m.reset_mock()
+        # m.reset_mock()
         self.clearTimelineLogs()
 
         form = response.forms["profile-edit"]
@@ -640,7 +664,7 @@ def test_modify_phone_updates_klant_api_but_skip_unchanged_phone(self, m):
         response = self.app.get(self.url, user=data.user)
 
         # reset noise from signals
-        m.reset_mock()
+        # m.reset_mock()
         self.clearTimelineLogs()
 
         form = response.forms["profile-edit"]