Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(ldap): add live integration coverage and fix sync paths

Browse Source
This commit is contained in:
lizerk 2026-04-12 03:51:52 +03:00
parent f5a908e443
commit b98a82176c
19 changed files with 976 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
backend/api/v1/directory/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Read-only API для directory/LDAP identity данных."""

13
backend/api/v1/directory/serializers.py Normal file
View File

@ -0,0 +1,13 @@
"""Serializers для directory API."""
from rest_framework import serializers
class DirectoryLoginResponseSerializer(serializers.Serializer):
username = serializers.CharField(allow_blank=True, allow_null=True)
source = serializers.ChoiceField(
choices=("db", "ldap", "none", "ldap_not_found")
)
is_cached = serializers.BooleanField()
is_ldap_managed = serializers.BooleanField()

84
backend/api/v1/directory/services.py Normal file
View File

@ -0,0 +1,84 @@
"""Сервисы для directory identity lookup."""
from __future__ import annotations
from dataclasses import dataclass
from django.conf import settings
from employees.models import LdapSyncState
@dataclass(frozen=True)
class DirectoryLoginResult:
username: str | None
source: str
is_cached: bool
is_ldap_managed: bool
def _find_ldap_user(employee):
from employees.ldap.orm_models import LdapUser
employee_number = str(employee.pk)
ldap_user = LdapUser.objects.filter(employee_number=employee_number).first()
if ldap_user:
return ldap_user
sync_state = LdapSyncState.objects.filter(
model="employee", object_pk=employee_number
).first()
if sync_state and sync_state.ldap_dn:
try:
return LdapUser.objects.get(dn=sync_state.ldap_dn)
except LdapUser.DoesNotExist:
return None
return None
def resolve_directory_login(employee, *, force_refresh: bool = False):
cached_username = (employee.username or "").strip()
if cached_username and not force_refresh:
return DirectoryLoginResult(
username=cached_username,
source="db",
is_cached=True,
is_ldap_managed=employee.is_ldap_managed,
)
if not employee.is_ldap_managed:
return DirectoryLoginResult(
username=cached_username or None,
source="db" if cached_username else "none",
is_cached=bool(cached_username),
is_ldap_managed=False,
)
if not getattr(settings, "LDAP_ENABLED", False):
return DirectoryLoginResult(
username=cached_username or None,
source="db" if cached_username else "none",
is_cached=bool(cached_username),
is_ldap_managed=True,
)
ldap_user = _find_ldap_user(employee)
if not ldap_user or not (ldap_user.sam_account_name or "").strip():
return DirectoryLoginResult(
username=None,
source="ldap_not_found",
is_cached=False,
is_ldap_managed=True,
)
username = ldap_user.sam_account_name.strip()
if employee.username != username:
employee.username = username
employee.save(update_fields=["username"])
return DirectoryLoginResult(
username=username,
source="ldap",
is_cached=False,
is_ldap_managed=True,
)

14
backend/api/v1/directory/urls.py Normal file
View File

@ -0,0 +1,14 @@
from django.urls import path
from .views import DirectoryMeLoginAPIView, DirectoryMeLoginRefreshAPIView
app_name = "directory"
urlpatterns = [
path("me/login/", DirectoryMeLoginAPIView.as_view(), name="me-login"),
path(
"me/login/refresh/",
DirectoryMeLoginRefreshAPIView.as_view(),
name="me-login-refresh",
),
]

53
backend/api/v1/directory/views.py Normal file
View File

@ -0,0 +1,53 @@
"""Read-only endpoints для directory identity данных."""
from __future__ import annotations
import logging
from employees.ldap.errors import DirectoryDbError, DirectoryLdapError
from rest_framework import status
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from .serializers import DirectoryLoginResponseSerializer
from .services import resolve_directory_login
logger = logging.getLogger(__name__)
class DirectoryMeLoginAPIView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
try:
result = resolve_directory_login(request.user, force_refresh=False)
except (DirectoryLdapError, DirectoryDbError) as exc:
logger.error("Directory login lookup failed: %s", exc, exc_info=True)
return Response(
{"detail": str(exc), "error": "directory_lookup_failed"},
status=status.HTTP_502_BAD_GATEWAY,
)
serializer = DirectoryLoginResponseSerializer(result)
return Response(serializer.data)
class DirectoryMeLoginRefreshAPIView(APIView):
permission_classes = [IsAuthenticated]
def post(self, request):
try:
result = resolve_directory_login(request.user, force_refresh=True)
except (DirectoryLdapError, DirectoryDbError) as exc:
logger.error(
"Directory login refresh failed: %s", exc, exc_info=True
)
return Response(
{"detail": str(exc), "error": "directory_refresh_failed"},
status=status.HTTP_502_BAD_GATEWAY,
)
serializer = DirectoryLoginResponseSerializer(result)
return Response(serializer.data)

4
backend/api/v1/employees/serializers/employee.py
View File

@ -42,6 +42,7 @@ class EmployeeSerializer(serializers.ModelSerializer):
avatar = Base64ImageField(required=False, allow_null=True)
actions = EmployeeActionSerializer(many=True, read_only=True)
username = serializers.CharField(read_only=True, allow_blank=True)
skills = SkillSerializer(many=True, read_only=True)
skills_ids = serializers.PrimaryKeyRelatedField(
@ -66,6 +67,7 @@ class EmployeeSerializer(serializers.ModelSerializer):
"avatar",
"actions",
"id",
"username",
"email",
"last_name",
"first_name",
@ -200,6 +202,8 @@ class EmployeeSerializer(serializers.ModelSerializer):
action = getattr(view, "action", None) if view else None
if not include_auth and action != "me":
fields.pop("auth", None)
if action != "me":
fields.pop("username", None)
return fields

6
backend/api/v1/employees/views/groups.py
View File

@ -113,7 +113,7 @@ class GroupViewSet(viewsets.ModelViewSet):
ser = self.get_serializer(data=request.data)
ser.is_valid(raise_exception=True)
grp = Group.objects.create(name=ser.validated_data["name"])
grp = Group(name=ser.validated_data["name"])
# Устанавливаем LDAP-специфичные атрибуты для сигнала
grp._ldap_parent_dn = request.data.get("ldap_parent_dn")
@ -124,12 +124,10 @@ class GroupViewSet(viewsets.ModelViewSet):
)
perms = ser.validated_data.get("permissions")
grp.save()
if perms:
grp.permissions.set(perms)
# Сохраняем для триггера сигнала создания
grp.save()
out = self.get_serializer(grp)
return Response(
out.data,

1
backend/api/v1/urls.py
View File

@ -88,6 +88,7 @@ router.register(r"communications/polls", PollViewSet, basename="polls")
urlpatterns = [
path("auth/", include("api.auth.urls")),
path("directory/", include("api.v1.directory.urls")),
# Notifications API (из самого модуля notifications)
path("notifications/", include("notifications.api.urls")),
# Procurement API

2
backend/employees/apps.py
View File

@ -42,10 +42,10 @@ class EmployeesConfig(AppConfig):
name = "employees"
def ready(self):
if getattr(settings, "LDAP_WRITE_ENABLED", False):
_ensure_ldapdb_timezone_compat()
self._patch_ldapdb_compiler()
from . import signals # noqa: F401
def _patch_ldapdb_compiler(self):
"""Патчит django-ldapdb для поддержки Value выражений.

