[#1874] Enable selection of company branch at eHerkenning login

src/eherkenning/tests/test_mock_views.py

@@ -7,6 +7,8 @@
 
 from furl import furl
 
+from open_inwoner.accounts.tests.factories import eHerkenningUserFactory
+
 RETURN_URL = "/"
 CANCEL_URL = reverse("login")
 
@@ -117,7 +119,7 @@ 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)
@@ -127,15 +129,14 @@ def test_post_redirects_and_authenticates(self):
         self.assertIn(reverse("eherkenning:acs"), response["Location"])
 
         # follow the ACS redirect and get/create the user
-        response = self.client.get(response["Location"], follow=False)
+        response = self.client.get(response["Location"], follow=True)
 
         User = get_user_model()
-        user = User.eherkenning_objects.get(kvk="29664887")
-
-        # follow redirect to 'next'
-        self.assertRedirects(response, RETURN_URL)
+        User.eherkenning_objects.get(kvk="29664887")
 
-        response = self.client.get(response["Location"], follow=False)
+        # TODO: double-check redirect flow
+        # follow redirect to "registration_necessary"
+        self.assertRedirects(response, reverse("profile:registration_necessary"))
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.context["user"].kvk, "29664887")
 

src/open_inwoner/accounts/tests/test_auth.py

@@ -618,7 +618,7 @@ def test_eherkenning_user_is_redirected_to_necessary_registration(self):
         """
         user = eHerkenningUserFactory(kvk="12345678", email="user-12345678@localhost")
 
-        response = self.app.get(reverse("pages-root"), user=user)
+        response = self.app.get(reverse("pages-root"), user=user).follow()
 
         self.assertRedirects(response, reverse("profile:registration_necessary"))
 

src/open_inwoner/accounts/tests/test_oidc_views.py

@@ -548,6 +548,7 @@ def test_logout(self, mock_get_solo):
         session = self.client.session
         session["oidc_states"] = {"mock": {"nonce": "nonce"}}
         session["oidc_id_token"] = "foo"
+        session["KVK_BRANCH_NUMBER"] = "1234"
         session.save()
         logout_url = reverse("eherkenning_oidc:logout")
 
@@ -563,7 +564,7 @@ def test_logout(self, mock_get_solo):
                 )
             )
             m.get(logout_endpoint_url)
-            logout_response = self.client.get(logout_url)
+            logout_response = self.client.get(logout_url, follow=False)
 
             self.assertEqual(len(m.request_history), 1)
             self.assertEqual(m.request_history[0].url, logout_endpoint_url)
@@ -575,6 +576,7 @@ def test_logout(self, mock_get_solo):
         self.assertNotIn("oidc_states", self.client.session)
         self.assertNotIn("oidc_id_token", self.client.session)
 
+    @patch("open_inwoner.contrib.kvk.client.KvKClient.get_all_company_branches")
     @patch("mozilla_django_oidc_db.backends.OIDCAuthenticationBackend.get_userinfo")
     @patch("mozilla_django_oidc_db.backends.OIDCAuthenticationBackend.store_tokens")
     @patch("mozilla_django_oidc_db.backends.OIDCAuthenticationBackend.verify_token")
@@ -590,11 +592,13 @@ def test_error_first_cleared_after_succesful_login(
         mock_verify_token,
         mock_store_tokens,
         mock_get_userinfo,
+        mock_kvk,
     ):
         mock_get_userinfo.return_value = {
             "sub": "some_username",
             "kvk": "12345678",
         }
+        mock_kvk.return_value = [{"vestigingsnummber": "1234"}]
         session = self.client.session
         session["oidc-error"] = "some error"
         session.save()
@@ -619,7 +623,14 @@ def test_error_first_cleared_after_succesful_login(
                 callback_response, reverse("pages-root"), fetch_redirect_response=False
             )
 
-            with self.subTest("check error page again"):
-                response = self.client.get(error_url)
+            # check redirect to /kvk/branches
+            response = self.client.get(callback_response["Location"])
+            self.assertRedirects(
+                response, reverse("kvk:branches"), fetch_redirect_response=False
+            )
 
-                self.assertEqual(response.status_code, 403)
+            # check final redirect to pages-root
+            response = self.client.get(response["Location"])
+            self.assertRedirects(
+                response, reverse("pages-root"), fetch_redirect_response=False
+            )

src/open_inwoner/accounts/tests/test_profile_views.py

@@ -2,6 +2,7 @@
 from datetime import date
 from unittest.mock import patch
 
+from django.conf import settings
 from django.template.defaultfilters import date as django_date
 from django.test import override_settings
 from django.urls import reverse
@@ -36,8 +37,17 @@
     eHerkenningUserFactory,
 )
 
+# Avoid redirects through `KvKLoginMiddleware`
+PATCHED_MIDDLEWARE = [
+    m
+    for m in settings.MIDDLEWARE
+    if m != "open_inwoner.contrib.kvk.middleware.KvKLoginMiddleware"
+]
 
-@override_settings(ROOT_URLCONF="open_inwoner.cms.tests.urls")
+
+@override_settings(
+    ROOT_URLCONF="open_inwoner.cms.tests.urls", MIDDLEWARE=PATCHED_MIDDLEWARE
+)
 class ProfileViewTests(WebTest):
     def setUp(self):
         self.url = reverse("profile:detail")

src/open_inwoner/accounts/views/auth.py

@@ -201,16 +201,3 @@ def get_success_url(self):
             del session["invite_url"]
 
         return super().get_success_url()
-
-    def get(self, request):
-        import pdbr
-
-        pdbr.set_trace()
-        return super().get(request)
-
-    # override get
-    # if there are branches:
-    # redirect: intermediate page, display branches
-    # select branch
-    # requestion.session["company_branch_number"] = branch_number
-    # redirect: get_success_url

src/open_inwoner/conf/base.py

@@ -257,6 +257,7 @@
     "open_inwoner.cms.utils.middleware.DropToolbarMiddleware",
     "django.contrib.flatpages.middleware.FlatpageFallbackMiddleware",
     "open_inwoner.extended_sessions.middleware.SessionTimeoutMiddleware",
+    "open_inwoner.contrib.kvk.middleware.KvKLoginMiddleware",
     "open_inwoner.accounts.middleware.NecessaryFieldsMiddleware",
     "open_inwoner.cms.utils.middleware.AnonymousHomePageRedirectMiddleware",
     "mozilla_django_oidc_db.middleware.SessionRefresh",

src/open_inwoner/contrib/kvk/client.py

@@ -13,7 +13,7 @@
 
 
 class KvKClient:
-    def __init__(self, config: KvKConfig = None):
+    def __init__(self, config: Optional[KvKConfig] = None):
         self.config = config or KvKConfig.get_solo()
 
     #
@@ -74,7 +74,7 @@ def _request(self, endpoint: str, params: dict) -> dict:
     def search_endpoint(self):
         return self._urljoin(self.config.api_root, "zoeken")
 
-    def search(self, **kwargs):
+    def search(self, **kwargs) -> dict:
         """
         Generic call to the 'Zoeken' endpoint of the KvK API
 

