draft: add endpoints to create and modify users

openedx/core/djangoapps/user_api/serializers.py

@@ -4,8 +4,12 @@
 
 
 from django.contrib.auth.models import User  # pylint: disable=imported-auth-user
+from django.core.exceptions import ValidationError
+from django.db import transaction
 from rest_framework import serializers
 
+from common.djangoapps.student.models.user import UserProfile, email_exists_or_retired, username_exists_or_retired
+from common.djangoapps.util.password_policy_validators import normalize_password, validate_password
 from lms.djangoapps.verify_student.models import ManualVerification, SoftwareSecurePhotoVerification
 
 from .models import UserPreference
@@ -126,3 +130,181 @@ def get_receipt_id(self, obj):
             return obj.receipt_id
         else:
             return None
+
+
+class UserProfileSerializer(serializers.Serializer):
+    """
+    Serializer used by the user modify endpoint to create or update a User and its related UserProfile.
+    """
+
+    email = serializers.EmailField(required=False, allow_blank=False)
+    username = serializers.CharField(required=False, allow_blank=False)
+    password = serializers.CharField(write_only=True, required=False, allow_blank=False)
+    is_active = serializers.BooleanField(required=False)
+    is_staff = serializers.BooleanField(required=False)
+    is_superuser = serializers.BooleanField(required=False)
+    name = serializers.CharField(required=False, allow_blank=False)
+    year_of_birth = serializers.IntegerField(required=False)
+    gender = serializers.CharField(required=False, allow_blank=False)
+    level_of_education = serializers.CharField(required=False, allow_blank=False)
+    country = serializers.CharField(required=False, allow_blank=False)
+
+    def create(self, validated_data: dict) -> User:
+        user_data = {}
+        profile_data = {}
+        password = validated_data.pop("password", None)
+        if password is not None:
+            password = normalize_password(password)
+
+        for field_name, field_value in validated_data.items():
+            if field_name in {"name", "year_of_birth", "gender", "level_of_education", "country"}:
+                profile_data[field_name] = field_value
+            else:
+                user_data[field_name] = field_value
+
+        with transaction.atomic():
+            user = User.objects.create_user(password=password, **user_data)
+            UserProfile.objects.create(user=user, **profile_data)
+
+        return user
+
+    def validate(self, attrs: dict) -> dict:
+        """
+        Validate the input data for the User serializer.
+
+        :param attrs: Dictionary with user's data.
+        :return: Validated data dictionary.
+        """
+
+        if self.instance is None:
+            if "username" not in attrs:
+                raise serializers.ValidationError(
+                    {"username": "Username is required."}
+                )
+
+            if "email" not in attrs:
+                raise serializers.ValidationError(
+                    {"email": "Email is required."}
+                )
+
+        return attrs
+
+    def validate_username(self, value: str) -> str:
+        """Validate the username.
+
+        Ensures that the username is unique if it is being set for a new user or changed for an existing user.
+
+        :param value: Username to validate.
+        :return: Validated username.
+        :raises serializers.ValidationError: If the username is not unique.
+        """
+        if self.instance is None or self.instance.username != value:
+            self._check_username_unique(value)
+        return value
+
+    def validate_email(self, value: str) -> str:
+        """Validate the email.
+
+        Ensures that the email is unique if it is being set for a new user or changed for an existing user.
+
+        :param value: Email to validate.
+        :return: Validated email.
+        :raises serializers.ValidationError: If the email is not unique.
+        """
+        if self.instance is None or self.instance.email != value:
+            self._check_email_unique(value)
+        return value
+
+    def validate_gender(self, value: str) -> str:
+        """Validate the gender.
+
+        Ensures that the gender is one of the valid choices defined in UserProfile.GENDER_CHOICES.
+
+        :param value: Gender to validate.
+        :return: Validated gender.
+        :raises serializers.ValidationError: If the gender is not one of the valid choices.
+        """
+        if value not in dict(UserProfile.GENDER_CHOICES):
+            choices = list(dict(UserProfile.GENDER_CHOICES).keys())
+            raise serializers.ValidationError(f"Gender must be one of {choices}")
+        return value
+
+    def validate_password(self, value: str) -> str:
+        """Validate the password.
+
+        Ensures that the password meets the required strength and complexity criteria.
+
+        :param value: Password to validate.
+        :return: Validated password.
+        :raises serializers.ValidationError: If the password does not meet the required criteria.
+        """
+        validate_password(value)
+        return value
+
+    def update(self, instance: User, validated_data: dict) -> User:
+        """
+        Update User instance by the data dictionary.
+
+        Method updates User instance if data is correct.
+        :param instance: User instance.
+        :param validated_data: Dictionary with user's data.
+        :return: User instance.
+        """
+
+        if not isinstance(instance, User):
+            raise ValidationError("The instance must be the User type.")
+
+        if not isinstance(validated_data, dict):
+            raise ValidationError("The data must be the Dictionary type.")
+
+        if "username" in validated_data:
+            raise ValidationError("Username cannot be changed.")
+
+        user_data = {}
+        profile_data = {}
+        password = validated_data.pop("password", None)
+
+        for field_name, field_value in validated_data.items():
+            if field_name in {"name", "year_of_birth", "gender", "level_of_education", "country"}:
+                profile_data[field_name] = field_value
+            else:
+                user_data[field_name] = field_value
+
+        with transaction.atomic():
+            if user_data:
+                for field_name, field_value in user_data.items():
+                    setattr(instance, field_name, field_value)
+                instance.save()
+
+            if password:
+                instance.set_password(normalize_password(password))
+                instance.save()
+
+            if profile_data:
+                profile = instance.profile
+                for field_name, field_value in profile_data.items():
+                    setattr(profile, field_name, field_value)
+                profile.save()
+
+        return instance
+
+    @staticmethod
+    def _check_email_unique(email):
+        if email_exists_or_retired(email=email):
+            raise serializers.ValidationError(
+                "User already exists with this email"
+            )
+
+    @staticmethod
+    def _check_username_unique(username):
+        if username_exists_or_retired(username=username):
+            raise serializers.ValidationError(
+                "User already exists with this username"
+            )
+
+    class Meta:
+        model = User
+        fields = ("id", "email", "password", "username", "is_active",
+                  "is_staff", "is_superuser", "name", "year_of_birth",
+                  "gender", "level_of_education", "country")
+        read_only_fields = ("id",)

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