6
backend/employees/ldap/mixins.py
View File

@ -93,7 +93,11 @@ class ModifyDnMixin:
"""
from .infrastructure.connections import _ldap
is_existing = hasattr(self, "_saved_dn") and self._saved_dn
is_existing = (
not getattr(self._state, "adding", True)
and hasattr(self, "_saved_dn")
and self._saved_dn
)
# --- 1) Перемещение между OU (base_dn изменился) ---
current_base_dn = getattr(self, "base_dn", None)

110
backend/employees/ldap/orm_models.py
View File

@ -108,36 +108,41 @@ import datetime as _dt
from django.conf import settings
from django.utils import timezone as _dj_tz
# Compatibility: django-ldapdb 1.5.1 uses timezone.utc removed in Django 5.x
if not hasattr(_dj_tz, "utc"):
class _UtcCompat(_dt.tzinfo):
"""pytz-compatible UTC wrapper for django-ldapdb.
class _UtcCompat(_dt.tzinfo):
"""pytz-compatible UTC wrapper for django-ldapdb.
Acts as a tzinfo subclass but delegates to datetime.timezone.utc
and provides pytz-compatible methods: localize() and normalize().
"""
django-ldapdb still expects timezone.utc to provide pytz-like
`localize()` and `normalize()` methods. Some Django/Python combinations
expose `timezone.utc`, but without these methods. In that case ORM
save/update paths for LDAP models crash deep inside field preparation.
"""
def localize(self, dt):
"""pytz-style localize method."""
return dt.replace(tzinfo=_dt.timezone.utc)
def localize(self, dt):
"""pytz-style localize method."""
return dt.replace(tzinfo=_dt.timezone.utc)
def normalize(self, dt):
"""pytz-style normalize method - same as localize for UTC."""
return dt.replace(tzinfo=_dt.timezone.utc)
def normalize(self, dt):
"""pytz-style normalize method - same as localize for UTC."""
return dt.replace(tzinfo=_dt.timezone.utc)
def utcoffset(self, dt):
return _dt.timedelta(0)
def utcoffset(self, dt):
return _dt.timedelta(0)
def tzname(self, dt):
return "UTC"
def tzname(self, dt):
return "UTC"
def dst(self, dt):
return _dt.timedelta(0)
def dst(self, dt):
return _dt.timedelta(0)
def __repr__(self):
return "UTC"
def __repr__(self):
return "UTC"
# Compatibility: django-ldapdb 1.5.1 expects timezone.utc with pytz methods.
if not hasattr(_dj_tz, "utc") or not all(
hasattr(_dj_tz.utc, attr) for attr in ("localize", "normalize")
):
_dj_tz.utc = _UtcCompat()
from ldapdb.models import Model as LdapModel
@ -444,6 +449,69 @@ class LdapGroup(ModifyDnMixin, LdapModel):
def __repr__(self):
return f"<LdapGroup: {self.cn}>"
# ==================== Membership Methods ==================== #
def list_members(self) -> list[str]:
"""Возвращает список DN участников группы."""
return list(self.member or [])
def add_member(self, member_dn: str) -> None:
"""Добавляет участника в глобальную группу."""
current = self.member or []
if member_dn in current:
return
from ldap3 import MODIFY_ADD
from .infrastructure.connections import _ldap
with _ldap() as conn:
ok = conn.modify(self.dn, {"member": [(MODIFY_ADD, [member_dn])]})
if not ok:
result_str = str(conn.result)
if "attributeOrValueExists" not in result_str:
raise RuntimeError(f"add_member failed: {conn.result}")
def remove_member(self, member_dn: str) -> None:
"""Удаляет участника из глобальной группы."""
from ldap3 import MODIFY_DELETE
from .infrastructure.connections import _ldap
with _ldap() as conn:
ok = conn.modify(
self.dn, {"member": [(MODIFY_DELETE, [member_dn])]}
)
if not ok:
result_str = str(conn.result)
if "noSuchAttribute" not in result_str:
raise RuntimeError(f"remove_member failed: {conn.result}")
def sync_members(self, desired_dns: list[str]) -> dict:
"""Синхронизирует состав группы к точному списку."""
current = set(self.member or [])
desired = set(desired_dns)
to_add = desired - current
to_remove = current - desired
from ldap3 import MODIFY_ADD, MODIFY_DELETE
from .infrastructure.connections import _ldap
with _ldap() as conn:
if to_remove:
conn.modify(
self.dn, {"member": [(MODIFY_DELETE, list(to_remove))]}
)
if to_add:
conn.modify(self.dn, {"member": [(MODIFY_ADD, list(to_add))]})
return {
"added": len(to_add),
"removed": len(to_remove),
}
class LdapOrganizationalUnitGroup(ModifyDnMixin, LdapModel):
"""LDAP модель группы организационной единицы (DEP_*).

