refactor

Author: arslanashraf7

src/ol_openedx_lti_utilities/ol_openedx_lti_utilities/views.py

@@ -3,25 +3,26 @@
 """
 
 import logging
-import re
 
-from django.contrib.auth.models import User
-from django.http import Http404
+from django.http import Http404, HttpResponseBadRequest
 from edx_rest_framework_extensions import permissions
 from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import (
     SessionAuthenticationAllowInactiveUser,
 )
 from lms.djangoapps.lti_provider.models import LtiUser
+from openedx.core.djangoapps.user_api.accounts.utils import (
+    create_retirement_request_and_deactivate_account,
+)
 from openedx.core.lib.api.authentication import BearerAuthenticationAllowInactiveUser
 from rest_framework import status
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 log = logging.getLogger(__name__)
 
-# Matches 40-character random username used by LTI users
-LTI_USERNAME_REGEX = r"^[A-Za-z0-9]{40}$"
+
+PLACEHOLDER_EMAIL_DOMAIN = "lti_example.com"
 
 
 class LtiUserFixView(APIView):
@@ -33,7 +34,6 @@ class LtiUserFixView(APIView):
     Request payload:
     {
         "email": "<user_email>",
-        "username": "<desired_username>"
     }
 
     Responses:
@@ -79,45 +79,32 @@ def post(self, request):
             If no LTI user exists for the provided email address
         """
         user_email = request.data.get("email")
-        user_username = request.data.get("username")
 
