feat(admin): add department role ldap sync action

This commit is contained in:
lizerk 2026-04-14 01:49:30 +03:00
parent 21db7f61a2
commit 8d22156b36
2 changed files with 149 additions and 0 deletions

View File

@ -1805,6 +1805,7 @@ class DepartmentRoleAdmin(admin.ModelAdmin):
search_fields = ("name", "department__name", "ldap_group_dn")
autocomplete_fields = ("department",)
filter_horizontal = ("scoped_permissions",)
actions = ["sync_to_ldap"]
fieldsets = (
(None, {"fields": ("department", "name")}),
@ -1818,6 +1819,48 @@ class DepartmentRoleAdmin(admin.ModelAdmin):
),
)
@admin.action(
description=(
"🔄 Пересинхронизировать выбранные роли в LDAP "
"(ROLE_* и вложение в DEP_*)"
)
)
def sync_to_ldap(self, request, queryset):
"""Приводит LDAP-состояние выбранных ролей к текущему состоянию БД."""
from employees.ldap.services.department_service import (
DepartmentService,
)
service = DepartmentService()
success_count = 0
error_count = 0
for role in queryset.select_related("department"):
try:
service.sync_role_state(role)
success_count += 1
except Exception as e:
error_count += 1
self.message_user(
request,
f"Ошибка синхронизации роли "
f"«{role.name}» ({role.department.name}): {e}",
level=messages.ERROR,
)
if success_count:
self.message_user(
request,
f"Успешно пересинхронизировано ролей: {success_count}",
level=messages.SUCCESS,
)
if error_count:
self.message_user(
request,
f"Ошибок синхронизации ролей: {error_count}",
level=messages.WARNING,
)
@admin.register(EmployeeDepartment)
class EmployeeDepartmentAdmin(admin.ModelAdmin):

View File

@ -0,0 +1,106 @@
import hashlib
from unittest.mock import Mock, call, patch
import pytest
from django.contrib import messages
from django.contrib.admin.sites import AdminSite
from django.test import RequestFactory
from employees.admin import DepartmentRoleAdmin
from employees.models import Department, DepartmentRole, Employee
def make_user(email: str, *, staff: bool = True, superuser: bool = True):
suffix = int(hashlib.sha256(email.encode("utf-8")).hexdigest(), 16) % 10**9
return Employee.objects.create_user(
email=email,
password="pwd12345",
phone_number=f"+79{suffix:09d}",
is_staff=staff,
is_superuser=superuser,
send_activation_email=False,
)
@pytest.mark.django_db
def test_department_role_admin_sync_to_ldap_calls_service_for_each_role():
request = RequestFactory().post("/admin/employees/departmentrole/")
request.user = make_user("admin-sync-roles@example.com")
dept = Department.objects.create(name="Dept Admin Sync")
role_a = DepartmentRole.objects.create(department=dept, name="Role A")
role_b = DepartmentRole.objects.create(department=dept, name="Role B")
model_admin = DepartmentRoleAdmin(DepartmentRole, AdminSite())
model_admin.message_user = Mock()
with patch(
"employees.ldap.services.department_service.DepartmentService.sync_role_state"
) as sync_role_state:
model_admin.sync_to_ldap(
request,
DepartmentRole.objects.filter(pk__in=[role_a.pk, role_b.pk]).order_by(
"pk"
),
)
assert sync_role_state.call_args_list == [call(role_a), call(role_b)]
model_admin.message_user.assert_called_once_with(
request,
"Успешно пересинхронизировано ролей: 2",
level=messages.SUCCESS,
)
@pytest.mark.django_db
def test_department_role_admin_sync_to_ldap_reports_errors():
request = RequestFactory().post("/admin/employees/departmentrole/")
request.user = make_user("admin-sync-role-errors@example.com")
dept = Department.objects.create(name="Dept Admin Sync Errors")
ok_role = DepartmentRole.objects.create(department=dept, name="Ok Role")
bad_role = DepartmentRole.objects.create(
department=dept,
name="Bad Role",
)
model_admin = DepartmentRoleAdmin(DepartmentRole, AdminSite())
model_admin.message_user = Mock()
def side_effect(role):
if role.pk == bad_role.pk:
raise RuntimeError("LDAP unavailable")
with patch(
"employees.ldap.services.department_service.DepartmentService.sync_role_state",
side_effect=side_effect,
) as sync_role_state:
model_admin.sync_to_ldap(
request,
DepartmentRole.objects.filter(pk__in=[ok_role.pk, bad_role.pk]).order_by(
"pk"
),
)
assert sync_role_state.call_args_list == [call(ok_role), call(bad_role)]
assert model_admin.message_user.call_count == 3
model_admin.message_user.assert_has_calls(
[
call(
request,
"Ошибка синхронизации роли "
"«Bad Role» (Dept Admin Sync Errors): LDAP unavailable",
level=messages.ERROR,
),
call(
request,
"Успешно пересинхронизировано ролей: 1",
level=messages.SUCCESS,
),
call(
request,
"Ошибок синхронизации ролей: 1",
level=messages.WARNING,
),
]
)