69
backend/employees/ldap/services/department_service.py
View File

@ -6,6 +6,7 @@
from __future__ import annotations
import hashlib
from typing import Any, Dict, List, Optional, Tuple
from django.conf import settings
@ -124,9 +125,15 @@ class DepartmentService(BaseService):
if current_parent.lower() == target_base_dn.lower():
return str(ldap_user.dn)
ldap_user.base_dn = target_base_dn
ldap_user.save()
return str(ldap_user.dn)
ldap_user.move_to(target_base_dn)
new_dn = str(ldap_user.dn)
self._touch_state(
model="employee",
object_pk=employee.pk,
ldap_dn=new_dn,
sync_dir=SyncDirection.AUTO,
)
return new_dn
def _load_ou_group(self, group_dn: str) -> LdapOrganizationalUnitGroup:
"""Возвращает high-level LDAP модель группы внутри OU отдела."""
@ -138,20 +145,12 @@ class DepartmentService(BaseService):
def _add_ou_group_member(self, group_dn: str, member_dn: str) -> None:
"""Добавляет DN в группу отдела/роли через ORM модель."""
group = self._load_ou_group(group_dn)
members = list(group.member or [])
if member_dn not in members:
members.append(member_dn)
group.member = members
group.save()
group.add_member(member_dn)
def _remove_ou_group_member(self, group_dn: str, member_dn: str) -> None:
"""Удаляет DN из группы отдела/роли через ORM модель."""
group = self._load_ou_group(group_dn)
members = list(group.member or [])
if member_dn in members:
members.remove(member_dn)
group.member = members
group.save()
group.remove_member(member_dn)
# ==================== Public API ==================== #
@ -756,17 +755,19 @@ class DepartmentService(BaseService):
role_name = self._sanitize_name(name)
cn = f"ROLE_{role_name}"
group_dn = f"CN={esc_rdn(cn)},{dept_dn}"
sam_account_name = self._build_role_sam_account_name(
department, name
)
try:
if not LdapOrganizationalUnitGroup.objects.filter(dn=group_dn).exists():
role_group = LdapOrganizationalUnitGroup(
LdapOrganizationalUnitGroup.objects.create(
dn=group_dn,
cn=cn,
sam_account_name=cn[:20],
sam_account_name=sam_account_name,
description=description
or f"Role: {name} in {department.name}",
)
role_group.save()
except Exception as e:
raise DirectoryLdapError(
f"LDAP create role group failed: {e}"
@ -820,10 +821,14 @@ class DepartmentService(BaseService):
try:
new_role_name = self._sanitize_name(new_name)
new_cn = f"ROLE_{new_role_name}"
new_sam = self._build_role_sam_account_name(
role.department, new_name
)
role_group = LdapOrganizationalUnitGroup.objects.get(
dn=role.ldap_group_dn
)
role_group.cn = new_cn
role_group.sam_account_name = new_sam
role_group.save()
new_dn = str(role_group.dn)
role.ldap_group_dn = new_dn
@ -1137,6 +1142,7 @@ class DepartmentService(BaseService):
role_name = self._sanitize_name(role.name)
expected_cn = f"ROLE_{role_name}"
expected_dn = f"CN={esc_rdn(expected_cn)},{dept_dn}"
expected_sam = self._build_role_sam_account_name(dept, role.name)
saved_dn = (role.ldap_group_dn or "").strip()
role_group: Optional[LdapOrganizationalUnitGroup] = None
@ -1153,13 +1159,12 @@ class DepartmentService(BaseService):
continue
if role_group is None:
role_group = LdapOrganizationalUnitGroup(
role_group = LdapOrganizationalUnitGroup.objects.create(
dn=expected_dn,
cn=expected_cn,
sam_account_name=expected_cn[:20],
sam_account_name=expected_sam,
description=f"Role: {role.name} in {dept.name}",
)
role_group.save()
else:
group_changed = False
_, current_parent = self._split_rdn_parent(str(role_group.dn))
@ -1169,6 +1174,9 @@ class DepartmentService(BaseService):
if role_group.cn != expected_cn:
role_group.cn = expected_cn
group_changed = True
if role_group.sam_account_name != expected_sam:
role_group.sam_account_name = expected_sam
group_changed = True
expected_description = f"Role: {role.name} in {dept.name}"
if (role_group.description or "") != expected_description:
role_group.description = expected_description
@ -1200,9 +1208,11 @@ class DepartmentService(BaseService):
new_role_name = self._sanitize_name(new_name)
new_cn = f"ROLE_{new_role_name}"
new_sam = self._build_role_sam_account_name(role.department, new_name)
role_group = LdapOrganizationalUnitGroup.objects.get(dn=role.ldap_group_dn)
role_group.cn = new_cn
role_group.sam_account_name = new_sam
role_group.save()
new_dn = str(role_group.dn)
@ -1248,6 +1258,21 @@ class DepartmentService(BaseService):
clean = re.sub(r"\s+", "_", clean)
return clean[:50] # Ограничение длины
def _build_role_sam_account_name(
self, department: Department, role_name: str
) -> str:
"""Строит детерминированный и глобально-уникальный sAMAccountName.
CN роли остаётся человекочитаемым (`ROLE_<Name>`), а sAMAccountName
должен быть уникален во всём домене и помещаться в legacy-лимит 20.
"""
sanitized = self._sanitize_name(role_name).upper() or "ROLE"
prefix = sanitized[:6]
digest = hashlib.sha1(
f"{department.pk}:{sanitized}".encode("utf-8")
).hexdigest()[:10].upper()
return f"R{prefix}{digest}"[:20]
# ==================== DN/Lookup Methods ==================== #
def _get_department_dn(self, dept: Department) -> str:
@ -1315,13 +1340,12 @@ class DepartmentService(BaseService):
try:
ou = LdapOrganizationalUnit.objects.get(dn=dn)
except LdapOrganizationalUnit.DoesNotExist:
ou = LdapOrganizationalUnit(
ou = LdapOrganizationalUnit.objects.create(
dn=dn,
ou=name,
description="",
managed_by="",
)
ou.save()
return str(ou.dn)
def _rename_department_ou(self, dept_dn: str, new_name: str) -> str:
@ -1446,13 +1470,12 @@ class DepartmentService(BaseService):
continue
if group is None:
group = LdapOrganizationalUnitGroup(
group = LdapOrganizationalUnitGroup.objects.create(
dn=expected_dn,
cn=expected_cn,
sam_account_name=expected_cn[:20],
description=expected_description,
)
group.save()
else:
group_changed = False
_, current_parent = self._split_rdn_parent(str(group.dn))

51
backend/employees/ldap/services/group_service.py
View File

@ -206,7 +206,7 @@ class GroupService(BaseService):
# ======================== Member Management ======================== #
def add_members(self, group_dn: str, member_dns: List[str]) -> None:
"""Добавляет участников в группу (ORM)."""
"""Добавляет участников в группу через high-level LDAP модель."""
if not member_dns:
return
@ -216,29 +216,26 @@ class GroupService(BaseService):
self._logger.warning(f"Group not found: {group_dn}")
return
members = list(group.member or [])
changed = False
added_count = 0
for dn in member_dns:
if dn not in members:
members.append(dn)
changed = True
before = set(group.list_members())
group.add_member(dn)
after = set(LdapGroup.objects.get(dn=group_dn).list_members())
if dn in after and dn not in before:
added_count += 1
if changed:
group.member = members
group.save()
if added_count:
total = len(LdapGroup.objects.get(dn=group_dn).list_members())
self._log_operation(
"add_members",
model="group",
dn=group_dn,
success=True,
extra={"added": added_count, "total": len(members)},
extra={"added": added_count, "total": total},
)
def remove_members(self, group_dn: str, member_dns: List[str]) -> None:
"""Удаляет участников из группы (ORM)."""
"""Удаляет участников из группы через high-level LDAP модель."""
if not member_dns:
return
@ -248,47 +245,47 @@ class GroupService(BaseService):
self._logger.warning(f"Group not found: {group_dn}")
return
members = list(group.member or [])
changed = False
removed_count = 0
for dn in member_dns:
if dn in members:
members.remove(dn)
changed = True
before = set(group.list_members())
group.remove_member(dn)
after = set(LdapGroup.objects.get(dn=group_dn).list_members())
if dn not in after and dn in before:
removed_count += 1
if changed:
group.member = members
group.save()
if removed_count:
total = len(LdapGroup.objects.get(dn=group_dn).list_members())
self._log_operation(
"remove_members",
model="group",
dn=group_dn,
success=True,
extra={"removed": removed_count, "total": len(members)},
extra={"removed": removed_count, "total": total},
)
def replace_members(
self, group_dn: str, exact_member_dns: List[str]
) -> None:
"""Полная замена состава группы (ORM)."""
"""Полная замена состава группы через high-level LDAP модель."""
try:
group = LdapGroup.objects.get(dn=group_dn)
except LdapGroup.DoesNotExist:
self._logger.warning(f"Group not found: {group_dn}")
return
old_count = len(group.member or [])
group.member = exact_member_dns
group.save()
old_count = len(group.list_members())
diff = group.sync_members(exact_member_dns)
self._log_operation(
"replace_members",
model="group",
dn=group_dn,
success=True,
extra={"old_count": old_count, "new_count": len(exact_member_dns)},
extra={
"old_count": old_count,
"new_count": len(exact_member_dns),
**diff,
},
)
def list_members(self, group_dn: str) -> List[str]:

15
backend/employees/ldap/services/position_service.py
View File

@ -54,6 +54,17 @@ class PositionService(BaseService):
self._group_service = group_service
self._user_service = user_service
def _ensure_runtime_services(self) -> None:
"""Ленивая инициализация зависимостей для sync/retry путей."""
if self._group_service is None:
from .group_service import GroupService
self._group_service = GroupService()
if self._user_service is None:
from .user_service import UserService
self._user_service = UserService(self._group_service)
# ======================== PUBLIC API ======================== #
def reconcile_position(self, pos: Position) -> str:
@ -75,6 +86,7 @@ class PositionService(BaseService):
RuntimeError: Если LDAP_POSITIONS_BASE не настроен
ValueError: Если имя должности пустое
"""
self._ensure_runtime_services()
from ..infrastructure.connections import _ldap
with _ldap() as conn:
@ -93,6 +105,7 @@ class PositionService(BaseService):
DirectoryServiceError: Если DN сотрудника не найден
RuntimeError: Если операция LDAP не удалась
"""
self._ensure_runtime_services()
from ..infrastructure.connections import _ldap
emp_dn = self._user_service._get_employee_dn(employee)
@ -121,6 +134,7 @@ class PositionService(BaseService):
employee: Сотрудник
position: Должность
"""
self._ensure_runtime_services()
from ..infrastructure.connections import _ldap
try:
@ -148,6 +162,7 @@ class PositionService(BaseService):
Args:
position: Должность с POS-группой для удаления
"""
self._ensure_runtime_services()
from ..infrastructure.connections import _ldap
from ..orm_models import LdapGroup

1
backend/employees/signals/__init__.py
View File

@ -19,6 +19,7 @@ if getattr(settings, 'LDAP_ENABLED', False):
import_module('employees.signals.ldap.employee')
import_module('employees.signals.ldap.department')
import_module('employees.signals.ldap.group')
import_module('employees.signals.ldap.position')
__all__ = ['common', 'birthday']
else:
__all__ = ['common', 'birthday']

8
backend/employees/signals/ldap/group.py
View File

@ -69,13 +69,19 @@ def sync_group_to_ldap_on_save(sender, instance, created, **kwargs):
security_enabled = getattr(instance, "_ldap_security_enabled", True)
try:
svc.create(
group_dn = svc.create(
cn=instance.name,
parent_dn=parent_dn,
description=description,
scope=scope,
security_enabled=security_enabled,
)
svc._touch_state(
model="group",
object_pk=instance.pk,
ldap_dn=group_dn,
sync_dir="auto",
)
logger.info(f"Created LDAP group: {instance.name}")
except Exception as e:
logger.error(

150
backend/tests/api/v1/test_directory_login.py Normal file
View File

@ -0,0 +1,150 @@
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from django.contrib.auth import get_user_model
from rest_framework import status
from rest_framework.test import APIClient
User = get_user_model()
def make_user(**overrides):
email = overrides.pop("email", "dir@example.com")
phone = overrides.pop("phone_number", "+79990000077")
user = User.objects.create_user(
email=email,
password="TestPass123!",
first_name=overrides.pop("first_name", "Dir"),
last_name=overrides.pop("last_name", "User"),
phone_number=phone,
send_activation_email=False,
**overrides,
)
user.email_verified = True
user.is_active = True
user.save(update_fields=["email_verified", "is_active"])
return user
@pytest.mark.django_db
def test_employees_me_returns_cached_username():
client = APIClient()
user = make_user(username="cached.login", is_ldap_managed=True)
client.force_authenticate(user=user)
response = client.get("/api/v1/employees/me/")
assert response.status_code == status.HTTP_200_OK
assert response.json()["username"] == "cached.login"
@pytest.mark.django_db
def test_directory_login_returns_cached_username_without_ldap_lookup(settings):
settings.LDAP_ENABLED = True
client = APIClient()
user = make_user(username="cached.login", is_ldap_managed=True)
client.force_authenticate(user=user)
with patch("api.v1.directory.services._find_ldap_user") as find_ldap_user:
response = client.get("/api/v1/directory/me/login/")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"username": "cached.login",
"source": "db",
"is_cached": True,
"is_ldap_managed": True,
}
find_ldap_user.assert_not_called()
@pytest.mark.django_db
def test_directory_login_uses_ldap_fallback_and_caches_result(settings):
settings.LDAP_ENABLED = True
client = APIClient()
user = make_user(username="", is_ldap_managed=True)
client.force_authenticate(user=user)
ldap_user = SimpleNamespace(sam_account_name="ldap.login")
with patch(
"api.v1.directory.services._find_ldap_user", return_value=ldap_user
) as find_ldap_user:
response = client.get("/api/v1/directory/me/login/")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"username": "ldap.login",
"source": "ldap",
"is_cached": False,
"is_ldap_managed": True,
}
find_ldap_user.assert_called_once()
user.refresh_from_db()
assert user.username == "ldap.login"
@pytest.mark.django_db
def test_directory_login_returns_empty_when_ldap_not_found(settings):
settings.LDAP_ENABLED = True
client = APIClient()
user = make_user(username="", is_ldap_managed=True)
client.force_authenticate(user=user)
with patch("api.v1.directory.services._find_ldap_user", return_value=None):
response = client.get("/api/v1/directory/me/login/")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"username": None,
"source": "ldap_not_found",
"is_cached": False,
"is_ldap_managed": True,
}
@pytest.mark.django_db
def test_directory_login_returns_empty_when_ldap_disabled(settings):
settings.LDAP_ENABLED = False
client = APIClient()
user = make_user(username="", is_ldap_managed=True)
client.force_authenticate(user=user)
with patch("api.v1.directory.services._find_ldap_user") as find_ldap_user:
response = client.get("/api/v1/directory/me/login/")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"username": None,
"source": "none",
"is_cached": False,
"is_ldap_managed": True,
}
find_ldap_user.assert_not_called()
@pytest.mark.django_db
def test_directory_login_refresh_forces_ldap_lookup(settings):
settings.LDAP_ENABLED = True
client = APIClient()
user = make_user(username="old.login", is_ldap_managed=True)
client.force_authenticate(user=user)
ldap_user = SimpleNamespace(sam_account_name="new.login")
with patch(
"api.v1.directory.services._find_ldap_user", return_value=ldap_user
) as find_ldap_user:
response = client.post("/api/v1/directory/me/login/refresh/")
assert response.status_code == status.HTTP_200_OK
assert response.json() == {
"username": "new.login",
"source": "ldap",
"is_cached": False,
"is_ldap_managed": True,
}
find_ldap_user.assert_called_once()
user.refresh_from_db()
assert user.username == "new.login"

143
backend/tests/manual/ldap_integration/conftest.py Normal file
View File

@ -0,0 +1,143 @@
from __future__ import annotations
import uuid
from collections.abc import Callable, Generator
import pytest
from django.contrib.auth import get_user_model
from django.test import override_settings
from ldap3 import SUBTREE
from rest_framework.test import APIClient
from employees.ldap import UserService, _ldap
from employees.ldap.domain.dtos import DirectoryUserDTO
from employees.models import LdapSyncState
pytestmark = [
pytest.mark.manual,
pytest.mark.integration,
pytest.mark.django_db(transaction=True, databases=["default", "ldap"]),
]
User = get_user_model()
def _uuid_suffix() -> str:
return uuid.uuid4().hex[:8]
@pytest.fixture
def ldap_runtime(settings) -> None:
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
@pytest.fixture
def ensure_live_ldap(ldap_runtime) -> None:
with _ldap() as conn:
assert conn.bound is True
@pytest.fixture
def unique_name() -> Callable[[str], str]:
def _make(prefix: str) -> str:
return f"{prefix}-{_uuid_suffix()}"
return _make
@pytest.fixture
def ldap_cleanup() -> Generator[Callable[[str], None], None, None]:
tracked_dns: set[str] = set()
def _track(dn: str) -> None:
if dn:
tracked_dns.add(dn)
yield _track
if not tracked_dns:
return
with _ldap() as conn:
for root_dn in sorted(tracked_dns, key=lambda value: value.count(","), reverse=True):
try:
ok = conn.search(
root_dn,
"(objectClass=*)",
search_scope=SUBTREE,
attributes=["distinguishedName"],
)
dns = (
{str(entry.entry_dn) for entry in conn.entries}
if ok and conn.entries
else {root_dn}
)
except Exception:
dns = {root_dn}
for entry_dn in sorted(dns, key=lambda value: value.count(","), reverse=True):
try:
conn.delete(entry_dn)
except Exception:
continue
@pytest.fixture
def superuser_client(user_factory) -> APIClient:
user = user_factory(
email=f"ldap-admin-{_uuid_suffix()}@example.com",
phone_number=f"+79991{uuid.uuid4().int % 10**6:06d}",
staff=True,
superuser=True,
verified=True,
active=True,
)
client = APIClient()
client.force_authenticate(user=user)
return client
@pytest.fixture
def create_ldap_user(
ensure_live_ldap, ldap_cleanup, unique_name
) -> Callable[..., User]:
def _make(
*,
first_name: str = "Ldap",
last_name: str = "User",
email_prefix: str = "ldap-user",
password: str = "TestPass123!",
is_active: bool = True,
) -> User:
suffix = _uuid_suffix()
dto = DirectoryUserDTO(
first_name=first_name,
last_name=f"{last_name}{suffix}",
email=f"{email_prefix}-{suffix}@example.com",
phone_e164=f"+7999{uuid.uuid4().int % 10**7:07d}",
department_dn=None,
group_cns=[],
initial_password=password,
is_active=is_active,
)
employee = UserService().create_user(dto)
employee.email_verified = True
employee.is_active = is_active
employee._skip_ldap_sync = True
employee.save(update_fields=["email_verified", "is_active"])
employee = User.objects.get(pk=employee.pk)
sync_state = LdapSyncState.objects.get(
model="employee", object_pk=str(employee.pk)
)
ldap_cleanup(sync_state.ldap_dn)
return employee
return _make
@pytest.fixture
def disable_ldap_for_cleanup() -> Generator[None, None, None]:
with override_settings(LDAP_ENABLED=False, LDAP_WRITE_ENABLED=False):
yield

323
backend/tests/manual/ldap_integration/test_entities_roundtrip.py Normal file
View File

@ -0,0 +1,323 @@
from __future__ import annotations
import pytest
from django.conf import settings
from django.contrib.auth.models import Group
from rest_framework.test import APIClient
from employees.ldap import (
DepartmentService,
LdapGroup,
LdapOrganizationalUnit,
LdapOrganizationalUnitGroup,
LdapUser,
)
from employees.models import Department, DepartmentRole, LdapSyncState, Position
"""
Живой LDAP integration-suite.
Запускать под обычными settings, а не settings_test:
LDAP_ENABLED=true LDAP_WRITE_ENABLED=true \
.venv/bin/python -m pytest \
-q -o addopts='' --ds=eusrr_backend.settings \
tests/manual/ldap_integration/test_entities_roundtrip.py
"""
pytestmark = [
pytest.mark.manual,
pytest.mark.integration,
pytest.mark.django_db(transaction=True, databases=["default", "ldap"]),
]
def _employee_dn(employee) -> str:
return LdapSyncState.objects.get(
model="employee", object_pk=str(employee.pk)
).ldap_dn
def _department_dn(department: Department) -> str:
return LdapSyncState.objects.get(
model="department", object_pk=str(department.pk)
).ldap_dn
def _norm_dn(value: str | None) -> str:
return (value or "").lower()
def _bootstrap_department_for_role_tests(name: str) -> Department:
return Department.objects.create(
name=name,
description="Bootstrapped for role tests",
)
def test_employee_profile_patch_updates_live_ldap(
ensure_live_ldap, create_ldap_user
):
user = create_ldap_user(first_name="Manual", last_name="Employee")
client = APIClient()
client.force_authenticate(user=user)
response = client.patch(
"/api/v1/employees/me/",
{
"first_name": "Updated",
"phone_number": "+79995554433",
},
format="json",
)
assert response.status_code == 200, response.content
ldap_user = LdapUser.objects.get(dn=_employee_dn(user))
assert ldap_user.given_name == "Updated"
assert ldap_user.telephone_number == "+79995554433"
def test_department_api_roundtrip_syncs_ou_head_and_membership(
ensure_live_ldap,
superuser_client,
create_ldap_user,
ldap_cleanup,
unique_name,
):
head = create_ldap_user(first_name="Head", last_name="Dept")
member = create_ldap_user(first_name="Member", last_name="Dept")
dept_name = unique_name("qa-dept")
create_response = superuser_client.post(
"/api/v1/departments/",
{"name": dept_name, "description": "Initial department"},
format="json",
)
assert create_response.status_code == 201, create_response.content
department = Department.objects.get(pk=create_response.json()["id"])
ldap_cleanup(_department_dn(department))
ou = LdapOrganizationalUnit.objects.get(dn=_department_dn(department))
assert ou.ou == dept_name
assert ou.description == "Initial department"
dept_group = LdapOrganizationalUnitGroup.objects.get(dn=department.ldap_group_dn)
assert dept_group.cn == f"DEP_{dept_name}"
set_head = superuser_client.post(
f"/api/v1/departments/{department.pk}/set_head/",
{"head_id": head.pk},
format="json",
)
assert set_head.status_code == 200, set_head.content
ou = LdapOrganizationalUnit.objects.get(dn=_department_dn(department))
assert _norm_dn(ou.managed_by) == _norm_dn(_employee_dn(head))
add_member = superuser_client.post(
f"/api/v1/departments/{department.pk}/add_member/",
{"employee_id": member.pk},
format="json",
)
assert add_member.status_code == 200, add_member.content
dept_group = LdapOrganizationalUnitGroup.objects.get(dn=department.ldap_group_dn)
member_dns = {_norm_dn(dn) for dn in list(dept_group.member or [])}
assert _norm_dn(_employee_dn(member)) in member_dns
rename_response = superuser_client.patch(
f"/api/v1/departments/{department.pk}/",
{"name": f"{dept_name}-renamed", "description": "Renamed department"},
format="json",
)
assert rename_response.status_code == 200, rename_response.content
department.refresh_from_db()
renamed_dn = _department_dn(department)
ldap_cleanup(renamed_dn)
renamed_ou = LdapOrganizationalUnit.objects.get(dn=renamed_dn)
assert renamed_ou.ou == f"{dept_name}-renamed"
assert renamed_ou.description == "Renamed department"
remove_member = superuser_client.post(
f"/api/v1/departments/{department.pk}/remove_member/",
{"employee_id": member.pk},
format="json",
)
assert remove_member.status_code == 200, remove_member.content
renamed_group = LdapOrganizationalUnitGroup.objects.get(dn=department.ldap_group_dn)
renamed_member_dns = {_norm_dn(dn) for dn in list(renamed_group.member or [])}
assert _norm_dn(_employee_dn(member)) not in renamed_member_dns
def test_group_api_roundtrip_syncs_live_ldap(
ensure_live_ldap,
superuser_client,
create_ldap_user,
ldap_cleanup,
unique_name,
):
member = create_ldap_user(first_name="Member", last_name="Group")
group_name = unique_name("qa-group")
group_dn = f"CN={group_name},{settings.LDAP_GROUPS_BASE}"
ldap_cleanup(group_dn)
create_response = superuser_client.post(
"/api/v1/groups/",
{
"name": group_name,
"ldap_description": "Initial LDAP group",
},
format="json",
)
assert create_response.status_code == 201, create_response.content
ldap_group = LdapGroup.objects.get(dn=group_dn)
assert ldap_group.description == "Initial LDAP group"
add_members = superuser_client.post(
f"/api/v1/groups/{create_response.json()['id']}/add-members/",
{"member_ids": [member.pk]},
format="json",
)
assert add_members.status_code == 200, add_members.content
ldap_group = LdapGroup.objects.get(dn=group_dn)
member_dns = {_norm_dn(dn) for dn in list(ldap_group.member or [])}
assert _norm_dn(_employee_dn(member)) in member_dns
rename_response = superuser_client.post(
f"/api/v1/groups/{create_response.json()['id']}/rename/",
{"new_name": f"{group_name}-renamed"},
format="json",
)
assert rename_response.status_code == 200, rename_response.content
renamed_dn = f"CN={group_name}-renamed,{settings.LDAP_GROUPS_BASE}"
ldap_cleanup(renamed_dn)
renamed_group = LdapGroup.objects.get(dn=renamed_dn)
assert renamed_group.cn == f"{group_name}-renamed"
description_response = superuser_client.post(
f"/api/v1/groups/{create_response.json()['id']}/set-description/",
{"description": "Updated LDAP description"},
format="json",
)
assert description_response.status_code == 200, description_response.content
renamed_group = LdapGroup.objects.get(dn=renamed_dn)
assert renamed_group.description == "Updated LDAP description"
remove_members = superuser_client.post(
f"/api/v1/groups/{create_response.json()['id']}/remove-members/",
{"member_ids": [member.pk]},
format="json",
)
assert remove_members.status_code == 200, remove_members.content
renamed_group = LdapGroup.objects.get(dn=renamed_dn)
renamed_member_dns = {_norm_dn(dn) for dn in list(renamed_group.member or [])}
assert _norm_dn(_employee_dn(member)) not in renamed_member_dns
def test_position_api_roundtrip_syncs_pos_group_and_nesting(
ensure_live_ldap,
superuser_client,
ldap_cleanup,
unique_name,
):
target_group_name = unique_name("qa-target")
target_group_dn = f"CN={target_group_name},{settings.LDAP_GROUPS_BASE}"
ldap_cleanup(target_group_dn)
group_response = superuser_client.post(
"/api/v1/groups/",
{
"name": target_group_name,
"ldap_description": "Target container group",
},
format="json",
)
assert group_response.status_code == 201, group_response.content
target_group = Group.objects.get(pk=group_response.json()["id"])
position_name = unique_name("qa-position")
create_position = superuser_client.post(
"/api/v1/positions/",
{"name": position_name, "description": "Position description"},
format="json",
)
assert create_position.status_code == 201, create_position.content
position = Position.objects.get(pk=create_position.json()["id"])
ldap_cleanup(position.ldap_group_dn)
pos_group = LdapGroup.objects.get(dn=position.ldap_group_dn)
assert pos_group.cn == f"POS_{position_name}"
set_groups = superuser_client.post(
f"/api/v1/positions/{position.pk}/set_groups/",
{"groups": [target_group.pk]},
format="json",
)
assert set_groups.status_code == 200, set_groups.content
target_ldap_group = LdapGroup.objects.get(dn=target_group_dn)
target_member_dns = {_norm_dn(dn) for dn in list(target_ldap_group.member or [])}
assert _norm_dn(position.ldap_group_dn) in target_member_dns
rename_position = superuser_client.patch(
f"/api/v1/positions/{position.pk}/",
{"name": f"{position_name}-renamed"},
format="json",
)
assert rename_position.status_code == 200, rename_position.content
position.refresh_from_db()
ldap_cleanup(position.ldap_group_dn)
renamed_pos_group = LdapGroup.objects.get(dn=position.ldap_group_dn)
assert renamed_pos_group.cn == f"POS_{position_name}-renamed"
def test_department_role_service_roundtrip_syncs_live_ldap(
ensure_live_ldap,
create_ldap_user,
ldap_cleanup,
unique_name,
):
service = DepartmentService()
employee = create_ldap_user(first_name="Role", last_name="Member")
department = _bootstrap_department_for_role_tests(unique_name("qa-role-dept"))
ldap_cleanup(_department_dn(department))
role = service.create_role(
department=department,
name="Operator",
description="Operators",
)
ldap_cleanup(role.ldap_group_dn)
role_group = LdapOrganizationalUnitGroup.objects.get(dn=role.ldap_group_dn)
assert role_group.cn == "ROLE_Operator"
assignment = service.assign_role(employee, role, assigned_by=None)
assert assignment.is_active is True
role_group = LdapOrganizationalUnitGroup.objects.get(dn=role.ldap_group_dn)
role_member_dns = {_norm_dn(dn) for dn in list(role_group.member or [])}
assert _norm_dn(_employee_dn(employee)) in role_member_dns
updated_role = service.update_role(role, {"name": "Lead Operator"})
ldap_cleanup(updated_role.ldap_group_dn)
renamed_group = LdapOrganizationalUnitGroup.objects.get(dn=updated_role.ldap_group_dn)
assert renamed_group.cn == "ROLE_Lead_Operator"
service.revoke_role(employee, updated_role)
assignment.refresh_from_db()
assert assignment.is_active is False
renamed_group = LdapOrganizationalUnitGroup.objects.get(dn=updated_role.ldap_group_dn)
renamed_role_member_dns = {_norm_dn(dn) for dn in list(renamed_group.member or [])}
assert _norm_dn(_employee_dn(employee)) not in renamed_role_member_dns
service.delete_role(updated_role)
assert not DepartmentRole.objects.filter(pk=updated_role.pk).exists()
assert not LdapOrganizationalUnitGroup.objects.filter(dn=updated_role.ldap_group_dn).exists()