draft: add endpoints to create and modify users

openedx/core/djangoapps/user_api/tests/test_views.py

@@ -1,34 +1,37 @@
 """Tests for the user API at the HTTP request level. """
 
+import json
+from unittest.mock import patch
+
 import ddt
 import pytest
+from django.contrib.auth import get_user_model
 from django.test.utils import override_settings
 from django.urls import reverse
 from opaque_keys.edx.keys import CourseKey
 from pytz import common_timezones, common_timezones_set, country_timezones
+from rest_framework import status
+from rest_framework.exceptions import ValidationError
 
 from common.djangoapps.student.tests.factories import UserFactory
 from openedx.core.djangoapps.django_comment_common import models
 from openedx.core.djangolib.testing.utils import CacheIsolationTestCase, skip_unless_lms
 from openedx.core.lib.api.test_utils import TEST_API_KEY, ApiTestCase
 from openedx.core.lib.time_zone_utils import get_display_time_zone
-from xmodule.modulestore.tests.django_utils import (
-    SharedModuleStoreTestCase,  # pylint: disable=wrong-import-order
-)
+from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase  # pylint: disable=wrong-import-order
 from xmodule.modulestore.tests.factories import CourseFactory  # pylint: disable=wrong-import-order
 
-from ..accounts.tests.retirement_helpers import (  # pylint: disable=unused-import
-    RetirementTestCase,  # noqa: F401
-    fake_requested_retirement,  # noqa: F401
-    setup_retirement_states,  # noqa: F401
-)
+from ..accounts.tests.retirement_helpers import RetirementTestCase  # pylint: disable=unused-import; noqa: F401
+from ..accounts.tests.retirement_helpers import fake_requested_retirement  # pylint: disable=unused-import; noqa: F401
+from ..accounts.tests.retirement_helpers import setup_retirement_states  # pylint: disable=unused-import; noqa: F401
 from ..models import UserOrgTag
 from ..tests.factories import UserPreferenceFactory
 
 USER_LIST_URI = "/api/user/v1/users/"
 USER_PREFERENCE_LIST_URI = "/api/user/v1/user_prefs/"
 ROLE_LIST_URI = "/api/user/v1/forum_roles/Moderator/users/"
 
