diff --git a/backend/employees/admin.py b/backend/employees/admin.py index 38df75d2..bbd59978 100644 --- a/backend/employees/admin.py +++ b/backend/employees/admin.py @@ -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): diff --git a/backend/tests/employees/test_admin_department_role_actions.py b/backend/tests/employees/test_admin_department_role_actions.py new file mode 100644 index 00000000..c582300f --- /dev/null +++ b/backend/tests/employees/test_admin_department_role_actions.py @@ -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, + ), + ] + )