fix(api): align admin users contracts
All checks were successful
CI/CD Pipeline / Code Quality Checks (push) Successful in 1m38s
CI/CD Pipeline / Run Tests (push) Successful in 4m12s
CI/CD Pipeline / Telegram Notify Success (push) Successful in 19s

This commit is contained in:
2026-03-23 10:24:08 +01:00
parent 1ec8c39470
commit c28b603831
3 changed files with 193 additions and 72 deletions

View File

@@ -0,0 +1,28 @@
from django.db import migrations
DEFAULT_REGISTER_NAMES = (
"Реестр предприятий ОПК",
"Реестр госкорпорации Роскосмос",
"Реестр госкорпорации Росатом",
)
def seed_default_registers(apps, schema_editor):
Register = apps.get_model("registers", "Register")
db_alias = schema_editor.connection.alias
for name in DEFAULT_REGISTER_NAMES:
Register.objects.using(db_alias).get_or_create(name=name)
class Migration(migrations.Migration):
dependencies = [
("registers", "0003_add_unique_active_membership_period"),
]
operations = [
migrations.RunPython(
seed_default_registers,
migrations.RunPython.noop,
),
]

View File

@@ -1,7 +1,6 @@
from apps.core.openapi import CommonResponses, ErrorResponses, swagger_tag
from django.contrib.auth import authenticate
from django.contrib.auth.hashers import check_password
from django.core.paginator import Paginator
from django.shortcuts import get_object_or_404
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
@@ -33,6 +32,79 @@ AUTH_TAG = swagger_tag("Аутентификация", "authentication")
USER_TAG = swagger_tag("Пользователь", "user")
USER_ADMIN_TAG = swagger_tag("Управление пользователями", "user_management")
ADMIN_USER_VALIDATION_ERROR_RESPONSE = openapi.Response(
description="Ошибка валидации пользователя",
schema=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"success": openapi.Schema(type=openapi.TYPE_BOOLEAN, default=False),
"data": openapi.Schema(type=openapi.TYPE_OBJECT, nullable=True),
"errors": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"code": openapi.Schema(
type=openapi.TYPE_STRING,
default="validation_error",
),
"message": openapi.Schema(
type=openapi.TYPE_STRING,
default="Validation failed",
),
"details": openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"fields": openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"email": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(
type=openapi.TYPE_STRING
),
),
"username": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(
type=openapi.TYPE_STRING
),
),
"password": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(
type=openapi.TYPE_STRING
),
),
"first_name": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(
type=openapi.TYPE_STRING
),
),
"last_name": openapi.Schema(
type=openapi.TYPE_ARRAY,
items=openapi.Schema(
type=openapi.TYPE_STRING
),
),
},
)
},
),
},
),
),
"meta": openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
"request_id": openapi.Schema(type=openapi.TYPE_STRING),
},
),
},
),
)
class RegisterView(APIView):
"""
@@ -162,24 +234,6 @@ class AdminUserListCreateView(APIView):
permission_classes = [IsAdminUser]
@staticmethod
def _get_positive_int(value, *, default: int, minimum: int = 1) -> int:
if value in (None, ""):
return default
try:
parsed = int(value)
except (TypeError, ValueError):
raise ValueError("must be integer") from None
if parsed < minimum:
raise ValueError("must be positive")
return parsed
@staticmethod
def _build_page_url(request, page_number: int) -> str:
query_params = request.query_params.copy()
query_params["page"] = page_number
return request.build_absolute_uri(f"{request.path}?{query_params.urlencode()}")
@swagger_auto_schema(
tags=[USER_ADMIN_TAG],
operation_summary="Список пользователей",
@@ -209,7 +263,7 @@ class AdminUserListCreateView(APIView):
),
],
responses={
200: FrontendUserWithProfileSerializer(many=True),
200: FrontendManagedUserSerializer(many=True),
**ErrorResponses.ADMIN,
},
)
@@ -218,47 +272,8 @@ class AdminUserListCreateView(APIView):
search=request.query_params.get("search", ""),
ordering=request.query_params.get("ordering", ""),
)
try:
page_number = self._get_positive_int(
request.query_params.get("page"),
default=1,
)
page_size = self._get_positive_int(
request.query_params.get("page_size"),
default=20,
)
except ValueError:
return Response(
{
"detail": (
"Параметры page и page_size должны быть положительными "
"целыми числами."
)
},
status=status.HTTP_400_BAD_REQUEST,
)
page_size = min(page_size, 100)
paginator = Paginator(queryset, page_size)
page_obj = paginator.get_page(page_number)
serializer = FrontendUserWithProfileSerializer(page_obj.object_list, many=True)
return Response(
{
"count": paginator.count,
"next": (
self._build_page_url(request, page_obj.next_page_number())
if page_obj.has_next()
else None
),
"previous": (
self._build_page_url(request, page_obj.previous_page_number())
if page_obj.has_previous()
else None
),
"results": serializer.data,
}
)
serializer = FrontendManagedUserSerializer(queryset, many=True)
return Response(serializer.data)
@swagger_auto_schema(
tags=[USER_ADMIN_TAG],
@@ -269,7 +284,7 @@ class AdminUserListCreateView(APIView):
request_body=AdminUserCreateSerializer,
responses={
201: FrontendManagedUserSerializer,
400: CommonResponses.BAD_REQUEST,
400: ADMIN_USER_VALIDATION_ERROR_RESPONSE,
**ErrorResponses.ADMIN,
},
)
@@ -314,7 +329,7 @@ class AdminUserDetailView(APIView):
request_body=AdminUserUpdateSerializer,
responses={
200: FrontendManagedUserSerializer,
400: CommonResponses.BAD_REQUEST,
400: ADMIN_USER_VALIDATION_ERROR_RESPONSE,
**ErrorResponses.ADMIN_NOT_FOUND,
},
)

View File

@@ -254,7 +254,7 @@ class AdminUserManagementViewTest(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
set(response.data["results"][0].keys()),
set(response.data[0].keys()),
{
"id",
"username",
@@ -263,14 +263,9 @@ class AdminUserManagementViewTest(APITestCase):
"is_active",
"role",
"role_label",
"profile",
},
)
self.assertEqual(
set(response.data["results"][0]["profile"].keys()),
{"first_name", "middle_name", "last_name", "full_name"},
)
usernames = {item["username"] for item in response.data["results"]}
usernames = {item["username"] for item in response.data}
self.assertIn(self.admin.username, usernames)
self.assertIn(self.user.username, usernames)
@@ -287,7 +282,7 @@ class AdminUserManagementViewTest(APITestCase):
response = self.client.get(self.list_url, {"search": "Петрович"})
self.assertEqual(response.status_code, status.HTTP_200_OK)
usernames = [item["username"] for item in response.data["results"]]
usernames = [item["username"] for item in response.data]
self.assertEqual(usernames, [self.user.username])
def test_admin_can_order_users(self):
@@ -299,7 +294,7 @@ class AdminUserManagementViewTest(APITestCase):
response = self.client.get(self.list_url, {"ordering": "first_name"})
self.assertEqual(response.status_code, status.HTTP_200_OK)
ordered_ids = [item["id"] for item in response.data["results"]]
ordered_ids = [item["id"] for item in response.data]
self.assertLess(ordered_ids.index(second.id), ordered_ids.index(first.id))
def test_admin_can_create_user_with_role(self):
@@ -336,6 +331,51 @@ class AdminUserManagementViewTest(APITestCase):
self.assertEqual(created.profile.first_name, "Петр")
self.assertEqual(created.profile.middle_name, "Петрович")
def test_admin_create_user_returns_duplicate_email_error(self):
payload = {
"email": self.user.email,
"username": fake.unique.user_name(),
"password": fake.password(length=12, special_chars=False),
"role": "user",
"first_name": "Петр",
"last_name": "Петров",
}
response = self.client.post(self.list_url, payload, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("email", response.data["errors"][0]["details"]["fields"])
def test_admin_create_user_returns_duplicate_username_error(self):
payload = {
"email": fake.unique.email(),
"username": self.user.username,
"password": fake.password(length=12, special_chars=False),
"role": "user",
"first_name": "Петр",
"last_name": "Петров",
}
response = self.client.post(self.list_url, payload, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("username", response.data["errors"][0]["details"]["fields"])
def test_admin_create_user_returns_password_validation_error(self):
payload = {
"email": fake.unique.email(),
"username": fake.unique.user_name(),
"password": "123",
"role": "user",
"first_name": "Петр",
"last_name": "Петров",
}
response = self.client.post(self.list_url, payload, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("password", response.data["errors"][0]["details"]["fields"])
def test_admin_can_update_user_and_role(self):
url = reverse("api_v1:user:admin-user-detail", args=[self.user.id])
@@ -382,6 +422,44 @@ class AdminUserManagementViewTest(APITestCase):
{"first_name", "middle_name", "last_name", "full_name"},
)
def test_admin_patch_user_returns_duplicate_email_error(self):
another = UserFactory.create_user()
url = reverse("api_v1:user:admin-user-detail", args=[another.id])
response = self.client.patch(
url,
{"email": self.user.email},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("email", response.data["errors"][0]["details"]["fields"])
def test_admin_patch_user_returns_duplicate_username_error(self):
another = UserFactory.create_user()
url = reverse("api_v1:user:admin-user-detail", args=[another.id])
response = self.client.patch(
url,
{"username": self.user.username},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("username", response.data["errors"][0]["details"]["fields"])
def test_admin_patch_user_returns_password_validation_error(self):
url = reverse("api_v1:user:admin-user-detail", args=[self.user.id])
response = self.client.patch(
url,
{"password": "123"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("password", response.data["errors"][0]["details"]["fields"])
def test_admin_cannot_patch_self_to_inactive(self):
url = reverse("api_v1:user:admin-user-detail", args=[self.admin.id])