feat: add admin user management API
All checks were successful
CI/CD Pipeline / Code Quality Checks (push) Successful in 2m59s
CI/CD Pipeline / Run Tests (push) Successful in 5m2s
CI/CD Pipeline / Build and Push Dev Images (push) Successful in 21m9s
CI/CD Pipeline / Deploy Dev via Compose (push) Successful in 54s

This commit is contained in:
2026-08-05 11:24:49 +02:00
parent 2b9ecc2681
commit 2259ee1209
5 changed files with 796 additions and 21 deletions

View File

@@ -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"""