[#1874] Enabled selection of company branch for eHerkenning users

src/eherkenning/tests/test_mock_views.py

@@ -19,6 +19,7 @@ def assertNoEHerkenningURLS(self, response):
 
 
 OVERRIDE_SETTINGS = dict(
+    ROOT_URLCONF="open_inwoner.cms.tests.urls",
     EHERKENNING_MOCK_APP_TITLE="FooBarBazz-MockApp",
     EHERKENNING_MOCK_RETURN_URL=RETURN_URL,  # url to redirect to after success
     EHERKENNING_MOCK_CANCEL_URL=CANCEL_URL,  # url to navigate to when users clicks 'cancel/annuleren'
@@ -106,7 +107,13 @@ def test_get_returns_valid_response(self):
         self.assertContains(response, reverse("eherkenning-mock:login"))
         self.assertNoEHerkenningURLS(response)
 
-    def test_post_redirects_and_authenticates(self):
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
+    def test_post_redirects_and_authenticates(self, mock_kvk):
+        mock_kvk.return_value = [
+            {"kvkNummer": "29664887", "vestigingsnummer": "1234"},
+            {"kvkNummer": "29664887", "vestigingsnummer": "5678"},
+        ]
+
         url = reverse("eherkenning-mock:password")
         params = {
             "acs": reverse("eherkenning:acs"),
@@ -117,8 +124,9 @@ def test_post_redirects_and_authenticates(self):
 
         data = {
             "auth_name": "29664887",
-            "auth_pass": "bar",
+            "auth_pass": "company@localhost",
         }
+
         # post our password to the IDP
         response = self.client.post(url, data, follow=False)
 
@@ -129,15 +137,98 @@ def test_post_redirects_and_authenticates(self):
         # follow the ACS redirect and get/create the user
         response = self.client.get(response["Location"], follow=False)
 
+        # redirect to kvk/branches
+        response = self.client.get(response["Location"])
+        self.assertIn(reverse("kvk:branches"), response["Location"])
+
+        # post branch_number, follow redirect back to register/necessary
+        response = self.client.post(
+            response["Location"], {"branch_number": "1234"}, follow=True
+        )
+        self.assertRedirects(response, reverse("profile:registration_necessary"))
+
+        # check user kvk
         User = get_user_model()
-        user = User.eherkenning_objects.get(kvk="29664887")
+        self.assertEqual(
+            response.context["user"].kvk, User.eherkenning_objects.get().kvk
+        )
 
-        # follow redirect to 'next'
-        self.assertRedirects(response, RETURN_URL)
+        # check company branch number in session
+        self.assertEqual(self.client.session["KVK_BRANCH_NUMBER"], "1234")
 
-        response = self.client.get(response["Location"], follow=False)
-        self.assertEqual(response.status_code, 200)
-        self.assertEqual(response.context["user"].kvk, "29664887")
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
+    def test_redirect_flow_with_single_company(self, mock_kvk):
+        """
+        Assert that if the KvK API returns only a single company:
+            1. the redirect flow passes automatically through `KvKLoginMiddleware`
+            2. the company vestigingsnummer is stored in the session
+        """
+        mock_kvk.return_value = [
+            {"kvkNummer": "29664887", "vestigingsnummer": "1234"},
+        ]
+
+        url = reverse("eherkenning-mock:password")
+        params = {
+            "acs": reverse("eherkenning:acs"),
+            "next": RETURN_URL,
+            "cancel": CANCEL_URL,
+        }
+        url = f"{url}?{urlencode(params)}"
+
+        data = {
+            "auth_name": "29664887",
+            "auth_pass": "company@localhost",
+        }
+
+        # post our password to the IDP
+        response = self.client.post(url, data, follow=False)
+
+        # it will redirect to our ACS
+        self.assertEqual(response.status_code, 302)
+        self.assertIn(reverse("eherkenning:acs"), response["Location"])
+
+        # follow the ACS redirect and get/create the user
+        response = self.client.get(response["Location"], follow=True)
+
+        # check company branch number in session
+        self.assertEqual(self.client.session["KVK_BRANCH_NUMBER"], "1234")
+
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
+    def test_redirect_flow_with_no_vestigingsnummer(self, mock_kvk):
+        """
+        Assert that if the KvK API returns only a single company without vestigingsnummer:
+            1. the redirect flow passes automatically through `KvKLoginMiddleware`
+            2. the company KvKNummer is stored in the session
+        """
+        mock_kvk.return_value = [
+            {"kvkNummer": "29664887"},
+        ]
+
+        url = reverse("eherkenning-mock:password")
+        params = {
+            "acs": reverse("eherkenning:acs"),
+            "next": RETURN_URL,
+            "cancel": CANCEL_URL,
+        }
+        url = f"{url}?{urlencode(params)}"
+
+        data = {
+            "auth_name": "29664887",
+            "auth_pass": "company@localhost",
+        }
+
+        # post our password to the IDP
+        response = self.client.post(url, data, follow=False)
+
+        # it will redirect to our ACS
+        self.assertEqual(response.status_code, 302)
+        self.assertIn(reverse("eherkenning:acs"), response["Location"])
+
+        # follow the ACS redirect and get/create the user
+        response = self.client.get(response["Location"], follow=True)
+
+        # check company branch number in session
+        self.assertEqual(self.client.session["KVK_BRANCH_NUMBER"], "29664887")
 
     def test_post_redirect_retains_acs_querystring_params(self):
         url = reverse("eherkenning-mock:password")

src/open_inwoner/accounts/middleware.py

@@ -43,6 +43,7 @@ def __call__(self, request):
                         reverse("logout"),
                         digid_logout,
                         digid_slo_redirect,
+                        reverse("kvk:branches"),
                     )
                 )
                 and request.user.require_necessary_fields()

src/open_inwoner/accounts/tests/test_auth.py

@@ -4,7 +4,7 @@
 
 from django.contrib.sites.models import Site
 from django.core import mail
-from django.test import override_settings
+from django.test import modify_settings, override_settings
 from django.urls import reverse, reverse_lazy
 from django.utils.translation import gettext as _
 
@@ -18,9 +18,9 @@
     OpenIDConnectEHerkenningConfig,
 )
 from open_inwoner.configurations.models import SiteConfiguration
