invitations_views: Add helper to accept all available invitations

Used in Inclusion Connect and ProConnect login flow

itou/openid_connect/inclusion_connect/views.py

@@ -18,6 +18,7 @@
 from itou.utils import constants as global_constants
 from itou.utils.constants import ITOU_HELP_CENTER_URL
 from itou.utils.urls import add_url_params, get_absolute_url
+from itou.www.invitations_views.helpers import accept_all_pending_invitations
 
 from ..errors import redirect_with_error_sso_email_conflict_on_registration
 from ..models import (
@@ -349,6 +350,8 @@ def inclusion_connect_callback(request):
     if prescriber_session_data := ic_state.data.get("prescriber_session_data"):
         request.session.update(prescriber_session_data)
 
+    accept_all_pending_invitations(request)
+
     next_url = ic_state.data["next_url"] or get_adapter(request).get_login_redirect_url(request)
     return HttpResponseRedirect(next_url)
 

itou/openid_connect/pro_connect/views.py

@@ -19,6 +19,7 @@
 from itou.utils import constants as global_constants
 from itou.utils.constants import ITOU_HELP_CENTER_URL
 from itou.utils.urls import add_url_params, get_absolute_url
+from itou.www.invitations_views.helpers import accept_all_pending_invitations
 
 from ..errors import redirect_with_error_sso_email_conflict_on_registration
 from ..models import (
@@ -298,6 +299,8 @@ def pro_connect_callback(request):
     if prescriber_session_data := pro_connect_state.data.get("prescriber_session_data"):
         request.session.update(prescriber_session_data)
 
+    accept_all_pending_invitations(request)
+
     next_url = pro_connect_state.data["next_url"] or get_adapter(request).get_login_redirect_url(request)
     return HttpResponseRedirect(next_url)
 

itou/www/invitations_views/helpers.py

@@ -0,0 +1,66 @@
+from django.contrib import messages
+from django.core.exceptions import PermissionDenied
+
+from itou.invitations.models import EmployerInvitation, LaborInspectorInvitation, PrescriberWithOrgInvitation
+from itou.users.enums import UserKind
+
+
+def handle_prescriber_intivation(invitation, request):
+    if not invitation.guest_can_join_organization(request):
+        raise PermissionDenied()
+
+    if invitation.can_be_accepted:
+        invitation.add_invited_user_to_organization()
+        # Send an email after the model changes
+        invitation.accept()
+        messages.success(
+            request, f"Vous êtes désormais membre de l'organisation {invitation.organization.display_name}."
+        )
+    elif not invitation.accepted_at:
+        messages.error(request, "Cette invitation n'est plus valide.")
+
+
+def handle_employer_invitation(invitation, request):
+    if not invitation.guest_can_join_company(request):
+        raise PermissionDenied()
+
+    if not invitation.company.is_active:
+        messages.error(request, "Cette structure n'est plus active.")
+    elif invitation.can_be_accepted:
+        invitation.add_invited_user_to_company()
+        invitation.accept()
+        messages.success(request, f"Vous êtes désormais membre de la structure {invitation.company.display_name}.")
+    elif not invitation.accepted_at:
+        messages.error(request, "Cette invitation n'est plus valide.")
+
+
+def handle_labor_inspector_invitation(invitation, request):
+    if not invitation.guest_can_join_institution(request):
+        raise PermissionDenied()
+
+    if invitation.can_be_accepted:
+        invitation.add_invited_user_to_institution()
+        invitation.accept()
+        messages.success(
+            request, f"Vous êtes désormais membre de l'organisation {invitation.institution.display_name}."
+        )
+    elif not invitation.accepted_at:
+        messages.error(request, "Cette invitation n'est plus valide.")
+
+
+def accept_all_pending_invitations(request):
+    if not request.user.is_authenticated:
+        return False
+
+    MAPPING = {
+        UserKind.PRESCRIBER: (PrescriberWithOrgInvitation, handle_prescriber_intivation),
+        UserKind.EMPLOYER: (EmployerInvitation, handle_employer_invitation),
+        UserKind.LABOR_INSPECTOR: (LaborInspectorInvitation, handle_labor_inspector_invitation),
+    }
+
+    InvitationModel, handle_invitation = MAPPING[request.user.kind]
+
+    invitations = list(InvitationModel.objects.pending().filter(email=request.user.email))
+    for invitation in invitations:
+        handle_invitation(invitation, request)
+    return invitations

itou/www/invitations_views/views.py

@@ -4,7 +4,6 @@
 from django.conf import settings
 from django.contrib import messages
 from django.contrib.auth.decorators import login_required
-from django.core.exceptions import PermissionDenied
 from django.http import HttpResponseRedirect
 from django.shortcuts import get_object_or_404, redirect, render, reverse
 from django.utils import formats, safestring
@@ -29,6 +28,11 @@
     NewUserInvitationForm,
     PrescriberWithOrgInvitationFormSet,
 )
+from itou.www.invitations_views.helpers import (
+    handle_employer_invitation,
+    handle_labor_inspector_invitation,
+    handle_prescriber_intivation,
+)
 from itou.www.signup import forms as signup_forms
 
 
@@ -178,19 +182,7 @@ def invite_prescriber_with_org(request, template_name="invitations_views/create.
 @login_required
 def join_prescriber_organization(request, invitation_id):
     invitation = get_object_or_404(PrescriberWithOrgInvitation, pk=invitation_id)
-    if not invitation.guest_can_join_organization(request):
-        raise PermissionDenied()
-
-    if invitation.can_be_accepted:
-        invitation.add_invited_user_to_organization()
-        # Send an email after the model changes
-        invitation.accept()
-        messages.success(
-            request, f"Vous êtes désormais membre de l'organisation {invitation.organization.display_name}."
-        )
-    else:
-        messages.error(request, "Cette invitation n'est plus valide.")
-
+    handle_prescriber_intivation(invitation, request)
     request.session[global_constants.ITOU_SESSION_CURRENT_ORGANIZATION_KEY] = invitation.organization.pk
     url = get_adapter(request).get_login_redirect_url(request)
     return HttpResponseRedirect(url)
@@ -224,18 +216,7 @@ def invite_employer(request, template_name="invitations_views/create.html"):
 @login_required
 def join_company(request, invitation_id):
     invitation = get_object_or_404(EmployerInvitation, pk=invitation_id)
-    if not invitation.guest_can_join_company(request):
-        raise PermissionDenied()
-
-    if not invitation.company.is_active:
-        messages.error(request, "Cette structure n'est plus active.")
-    elif invitation.can_be_accepted:
-        invitation.add_invited_user_to_company()
-        invitation.accept()
-        messages.success(request, f"Vous êtes désormais membre de la structure {invitation.company.display_name}.")
-    else:
-        messages.error(request, "Cette invitation n'est plus valide.")
-
+    handle_employer_invitation(invitation, request)
     request.session[global_constants.ITOU_SESSION_CURRENT_ORGANIZATION_KEY] = invitation.company.pk
     url = get_adapter(request).get_login_redirect_url(request)
     return HttpResponseRedirect(url)
@@ -285,17 +266,6 @@ def invite_labor_inspector(request, template_name="invitations_views/create.html
 @login_required
 def join_institution(request, invitation_id):
     invitation = get_object_or_404(LaborInspectorInvitation, pk=invitation_id)
-    if not invitation.guest_can_join_institution(request):
-        raise PermissionDenied()
-
-    if invitation.can_be_accepted:
-        invitation.add_invited_user_to_institution()
-        invitation.accept()
-        messages.success(
-            request, f"Vous êtes désormais membre de l'organisation {invitation.institution.display_name}."
-        )
-    else:
-        messages.error(request, "Cette invitation n'est plus valide.")
-
+    handle_labor_inspector_invitation(invitation, request)
     request.session[global_constants.ITOU_SESSION_CURRENT_ORGANIZATION_KEY] = invitation.institution.pk
     return redirect("dashboard:index")

tests/www/invitations_views/test_company_accept.py

@@ -37,6 +37,7 @@ def assert_accepted_invitation(self, response, invitation, user, mailoutbox):
         assertContains(
             response, escape(f"Vous êtes désormais membre de la structure {invitation.company.display_name}.")
         )
+        assertNotContains(response, escape("Cette invitation n'est plus valide."))
 
         # Assert the user sees his new siae dashboard
         current_company = get_current_company_or_404(response.wsgi_request)
@@ -106,6 +107,7 @@ def test_accept_invitation_signup_returns_on_other_browser(self, client, mailout
 
         total_users_before = User.objects.count()
 
+        # coming back from another browser without the next_url
         other_client = ItouClient()
         response = sso_setup.mock_oauth_dance(
             client,
@@ -117,9 +119,27 @@ def test_accept_invitation_signup_returns_on_other_browser(self, client, mailout
             other_client=other_client,
         )
         response = other_client.get(response.url, follow=True)
-        # Check user is redirected to the welcoming tour
-        last_url, _status_code = response.redirect_chain[-1]
-        assert last_url == reverse("welcoming_tour:index")
+
+        total_users_after = User.objects.count()
+        assert (total_users_before + 1) == total_users_after
+
+        user = User.objects.get(email=invitation.email)
+        self.assert_accepted_invitation(response, invitation, user, mailoutbox)
+
+    @sso_parametrize
+    @respx.mock
+    def test_accept_invitation_signup_without_link(self, client, mailoutbox, sso_setup):
+        # The user's invitations are automatically accepted at login
+        invitation = EmployerInvitationFactory(email=sso_setup.oidc_userinfo["email"])
+
+        total_users_before = User.objects.count()
+
+        response = sso_setup.mock_oauth_dance(
+            client,
+            KIND_EMPLOYER,
+            user_email=invitation.email,
+        )
+        response = client.get(response.url, follow=True)
 
         total_users_after = User.objects.count()
         assert (total_users_before + 1) == total_users_after

tests/www/invitations_views/test_helpers.py

@@ -0,0 +1,78 @@
+from django.contrib.auth.models import AnonymousUser
+from django.contrib.messages.middleware import MessageMiddleware
+from django.contrib.sessions.middleware import SessionMiddleware
+from django.test import RequestFactory
+from django.utils import timezone
+
+from itou.www.invitations_views.helpers import accept_all_pending_invitations
+from tests.invitations.factories import (
+    EmployerInvitationFactory,
+    LaborInspectorInvitationFactory,
+    PrescriberWithOrgSentInvitationFactory,
+)
+from tests.users.factories import EmployerFactory, LaborInspectorFactory, PrescriberFactory
+from tests.utils.tests import get_response_for_middlewaremixin
+
+
+def test_anonymous_user():
+    request = RequestFactory()
+    request.user = AnonymousUser()
+    assert accept_all_pending_invitations(request) == 0
+
+
+def fake_request():
+    request = RequestFactory().get("/")
+    SessionMiddleware(get_response_for_middlewaremixin).process_request(request)
+    MessageMiddleware(get_response_for_middlewaremixin).process_request(request)
+    return request
+
+
+def test_accept_prescriber_invitations():
+    request = fake_request()
+    prescriber = PrescriberFactory()
+    request.user = prescriber
+
+    valid_invitation_1 = PrescriberWithOrgSentInvitationFactory(email=prescriber.email)
+    valid_invitation_2 = PrescriberWithOrgSentInvitationFactory(email=prescriber.email)
+    # non pending invitations
+    PrescriberWithOrgSentInvitationFactory(email=prescriber.email, expired=True)
+    PrescriberWithOrgSentInvitationFactory(email=prescriber.email, accepted=True, accepted_at=timezone.now())
+    # other user kind invitations
+    EmployerInvitationFactory(email=prescriber.email)
+    LaborInspectorInvitationFactory(email=prescriber.email)
+
+    assert accept_all_pending_invitations(request) == [valid_invitation_1, valid_invitation_2]
+
+
+def test_accept_employer_invitations():
+    request = fake_request()
+    prescriber = EmployerFactory()
+    request.user = prescriber
+
+    valid_invitation_1 = EmployerInvitationFactory(email=prescriber.email)
+    valid_invitation_2 = EmployerInvitationFactory(email=prescriber.email)
+    # non pending invitations
+    EmployerInvitationFactory(email=prescriber.email, expired=True)
+    EmployerInvitationFactory(email=prescriber.email, accepted=True, accepted_at=timezone.now())
+    # other user kind invitations
+    PrescriberWithOrgSentInvitationFactory(email=prescriber.email)
+    LaborInspectorInvitationFactory(email=prescriber.email)
+
+    assert accept_all_pending_invitations(request) == [valid_invitation_1, valid_invitation_2]
+
+
+def test_accept_labor_inspector_invitations():
+    request = fake_request()
+    prescriber = LaborInspectorFactory()
+    request.user = prescriber
+
+    valid_invitation_1 = LaborInspectorInvitationFactory(email=prescriber.email)
+    valid_invitation_2 = LaborInspectorInvitationFactory(email=prescriber.email)
+    # non pending invitations
+    LaborInspectorInvitationFactory(email=prescriber.email, expired=True)
+    LaborInspectorInvitationFactory(email=prescriber.email, accepted=True, accepted_at=timezone.now())
+    # other user kind invitations
+    EmployerInvitationFactory(email=prescriber.email)
+    PrescriberWithOrgSentInvitationFactory(email=prescriber.email)
+
+    assert accept_all_pending_invitations(request) == [valid_invitation_1, valid_invitation_2]