src/open_inwoner/contrib/kvk/forms.py

@@ -0,0 +1,5 @@
+from django import forms
+
+
+class CompanyBranchChoiceForm(forms.Form):
+    branch_number = forms.CharField(widget=forms.HiddenInput())

src/open_inwoner/contrib/kvk/middleware.py

@@ -0,0 +1,49 @@
+import logging
+
+from django.http import HttpResponseRedirect
+from django.urls import NoReverseMatch, reverse
+
+from furl import furl
+
+logger = logging.getLogger(__name__)
+
+
+class KvKLoginMiddleware:
+    """Redirect authenticated eHerkenning users to select a company branch"""
+
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        user = request.user
+
+        # pass through
+        if (
+            not user.is_authenticated
+            or not user.kvk
+            or request.session.get("KVK_BRANCH_NUMBER")
+        ):
+            return self.get_response(request)
+
+        # let the user logout and avoid redirect circles
+        try:
+            logout = reverse("logout")
+            eherkenning_logout = reverse("eherkenning:logout")
+            registration = reverse("profile:registration_necessary")
+            branches = reverse("kvk:branches")
+        except NoReverseMatch:
+            logout = "/logout/"
+            eherkenning_logout = "/eherkenning/logout/"
+            registration = "/mijn-profiel/register/necessary/"
+            branches = "/kvk/branches/"
+
+        if request.path.startswith(
+            (logout, eherkenning_logout, registration, branches)
+        ):
+            return self.get_response(request)
+
+        # redirect to company branch choice, passing on next url
+        redirect = furl(reverse("kvk:branches"))
+        redirect.args.update(request.GET)
+
+        return HttpResponseRedirect(redirect.url)

src/open_inwoner/contrib/kvk/urls.py

@@ -0,0 +1,9 @@
+from django.urls import path
+
+from .views import CompanyBranchChoiceView
+
+app_name = "kvk"
+
+urlpatterns = [
+    path("branches/", CompanyBranchChoiceView.as_view(), name="branches"),
+]

src/open_inwoner/contrib/kvk/views.py