+User = get_user_model()
 
 class UserAPITestCase(ApiTestCase):
     """
@@ -649,3 +652,130 @@
         assert len(results) == len(common_timezones)
         for time_zone_info in results:
             self._assert_time_zone_is_valid(time_zone_info)
+
+
+@override_settings(ENABLE_AUTHN_REGISTER_HIBP_POLICY=False)
+@ddt.ddt
+class TestUserModifyAPI(ApiTestCase):
+    """Test cases covering the user modification API"""
+
+    PATH = "/api/user/v1/modify"
+
+    DATA = {
+        "name": "Test User",
+        "username": "testuser",
+        "password": "Password1234",
+        "email": "e@mail.com",
+        "terms_of_service": "true",
+        "honor_code": "true",
+    }
+
+    def setUp(self):
+        """Create a test user and log in with that user"""
+        super().setUp()
+
+        self.test_user = UserFactory.create(
+            username="user",
+            email="user@example.com",
+            password="pass",
+            is_staff=True,
+            is_superuser=True,
+        )
+        self.client.login(username="user", password="pass")
+
+    @override_settings(ENFORCE_SAFE_SESSIONS=False)
+    def test_create_new_user_success(self):
+        """Test creating a user with valid information"""
+        response = self.client.post(self.PATH, self.DATA)
+        assert response.status_code == status.HTTP_201_CREATED
+        created_user = User.objects.get(username=self.DATA["username"])
+        assert response.json() == {
+            "user_id": created_user.id,
+            "username": created_user.username,
+        }
+
+    @ddt.data("username", "password", "email", "terms_of_service", "honor_code")
+    def test_create_new_user_error_missing_info(self, missing_field):
+        """Test creating a user with missing required information"""
+        data = self.DATA.copy()
+        data.pop(missing_field)
+        response = self.client.post(self.PATH, data)
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "error" in response.json()
+        assert missing_field in str(response.json()["error"])
+
+    def test_create_new_user_error_invalid_attribute(self):
+        """Test creating a user with an invalid attribute"""
+        data = self.DATA.copy()
+        data["email"] = "invalid-email"
+        response = self.client.post(self.PATH, data)
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "error" in response.json()
+        assert "email" in str(response.json()["error"])
+
+    @ddt.data(
+        ("email", "user@example.com", "existing account"),
+        ("username", "user", "already exists"),
+    )
+    @ddt.unpack
+    def test_create_new_user_error_already_exists(self, field, value, error):
+        """Test creating a user with an invalid attribute"""
+        data = self.DATA.copy()
+        data[field] = value
+        response = self.client.post(self.PATH, data)
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "error" in response.json()
+        assert error in str(response.json()["error"])
+
+    def test_patch_user_success(self):
+        """Test updating a user with a valid lookup field"""
+        user = UserFactory.create(
+            username="patch-user",
+            email="patch@example.com",
+        )
+
+        response = self.client.patch(
+            self.PATH,
+            data=json.dumps({"email": user.email, "first_name": "Updated Name"}),
+            content_type="application/json",
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json() == {"user_id": user.id, "username": user.username}
+        user = User.objects.get(id=user.id)
+        assert user.first_name == "Updated Name"
+
+    def test_patch_user_not_found(self):
+        """Test patch returns 404 when no user matches the lookup fields"""
+        response = self.client.patch(
+            self.PATH,
+            data=json.dumps(
+                {"email": "missing@example.com", "first_name": "Updated Name"}
+            ),
+            content_type="application/json",
+        )
+
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        assert response.json() == {"error": "User not found."}
+
+    @patch("openedx.core.djangoapps.user_api.views.UserSerializer.update")
+    def test_patch_user_validation_error(self, serializer_update):
+        """Test serializer validation errors are returned with status 400"""
+        user = UserFactory.create(
+            username="test-user",
+            email="patch-error@example.com",
+        )
+        serializer_update.side_effect = ValidationError(
+            {"email": ["Invalid email address."]}
+        )
+
+        response = self.client.patch(
+            self.PATH,
+            data=json.dumps({"email": user.email, "first_name": "Updated Name"}),
+            content_type="application/json",
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json() == {
+            "error": {"email": ["Invalid email address."]}
+        }

openedx/core/djangoapps/user_api/urls.py

@@ -178,6 +178,9 @@
     path('v1/accounts/replace_usernames/', UsernameReplacementView.as_view(),
          name='username_replacement'
          ),
+    path('v1/modify', user_api_views.UserModifyView.as_view(),
+         name='user_account_modify'
+         ),
     re_path(
         fr'^v1/preferences/{settings.USERNAME_PATTERN}$',
         PreferencesView.as_view(),

openedx/core/djangoapps/user_api/views.py

@@ -1,19 +1,24 @@
 """HTTP end-points for the User API. """
 
-from django.contrib.auth.models import User  # pylint: disable=imported-auth-user
+from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
+from django.core.exceptions import ValidationError as DjangoValidationError
+from django.db import transaction
 from django.http import HttpResponse
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import ensure_csrf_cookie
 from django_filters.rest_framework import DjangoFilterBackend
+from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
 from edx_rest_framework_extensions.auth.session.authentication import SessionAuthenticationAllowInactiveUser
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx import locator
 from opaque_keys.edx.keys import CourseKey
 from rest_framework import generics, status, viewsets
-from rest_framework.exceptions import ParseError
-from rest_framework.permissions import IsAuthenticated
+from rest_framework.exceptions import ParseError, ValidationError
+from rest_framework.permissions import IsAdminUser, IsAuthenticated
+from rest_framework.response import Response
 from rest_framework.views import APIView
 
+from common.djangoapps.student.helpers import AccountValidationError
 from openedx.core.djangoapps.django_comment_common.models import Role
 from openedx.core.djangoapps.user_api.models import UserPreference
 from openedx.core.djangoapps.user_api.preferences.api import get_country_time_zones, update_email_opt_in
@@ -22,6 +27,8 @@
     UserPreferenceSerializer,
     UserSerializer,
 )
+from openedx.core.djangoapps.user_authn.views.register import create_account_with_params
+from openedx.core.lib.api.authentication import BearerAuthenticationAllowInactiveUser
 from openedx.core.lib.api.permissions import ApiKeyHeaderPermission
 from openedx.core.lib.api.view_utils import require_post_params
 
@@ -161,3 +168,128 @@ class CountryTimeZoneListView(generics.ListAPIView):
     def get_queryset(self):
         country_code = self.request.GET.get("country_code", None)
         return get_country_time_zones(country_code)
+
+
+class UserModifyView(APIView):
+    """
+    **Use Cases**
+
+        Creates a user with email and username, or updates user information by
+        email or username.
+
+    **Example Requests**
+
+        POST /api/user/v1/modify/
+
+        PATCH /api/user/v1/modify/
+
+    **Example POST Response**
+
+        If the request is successful, an HTTP 201 "Created" response is
+        returned along with the user id and username, e.g.:
+
+        {
+            "user_id": 5,
+            "username": "newuser"
+        }
+
+    """
+
+    authentication_classes = (
+        JwtAuthentication,
+        BearerAuthenticationAllowInactiveUser,
+        SessionAuthenticationAllowInactiveUser,
+    )
+    permission_classes = (IsAdminUser,)
+
+    @method_decorator(transaction.non_atomic_requests)
+    def dispatch(self, request, *args, **kwargs):
+        """
+        Decorate with `non_atomic_requests` to work on newer versions of platform.
+        """
+        return super().dispatch(request, *args, **kwargs)
+
+    def post(self, request):
+        """
+        Create a user with email and username.
+        """
+        try:
+            user = create_account_with_params(request, request.data)
+        except (
+            AccountValidationError,
+            ValueError,
+            ValidationError,
+            DjangoValidationError,
+        ) as e:
+            if isinstance(e, ValidationError):
+                message = e.detail
+            else:
+                message = str(e)
+            return Response(
+                data={"error": message},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        return Response(
+            data={"user_id": user.id, "username": user.username},
+            status=status.HTTP_201_CREATED,
+        )
+
+    def patch(self, request):
+        """
+        Update user information by email or username.
+        """
+        try:
+            user = self._get_user_by_email_or_username(request)
+            if not user:
+                return Response(
+                    data={"error": "User not found."},
+                    status=status.HTTP_404_NOT_FOUND,
+                )
+
+            data = request.data.copy()
+            # Remove email and username from the data to prevent changing them
+            data.pop("email", None)
+            data.pop("username", None)
+            # Remove password from the data and handle it separately
+            password = data.pop("password", None)
+            if password:
+                user.set_password(password)
+            user = UserSerializer().update(user, data)
+        except (
+            AccountValidationError,
+            ValueError,
+            ValidationError,
+            DjangoValidationError,
+        ) as e:
+            if isinstance(e, ValidationError):
+                message = e.detail
+            else:
+                message = str(e)
+            return Response(
+                data={"error": message},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        return Response(
+            data={"user_id": user.id, "username": user.username},
+            status=status.HTTP_200_OK,
+        )
+
+    @staticmethod
+    def _get_user_by_email_or_username(request):
+        """
+        Get a user by email and/or username.
+
+        Returns None if one doesn't exist.
+        """
+
+        email = request.data.get("email")
+        username = request.data.get("username")
+        if not email and not username:
+            raise ValidationError("email or username must be specified")
+        query = {}
+        if email:
+            query["email"] = email
+        if username:
+            query["username"] = username
+        return User.objects.filter(**query).first()