diff --git a/openedx/core/djangoapps/user_api/serializers.py b/openedx/core/djangoapps/user_api/serializers.py index 5b4dda5e2d3d..4afd4d2e8cef 100644 --- a/openedx/core/djangoapps/user_api/serializers.py +++ b/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",) diff --git a/openedx/core/djangoapps/user_api/tests/test_serializers.py b/openedx/core/djangoapps/user_api/tests/test_serializers.py new file mode 100644 index 000000000000..9ddaea413edd --- /dev/null +++ b/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"}) diff --git a/openedx/core/djangoapps/user_api/tests/test_views.py b/openedx/core/djangoapps/user_api/tests/test_views.py index 4b06ae0c5ef0..623803e6973e 100644 --- a/openedx/core/djangoapps/user_api/tests/test_views.py +++ b/openedx/core/djangoapps/user_api/tests/test_views.py @@ -1,8 +1,11 @@ """Tests for the user API at the HTTP request level. """ +import json + import ddt import pytest from django.contrib import messages as django_messages +from django.contrib.auth import get_user_model from django.contrib.messages.storage.session import SessionStorage from django.http import HttpResponse from django.test import RequestFactory @@ -10,22 +13,16 @@ 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 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 ..models import UserOrgTag from ..tests.factories import UserPreferenceFactory @@ -33,6 +30,7 @@ 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): """ @@ -764,3 +762,384 @@ def test_get_all_common_timezones(self): 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", + "is_staff": False, + "is_superuser": False, + } + + 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, + } + + def test_create_new_user_defaults_to_non_superuser_when_flag_is_omitted(self): + """A non-superuser can create a regular user without the flag.""" + self.test_user.is_superuser = False + self.test_user.save(update_fields=["is_superuser"]) + + data = self.DATA.copy() + data.pop("is_superuser") + response = self.client.post(self.PATH, data) + + assert response.status_code == status.HTTP_201_CREATED + created_user = User.objects.get(username=data["username"]) + assert created_user.is_superuser is False + + def test_create_new_user_forbidden_when_superuser_flag_is_set_by_non_superuser(self): + """Test a non-superuser cannot set the superuser flag on a new user.""" + self.test_user.is_superuser = False + self.test_user.save(update_fields=["is_superuser"]) + + data = self.DATA.copy() + data["is_superuser"] = True + response = self.client.post(self.PATH, data) + + assert response.status_code == status.HTTP_403_FORBIDDEN + assert response.json() == {"error": ["You must be a superuser to perform this action."]} + + @ddt.data(True, False) + def test_create_new_user_superuser_success(self, is_superuser): + """Test creating a new superuser successfully""" + + self.test_user.is_superuser = is_superuser + self.test_user.save(update_fields=["is_superuser"]) + + data = self.DATA.copy() + data["is_superuser"] = is_superuser + response = self.client.post(self.PATH, data) + + assert response.status_code == status.HTTP_201_CREATED + user = User.objects.get(username=data["username"]) + assert user.is_superuser == is_superuser + + @ddt.data("username", "email") + 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"]).lower() + + 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", "User already exists with this email"), + ("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( + {"username_or_email": user.email, "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.profile.name == "Updated Name" + + @ddt.data(False, True) + def test_patch_user_without_superuser_flag_by_non_superuser(self, is_superuser): + """A non-superuser can update other fields without submitting the flag.""" + self.test_user.is_superuser = False + self.test_user.save(update_fields=["is_superuser"]) + + user = UserFactory.create( + username="patch-user", + email="patch@example.com", + is_superuser=is_superuser + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "name": "Updated Name"} + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_200_OK + user.profile.refresh_from_db() + assert user.profile.name == "Updated Name" + assert user.is_superuser is is_superuser + + def test_patch_user_success_by_username(self): + """Test updating a user using their username as the lookup field""" + user = UserFactory.create( + username="patch-user", + email="patch@example.com", + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.username, "name": "Updated Name"} + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_200_OK + user.profile.refresh_from_db() + assert user.profile.name == "Updated Name" + + def test_patch_user_updates_user_and_profile_fields(self): + """Test updating fields stored on both the user and profile models""" + user = UserFactory.create( + username="patch-user", + email="patch@example.com", + profile__name="Original Name", + ) + data = { + "username_or_email": user.email, + "email": "updated@example.com", + "name": "Updated Name", + "password": "new-password", + "year_of_birth": 2000, + "gender": "f", + "level_of_education": "b", + "country": "US", + } + + response = self.client.patch( + self.PATH, + data=json.dumps(data), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_200_OK + 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_patch_user_missing_username_or_email(self): + """Test patch returns 400 when username_or_email is missing""" + response = self.client.patch( + self.PATH, + data=json.dumps({"name": "Updated Name"}), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert response.json() == {"error": ["username_or_email is required."]} + + def test_patch_user_invalid_profile_field_does_not_update_user(self): + """Test invalid profile data does not partially update the user""" + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + { + "username_or_email": user.email, + "email": "updated@example.com", + "gender": "invalid", + } + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + user.refresh_from_db() + assert user.email == "patch-error@example.com" + + def test_patch_user_invalid_email(self): + """Test invalid user data is returned as a validation error""" + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "email": "invalid-email"} + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "email" in str(response.json()["error"]) + + 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( + {"username_or_email": "missing@example.com", "name": "Updated Name"} + ), + content_type="application/json", + ) + + assert response.json() == {"error": ["User not found."]} + assert response.status_code == status.HTTP_404_NOT_FOUND + + def test_patch_user_validation_read_only_field(self): + """Test serializer validation errors are returned with status 400""" + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "username": "modifieduser"} + ), + content_type="application/json", + ) + + assert response.json() == {'error': ['Username cannot be changed.']} + assert response.status_code == status.HTTP_400_BAD_REQUEST + + def test_patch_user_validation_inexistent_field(self): + """Test serializer validation errors are returned with status 400""" + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "non_existent_field": "value"} + ), + content_type="application/json", + ) + + assert response.json() == {'error': ['Unexpected field: non_existent_field']} + assert response.status_code == status.HTTP_400_BAD_REQUEST + + @ddt.data(True, False) + def test_patch_superuser_with_permission(self, is_superuser): + """A superuser can set the flag to either value.""" + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + is_superuser=not is_superuser, + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "is_superuser": is_superuser} + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_200_OK + user.refresh_from_db() + assert user.is_superuser is is_superuser + + @ddt.data(True, False) + def test_patch_superuser_without_permission(self, is_superuser): + """Test that a non-superuser cannot set the superuser flag""" + self.test_user.is_superuser = False + self.test_user.save(update_fields=["is_superuser"]) + + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + is_superuser=not is_superuser, + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "is_superuser": is_superuser} + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_403_FORBIDDEN + assert response.json() == {"error": ["You must be a superuser to perform this action."]} + + @ddt.data(True, False) + def test_patch_superuser_without_permission_success(self, is_superuser): + """A non-superuser can successfully update a user without changing the superuser flag.""" + self.test_user.is_superuser = False + self.test_user.save(update_fields=["is_superuser"]) + + user = UserFactory.create( + username="test-user", + email="patch-error@example.com", + is_superuser=is_superuser, + ) + + response = self.client.patch( + self.PATH, + data=json.dumps( + {"username_or_email": user.email, "is_superuser": is_superuser} + ), + content_type="application/json", + ) + + assert response.status_code == status.HTTP_200_OK + user.refresh_from_db() + assert user.is_superuser is is_superuser diff --git a/openedx/core/djangoapps/user_api/urls.py b/openedx/core/djangoapps/user_api/urls.py index 04861ef27b71..1924f2048253 100644 --- a/openedx/core/djangoapps/user_api/urls.py +++ b/openedx/core/djangoapps/user_api/urls.py @@ -183,6 +183,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(), diff --git a/openedx/core/djangoapps/user_api/views.py b/openedx/core/djangoapps/user_api/views.py index e72b5789d334..39e2adcb8c69 100644 --- a/openedx/core/djangoapps/user_api/views.py +++ b/openedx/core/djangoapps/user_api/views.py @@ -1,30 +1,36 @@ """HTTP end-points for the User API. """ from django.contrib import messages as django_messages -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.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 drf_spectacular.utils import OpenApiResponse, extend_schema, inline_serializer +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, serializers, status, viewsets -from rest_framework.exceptions import ParseError -from rest_framework.permissions import IsAuthenticated +from rest_framework.exceptions import ParseError, PermissionDenied, 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 common.djangoapps.student.models.user import get_user_by_username_or_email 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 from openedx.core.djangoapps.user_api.serializers import ( CountryTimeZoneSerializer, UserPreferenceSerializer, + UserProfileSerializer, UserSerializer, ) +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 @@ -227,3 +233,153 @@ def get(self, request): for message in other_messages: django_messages.add_message(request, message.level, message.message, extra_tags=message.extra_tags) return Response({"user_message": user_message}) + + +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,) + + def post(self, request): + """ + Create a user with email and username. + """ + data = request.data.copy() + allowed_fields = UserProfileSerializer.Meta.fields + try: + for key in data: + if key not in allowed_fields: + raise ValidationError(f"Unexpected field: {key}") + if key in UserProfileSerializer.Meta.read_only_fields: + raise ValidationError(f"Read-only field: {key}") + serializer = UserProfileSerializer(data=data) + serializer.is_valid(raise_exception=True) + self._check_superuser(serializer, request) + user = serializer.save() + except ( + AccountValidationError, + ValueError, + ValidationError, + DjangoValidationError, + PermissionDenied, + ) as e: + return self._build_error_response(e) + + 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: + data = request.data.copy() + username_or_email = data.pop("username_or_email", None) + if not username_or_email: + return Response( + data={"error": ["username_or_email is required."]}, + status=status.HTTP_400_BAD_REQUEST, + ) + + for key in data: + if key not in UserProfileSerializer.Meta.fields: + raise ValidationError(f"Unexpected field: {key}") + if key in UserProfileSerializer.Meta.read_only_fields: + raise ValidationError(f"Read-only field: {key}") + try: + # Fetch user based on username or email + user = get_user_by_username_or_email( + username_or_email=username_or_email + ) + except User.MultipleObjectsReturned: + return Response( + data={"error": ["Multiple users found."]}, + status=status.HTTP_400_BAD_REQUEST, + ) + except User.DoesNotExist: + return Response( + data={"error": ["User not found."]}, + status=status.HTTP_404_NOT_FOUND, + ) + + serializer = UserProfileSerializer( + instance=user, data=data, partial=True + ) + serializer.is_valid(raise_exception=True) + self._check_superuser(serializer, request) + serializer.save() + + return Response( + data={"user_id": user.id, "username": user.username}, + status=status.HTTP_200_OK, + ) + except ( + AccountValidationError, + ValueError, + ValidationError, + DjangoValidationError, + PermissionDenied, + ) as e: + return self._build_error_response(e) + + def _check_superuser(self, serializer, request): + """Check if the current user is allowed to set the superuser flag.""" + if request.user.is_superuser: + # Superusers are allowed to set the superuser flag without restriction. + return + is_superuser = serializer.validated_data.get("is_superuser") + if is_superuser is None: + # Requests that don't include the superuser flag are allowed. + return + current = serializer.instance.is_superuser if serializer.instance else False + if current == is_superuser: + # No change in the superuser flag, allowed for non-superusers. + return + raise PermissionDenied("You must be a superuser to perform this action.") + + def _build_error_response(self, e: Exception) -> Response: + """Build an appropriate error response based on the type of exception.""" + if isinstance(e, ValidationError): + message = e.detail + elif isinstance(e, DjangoValidationError): + message = e.messages + elif isinstance(e, PermissionDenied): + return Response( + data={"error": [e.detail]}, + status=status.HTTP_403_FORBIDDEN, + ) + else: + message = [str(e)] + return Response( + data={"error": message}, + status=status.HTTP_400_BAD_REQUEST, + )