-        if not user_email or not user_username:
-            log.error("email and username are required")
-            return Response(
-                {"detail": "email and username are required"},
-                status=status.HTTP_400_BAD_REQUEST,
+        if not user_email:
+            log.error("email is required")
+            return HttpResponseBadRequest("email is required")
+
+        # A user that is created by LTI will always have the same username as
+        # lti_user_id in LtiUser table.
+        lti_user = LtiUser.objects.filter(edx_user__email=user_email).first()
+        if not lti_user:
+            log.error("No user was found against the given email (%s)", user_email)
+            raise Http404
+        if lti_user.lti_user_id != lti_user.edx_user.username:
+            log.error(
+                "User with email (%s) does not appear to be an LTI-created user",
+                user_email,
             )
-
-        def get_bad_lti_user_or_raise(email):
-            """Get LTI user or raise if none exists."""
-            qs = LtiUser.objects.filter(edx_user__email=email)
-            if not qs.exists():
-                raise LtiUser.DoesNotExist
-            return qs.first()
-
-        try:
-            lti_user = get_bad_lti_user_or_raise(user_email)
-
-            # LTI-created users have a 40-char random username → "bad" username
-            bad_lti_user = (
-                lti_user.edx_user
-                if re.match(LTI_USERNAME_REGEX, lti_user.edx_user.username)
-                else None
+            return HttpResponseBadRequest(
+                "User with the given email does not appear to be an LTI-created user."
             )
-
-            if not bad_lti_user:
-                return Response(
-                    {"detail": "User does not need fixing"},
-                    status=status.HTTP_400_BAD_REQUEST,
-                )
-
-        except LtiUser.DoesNotExist as exc:
-            log.error("No user was found against the given email (%s)", user_email)  # noqa: TRY400
-            raise Http404 from exc
-
-        # Fix username
-        user = User.objects.get(email=user_email)
-        user.username = user_username
+        # Fix username of the LTI created user
+        user = lti_user.edx_user
+        user.email = user.email.split("@")[0] + "@" + PLACEHOLDER_EMAIL_DOMAIN
         user.save()
-
+        # Send the user for retirement and deactivate the account
+        try:
+            create_retirement_request_and_deactivate_account(user)
+        except Exception as e:  # noqa: BLE001
+            log.error("Error retiring and deactivating user: %s", e)  # noqa: TRY400
         return Response(status=status.HTTP_200_OK)

src/ol_openedx_lti_utilities/tests/test_views.py

@@ -3,7 +3,6 @@
 from ddt import data, ddt, unpack
 from django.contrib.auth.models import User
 from django.test import TestCase
-from lms.djangoapps.lti_provider.models import LtiUser
 from rest_framework import status
 from rest_framework.test import APIClient
 
@@ -36,25 +35,25 @@ def setUp(self):
     def test_successful_lti_user_fix(self, mock_user_model, mock_lti_user_model):
         """Test successful LTI user fix."""
         # Mock LTI user with bad username
+        mock_lti_user = Mock()
+        mock_lti_user.lti_user_id = "TEST_LTI_USER_ID"
+
         mock_edx_user = Mock()
-        mock_edx_user.username = self.lti_username
+        mock_edx_user.username = "TEST_LTI_USER_ID"
         mock_edx_user.email = self.valid_email
-
-        mock_lti_user = Mock()
         mock_lti_user.edx_user = mock_edx_user
 
         mock_qs = Mock()
-        mock_qs.exists.return_value = True
         mock_qs.first.return_value = mock_lti_user
         mock_lti_user_model.objects.filter.return_value = mock_qs
 
         # Mock User.objects.get
         mock_user = Mock()
         mock_user_model.objects.get.return_value = mock_user
 
-        data = {"email": self.valid_email, "username": self.valid_username}
+        payload = {"email": self.valid_email, "username": self.valid_username}
 
-        response = self.client.post(self.url, data)
+        response = self.client.post(self.url, payload)
 
         assert response.status_code == status.HTTP_200_OK
         mock_user_model.objects.get.assert_called_once_with(email=self.valid_email)
@@ -69,23 +68,19 @@ def test_successful_lti_user_fix(self, mock_user_model, mock_lti_user_model):
     def test_missing_required_fields(self, test_data, expected_message):
         """Test request with missing required fields."""
         response = self.client.post(self.url, test_data)
-
-        assert response.status_code == status.HTTP_400_BAD_REQUEST
-        assert expected_message in response.data["detail"]
         assert response.status_code == status.HTTP_400_BAD_REQUEST
-        assert expected_message in response.data["detail"]
+        assert expected_message in response.content.decode()
 
     @patch("ol_openedx_lti_utilities.views.LtiUser")
     def test_no_lti_user_found(self, mock_lti_user_model):
         """Test when no LTI user exists for given email."""
         mock_qs = Mock()
-        mock_qs.exists.return_value = False
+        mock_qs.first.return_value = None
         mock_lti_user_model.objects.filter.return_value = mock_qs
-        mock_lti_user_model.DoesNotExist = LtiUser.DoesNotExist
 
-        data = {"email": self.valid_email, "username": self.valid_username}
+        payload = {"email": self.valid_email, "username": self.valid_username}
 
-        response = self.client.post(self.url, data)
+        response = self.client.post(self.url, payload)
 
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
@@ -105,97 +100,43 @@ def test_user_does_not_need_fixing(self, mock_lti_user_model):
         mock_qs.first.return_value = mock_lti_user
         mock_lti_user_model.objects.filter.return_value = mock_qs
 
-        data = {"email": self.valid_email, "username": self.valid_username}
+        payload = {"email": self.valid_email, "username": self.valid_username}
 
-        response = self.client.post(self.url, data)
+        response = self.client.post(self.url, payload)
 
         assert response.status_code == status.HTTP_400_BAD_REQUEST
-        assert "User does not need fixing" in response.data["detail"]
-
-    @patch("ol_openedx_lti_utilities.views.LtiUser")
-    def test_lti_username_regex_matching(self, mock_lti_user_model):
-        """Test various username formats against LTI username regex."""
-        test_cases = [
-            (
-                "A1B2C3D4E5F6G7H8I9J0K1L2M3N4O5P6Q7R8S9T0",  # pragma: allowlist secret
-                True,
-            ),  # 40 chars, alphanumeric
-            (
-                "abcdefghijklmnopqrstuvwxyz1234567890ABCD",  # pragma: allowlist secret
-                True,
-            ),  # pragma: allowlist secret
-            (
-                "A1B2C3D4E5F6G7H8I9J0K1L2M3N4O5P6Q7R8S9",  # pragma: allowlist secret
-                False,
-            ),  # pragma: allowlist secret
-            (
-                "A1B2C3D4E5F6G7H8I9J0K1L2M3N4O5P6Q7R8S9T0X",  # pragma: allowlist secret
-                False,
-            ),  # pragma: allowlist secret
-            ("normalusername", False),  # Normal username
-            (
-                "A1B2-C3D4E5F6G7H8I9J0K1L2M3N4O5P6Q7R8S9T0",  # pragma: allowlist secret
-                False,
-            ),  # pragma: allowlist secret
-            ("", False),  # Empty string
-        ]
-
-        for username, should_match in test_cases:
-            with self.subTest(username=username):
-                mock_edx_user = Mock()
-                mock_edx_user.username = username
-                mock_edx_user.email = self.valid_email
-
-                mock_lti_user = Mock()
-                mock_lti_user.edx_user = mock_edx_user
-
-                mock_qs = Mock()
-                mock_qs.exists.return_value = True
-                mock_qs.first.return_value = mock_lti_user
-                mock_lti_user_model.objects.filter.return_value = mock_qs
-
-                data = {"email": self.valid_email, "username": self.valid_username}
-
-                with patch("ol_openedx_lti_utilities.views.User") as mock_user_model:
-                    mock_user = Mock()
-                    mock_user_model.objects.get.return_value = mock_user
-
-                    response = self.client.post(self.url, data)
-
-                    if should_match:
-                        assert response.status_code == status.HTTP_200_OK
-                        mock_user.save.assert_called_once()
-                    else:
-                        assert response.status_code == status.HTTP_400_BAD_REQUEST
-                        assert "User does not need fixing" in response.data["detail"]
+        assert (
+            "User with the given email does not appear to be an LTI-created user."
+            in response.content.decode()
+        )
 
     def test_only_post_method_allowed(self):
         """Test that only POST method is allowed."""
-        data = {"email": self.valid_email, "username": self.valid_username}
+        payload = {"email": self.valid_email, "username": self.valid_username}
 
         # Test GET
         response = self.client.get(self.url)
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
 
         # Test PUT
-        response = self.client.put(self.url, data)
+        response = self.client.put(self.url, payload)
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
 
         # Test DELETE
         response = self.client.delete(self.url)
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
 
         # Test PATCH
-        response = self.client.patch(self.url, data)
+        response = self.client.patch(self.url, payload)
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
 
     @patch("ol_openedx_lti_utilities.views.log")
     @patch("ol_openedx_lti_utilities.views.LtiUser")
     def test_logging_on_missing_fields(self, mock_lti_user_model, mock_log):  # noqa: ARG002
         """Test that error is logged when required fields are missing."""
-        data = {}
+        payload = {}
 
-        response = self.client.post(self.url, data)
+        response = self.client.post(self.url, payload)
 
         assert response.status_code == status.HTTP_400_BAD_REQUEST
         mock_log.error.assert_called_once_with("email and username are required")
@@ -205,13 +146,12 @@ def test_logging_on_missing_fields(self, mock_lti_user_model, mock_log):  # noqa
     def test_logging_on_no_user_found(self, mock_lti_user_model, mock_log):
         """Test that error is logged when no LTI user is found."""
         mock_qs = Mock()
-        mock_qs.exists.return_value = False
+        mock_qs.first.return_value = None
         mock_lti_user_model.objects.filter.return_value = mock_qs
-        mock_lti_user_model.DoesNotExist = LtiUser.DoesNotExist
 
-        data = {"email": self.valid_email, "username": self.valid_username}
+        payload = {"email": self.valid_email, "username": self.valid_username}
 
-        response = self.client.post(self.url, data)
+        response = self.client.post(self.url, payload)
 
         assert response.status_code == status.HTTP_404_NOT_FOUND
         mock_log.error.assert_called_once_with(