@@ -0,0 +1,150 @@
+"""Unit tests for the combined User/UserProfile serializer."""
+
+import pytest
+from django.contrib.auth import get_user_model
+from django.core.exceptions import ValidationError as DjangoValidationError
+from django.test import TestCase
+
+from common.djangoapps.student.models.user import UserProfile
+from common.djangoapps.student.tests.factories import UserFactory
+
+from ..serializers import UserProfileSerializer
+
+User = get_user_model()
+
+
+class UserProfileSerializerTest(TestCase):
+    """Tests for updates spanning the User and UserProfile models."""
+
+    def test_create_user_and_profile_fields(self):
+        data = {
+            "username": "new-user",
+            "email": "new@example.com",
+            "password": "Password1234",
+            "name": "New User",
+            "year_of_birth": UserProfile.VALID_YEARS[-1],
+            "gender": "f",
+            "level_of_education": "b",
+            "country": "US",
+        }
+
+        serializer = UserProfileSerializer(data=data)
+        assert serializer.is_valid(), serializer.errors
+        user = serializer.save()
+
+        assert user.username == data["username"]
+        assert user.email == data["email"]
+        assert user.check_password(data["password"])
+        assert user.profile.name == data["name"]
+        assert user.profile.year_of_birth == data["year_of_birth"]
+        assert user.profile.gender == data["gender"]
+        assert user.profile.level_of_education == data["level_of_education"]
+        assert user.profile.country == data["country"]
+
+    def test_create_user_with_no_profile_fields(self):
+        data = {
+            "username": "new-user",
+            "email": "new@example.com",
+            "password": "Password1234",
+        }
+
+        serializer = UserProfileSerializer(data=data)
+        assert serializer.is_valid(), serializer.errors
+        user = serializer.save()
+
+        assert user.username == data["username"]
+        assert user.email == data["email"]
+        assert user.check_password(data["password"])
+        assert UserProfile.objects.filter(user=user).exists()
+        assert user.profile.name == ""
+        assert user.profile.year_of_birth is None
+        assert user.profile.gender is None
+        assert user.profile.level_of_education is None
+        assert user.profile.country is None
+
+    def test_update_user_and_profile_fields(self):
+        user = UserFactory.create(email="original@example.com", profile__name="Original Name")
+        data = {
+            "email": "updated@example.com",
+            "name": "Updated Name",
+            "password": "new-password",
+            "year_of_birth": UserProfile.VALID_YEARS[-1],
+            "gender": "f",
+            "level_of_education": "b",
+            "country": "US",
+        }
+
+        serializer = UserProfileSerializer(instance=user, data=data, partial=True)
+        assert serializer.is_valid(), serializer.errors
+        serializer.save()
+
+        user.refresh_from_db()
+        user.profile.refresh_from_db()
+
+        assert user.email == data["email"]
+        assert user.check_password(data["password"])
+        assert user.profile.name == data["name"]
+        assert user.profile.year_of_birth == data["year_of_birth"]
+        assert user.profile.gender == data["gender"]
+        assert user.profile.level_of_education == data["level_of_education"]
+        assert user.profile.country == data["country"]
+
+    def test_blank_user_fields_are_rejected(self):
+        for field in ("email", "username", "password", "name"):
+            serializer = UserProfileSerializer(data={field: ""}, partial=True)
+            assert not serializer.is_valid()
+            assert field in serializer.errors
+
+    def test_whitespace_only_fields_are_rejected(self):
+        for field in ("email", "username", "name"):
+            serializer = UserProfileSerializer(data={field: "   "}, partial=True)
+            assert not serializer.is_valid()
+            assert field in serializer.errors
+
+    def test_missing_required_fields_are_rejected(self):
+        for field, message in (("email", "Email is required."), ("username", "Username is required.")):
+            data = {"email": "new@example.com", "username": "new-user"}
+            data.pop(field)
+            serializer = UserProfileSerializer(data=data)
+            assert not serializer.is_valid(), serializer.errors
+            assert field in serializer.errors
+            assert message in serializer.errors[field]
+
+    def test_duplicate_email_and_username_are_rejected(self):
+        existing = UserFactory.create(username="existing-user", email="existing@example.com")
+
+        cases = (
+            ("email", existing.email, "User already exists with this email"),
+            ("username", existing.username, "User already exists with this username"),
+        )
+
+        for field, value, message in cases:
+            serializer = UserProfileSerializer(
+                data={"email": "new@example.com", "username": "new-user", field: value}
+            )
+            assert not serializer.is_valid(), serializer.errors
+            assert field in serializer.errors
+            assert message in serializer.errors[field]
+
+    def test_weak_password_is_rejected(self):
+        serializer = UserProfileSerializer(data={
+            "username": "new-user",
+            "email": "new@example.com",
+            "password": "weak",
+        })
+        assert not serializer.is_valid()
+        assert "password" in serializer.errors, serializer.errors
+
+    def test_no_password_is_accepted(self):
+        serializer = UserProfileSerializer(data={
+            "username": "new-user",
+            "email": "new@example.com",
+        })
+        assert serializer.is_valid(), serializer.errors
+        user = serializer.save()
+        assert not user.has_usable_password()
+
+    def test_username_cannot_be_changed(self):
+        user = UserFactory.create(username="original-user", email="original@example.com")
+        with pytest.raises(DjangoValidationError, match=r"Username cannot be changed\."):
+            UserProfileSerializer().update(user, {"username": "new-user"})