-from open_inwoner.contrib.kvk.models import KvKConfig
-from open_inwoner.contrib.kvk.tests.factories import CertificateFactory
 from open_inwoner.haalcentraal.tests.mixins import HaalCentraalMixin
+from open_inwoner.kvk.models import KvKConfig
+from open_inwoner.kvk.tests.factories import CertificateFactory
 
 from ...cms.tests import cms_tools
 from ...utils.test import ClearCachesMixin
@@ -573,10 +573,18 @@ def test_eherkenning_fail_with_invite_redirects_to_register_page(self, m):
             f"http://testserver{reverse('django_registration_register')}?invite={invite.key}",
         )
 
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
     @patch(
-        "open_inwoner.contrib.kvk.models.KvKConfig.get_solo",
+        "open_inwoner.kvk.models.KvKConfig.get_solo",
     )
-    def test_invite_url_not_in_session_after_successful_login(self, mock_solo):
+    def test_invite_url_not_in_session_after_successful_login(
+        self, mock_solo, mock_kvk
+    ):
+        mock_kvk.return_value = [
+            {"kvkNummer": "12345678", "vestigingsnummer": "1234"},
+        ]
+
+        mock_solo.return_value.api_key = "123"
         mock_solo.return_value.api_root = "http://foo.bar/api/v1/"
         mock_solo.return_value.client_certificate = CertificateFactory()
         mock_solo.return_value.server_certificate = CertificateFactory()
@@ -603,24 +611,97 @@ def test_invite_url_not_in_session_after_successful_login(self, mock_solo):
         self.assertIn("invite_url", self.client.session.keys())
 
         # post our password to the IDP
-        response = self.client.post(url, data, follow=True)
+        response = self.client.post(url, data, follow=False)
+
+        # follow redirect flow
+        res = self.client.get(response["Location"])
+        res = self.client.get(res["Location"])
+        res = self.client.get(res["Location"])
 
         self.assertRedirects(
-            response,
-            f"{reverse('profile:registration_necessary')}?invite={invite.key}",
+            res, f"{reverse('profile:registration_necessary')}?invite={invite.key}"
         )
         self.assertNotIn("invite_url", self.client.session.keys())
 
