Jlc/change staff admin msg pipeline

eox_nelp/locale/ar/LC_MESSAGES/django.po

@@ -26,3 +26,17 @@ msgstr "أدخل"
 #: eox_nelp/middleware.py:111
 msgid "Enter your"
 msgstr "حدد"
+
+#: eox_nelp/third_party_auth/pipeline.py:99
+msgid "It is not allowed to auto associate staff or admin users"
+msgstr "لا يجوز ربط الموظفين أو المستخدمين الإداريين تلقائيًا"
+
+#: eox_nelp/third_party_auth/pipeline.py:99
+msgid ""
+"You are trying to access with a user that already exists with privileged permissions."
+" Please try to authenticate on the site where you registered for the first time "
+"or contact support to change permissions."
+msgstr ""
+"أنت تحاول الوصول مع مستخدم موجود بالفعل بأذونات مميزة."
+"يرجى محاولة المصادقة على الموقع الذي قمت بالتسجيل فيه لأول مرة"
+"أو اتصل بالدعم لتغيير الأذونات."

eox_nelp/third_party_auth/pipeline.py

@@ -6,9 +6,11 @@
 """
 from django.conf import settings
 from django.contrib.auth import logout
+from django.http import HttpResponseForbidden
+from django.utils.translation import gettext_lazy as _
 from social_core.pipeline.social_auth import social_details as social_core_details
 
-from eox_nelp.third_party_auth.exceptions import EoxNelpAuthException
+from eox_nelp.edxapp_wrapper.edxmako import edxmako
 
 
 def social_details(backend, details, response, *args, **kwargs):
@@ -70,7 +72,7 @@ def close_mismatch_session(request, *args, user=None, **kwargs):  # pylint: disa
 
 
 def safer_associate_username_by_uid(  # pylint: disable=unused-argument
-    backend, details, response, *args, user=None, **kwargs,
+    request, backend, details, response, *args, user=None, **kwargs,
 ):
     """Pipeline to retrieve user if possible matching uid with the username of a user.
     The uid is based in the configuration of the saml with the field `attr_user_permanent_id`:
@@ -94,8 +96,20 @@ def safer_associate_username_by_uid(  # pylint: disable=unused-argument
     if not user_match:
         return None
     if user_match.is_staff or user_match.is_superuser:
-        raise EoxNelpAuthException(backend, "It is not allowed to auto associate staff or admin users")
-
+        return HttpResponseForbidden(
+            edxmako.shortcuts.render_to_string(
+                "static_templates/server-error.html",
+                {
+                    "page_header": _("It is not allowed to auto associate staff or admin users"),
+                    "page_content": _(
+                        "You are trying to access with a user that already exists with privileged permissions."
+                        " Please try to authenticate on the site where you registered for the first time "
+                        "or contact support to change permissions."
+                    ),
+                },
+                request=request,
+            )
+        )
     return {
         "user": user_match,
         "is_new": False,

eox_nelp/third_party_auth/tests/test_pipeline.py

@@ -1,10 +1,11 @@
 """ Test file for third_party_auth pipeline functions."""
 from ddt import data, ddt
 from django.contrib.auth import get_user_model
+from django.http import HttpResponseForbidden
 from django.test import TestCase
 from mock import Mock
+from rest_framework import status
 
-from eox_nelp.third_party_auth.exceptions import EoxNelpAuthException
 from eox_nelp.third_party_auth.pipeline import safer_associate_username_by_uid
 
 User = get_user_model()
@@ -24,70 +25,76 @@ def setUp(self):  # pylint: disable=invalid-name
             "attributes": {},
         }
         self.user, _ = User.objects.get_or_create(username="vader")
+        self.backend = Mock()
+        self.request = Mock()
 
     def test_user_already_matched(self):
         """Test the pipeline method is called with already matched user.
         Expected behavior:
             - Return None.
         """
-        backend = Mock()
-
-        pipe_output = safer_associate_username_by_uid(backend, self.details, self.response, user=self.user)
+        pipe_output = safer_associate_username_by_uid(
+            self.request, self.backend, self.details, self.response, user=self.user,
+        )
 
-        self.assertEqual(None, pipe_output)
+        self.assertIsNone(pipe_output)
 
     def test_user_not_associated(self):
         """Test the pipeline method try to match with uid but there is not user with that username.
         Expected behavior:
-            - Pipe Return None.
+            - Pipe method returns None.
             - Strategy storage get_user method is called with desired params.
         """
         test_uid = "1888222999"
-        backend = Mock()
-        backend.get_idp.return_value.get_user_permanent_id.return_value = test_uid
-        backend.strategy.storage.user.get_user.return_value = None
 
-        pipe_output = safer_associate_username_by_uid(backend, self.details, self.response)
+        self.backend.get_idp.return_value.get_user_permanent_id.return_value = test_uid
+        self.backend.strategy.storage.user.get_user.return_value = None
+
+        pipe_output = safer_associate_username_by_uid(self.request, self.backend, self.details, self.response)
 
         self.assertEqual(None, pipe_output)
-        backend.strategy.storage.user.get_user.assert_called_with(username=test_uid)
+        self.backend.strategy.storage.user.get_user.assert_called_with(username=test_uid)
 
     @data(
         {"is_staff": True, "username": "1222333444"},
         {"is_superuser": True, "username": "1222333555"},
     )
-    def test_staff_user_raise_exc(self, user_kwargs):
+    def test_staff_user_return_forbidden(self, user_kwargs):
         """Test the pipeline method try to match with uid, the user with matched username exists
         but is staff or superuser.
         Expected behavior:
-            - Raise EoxNelpAuthException exception
+            - The pipe method returns an HttpResponseForbidden object.
+            - The pipe method response has a 403 forbidden status.
             - Strategy storage get_user method is called with desired params.
         """
         test_uid = user_kwargs["username"]
         past_user, _ = User.objects.get_or_create(**user_kwargs)
-        backend = Mock()
-        backend.get_idp.return_value.get_user_permanent_id.return_value = test_uid
-        backend.strategy.storage.user.get_user.return_value = past_user
 
-        self.assertRaises(EoxNelpAuthException, safer_associate_username_by_uid, backend, self.details, self.response)
-        backend.strategy.storage.user.get_user.assert_called_with(username=test_uid)
+        self.backend.get_idp.return_value.get_user_permanent_id.return_value = test_uid
+        self.backend.strategy.storage.user.get_user.return_value = past_user
+
+        pipe_output = safer_associate_username_by_uid(self.request, self.backend, self.details, self.response)
+
+        self.assertIsInstance(pipe_output, HttpResponseForbidden)
+        self.assertEqual(status.HTTP_403_FORBIDDEN, pipe_output.status_code)
+        self.backend.strategy.storage.user.get_user.assert_called_with(username=test_uid)
 
     def test_user_associate_username_with_uid(self):
         """Test the pipeline method try to match with uid, the user with matched username exists
         and is returned.
         Expected behavior:
             - Strategy storage get_user method is called with desired params.
-            - The method return the desirect with `user` and `is_new` keys.
+            - The method returns the desirect with `user` and `is_new` keys.
         """
         test_uid = "1777888999"
         past_user, _ = User.objects.get_or_create(
             username=test_uid,
         )
-        backend = Mock()
-        backend.get_idp.return_value.get_user_permanent_id.return_value = test_uid
-        backend.strategy.storage.user.get_user.return_value = past_user
 
-        pipe_output = safer_associate_username_by_uid(backend, self.details, self.response)
+        self.backend.get_idp.return_value.get_user_permanent_id.return_value = test_uid
+        self.backend.strategy.storage.user.get_user.return_value = past_user
+
+        pipe_output = safer_associate_username_by_uid(self.request, self.backend, self.details, self.response)
 
         self.assertEqual({"user": past_user, "is_new": False}, pipe_output)
-        backend.strategy.storage.user.get_user.assert_called_with(username=test_uid)
+        self.backend.strategy.storage.user.get_user.assert_called_with(username=test_uid)