diff --git a/src/apps/user/serializers.py b/src/apps/user/serializers.py index 6c79f9d..57733fa 100644 --- a/src/apps/user/serializers.py +++ b/src/apps/user/serializers.py @@ -7,6 +7,7 @@ from rest_framework import serializers from rest_framework.validators import UniqueValidator from .models import Profile +from .services import UserService User = get_user_model() @@ -126,25 +127,14 @@ class CurrentUserSerializer(serializers.ModelSerializer): @staticmethod def get_role(obj) -> str: - if obj.is_superuser: - return "admin" - if obj.is_staff: - return "staff" - return "user" + return UserService.get_user_role(obj) def get_role_label(self, obj) -> str: - labels = { - "admin": "Администратор системы", - "staff": "Сотрудник системы", - "user": "Пользователь системы", - } - return labels[self.get_role(obj)] + return UserService.get_role_label(self.get_role(obj)) @staticmethod def get_capabilities(obj) -> dict[str, bool]: - return { - "can_access_admin_page": bool(obj.is_staff or obj.is_superuser), - } + return UserService.get_user_capabilities(obj) class Meta: model = User @@ -169,6 +159,11 @@ class UserManagementSerializer(serializers.Serializer): email = serializers.EmailField(read_only=True) phone = serializers.CharField(read_only=True, allow_null=True) is_active = serializers.BooleanField(read_only=True) + is_verified = serializers.BooleanField(read_only=True) + role = serializers.SerializerMethodField() + role_label = serializers.SerializerMethodField() + capabilities = serializers.SerializerMethodField() + profile = CurrentUserProfileSerializer(read_only=True) first_name = serializers.SerializerMethodField() middle_name = serializers.SerializerMethodField() last_name = serializers.SerializerMethodField() @@ -196,6 +191,17 @@ class UserManagementSerializer(serializers.Serializer): def get_last_name(self, obj: User) -> str: return self._get_profile_value(getattr(obj, "profile", None), "last_name") + @staticmethod + def get_role(obj: User) -> str: + return UserService.get_user_role(obj) + + def get_role_label(self, obj: User) -> str: + return UserService.get_role_label(self.get_role(obj)) + + @staticmethod + def get_capabilities(obj: User) -> dict[str, bool]: + return UserService.get_user_capabilities(obj) + @staticmethod def _get_latest_job(obj: User) -> Any | None: return getattr(obj, "latest_job", None) @@ -243,6 +249,157 @@ class UserManagementSerializer(serializers.Serializer): return job.is_successful +class AdminUserListResponseSerializer(serializers.Serializer): + """Пагинированный ответ списка пользователей для административного UI.""" + + count = serializers.IntegerField(read_only=True) + next = serializers.URLField(read_only=True, allow_null=True) + previous = serializers.URLField(read_only=True, allow_null=True) + results = UserManagementSerializer(many=True, read_only=True) + + +class ManagedUserSerializer(serializers.ModelSerializer): + """Короткий ответ мутаций административного управления пользователями.""" + + role = serializers.SerializerMethodField() + role_label = serializers.SerializerMethodField() + + class Meta: + model = User + fields = ( + "id", + "username", + "email", + "phone", + "role", + "role_label", + "is_active", + ) + read_only_fields = fields + + @staticmethod + def get_role(obj: User) -> str: + return UserService.get_user_role(obj) + + def get_role_label(self, obj: User) -> str: + return UserService.get_role_label(self.get_role(obj)) + + +class AdminUserCreateSerializer(serializers.ModelSerializer): + """Входные данные для создания пользователя администратором.""" + + password = serializers.CharField( + write_only=True, + min_length=8, + help_text="Пароль (минимум 8 символов)", + ) + role = serializers.ChoiceField( + choices=UserService.ROLE_CHOICES, + default=UserService.ROLE_USER, + help_text="Прикладная роль пользователя", + ) + first_name = serializers.CharField(max_length=50, allow_blank=False) + middle_name = serializers.CharField( + max_length=50, + required=False, + allow_blank=True, + ) + last_name = serializers.CharField(max_length=50, allow_blank=False) + + class Meta: + model = User + fields = ( + "email", + "username", + "phone", + "password", + "role", + "is_active", + "is_verified", + "first_name", + "middle_name", + "last_name", + ) + extra_kwargs = { + "email": { + "validators": [UniqueValidator(queryset=User.objects.all())], + }, + "username": { + "validators": [UniqueValidator(queryset=User.objects.all())], + }, + } + + +class AdminUserUpdateSerializer(serializers.ModelSerializer): + """Входные данные частичного обновления пользователя администратором.""" + + password = serializers.CharField( + write_only=True, + required=False, + min_length=8, + help_text="Новый пароль (опционально)", + ) + role = serializers.ChoiceField( + choices=UserService.ROLE_CHOICES, + required=False, + help_text="Прикладная роль пользователя", + ) + first_name = serializers.CharField( + max_length=50, + required=False, + allow_blank=True, + ) + middle_name = serializers.CharField( + max_length=50, + required=False, + allow_blank=True, + ) + last_name = serializers.CharField( + max_length=50, + required=False, + allow_blank=True, + ) + + class Meta: + model = User + fields = ( + "email", + "username", + "phone", + "password", + "role", + "is_active", + "is_verified", + "first_name", + "middle_name", + "last_name", + ) + + def validate(self, attrs): + profile_fields = {"first_name", "middle_name", "last_name"} + if not any(field in attrs for field in profile_fields): + return attrs + + profile = getattr(self.instance, "profile", None) if self.instance else None + first_name = attrs.get( + "first_name", + profile.first_name if profile is not None else None, + ) + last_name = attrs.get( + "last_name", + profile.last_name if profile is not None else None, + ) + + errors = {} + if not str(first_name or "").strip(): + errors["first_name"] = "Обязательное поле." + if not str(last_name or "").strip(): + errors["last_name"] = "Обязательное поле." + if errors: + raise serializers.ValidationError(errors) + return attrs + + class UserUpdateSerializer(serializers.ModelSerializer): """Сериализатор для обновления данных пользователя""" diff --git a/src/apps/user/services.py b/src/apps/user/services.py index de36d1b..7474116 100644 --- a/src/apps/user/services.py +++ b/src/apps/user/services.py @@ -2,7 +2,9 @@ from typing import Any from apps.core.exceptions import NotFoundError from django.contrib.auth import get_user_model +from django.contrib.auth.models import Group from django.db import transaction +from django.db.models import F, Q from rest_framework_simplejwt.tokens import RefreshToken from .models import Profile @@ -13,6 +15,14 @@ User = get_user_model() class UserService: """Сервисный слой для работы с пользователями""" + ROLE_USER = "user" + ROLE_ADMIN = "admin" + ROLE_CHOICES = ( + (ROLE_USER, "Пользователь"), + (ROLE_ADMIN, "Администратор"), + ) + ROLE_LABELS = dict(ROLE_CHOICES) + @classmethod def create_user( cls, *, email: str, username: str, password: str, **extra_fields @@ -84,6 +94,74 @@ class UserService: except User.DoesNotExist: return None + @classmethod + def get_users_queryset(cls): + """Базовый queryset для административного управления пользователями.""" + return ( + User.objects.all() + .select_related("profile") + .prefetch_related("groups") + .order_by("-created_at") + ) + + @classmethod + def get_filtered_users_queryset( + cls, + *, + search: str = "", + ordering: str = "", + ): + """Фильтрует и сортирует пользователей для административного API.""" + queryset = cls.get_users_queryset() + search_term = search.strip() + + if search_term: + queryset = queryset.filter( + Q(username__icontains=search_term) + | Q(email__icontains=search_term) + | Q(phone__icontains=search_term) + | Q(profile__first_name__icontains=search_term) + | Q(profile__mid_name__icontains=search_term) + | Q(profile__last_name__icontains=search_term) + ).distinct() + + ordering_fields = [] + ordering_map = { + "id": ("id", False), + "email": ("email", False), + "username": ("username", False), + "phone": ("phone", False), + "is_active": ("is_active", False), + "is_verified": ("is_verified", False), + "created_at": ("created_at", False), + "updated_at": ("updated_at", False), + "first_name": ("profile__first_name", True), + "middle_name": ("profile__mid_name", True), + "last_name": ("profile__last_name", True), + "role": ("is_staff", False), + } + + for raw_field in (item.strip() for item in ordering.split(",") if item.strip()): + is_desc = raw_field.startswith("-") + field_name = raw_field[1:] if is_desc else raw_field + mapped_config = ordering_map.get(field_name) + if not mapped_config: + continue + mapped_field, nulls_last = mapped_config + if nulls_last: + ordering_fields.append( + F(mapped_field).desc(nulls_last=True) + if is_desc + else F(mapped_field).asc(nulls_last=True) + ) + continue + ordering_fields.append(f"-{mapped_field}" if is_desc else mapped_field) + + if ordering_fields: + queryset = queryset.order_by(*ordering_fields, "-created_at") + + return queryset + @classmethod def update_user(cls, user_id: int, **fields) -> User: """ @@ -107,6 +185,79 @@ class UserService: user.save() return user + @classmethod + @transaction.atomic + def create_managed_user( + cls, + *, + email: str, + username: str, + password: str, + role: str, + first_name: str, + last_name: str, + middle_name: str | None = None, + **extra_fields, + ) -> User: + """Создаёт пользователя администратором и назначает ему роль.""" + user = User.objects.create_user( + email=email, + username=username, + password=password, + **extra_fields, + ) + cls.assign_role(user, role) + cls._update_or_create_profile( + user=user, + first_name=first_name, + middle_name=middle_name or "", + last_name=last_name, + ) + return cls.get_users_queryset().get(id=user.id) + + @classmethod + @transaction.atomic + def update_managed_user(cls, user_id: int, **fields) -> User: + """Обновляет учётные данные, профиль, пароль и роль пользователя.""" + user = cls.get_user_by_id(user_id) + role = fields.pop("role", None) + password = fields.pop("password", None) + profile_fields = { + key: fields.pop(key) + for key in ("first_name", "middle_name", "last_name") + if key in fields + } + + for field, value in fields.items(): + setattr(user, field, value) + + if password: + user.set_password(password) + user.save() + + if role is not None: + cls.assign_role(user, role) + if profile_fields: + cls._update_or_create_profile(user=user, **profile_fields) + + return cls.get_users_queryset().get(id=user.id) + + @classmethod + def deactivate_user(cls, user_id: int) -> User: + """Деактивирует пользователя без удаления записи.""" + user = cls.get_user_by_id(user_id) + user.is_active = False + user.save(update_fields=["is_active"]) + return user + + @classmethod + def activate_user(cls, user_id: int) -> User: + """Возвращает пользователя в активное состояние.""" + user = cls.get_user_by_id(user_id) + user.is_active = True + user.save(update_fields=["is_active"]) + return user + @classmethod def delete_user(cls, user_id: int) -> None: """ @@ -138,6 +289,74 @@ class UserService: "access": str(refresh.access_token), } + @classmethod + def ensure_role_groups(cls) -> dict[str, Group]: + """Гарантирует наличие системных групп ролей.""" + groups: dict[str, Group] = {} + for role, _label in cls.ROLE_CHOICES: + group, _ = Group.objects.get_or_create(name=role) + groups[role] = group + return groups + + @classmethod + def get_user_role(cls, user: User) -> str: + """Возвращает прикладную роль пользователя.""" + if user.is_superuser or user.is_staff: + return cls.ROLE_ADMIN + group_names = {group.name for group in user.groups.all()} + if cls.ROLE_ADMIN in group_names: + return cls.ROLE_ADMIN + return cls.ROLE_USER + + @classmethod + def get_role_label(cls, role: str) -> str: + """Возвращает человекочитаемое название роли.""" + return cls.ROLE_LABELS.get(role, role) + + @classmethod + def get_user_capabilities(cls, user: User) -> dict[str, bool]: + """Возвращает доступные пользователю административные возможности.""" + return { + "can_access_admin_page": cls.get_user_role(user) == cls.ROLE_ADMIN, + } + + @classmethod + def assign_role(cls, user: User, role: str) -> User: + """Назначает роль через auth.Group и согласует staff-флаги.""" + if role not in cls.ROLE_LABELS: + raise ValueError(f"Unsupported role: {role}") + + groups = cls.ensure_role_groups() + current_role_groups = list(user.groups.filter(name__in=groups)) + if current_role_groups: + user.groups.remove(*current_role_groups) + user.groups.add(groups[role]) + + user.is_staff = role == cls.ROLE_ADMIN + if role == cls.ROLE_USER: + user.is_superuser = False + user.save(update_fields=["is_staff", "is_superuser"]) + return user + + @classmethod + def _update_or_create_profile( + cls, + *, + user: User, + first_name: str | None = None, + middle_name: str | None = None, + last_name: str | None = None, + ) -> Profile: + profile, _ = Profile.objects.get_or_create(user=user) + if first_name is not None: + profile.first_name = first_name + if middle_name is not None: + profile.mid_name = middle_name + if last_name is not None: + profile.last_name = last_name + profile.save() + return profile + @classmethod def verify_email(cls, user_id: int) -> User: """ diff --git a/src/apps/user/urls.py b/src/apps/user/urls.py index c25dd3b..2e47a17 100644 --- a/src/apps/user/urls.py +++ b/src/apps/user/urls.py @@ -15,6 +15,21 @@ urlpatterns = [ # Пользовательские данные path("me/", views.CurrentUserView.as_view(), name="current_user"), path("admin/users/", views.AdminUsersManagementView.as_view(), name="admin_users"), + path( + "admin/users//", + views.AdminUserDetailView.as_view(), + name="admin-user-detail", + ), + path( + "admin/users//deactivate/", + views.AdminUserDeactivateView.as_view(), + name="admin-user-deactivate", + ), + path( + "admin/users//activate/", + views.AdminUserActivateView.as_view(), + name="admin-user-activate", + ), path("me/update/", views.UserUpdateView.as_view(), name="user_update"), path("profile/", views.ProfileDetailView.as_view(), name="profile_detail"), path("profile/full/", views.user_profile_detail, name="profile_full"), diff --git a/src/apps/user/views.py b/src/apps/user/views.py index 6b7d11e..18dbd06 100644 --- a/src/apps/user/views.py +++ b/src/apps/user/views.py @@ -1,13 +1,18 @@ +from urllib.parse import urlencode + from apps.core.models import BackgroundJob from apps.core.services import BackgroundJobService from django.contrib.auth import authenticate, get_user_model from django.contrib.auth.hashers import check_password +from django.core.paginator import Paginator from django.db.models import F from django.db.models.functions import Coalesce +from django.shortcuts import get_object_or_404 from drf_yasg import openapi from drf_yasg.utils import swagger_auto_schema from rest_framework import generics, status from rest_framework.decorators import api_view, permission_classes +from rest_framework.exceptions import ValidationError from rest_framework.permissions import AllowAny, IsAdminUser, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView @@ -15,8 +20,12 @@ from rest_framework_simplejwt.exceptions import TokenError from rest_framework_simplejwt.tokens import RefreshToken from .serializers import ( + AdminUserCreateSerializer, + AdminUserListResponseSerializer, + AdminUserUpdateSerializer, CurrentUserSerializer, LoginSerializer, + ManagedUserSerializer, PasswordChangeSerializer, ProfileUpdateSerializer, TokenSerializer, @@ -34,6 +43,56 @@ AUTH_TAG = "Аутентификация" USER_TAG = "Пользователь" +def _build_page_url(request, page_number: int) -> str: + query_params = request.query_params.copy() + query_params["page"] = page_number + encoded_query = urlencode(query_params, doseq=True) + return request.build_absolute_uri(f"{request.path}?{encoded_query}") + + +def _paginate_user_queryset(request, queryset): + """Пагинирует только запросы с явными page/page_size для legacy-совместимости.""" + if "page" not in request.query_params and "page_size" not in request.query_params: + return { + "count": queryset.count(), + "next": None, + "previous": None, + "results": list(queryset), + } + + page_size_raw = request.query_params.get("page_size", "20") + page_raw = request.query_params.get("page", "1") + try: + page_size = max(1, min(int(page_size_raw), 100)) + page_number = max(1, int(page_raw)) + except (TypeError, ValueError) as exc: + raise ValidationError( + { + "detail": ( + "Параметры page и page_size должны быть положительными " + "целыми числами." + ) + } + ) from exc + + paginator = Paginator(queryset, page_size) + page_obj = paginator.get_page(page_number) + return { + "count": paginator.count, + "next": ( + _build_page_url(request, page_obj.next_page_number()) + if page_obj.has_next() + else None + ), + "previous": ( + _build_page_url(request, page_obj.previous_page_number()) + if page_obj.has_previous() + else None + ), + "results": list(page_obj.object_list), + } + + class RegisterView(APIView): """ Регистрация нового пользователя. @@ -239,7 +298,7 @@ class PasswordChangeView(APIView): class AdminUsersManagementView(APIView): - """Список пользователей для административной страницы управления.""" + """Список и создание пользователей в административной панели.""" permission_classes = [IsAdminUser] @@ -276,15 +335,152 @@ class AdminUsersManagementView(APIView): operation_summary="Список пользователей (admin)", operation_description=( "Возвращает пользователей для административной панели управления. " - "Включает метрики последней фоновой задачи пользователя." + "Включает роль, профиль и метрики последней фоновой задачи пользователя." ), - responses={200: UserManagementSerializer(many=True)}, + manual_parameters=[ + openapi.Parameter( + "search", + openapi.IN_QUERY, + type=openapi.TYPE_STRING, + required=False, + ), + openapi.Parameter( + "ordering", + openapi.IN_QUERY, + type=openapi.TYPE_STRING, + required=False, + ), + openapi.Parameter( + "page", + openapi.IN_QUERY, + type=openapi.TYPE_INTEGER, + required=False, + ), + openapi.Parameter( + "page_size", + openapi.IN_QUERY, + type=openapi.TYPE_INTEGER, + required=False, + ), + ], + responses={200: AdminUserListResponseSerializer}, ) def get(self, request): - users = User.objects.all().select_related("profile").order_by("id") - self._attach_latest_jobs(list(users)) + queryset = UserService.get_filtered_users_queryset( + search=request.query_params.get("search", ""), + ordering=request.query_params.get("ordering", ""), + ) + paginated = _paginate_user_queryset(request, queryset) + users = paginated["results"] + self._attach_latest_jobs(users) serializer = UserManagementSerializer(users, many=True) - return Response({"results": serializer.data}) + return Response( + { + "count": paginated["count"], + "next": paginated["next"], + "previous": paginated["previous"], + "results": serializer.data, + } + ) + + @swagger_auto_schema( + tags=[USER_TAG], + operation_summary="Создать пользователя (admin)", + request_body=AdminUserCreateSerializer, + responses={201: ManagedUserSerializer}, + ) + def post(self, request): + serializer = AdminUserCreateSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + user = UserService.create_managed_user(**serializer.validated_data) + return Response( + ManagedUserSerializer(user).data, + status=status.HTTP_201_CREATED, + ) + + +class AdminUserDetailView(APIView): + """Просмотр и частичное обновление пользователя администратором.""" + + permission_classes = [IsAdminUser] + + @staticmethod + def _get_user(user_id: int): + return get_object_or_404(UserService.get_users_queryset(), id=user_id) + + @swagger_auto_schema( + tags=[USER_TAG], + operation_summary="Детали пользователя (admin)", + responses={200: UserManagementSerializer}, + ) + def get(self, request, user_id: int): + user = self._get_user(user_id) + AdminUsersManagementView._attach_latest_jobs([user]) + return Response(UserManagementSerializer(user).data) + + @swagger_auto_schema( + tags=[USER_TAG], + operation_summary="Обновить пользователя (admin)", + request_body=AdminUserUpdateSerializer, + responses={200: ManagedUserSerializer}, + ) + def patch(self, request, user_id: int): + user = self._get_user(user_id) + serializer = AdminUserUpdateSerializer(user, data=request.data, partial=True) + serializer.is_valid(raise_exception=True) + + if request.user.id == user.id: + if serializer.validated_data.get("is_active") is False: + return Response( + {"detail": "Нельзя деактивировать самого себя."}, + status=status.HTTP_400_BAD_REQUEST, + ) + if serializer.validated_data.get("role") == UserService.ROLE_USER: + return Response( + {"detail": "Нельзя снять у себя роль администратора."}, + status=status.HTTP_400_BAD_REQUEST, + ) + + updated_user = UserService.update_managed_user( + user_id=user.id, + **serializer.validated_data, + ) + return Response(ManagedUserSerializer(updated_user).data) + + +class AdminUserDeactivateView(APIView): + """Деактивация пользователя администратором.""" + + permission_classes = [IsAdminUser] + + @swagger_auto_schema( + tags=[USER_TAG], + operation_summary="Деактивировать пользователя (admin)", + responses={200: ManagedUserSerializer}, + ) + def post(self, request, user_id: int): + if request.user.id == user_id: + return Response( + {"detail": "Нельзя деактивировать самого себя."}, + status=status.HTTP_400_BAD_REQUEST, + ) + user = UserService.deactivate_user(user_id) + return Response(ManagedUserSerializer(user).data) + + +class AdminUserActivateView(APIView): + """Активация пользователя администратором.""" + + permission_classes = [IsAdminUser] + + @swagger_auto_schema( + tags=[USER_TAG], + operation_summary="Активировать пользователя (admin)", + responses={200: ManagedUserSerializer}, + ) + def post(self, request, user_id: int): + user = UserService.activate_user(user_id) + return Response(ManagedUserSerializer(user).data) @swagger_auto_schema( diff --git a/tests/apps/user/test_views.py b/tests/apps/user/test_views.py index 153e05e..c20656d 100644 --- a/tests/apps/user/test_views.py +++ b/tests/apps/user/test_views.py @@ -152,7 +152,7 @@ class CurrentUserViewTest(APITestCase): response = self.client.get(self.current_user_url) self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data["role"], "staff") + self.assertEqual(response.data["role"], "admin") self.assertEqual(response.data["capabilities"]["can_access_admin_page"], True) def test_get_current_user_unauthenticated(self): @@ -224,6 +224,10 @@ class AdminUsersManagementViewTest(APITestCase): self.assertEqual(job_user_row["first_name"], "Иван") self.assertEqual(job_user_row["middle_name"], "Сергеевич") self.assertEqual(job_user_row["last_name"], "Петров") + self.assertEqual(job_user_row["profile"]["middle_name"], "Сергеевич") + self.assertEqual(job_user_row["role"], "user") + self.assertEqual(job_user_row["role_label"], "Пользователь") + self.assertFalse(job_user_row["capabilities"]["can_access_admin_page"]) self.assertEqual(job_user_row["progress_message"], "Готово") self.assertEqual(job_user_row["result"], {"processed": 10}) self.assertIsNone(job_user_row["error"]) @@ -237,6 +241,29 @@ class AdminUsersManagementViewTest(APITestCase): self.assertIsNone(regular_user_row["error"]) self.assertIsNone(regular_user_row["is_successful"]) + def test_admin_can_search_order_and_paginate_users(self): + first = UserFactory.create_user(username="alpha-user") + second = UserFactory.create_user(username="beta-user") + first.profile.first_name = "Яков" + first.profile.save(update_fields=["first_name"]) + second.profile.first_name = "Алексей" + second.profile.save(update_fields=["first_name"]) + + search_response = self.client.get(self.url, {"search": "beta-user"}) + ordering_response = self.client.get( + self.url, + {"ordering": "first_name", "page": 1, "page_size": 100}, + ) + + self.assertEqual(search_response.status_code, status.HTTP_200_OK) + self.assertEqual( + [item["id"] for item in search_response.data["results"]], + [second.id], + ) + ordered_ids = [item["id"] for item in ordering_response.data["results"]] + self.assertLess(ordered_ids.index(second.id), ordered_ids.index(first.id)) + self.assertEqual(ordering_response.data["count"], User.objects.count()) + def test_non_admin_cannot_access_endpoint(self): self.client.force_authenticate(self.regular_user) response = self.client.get(self.url) @@ -244,6 +271,167 @@ class AdminUsersManagementViewTest(APITestCase): self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) +class AdminUserCrudViewTest(APITestCase): + """Tests for admin create, detail, update and activation endpoints.""" + + def setUp(self): + self.admin_user = UserFactory.create_user(is_staff=True) + self.regular_user = UserFactory.create_user() + self.client.force_authenticate(self.admin_user) + self.list_url = reverse("api_v1:user:admin_users") + + def test_admin_can_create_user_with_role_and_profile(self): + password = fake.password(length=12, special_chars=False) + payload = { + "email": fake.unique.email(), + "username": fake.unique.user_name(), + "phone": f"+7{fake.numerify('##########')}", + "password": password, + "role": "admin", + "is_verified": True, + "first_name": "Пётр", + "middle_name": "Петрович", + "last_name": "Петров", + } + + response = self.client.post(self.list_url, payload, format="json") + + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + created = User.objects.get(username=payload["username"]) + self.assertTrue(created.check_password(password)) + self.assertTrue(created.is_staff) + self.assertFalse(created.is_superuser) + self.assertTrue(created.is_verified) + self.assertTrue(created.groups.filter(name="admin").exists()) + self.assertEqual(created.profile.first_name, "Пётр") + self.assertEqual(created.profile.mid_name, "Петрович") + self.assertEqual(response.data["role"], "admin") + + def test_admin_create_validates_unique_fields_and_password(self): + duplicate_response = self.client.post( + self.list_url, + { + "email": self.regular_user.email, + "username": fake.unique.user_name(), + "password": "valid-pass", + "first_name": "Иван", + "last_name": "Иванов", + }, + format="json", + ) + password_response = self.client.post( + self.list_url, + { + "email": fake.unique.email(), + "username": fake.unique.user_name(), + "password": "short", + "first_name": "Иван", + "last_name": "Иванов", + }, + format="json", + ) + + self.assertEqual(duplicate_response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertEqual(password_response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_admin_can_get_and_update_user(self): + detail_url = reverse( + "api_v1:user:admin-user-detail", + args=[self.regular_user.id], + ) + new_password = fake.password(length=12, special_chars=False) + + detail_response = self.client.get(detail_url) + update_response = self.client.patch( + detail_url, + { + "role": "admin", + "password": new_password, + "is_verified": True, + "first_name": "Иван", + "middle_name": "Иванович", + "last_name": "Иванов", + }, + format="json", + ) + + self.assertEqual(detail_response.status_code, status.HTTP_200_OK) + self.assertIn("profile", detail_response.data) + self.assertIn("progress_message", detail_response.data) + self.assertEqual(update_response.status_code, status.HTTP_200_OK) + self.regular_user.refresh_from_db() + self.assertTrue(self.regular_user.is_staff) + self.assertTrue(self.regular_user.is_verified) + self.assertTrue(self.regular_user.check_password(new_password)) + self.assertEqual(self.regular_user.profile.first_name, "Иван") + self.assertEqual(self.regular_user.profile.mid_name, "Иванович") + self.assertEqual(update_response.data["role"], "admin") + + def test_admin_can_deactivate_and_activate_user(self): + deactivate_url = reverse( + "api_v1:user:admin-user-deactivate", + args=[self.regular_user.id], + ) + activate_url = reverse( + "api_v1:user:admin-user-activate", + args=[self.regular_user.id], + ) + + deactivate_response = self.client.post(deactivate_url, {}, format="json") + self.regular_user.refresh_from_db() + self.assertEqual(deactivate_response.status_code, status.HTTP_200_OK) + self.assertFalse(self.regular_user.is_active) + + activate_response = self.client.post(activate_url, {}, format="json") + self.regular_user.refresh_from_db() + self.assertEqual(activate_response.status_code, status.HTTP_200_OK) + self.assertTrue(self.regular_user.is_active) + + def test_admin_cannot_deactivate_or_demote_self(self): + detail_url = reverse( + "api_v1:user:admin-user-detail", + args=[self.admin_user.id], + ) + deactivate_url = reverse( + "api_v1:user:admin-user-deactivate", + args=[self.admin_user.id], + ) + + demote_response = self.client.patch( + detail_url, + {"role": "user"}, + format="json", + ) + patch_inactive_response = self.client.patch( + detail_url, + {"is_active": False}, + format="json", + ) + deactivate_response = self.client.post(deactivate_url, {}, format="json") + + self.assertEqual(demote_response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertEqual( + patch_inactive_response.status_code, + status.HTTP_400_BAD_REQUEST, + ) + self.assertEqual(deactivate_response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_regular_user_cannot_manage_users(self): + self.client.force_authenticate(self.regular_user) + detail_url = reverse( + "api_v1:user:admin-user-detail", + args=[self.admin_user.id], + ) + + response = self.client.patch( + detail_url, + {"is_active": False}, + format="json", + ) + + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + + class UserUpdateViewTest(APITestCase): """Tests for UserUpdateView"""