-    def test_eherkenning_user_is_redirected_to_necessary_registration(self):
+        # check company branch number in session
+        self.assertEqual(self.client.session["KVK_BRANCH_NUMBER"], "1234")
+
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
+    @patch(
+        "open_inwoner.kvk.models.KvKConfig.get_solo",
+    )
+    def test_redirect_flow_with_no_vestigingsnummer(self, mock_solo, mock_kvk):
+        """
+        Assert that if the KvK API returns only a single company without vestigingsnummer:
+            1. the redirect flow passes automatically through `KvKLoginMiddleware`
+            2. the company KvKNummer is stored in the session
+        """
+        mock_kvk.return_value = [
+            {"kvkNummer": "12345678"},
+        ]
+
+        mock_solo.return_value.api_key = "123"
+        mock_solo.return_value.api_root = "http://foo.bar/api/v1/"
+        mock_solo.return_value.client_certificate = CertificateFactory()
+        mock_solo.return_value.server_certificate = CertificateFactory()
+
+        user = eHerkenningUserFactory.create(
+            kvk="12345678", email="user-12345678@localhost"
+        )
+
+        url = reverse("eherkenning-mock:password")
+        params = {
+            "acs": reverse("eherkenning:acs"),
+            "next": "/",
+        }
+        url = f"{url}?{urlencode(params)}"
+
+        data = {
+            "auth_name": "12345678",
+            "auth_pass": "foo",
+        }
+
+        response = self.client.get(url, user=user)
+
+        # post our password to the IDP
+        response = self.client.post(url, data, user=user, follow=True)
+        self.assertRedirects(
+            response,
+            f"{reverse('profile:registration_necessary')}",
+        )
+
+        # check company branch number in session
+        self.assertEqual(self.client.session["KVK_BRANCH_NUMBER"], "12345678")
+
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
+    @patch(
+        "open_inwoner.kvk.models.KvKConfig.get_solo",
+    )
+    def test_eherkenning_user_is_redirected_to_necessary_registration(
+        self, mock_solo, mock_kvk
+    ):
         """
         eHerkenning users that do not have their email filled in should be redirected to
         the registration form
         """
-        user = eHerkenningUserFactory(kvk="12345678", email="user-12345678@localhost")
+        mock_kvk.return_value = [
+            {"kvkNummer": "12345678", "vestigingsnummer": "1234"},
+        ]
+
+        mock_solo.return_value.api_key = "123"
+        mock_solo.return_value.api_root = "http://foo.bar/api/v1/"
+        mock_solo.return_value.client_certificate = CertificateFactory()
+        mock_solo.return_value.server_certificate = CertificateFactory()
+
+        user = eHerkenningUserFactory.create(kvk="12345678", email="example@localhost")
 
         response = self.app.get(reverse("pages-root"), user=user)
 
-        self.assertRedirects(response, reverse("profile:registration_necessary"))
+        # redirect to /kvk/branches/
+        res = self.app.post(response["Location"], {"branch_number": "1234"})
+
+        # redirect to /register/necessary/
+        self.assertRedirects(res, reverse("profile:registration_necessary"))
 
 
 @override_settings(ROOT_URLCONF="open_inwoner.cms.tests.urls")
@@ -914,9 +995,6 @@ def setUpTestData(cls):
         cls.msg_dupes = _("This email is already taken.")
         cls.msg_inactive = _("This account has been deactivated")
 
-    #
-    # digid users
-    #
     def test_digid_user_success(self):
         """Assert that digid users can register with duplicate emails"""
         test_user = DigidUserFactory.create(
@@ -952,8 +1030,15 @@ def test_digid_user_success(self):
         self.assertEqual(users.first().email, "[email protected]")
         self.assertEqual(users.last().email, "[email protected]")
 
-    def test_eherkenning_user_success(self):
+    @patch("open_inwoner.kvk.client.KvKClient.get_all_company_branches")
+    def test_eherkenning_user_success(self, mock_kvk):
         """Assert that eHerkenning users can register with duplicate emails"""
+
+        mock_kvk.return_value = [
+            {"kvkNummer": "12345678", "vestigingsnummer": "1234"},
+            {"kvkNummer": "12345678", "vestigingsnummer": "5678"},
+        ]
+
         test_user = eHerkenningUserFactory.create(
             email="test@localhost",
             kvk="64819772",
@@ -974,6 +1059,13 @@ def test_eherkenning_user_success(self):
         # post our password to the IDP
         response = self.app.post(url, data).follow().follow()
 
+        # select company branch
+        response = self.app.get(response["Location"])
+        form = response.forms["eherkenning-branch-form"]
+        form["branch_number"] = "5678"
+        response = form.submit().follow()
+
+        # fill in necessary fields form
         form = response.forms["necessary-form"]
 
         self.assertEqual(form["email"].value, "")