avm/state-corp-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(user): users/me and admin user-management contract fields

Browse Source
This commit is contained in:
Aleksandr Meshchriakov
2026-04-14 10:41:18 +02:00
parent 97e269fe1a
commit 9ef88c23a0
5 changed files with 318 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
src/apps/user/serializers.py
View File

@@ -1,3 +1,7 @@
from __future__ import annotations
from typing import Any
from django.contrib.auth import get_user_model
from rest_framework import serializers
from rest_framework.validators import UniqueValidator
@@ -89,7 +93,7 @@ class UserSerializer(serializers.ModelSerializer):
class CurrentUserProfileSerializer(serializers.ModelSerializer):
"""Профиль текущего пользователя в контракте `/users/me/`."""
middle_name = serializers.CharField(source="mid_name", allow_null=True)
middle_name = serializers.SerializerMethodField()
full_name = serializers.ReadOnlyField()
class Meta:
@@ -101,6 +105,10 @@ class CurrentUserProfileSerializer(serializers.ModelSerializer):
"full_name",
)
@staticmethod
def get_middle_name(obj: Profile) -> str:
return obj.mid_name or ""
class CurrentUserCapabilitiesSerializer(serializers.Serializer):
"""Возможности пользователя для UI-контрактов."""
@@ -153,6 +161,88 @@ class CurrentUserSerializer(serializers.ModelSerializer):
)
class UserManagementSerializer(serializers.Serializer):
"""Serializer for users listed in admin management endpoint."""
id = serializers.IntegerField(read_only=True)
username = serializers.CharField(read_only=True)
email = serializers.EmailField(read_only=True)
phone = serializers.CharField(read_only=True, allow_null=True)
is_active = serializers.BooleanField(read_only=True)
first_name = serializers.SerializerMethodField()
middle_name = serializers.SerializerMethodField()
last_name = serializers.SerializerMethodField()
progress_message = serializers.SerializerMethodField()
result = serializers.SerializerMethodField()
error = serializers.SerializerMethodField()
started_at = serializers.SerializerMethodField()
completed_at = serializers.SerializerMethodField()
duration = serializers.SerializerMethodField()
is_successful = serializers.SerializerMethodField()
@staticmethod
def _get_profile_value(profile: Profile | None, field_name: str) -> str:
if profile is None:
return ""
value = getattr(profile, field_name, "")
return value or ""
def get_first_name(self, obj: User) -> str:
return self._get_profile_value(getattr(obj, "profile", None), "first_name")
def get_middle_name(self, obj: User) -> str:
return self._get_profile_value(getattr(obj, "profile", None), "mid_name")
def get_last_name(self, obj: User) -> str:
return self._get_profile_value(getattr(obj, "profile", None), "last_name")
@staticmethod
def _get_latest_job(obj: User) -> Any | None:
return getattr(obj, "latest_job", None)
def get_progress_message(self, obj: User) -> str | None:
job = self._get_latest_job(obj)
if job is None:
return None
return job.progress_message or None
def get_result(self, obj: User):
job = self._get_latest_job(obj)
if job is None:
return None
return job.result
def get_error(self, obj: User) -> str | None:
job = self._get_latest_job(obj)
if job is None:
return None
return job.error or None
def get_started_at(self, obj: User):
job = self._get_latest_job(obj)
if job is None:
return None
return job.started_at
def get_completed_at(self, obj: User):
job = self._get_latest_job(obj)
if job is None:
return None
return job.completed_at
def get_duration(self, obj: User) -> float | None:
job = self._get_latest_job(obj)
if job is None:
return None
return job.duration
def get_is_successful(self, obj: User) -> bool | None:
job = self._get_latest_job(obj)
if job is None:
return None
return job.is_successful
class UserUpdateSerializer(serializers.ModelSerializer):
"""Сериализатор для обновления данных пользователя"""

1
src/apps/user/urls.py
View File