@@ -0,0 +1,75 @@
+from django.http import HttpResponseRedirect
+from django.shortcuts import render
+from django.urls import reverse
+from django.views.generic import FormView
+
+from .client import KvKClient
+from .forms import CompanyBranchChoiceForm
+
+
+class CompanyBranchChoiceView(FormView):
+    """Choose the branch ("vestiging") of a company"""
+
+    template_name = "pages/kvk/branches.html"
+    form_class = CompanyBranchChoiceForm
+
+    def get_context_data(self, **kwargs):
+        context = super().get_context_data(**kwargs)
+
+        kvk_client = KvKClient()
+
+        company_branches = kvk_client.get_all_company_branches(
+            kvk=self.request.user.kvk
+        )
+        form = self.get_form()
+        target_url = self.request.GET.get("next") or reverse("pages-root")
+
+        context.update(
+            company_branches=company_branches,
+            form=form,
+            target_url=target_url,
+        )
+
+        return context
+
+    def get(self, request):
+        context = self.get_context_data()
+
+        target_url = context["target_url"]
+        company_branches = context["company_branches"]
+
+        if not company_branches:
+            return HttpResponseRedirect(target_url)
+
+        if len(company_branches) == 1:
+            request.session["KVK_BRANCH_NUMBER"] = (
+                company_branches[0].get("vestigingsnummer") or request.user.kvk
+            )
+            return HttpResponseRedirect(target_url)
+
+        form = context["form"]
+
+        return render(
+            request,
+            self.template_name,
+            {
+                "company_branches": company_branches,
+                "form": form,
+            },
+        )
+
+    def post(self, request):
+        form = self.get_form()
+
+        if form.is_valid():
+            context = self.get_context_data()
+            target_url = context["target_url"]
+
+            cleaned = form.cleaned_data
+            branch_number = cleaned["branch_number"]
+
+            request.session["KVK_BRANCH_NUMBER"] = branch_number
+
+            return HttpResponseRedirect(target_url)
+
+        return super().form_invalid(form)

src/open_inwoner/openzaak/tests/test_case_detail.py

@@ -1,6 +1,7 @@
 import datetime
 from unittest.mock import patch
 
+from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from django.test import RequestFactory
 from django.test.utils import override_settings
@@ -40,9 +41,18 @@
 from .factories import CatalogusConfigFactory, ServiceFactory
 from .shared import CATALOGI_ROOT, DOCUMENTEN_ROOT, ZAKEN_ROOT
 
+PATCHED_MIDDLEWARE = [
+    m
+    for m in settings.MIDDLEWARE
+    if m != "open_inwoner.contrib.kvk.middleware.KvKLoginMiddleware"
+]
+
 
 @requests_mock.Mocker()
-@override_settings(ROOT_URLCONF="open_inwoner.cms.tests.urls")
+@override_settings(
+    ROOT_URLCONF="open_inwoner.cms.tests.urls",
+    MIDDLEWARE=PATCHED_MIDDLEWARE,
+)
 class TestCaseDetailView(AssertRedirectsMixin, ClearCachesMixin, WebTest):
     @classmethod
     def setUpTestData(cls):
@@ -710,7 +720,8 @@ def test_page_displays_expected_data_for_eherkenning_user(self, m):
                 self.config.save()
 
                 response = self.app.get(
-                    self.eherkenning_case_detail_url, user=self.eherkenning_user
+                    self.eherkenning_case_detail_url,
+                    user=self.eherkenning_user,
                 )
 
                 self.assertContains(response, "ZAAK-2022-0000000025")

src/open_inwoner/openzaak/tests/test_cases.py

@@ -1,6 +1,7 @@
 import datetime
 from unittest.mock import patch
 
+from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from django.test.utils import override_settings
 from django.urls import reverse_lazy
@@ -34,6 +35,13 @@
 from .mocks import ESuiteData
 from .shared import CATALOGI_ROOT, ZAKEN_ROOT
 
+# Avoid redirects through `KvKLoginMiddleware`
+PATCHED_MIDDLEWARE = [
+    m
+    for m in settings.MIDDLEWARE
+    if m != "open_inwoner.contrib.kvk.middleware.KvKLoginMiddleware"
+]
+
 
 @override_settings(ROOT_URLCONF="open_inwoner.cms.tests.urls")
 class CaseListAccessTests(AssertRedirectsMixin, ClearCachesMixin, WebTest):
@@ -129,7 +137,9 @@ def test_no_cases_are_retrieved_when_http_500(self, m):
 
 
 @requests_mock.Mocker()
-@override_settings(ROOT_URLCONF="open_inwoner.cms.tests.urls")
+@override_settings(
+    ROOT_URLCONF="open_inwoner.cms.tests.urls", MIDDLEWARE=PATCHED_MIDDLEWARE
+)
 class CaseListViewTests(AssertTimelineLogMixin, ClearCachesMixin, WebTest):
     inner_url = reverse_lazy("cases:cases_content")
     maxDiff = None
@@ -384,6 +394,9 @@ def _setUpMocks(self, m):
         ]:
             m.get(resource["url"], json=resource)
 
+        # TODO
+        # m.get(reverse("kvk:branches"))
+
     def test_list_cases(self, m):
         self._setUpMocks(m)