@@ -14,6 +14,7 @@ urlpatterns = [
path("token/verify/", TokenVerifyView.as_view(), name="token_verify"),
# Пользовательские данные
path("me/", views.CurrentUserView.as_view(), name="current_user"),
path("admin/users/", views.AdminUsersManagementView.as_view(), name="admin_users"),
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"),

48
src/apps/user/views.py
View File

@@ -1,10 +1,12 @@
from django.contrib.auth import authenticate
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 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.permissions import AllowAny, IsAuthenticated
from rest_framework.permissions import AllowAny, IsAdminUser, IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_simplejwt.tokens import RefreshToken
@@ -15,12 +17,15 @@ from .serializers import (
PasswordChangeSerializer,
ProfileUpdateSerializer,
TokenSerializer,
UserManagementSerializer,
UserRegistrationSerializer,
UserSerializer,
UserUpdateSerializer,
)
from .services import ProfileService, UserService
User = get_user_model()
# Swagger теги для группировки
AUTH_TAG = "Аутентификация"
USER_TAG = "Пользователь"
@@ -230,6 +235,45 @@ class PasswordChangeView(APIView):
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class AdminUsersManagementView(APIView):
"""Список пользователей для административной страницы управления."""
permission_classes = [IsAdminUser]
@staticmethod
def _attach_latest_jobs(users: list[User]):
user_ids = [user.id for user in users]
if not user_ids:
return
latest_jobs: dict[int, BackgroundJob] = {}
jobs = BackgroundJobService.get_queryset().filter(
user_id__in=user_ids
).order_by("user_id", "-created_at")
for job in jobs:
if job.user_id not in latest_jobs:
latest_jobs[job.user_id] = job
for user in users:
user.latest_job = latest_jobs.get(user.id)
@swagger_auto_schema(
tags=[USER_TAG],
operation_summary="Список пользователей (admin)",
operation_description=(
"Возвращает пользователей для административной панели управления. "
"Включает метрики последней фоновой задачи пользователя."
),
responses={200: UserManagementSerializer(many=True)},
)
def get(self, request):
users = User.objects.all().select_related("profile").order_by("id")
self._attach_latest_jobs(list(users))
serializer = UserManagementSerializer(users, many=True)
return Response({"results": serializer.data})
@swagger_auto_schema(
method="get",
tags=[USER_TAG],

93
tests/apps/user/test_serializers.py
View File

@@ -1,10 +1,16 @@
"""Tests for user serializers"""
from datetime import datetime, timedelta
from apps.core.models import BackgroundJob, JobStatus
from apps.user.serializers import (
CurrentUserProfileSerializer,
CurrentUserSerializer,
LoginSerializer,
PasswordChangeSerializer,
ProfileUpdateSerializer,
TokenSerializer,
UserManagementSerializer,
UserRegistrationSerializer,
UserSerializer,
UserUpdateSerializer,
@@ -124,6 +130,93 @@ class UserSerializerTest(TestCase):
self.assertIn(field_name, serializer.Meta.read_only_fields)
class CurrentUserProfileSerializerTest(TestCase):
"""Tests for current user profile serializer"""
def setUp(self):
self.user = UserFactory.create_user()
self.profile = ProfileFactory.create_profile(
user=self.user,
first_name="Ivan",
mid_name=None,
last_name="Petrov",
)
def test_middle_name_falls_back_to_empty_string(self):
serializer = CurrentUserProfileSerializer(self.profile)
self.assertEqual(serializer.data["middle_name"], "")
class CurrentUserSerializerTest(TestCase):
"""Tests for current user serializer fields."""
def setUp(self):
self.user = UserFactory.create_user()
def test_current_user_contains_is_active(self):
serializer = CurrentUserSerializer(self.user)
self.assertIn("is_active", serializer.data)
self.assertEqual(serializer.data["is_active"], self.user.is_active)
class UserManagementSerializerTest(TestCase):
"""Tests for user management serializer."""
def setUp(self):
self.user = UserFactory.create_user()
self.user.profile.first_name = None
self.user.profile.mid_name = None
self.user.profile.last_name = None
self.user.profile.save()
def test_profile_fields_fallback_to_empty_string(self):
serializer = UserManagementSerializer(self.user)
self.assertEqual(serializer.data["first_name"], "")
self.assertEqual(serializer.data["middle_name"], "")
self.assertEqual(serializer.data["last_name"], "")
def test_metric_fields_are_derived_from_latest_job(self):
now = datetime(2026, 4, 14, 10, 0, 0)
latest_job = BackgroundJob.objects.create(
task_id="admin-management-latest",
task_name="apps.forms.process",
user_id=self.user.id,
status=JobStatus.SUCCESS,
progress=100,
progress_message="Готово",
result={"processed": 10},
started_at=now,
completed_at=now + timedelta(minutes=3),
created_at=now,
updated_at=now + timedelta(minutes=3),
)
BackgroundJob.objects.create(
task_id="admin-management-old",
task_name="apps.forms.process",
user_id=self.user.id,
status=JobStatus.FAILURE,
progress=100,
progress_message="Ошибка",
error="old-error",
started_at=now - timedelta(minutes=15),
completed_at=now - timedelta(minutes=10),
created_at=now - timedelta(minutes=15),
updated_at=now - timedelta(minutes=10),
)
self.user.latest_job = latest_job
serializer = UserManagementSerializer(self.user)
self.assertEqual(serializer.data["progress_message"], "Готово")
self.assertEqual(serializer.data["result"], {"processed": 10})
self.assertIsNone(serializer.data["error"])
self.assertEqual(serializer.data["is_successful"], True)
self.assertEqual(serializer.data["duration"], 180.0)
class UserUpdateSerializerTest(TestCase):
"""Tests for UserUpdateSerializer"""

88
tests/apps/user/test_views.py
View File

@@ -1,5 +1,8 @@
"""Tests for user DRF views"""
from datetime import datetime, timedelta
from apps.core.models import BackgroundJob, JobStatus
from apps.user.models import Profile
from apps.user.services import UserService
from django.contrib.auth import get_user_model
@@ -123,7 +126,7 @@ class CurrentUserViewTest(APITestCase):
def setUp(self):
self.user = UserFactory.create_user()
ProfileFactory.create_profile(user=self.user)
ProfileFactory.create_profile(user=self.user, mid_name="")
self.current_user_url = reverse("api_v1:user:current_user")
self.tokens = UserService.get_tokens_for_user(self.user)
self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {self.tokens['access']}")
@@ -135,7 +138,9 @@ class CurrentUserViewTest(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["id"], self.user.id)
self.assertEqual(response.data["email"], self.user.email)
self.assertEqual(response.data["is_active"], self.user.is_active)
self.assertIn("profile", response.data)
self.assertEqual(response.data["profile"]["middle_name"], "")
self.assertEqual(response.data["role"], "user")
self.assertEqual(response.data["capabilities"]["can_access_admin_page"], False)
@@ -157,6 +162,87 @@ class CurrentUserViewTest(APITestCase):
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class AdminUsersManagementViewTest(APITestCase):
"""Tests for admin users endpoint."""
def setUp(self):
self.admin_user = UserFactory.create_user(is_staff=True)
self.regular_user = UserFactory.create_user()
self.job_user = UserFactory.create_user()
self.client.force_authenticate(self.admin_user)
self.url = reverse("api_v1:user:admin_users")
ProfileFactory.create_profile(
user=self.job_user,
first_name="Иван",
mid_name="Сергеевич",
last_name="Петров",
)
now = datetime(2026, 4, 14, 10, 0, 0)
completed_job = now + timedelta(minutes=3)
failed_job = now - timedelta(minutes=10)
BackgroundJob.objects.create(
task_id="admin-management-task",
task_name="apps.forms.process",
user_id=self.job_user.id,
status=JobStatus.SUCCESS,
progress=100,
progress_message="Готово",
result={"processed": 10},
started_at=now,
completed_at=completed_job,
created_at=completed_job,
updated_at=completed_job,
)
BackgroundJob.objects.create(
task_id="admin-management-task-failed",
task_name="apps.forms.process",
user_id=self.job_user.id,
status=JobStatus.FAILURE,
progress=100,
progress_message="Неуспешно",
error="error",
started_at=failed_job,
completed_at=now - timedelta(minutes=50),
created_at=failed_job,
updated_at=failed_job,
)
def test_admin_users_list_contains_import_metrics_and_profile_fields(self):
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("results", response.data)
results = response.data["results"]
row_map = {row["id"]: row for row in results}
job_user_row = row_map[self.job_user.id]
regular_user_row = row_map[self.regular_user.id]
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["progress_message"], "Готово")
self.assertEqual(job_user_row["result"], {"processed": 10})
self.assertIsNone(job_user_row["error"])
self.assertEqual(job_user_row["is_successful"], True)
self.assertEqual(job_user_row["duration"], 180.0)
self.assertIsNotNone(job_user_row["started_at"])
self.assertIsNotNone(job_user_row["completed_at"])
self.assertIsNone(regular_user_row["progress_message"])
self.assertIsNone(regular_user_row["result"])
self.assertIsNone(regular_user_row["error"])
self.assertIsNone(regular_user_row["is_successful"])
def test_non_admin_cannot_access_endpoint(self):
self.client.force_authenticate(self.regular_user)
response = self.client.get(self.url)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
class UserUpdateViewTest(APITestCase):
"""Tests for UserUpdateView"""