feat: add guest module backend

This commit is contained in:
lizerk 2026-06-20 12:20:52 +03:00
parent c51ca2e247
commit dc35680156
27 changed files with 4018 additions and 11 deletions

View File

@ -0,0 +1 @@

View File

@ -0,0 +1,59 @@
from rest_framework.permissions import BasePermission, SAFE_METHODS
from guests.constants import EDITABLE_BY_AUTHOR_STATUSES
from guests.permissions import (
can_decide_guest_visit,
can_manage_guest_account,
is_guest_admin,
)
class GuestVisitPermission(BasePermission):
def has_permission(self, request, view):
return bool(request.user and request.user.is_authenticated)
def has_object_permission(self, request, view, obj):
user = request.user
action = getattr(view, "action", "")
if is_guest_admin(user):
return True
if request.method in SAFE_METHODS:
return obj.inviter_id == user.id
if action in {"submit", "provide_info", "cancel"}:
return obj.inviter_id == user.id
if action in {"approve", "reject", "request_info", "revoke", "sync_ldap"}:
return False
if action in {"update", "partial_update"}:
return (
obj.inviter_id == user.id
and obj.status in EDITABLE_BY_AUTHOR_STATUSES
)
if action in {"comments", "delete_comment"}:
return obj.inviter_id == user.id
if action in {"attach_document", "remove_document"}:
return obj.inviter_id == user.id
return False
class GuestAdminPermission(BasePermission):
def has_permission(self, request, view):
if not request.user or not request.user.is_authenticated:
return False
action = getattr(view, "action", "")
if action in {"sync_ldap", "deactivate"}:
return can_manage_guest_account(request.user)
return is_guest_admin(request.user)
class GuestDecisionPermission(BasePermission):
def has_permission(self, request, view):
return can_decide_guest_visit(request.user)

View File

@ -0,0 +1,287 @@
from __future__ import annotations
from django.contrib.auth import get_user_model
from rest_framework import serializers
from api.v1.documents.serializers import DocumentReadSerializer
from api.v1.employees.serializers import EmployeeBriefSerializer
from documents.models import Document
from guests.constants import GuestVisitStatus
from guests.models import Guest, GuestVisit, GuestVisitEvent
from guests.permissions import (
can_decide_guest_visit,
can_manage_guest_account,
is_guest_admin,
)
from guests.services import GuestVisitWorkflow, normalize_all_day_range
User = get_user_model()
class GuestBriefSerializer(serializers.ModelSerializer):
full_name = serializers.CharField(read_only=True)
class Meta:
model = Guest
fields = (
"id",
"full_name",
"last_name",
"first_name",
"patronymic",
"organization",
"email",
"phone",
"ldap_enabled",
"is_active",
)
class GuestSerializer(serializers.ModelSerializer):
full_name = serializers.CharField(read_only=True)
created_by = EmployeeBriefSerializer(read_only=True)
class Meta:
model = Guest
fields = (
"id",
"full_name",
"last_name",
"first_name",
"patronymic",
"birth_date",
"phone",
"email",
"organization",
"position",
"comment",
"created_by",
"is_active",
"ldap_enabled",
"ldap_username",
"ldap_upn",
"ldap_last_synced_at",
"ldap_last_error",
"created_at",
"updated_at",
)
read_only_fields = (
"id",
"full_name",
"created_by",
"ldap_enabled",
"ldap_username",
"ldap_upn",
"ldap_last_synced_at",
"ldap_last_error",
"created_at",
"updated_at",
)
class GuestVisitEventSerializer(serializers.ModelSerializer):
actor = EmployeeBriefSerializer(read_only=True)
class Meta:
model = GuestVisitEvent
fields = (
"id",
"event_type",
"actor",
"from_status",
"to_status",
"comment",
"metadata",
"created_at",
)
class GuestVisitReadSerializer(serializers.ModelSerializer):
guest = GuestBriefSerializer(read_only=True)
inviter = EmployeeBriefSerializer(read_only=True)
decided_by = EmployeeBriefSerializer(read_only=True)
documents = DocumentReadSerializer(many=True, read_only=True)
events = GuestVisitEventSerializer(many=True, read_only=True)
status_display = serializers.CharField(source="get_status_display", read_only=True)
is_active_now = serializers.BooleanField(read_only=True)
is_expired = serializers.BooleanField(read_only=True)
can_edit = serializers.SerializerMethodField()
can_submit = serializers.SerializerMethodField()
can_decide = serializers.SerializerMethodField()
can_request_info = serializers.SerializerMethodField()
can_cancel = serializers.SerializerMethodField()
can_revoke = serializers.SerializerMethodField()
can_sync_ldap = serializers.SerializerMethodField()
comments_count = serializers.IntegerField(read_only=True, default=0)
class Meta:
model = GuestVisit
fields = (
"id",
"guest",
"inviter",
"inviter_snapshot_name",
"inviter_snapshot_email",
"host_department",
"purpose",
"visit_comment",
"admin_comment",
"status",
"status_display",
"access_starts_at",
"access_expires_at",
"all_day",
"unlimited",
"documents",
"submitted_at",
"decided_by",
"decided_at",
"decision_comment",
"cancelled_at",
"cancel_reason",
"revoked_at",
"revoke_reason",
"expired_at",
"inviter_inactive",
"created_at",
"updated_at",
"is_active_now",
"is_expired",
"can_edit",
"can_submit",
"can_decide",
"can_request_info",
"can_cancel",
"can_revoke",
"can_sync_ldap",
"comments_count",
"events",
)
def _user(self):
request = self.context.get("request")
return getattr(request, "user", None)
def get_can_edit(self, obj):
return obj.can_edit_by(self._user())
def get_can_submit(self, obj):
user = self._user()
return bool(
user
and obj.inviter_id == user.id
and obj.status in {GuestVisitStatus.DRAFT, GuestVisitStatus.NEEDS_INFO}
)
def get_can_decide(self, obj):
return can_decide_guest_visit(self._user())
def get_can_request_info(self, obj):
return can_decide_guest_visit(self._user())
def get_can_cancel(self, obj):
user = self._user()
return bool(user and (is_guest_admin(user) or obj.inviter_id == user.id))
def get_can_revoke(self, obj):
return can_decide_guest_visit(self._user())
def get_can_sync_ldap(self, obj):
return can_manage_guest_account(self._user())
class GuestVisitWriteSerializer(serializers.Serializer):
guest_id = serializers.IntegerField(required=False)
guest = GuestSerializer(required=False)
purpose = serializers.CharField(required=False, allow_blank=True)
visit_comment = serializers.CharField(required=False, allow_blank=True)
all_day = serializers.BooleanField(required=False, default=True)
unlimited = serializers.BooleanField(required=False, default=False)
access_starts_at = serializers.DateTimeField(required=False, allow_null=True)
access_expires_at = serializers.DateTimeField(required=False, allow_null=True)
date_from = serializers.DateField(required=False)
date_to = serializers.DateField(required=False)
document_ids = serializers.ListField(
child=serializers.IntegerField(min_value=1),
required=False,
allow_empty=True,
)
def validate(self, attrs):
instance = getattr(self, "instance", None)
unlimited = attrs.get("unlimited", getattr(instance, "unlimited", False))
all_day = attrs.get("all_day", getattr(instance, "all_day", True))
if all_day and "date_from" in attrs and "date_to" in attrs:
starts_at, expires_at = normalize_all_day_range(
attrs.pop("date_from"),
attrs.pop("date_to"),
)
attrs["access_starts_at"] = starts_at
attrs["access_expires_at"] = expires_at
starts_at = attrs.get("access_starts_at", getattr(instance, "access_starts_at", None))
expires_at = attrs.get("access_expires_at", getattr(instance, "access_expires_at", None))
if not unlimited:
if starts_at and expires_at and starts_at >= expires_at:
raise serializers.ValidationError(
{"access_expires_at": "Окончание должно быть позже начала."}
)
else:
attrs["access_expires_at"] = None
if not instance and not attrs.get("guest_id") and not attrs.get("guest"):
raise serializers.ValidationError({"guest": "Укажите гостя."})
return attrs
def _get_or_create_guest(self, attrs):
request = self.context["request"]
guest_id = attrs.pop("guest_id", None)
guest_data = attrs.pop("guest", None)
if guest_id:
try:
return Guest.objects.get(pk=guest_id)
except Guest.DoesNotExist as exc:
raise serializers.ValidationError(
{"guest_id": "Гость не найден."}
) from exc
serializer = GuestSerializer(data=guest_data)
serializer.is_valid(raise_exception=True)
return serializer.save(created_by=request.user)
def create(self, validated_data):
document_ids = validated_data.pop("document_ids", [])
guest = self._get_or_create_guest(validated_data)
visit = GuestVisitWorkflow.create_visit(
actor=self.context["request"].user,
guest=guest,
**validated_data,
)
if document_ids:
visit.documents.set(Document.objects.filter(id__in=document_ids))
return visit
def update(self, instance, validated_data):
document_ids = validated_data.pop("document_ids", None)
validated_data.pop("guest", None)
validated_data.pop("guest_id", None)
for field in (
"purpose",
"visit_comment",
"all_day",
"unlimited",
"access_starts_at",
"access_expires_at",
):
if field in validated_data:
setattr(instance, field, validated_data[field])
instance.save()
if document_ids is not None:
instance.documents.set(Document.objects.filter(id__in=document_ids))
return instance
class GuestSearchSerializer(GuestBriefSerializer):
pass

View File

@ -0,0 +1,14 @@
from django.urls import include, path
from rest_framework.routers import DefaultRouter
from .views import GuestViewSet, GuestVisitViewSet
app_name = "guests"
router = DefaultRouter()
router.register(r"visits", GuestVisitViewSet, basename="guest-visits")
router.register(r"", GuestViewSet, basename="guests")
urlpatterns = [
path("", include(router.urls)),
]

View File

@ -0,0 +1,403 @@
from __future__ import annotations
from django.contrib.contenttypes.models import ContentType
from django.db.models import Count, IntegerField, OuterRef, Q, Subquery, Value
from django.db.models.functions import Coalesce
from django.shortcuts import get_object_or_404
from rest_framework import mixins, status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied, ValidationError
from rest_framework.response import Response
from communications import comments_helpers
from communications.models import Message
from documents.models import Document
from guests.constants import GuestVisitEventType, GuestVisitStatus
from guests.models import Guest, GuestVisit
from guests.permissions import (
can_decide_guest_visit,
can_manage_guest_account,
is_guest_admin,
)
from guests.services import (
GuestLdapService,
GuestVisitWorkflow,
record_guest_event,
)
from .permissions import GuestAdminPermission, GuestVisitPermission
from .serializers import (
GuestSerializer,
GuestSearchSerializer,
GuestVisitReadSerializer,
GuestVisitWriteSerializer,
)
from ..employees.serializers import EmployeeBriefSerializer
class GuestVisitViewSet(
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.CreateModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
queryset = (
GuestVisit.objects.select_related(
"guest",
"inviter",
"decided_by",
"host_department",
)
.prefetch_related("documents", "events__actor")
.order_by("-created_at")
)
permission_classes = [GuestVisitPermission]
def get_serializer_class(self):
if self.action in {"list", "retrieve"}:
return GuestVisitReadSerializer
return GuestVisitWriteSerializer
def get_queryset(self):
qs = super().get_queryset()
user = self.request.user
params = self.request.query_params
request_ct = ContentType.objects.get_for_model(GuestVisit)
comments_subquery = (
Message.objects.filter(
chat__type="comments",
chat__context_content_type=request_ct,
chat__context_object_id=OuterRef("pk"),
is_deleted=False,
)
.values("chat")
.annotate(count=Count("id"))
.values("count")
)
qs = qs.annotate(
comments_count=Coalesce(
Subquery(comments_subquery, output_field=IntegerField()),
Value(0),
)
)
scope = params.get("scope")
if not is_guest_admin(user):
qs = qs.filter(inviter=user)
elif scope == "mine":
qs = qs.filter(inviter=user)
elif scope == "pending_decision":
qs = qs.filter(status=GuestVisitStatus.PENDING)
elif scope == "active":
qs = qs.filter(status=GuestVisitStatus.APPROVED)
elif scope == "expired":
qs = qs.filter(status=GuestVisitStatus.EXPIRED)
elif scope == "risk":
qs = qs.filter(inviter_inactive=True)
if status_value := params.get("status"):
qs = qs.filter(status=status_value)
if guest_id := params.get("guest_id"):
qs = qs.filter(guest_id=guest_id)
if inviter_id := params.get("inviter_id"):
qs = qs.filter(inviter_id=inviter_id)
if created_from := params.get("created_from"):
qs = qs.filter(created_at__date__gte=created_from)
if created_to := params.get("created_to"):
qs = qs.filter(created_at__date__lte=created_to)
if access_from := params.get("access_from"):
qs = qs.filter(
Q(access_expires_at__isnull=True)
| Q(access_expires_at__gte=access_from)
)
if access_to := params.get("access_to"):
qs = qs.filter(
Q(access_starts_at__isnull=True)
| Q(access_starts_at__lte=access_to)
)
if params.get("unlimited") in {"true", "false"}:
qs = qs.filter(unlimited=params.get("unlimited") == "true")
if params.get("ldap_enabled") in {"true", "false"}:
qs = qs.filter(guest__ldap_enabled=params.get("ldap_enabled") == "true")
if q := (params.get("q") or "").strip():
qs = qs.filter(
Q(guest__last_name__icontains=q)
| Q(guest__first_name__icontains=q)
| Q(guest__organization__icontains=q)
| Q(guest__email__icontains=q)
| Q(guest__phone__icontains=q)
| Q(purpose__icontains=q)
)
ordering = params.get("ordering")
allowed_ordering = {
"created_at",
"-created_at",
"access_starts_at",
"-access_starts_at",
"access_expires_at",
"-access_expires_at",
}
if ordering in allowed_ordering:
qs = qs.order_by(ordering)
return qs.distinct()
def _read_response(self, visit, response_status=status.HTTP_200_OK):
visit.refresh_from_db()
serializer = GuestVisitReadSerializer(
visit,
context=self.get_serializer_context(),
)
return Response(serializer.data, status=response_status)
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
visit = serializer.save()
return self._read_response(visit, status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
partial = kwargs.pop("partial", False)
instance = self.get_object()
if not instance.can_edit_by(request.user):
raise PermissionDenied("Недостаточно прав для изменения визита.")
serializer = self.get_serializer(
instance,
data=request.data,
partial=partial,
)
serializer.is_valid(raise_exception=True)
visit = serializer.save()
return self._read_response(visit)
@action(detail=True, methods=["post"])
def submit(self, request, pk=None):
visit = self.get_object()
try:
GuestVisitWorkflow.submit(visit, actor=request.user)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"])
def approve(self, request, pk=None):
if not can_decide_guest_visit(request.user):
raise PermissionDenied("Недостаточно прав для решения.")
visit = self.get_object()
try:
GuestVisitWorkflow.approve(
visit,
actor=request.user,
comment=request.data.get("comment", ""),
)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"])
def reject(self, request, pk=None):
if not can_decide_guest_visit(request.user):
raise PermissionDenied("Недостаточно прав для решения.")
visit = self.get_object()
try:
GuestVisitWorkflow.reject(
visit,
actor=request.user,
comment=request.data.get("comment", ""),
)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"], url_path="request-info")
def request_info(self, request, pk=None):
if not can_decide_guest_visit(request.user):
raise PermissionDenied("Недостаточно прав для решения.")
visit = self.get_object()
try:
GuestVisitWorkflow.request_info(
visit,
actor=request.user,
comment=request.data.get("comment", ""),
)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"], url_path="provide-info")
def provide_info(self, request, pk=None):
visit = self.get_object()
comment = (request.data.get("comment") or "").strip()
if not comment:
raise ValidationError("Комментарий обязателен.")
comments_helpers.create_comment(visit, request.user, comment)
try:
GuestVisitWorkflow.submit(visit, actor=request.user)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"])
def cancel(self, request, pk=None):
visit = self.get_object()
if visit.inviter_id != request.user.id and not is_guest_admin(request.user):
raise PermissionDenied("Недостаточно прав для отмены.")
try:
GuestVisitWorkflow.cancel(
visit,
actor=request.user,
comment=request.data.get("comment", ""),
)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"])
def revoke(self, request, pk=None):
if not can_decide_guest_visit(request.user):
raise PermissionDenied("Недостаточно прав для отзыва.")
visit = self.get_object()
try:
GuestVisitWorkflow.revoke(
visit,
actor=request.user,
comment=request.data.get("comment", ""),
)
except ValueError as exc:
raise ValidationError(str(exc)) from exc
return self._read_response(visit)
@action(detail=True, methods=["post"], url_path="sync-ldap")
def sync_ldap(self, request, pk=None):
if not can_manage_guest_account(request.user):
raise PermissionDenied("Недостаточно прав для LDAP.")
visit = self.get_object()
GuestLdapService().sync_guest_for_visit(visit)
return self._read_response(visit)
@action(detail=True, methods=["get", "post"])
def comments(self, request, pk=None):
visit = self.get_object()
if request.method == "GET":
messages = comments_helpers.get_comments(visit)
return Response(
[
{
"id": msg.id,
"author": EmployeeBriefSerializer(msg.author).data,
"text": msg.content,
"created_at": msg.created_at,
}
for msg in messages
]
)
text = (request.data.get("text") or "").strip()
if not text:
raise ValidationError({"text": "Это поле не может быть пустым."})
msg = comments_helpers.create_comment(visit, request.user, text)
return Response(
{
"id": msg.id,
"author": EmployeeBriefSerializer(msg.author).data,
"text": msg.content,
"created_at": msg.created_at,
},
status=status.HTTP_201_CREATED,
)
@action(
detail=True,
methods=["delete"],
url_path="comments/(?P<comment_id>[^/.]+)",
)
def delete_comment(self, request, pk=None, comment_id=None):
visit = self.get_object()
chat = comments_helpers.get_or_create_comments_chat(visit)
message = get_object_or_404(Message, id=comment_id, chat=chat)
if message.author_id != request.user.id and not is_guest_admin(request.user):
raise PermissionDenied("Недостаточно прав для удаления комментария.")
comments_helpers.delete_comment(message, request.user, soft_delete=True)
return Response(status=status.HTTP_204_NO_CONTENT)
@action(detail=True, methods=["post"], url_path="documents")
def attach_document(self, request, pk=None):
visit = self.get_object()
document_id = request.data.get("document_id")
document = get_object_or_404(Document, pk=document_id)
visit.documents.add(document)
record_guest_event(
visit,
GuestVisitEventType.DOCUMENT_ATTACHED,
actor=request.user,
metadata={"document_id": document.id},
)
return self._read_response(visit)
@action(
detail=True,
methods=["delete"],
url_path="documents/(?P<document_id>[^/.]+)",
)
def remove_document(self, request, pk=None, document_id=None):
visit = self.get_object()
document = get_object_or_404(Document, pk=document_id)
visit.documents.remove(document)
record_guest_event(
visit,
GuestVisitEventType.DOCUMENT_REMOVED,
actor=request.user,
metadata={"document_id": document.id},
)
return self._read_response(visit)
class GuestViewSet(
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
viewsets.GenericViewSet,
):
queryset = Guest.objects.all().order_by("last_name", "first_name", "id")
serializer_class = GuestSerializer
permission_classes = [GuestAdminPermission]
def get_queryset(self):
qs = super().get_queryset()
q = (self.request.query_params.get("q") or "").strip()
if q:
text_q = (
Q(last_name__icontains=q)
| Q(first_name__icontains=q)
| Q(patronymic__icontains=q)
| Q(email__icontains=q)
| Q(phone__icontains=q)
| Q(organization__icontains=q)
)
if q.isdigit():
text_q |= Q(id=int(q))
qs = qs.filter(text_q)
return qs
@action(detail=False, methods=["get"])
def search(self, request):
serializer = GuestSearchSerializer(
self.get_queryset()[:20],
many=True,
context=self.get_serializer_context(),
)
return Response(serializer.data)
@action(detail=True, methods=["post"])
def deactivate(self, request, pk=None):
guest = self.get_object()
guest.is_active = False
guest.save(update_fields=["is_active", "updated_at"])
GuestLdapService().disable_guest(guest)
return Response(self.get_serializer(guest).data)
@action(detail=True, methods=["post"], url_path="sync-ldap")
def sync_ldap(self, request, pk=None):
guest = self.get_object()
GuestLdapService().sync_guest(guest)
return Response(self.get_serializer(guest).data)

View File

@ -179,6 +179,8 @@ urlpatterns = [
path("notifications/", include("notifications.api.urls")),
# Procurement API
path("procurement/", include("api.v1.procurement.urls")),
# Guests API
path("guests/", include("api.v1.guests.urls")),
# Search API
path("search/", search_api_view, name="search"),
# Router URLs (включая все ViewSets)

View File

@ -0,0 +1,49 @@
from django.db import migrations, models
GUEST_ID_MIN = 900_000_000_000_000
EMPLOYEE_ID_MAX = GUEST_ID_MIN - 1
def cap_employee_sequence(apps, schema_editor):
if schema_editor.connection.vendor != "postgresql":
return
with schema_editor.connection.cursor() as cursor:
cursor.execute(
"SELECT pg_get_serial_sequence('employees_employee', 'id')"
)
row = cursor.fetchone()
sequence_name = row[0] if row else None
if sequence_name:
cursor.execute(f"ALTER SEQUENCE {sequence_name} MAXVALUE {EMPLOYEE_ID_MAX}")
def uncap_employee_sequence(apps, schema_editor):
if schema_editor.connection.vendor != "postgresql":
return
with schema_editor.connection.cursor() as cursor:
cursor.execute(
"SELECT pg_get_serial_sequence('employees_employee', 'id')"
)
row = cursor.fetchone()
sequence_name = row[0] if row else None
if sequence_name:
cursor.execute(f"ALTER SEQUENCE {sequence_name} NO MAXVALUE")
class Migration(migrations.Migration):
dependencies = [
("employees", "0053_cleanup_department_role_permissions"),
]
operations = [
migrations.AddConstraint(
model_name="employee",
constraint=models.CheckConstraint(
condition=models.Q(("id__lt", GUEST_ID_MIN)),
name="employee_id_below_guest_range",
),
),
migrations.RunPython(cap_employee_sequence, uncap_employee_sequence),
]

View File

@ -16,6 +16,8 @@ from .constants import (
DeptPerm,
)
GUEST_ID_MIN = 900_000_000_000_000
def _sync_department_calendar_after_commit(department_id: int) -> None:
from scheduling.services import (
@ -276,6 +278,12 @@ class Employee(AbstractUser):
"Может просматривать LDAP информацию сотрудников",
),
]
constraints = [
models.CheckConstraint(
name="employee_id_below_guest_range",
condition=models.Q(id__lt=GUEST_ID_MIN),
),
]
# Контактные поля telegram/whatsapp/wechat опциональны
# Валидация "хотя бы одно поле" удалена

View File

@ -325,6 +325,14 @@ def _execute_role_assignment(payload: dict) -> None:
)
def _execute_guest_operation(payload: dict) -> None:
"""Повторяет синхронизацию гостевой LDAP-учетки."""
from guests.tasks import execute_guest_queue_operation
operation = payload.get("_operation") or payload.get("operation") or "guest_sync"
execute_guest_queue_operation(operation, payload)
# Реестр: operation → callable
_EXECUTORS = {
"employee_save": _execute_employee_save,
@ -340,6 +348,9 @@ _EXECUTORS = {
"department_role_save": _execute_department_role_save,
"department_role_delete": _execute_department_role_delete,
"role_assignment": _execute_role_assignment,
"guest_sync": _execute_guest_operation,
"guest_disable": _execute_guest_operation,
"guest_delete": _execute_guest_operation,
}

View File

@ -68,6 +68,7 @@ INSTALLED_APPS = [
"search.apps.SearchConfig",
"finance.apps.FinanceConfig",
"procurement.apps.ProcurementConfig",
"guests.apps.GuestsConfig",
"push_notifications", # django-push-notifications для Web Push
]
@ -606,6 +607,31 @@ LDAP_GROUPS_BASE = os.getenv("LDAP_GROUPS_BASE", "OU=Groups,DC=eusrr,DC=local")
LDAP_POSITIONS_BASE = os.getenv(
"LDAP_POSITIONS_BASE", "OU=Positions,DC=eusrr,DC=local"
)
LDAP_GUESTS_BASE = os.getenv(
"LDAP_GUESTS_BASE", "OU=Guests,DC=eusrr,DC=local"
)
LDAP_GUESTS_ACTIVE_BASE = os.getenv(
"LDAP_GUESTS_ACTIVE_BASE", f"OU=Active,{LDAP_GUESTS_BASE}"
)
LDAP_GUESTS_DEACTIVATED_BASE = os.getenv(
"LDAP_GUESTS_DEACTIVATED_BASE", f"OU=Deactivated,{LDAP_GUESTS_BASE}"
)
GUESTS_ID_MIN = int(os.getenv("GUESTS_ID_MIN", "900000000000000"))
GUESTS_ID_START = int(os.getenv("GUESTS_ID_START", "900000000000001"))
GUESTS_ID_MAX = int(os.getenv("GUESTS_ID_MAX", "999999999999999"))
GUESTS_ID_PREFIX = os.getenv("GUESTS_ID_PREFIX", "9")
GUESTS_ID_LENGTH = int(os.getenv("GUESTS_ID_LENGTH", "15"))
GUESTS_REQUIRE_ID_DOCUMENT = (
os.getenv("GUESTS_REQUIRE_ID_DOCUMENT", "false").lower() == "true"
)
GUESTS_NOTIFY_ON_EXPIRATION = (
os.getenv("GUESTS_NOTIFY_ON_EXPIRATION", "true").lower() == "true"
)
GUESTS_AUTO_DISABLE_ON_INVITER_INACTIVE = (
os.getenv("GUESTS_AUTO_DISABLE_ON_INVITER_INACTIVE", "false").lower()
== "true"
)
BRAND_NAME = os.getenv("BRAND_NAME", "CORP - HiRo")
BRAND_LOGO = "img/logo.png"
@ -689,6 +715,18 @@ CELERY_BEAT_SCHEDULE = {
"task": "attendance.tasks.dispatch_attendance_auto_sync",
"schedule": 60.0,
},
"activate-due-guest-visits": {
"task": "guests.tasks.activate_due_guest_visits",
"schedule": 60.0,
},
"expire-guest-visits": {
"task": "guests.tasks.expire_guest_visits",
"schedule": 60.0,
},
"detect-inactive-guest-inviters": {
"task": "guests.tasks.detect_inactive_inviters",
"schedule": 300.0,
},
}
# Опции по умолчанию для всех задач

View File

@ -0,0 +1 @@

101
backend/guests/admin.py Normal file
View File

@ -0,0 +1,101 @@
from django.contrib import admin
from .models import Guest, GuestVisit, GuestVisitEvent
@admin.register(Guest)
class GuestAdmin(admin.ModelAdmin):
list_display = (
"id",
"full_name",
"organization",
"email",
"phone",
"ldap_enabled",
"is_active",
)
list_filter = ("ldap_enabled", "is_active", "organization")
search_fields = (
"id",
"last_name",
"first_name",
"patronymic",
"email",
"phone",
"organization",
)
readonly_fields = (
"id",
"ldap_last_synced_at",
"ldap_last_error",
"created_at",
"updated_at",
)
class GuestVisitEventInline(admin.TabularInline):
model = GuestVisitEvent
extra = 0
readonly_fields = (
"actor",
"event_type",
"from_status",
"to_status",
"comment",
"metadata",
"created_at",
)
can_delete = False
@admin.register(GuestVisit)
class GuestVisitAdmin(admin.ModelAdmin):
list_display = (
"id",
"guest",
"inviter",
"status",
"access_starts_at",
"access_expires_at",
"unlimited",
"inviter_inactive",
)
list_filter = ("status", "unlimited", "all_day", "inviter_inactive")
search_fields = (
"guest__id",
"guest__last_name",
"guest__first_name",
"guest__organization",
"inviter__email",
"purpose",
)
readonly_fields = (
"created_at",
"updated_at",
"submitted_at",
"decided_at",
"cancelled_at",
"revoked_at",
"expired_at",
"inviter_snapshot_name",
"inviter_snapshot_email",
)
inlines = [GuestVisitEventInline]
@admin.register(GuestVisitEvent)
class GuestVisitEventAdmin(admin.ModelAdmin):
list_display = ("id", "visit", "event_type", "actor", "created_at")
list_filter = ("event_type", "created_at")
search_fields = ("visit__guest__last_name", "visit__guest__id", "comment")
readonly_fields = (
"visit",
"actor",
"event_type",
"from_status",
"to_status",
"comment",
"metadata",
"created_at",
)

8
backend/guests/apps.py Normal file
View File

@ -0,0 +1,8 @@
from django.apps import AppConfig
class GuestsConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "guests"
verbose_name = "Гости"

View File

@ -0,0 +1,53 @@
from django.db import models
GUEST_ID_MIN = 900_000_000_000_000
GUEST_ID_START = 900_000_000_000_001
GUEST_ID_MAX = 999_999_999_999_999
class GuestVisitStatus(models.TextChoices):
DRAFT = "draft", "Черновик"
PENDING = "pending", "На рассмотрении"
NEEDS_INFO = "needs_info", "Требуется информация"
APPROVED = "approved", "Одобрено"
REJECTED = "rejected", "Отклонено"
CANCELLED = "cancelled", "Отменено"
EXPIRED = "expired", "Истекло"
REVOKED = "revoked", "Отозвано"
class GuestVisitEventType(models.TextChoices):
CREATED = "created", "Создано"
SUBMITTED = "submitted", "Отправлено"
NEEDS_INFO_REQUESTED = "needs_info_requested", "Запрошена информация"
INFO_PROVIDED = "info_provided", "Информация предоставлена"
APPROVED = "approved", "Одобрено"
REJECTED = "rejected", "Отклонено"
DECISION_CHANGED = "decision_changed", "Решение изменено"
CANCELLED = "cancelled", "Отменено"
REVOKED = "revoked", "Отозвано"
EXPIRED = "expired", "Истекло"
LDAP_CREATED = "ldap_created", "LDAP создан"
LDAP_UPDATED = "ldap_updated", "LDAP обновлен"
LDAP_ENABLED = "ldap_enabled", "LDAP перемещен в Active OU"
LDAP_DISABLED = "ldap_disabled", "LDAP перемещен в Deactivated OU"
LDAP_FAILED = "ldap_failed", "Ошибка LDAP"
LDAP_SKIPPED = "ldap_skipped", "LDAP пропущен"
DOCUMENT_ATTACHED = "document_attached", "Документ прикреплен"
DOCUMENT_REMOVED = "document_removed", "Документ удален"
INVITER_INACTIVE_DETECTED = (
"inviter_inactive_detected",
"Приглашающий неактивен",
)
ACTIVE_VISIT_STATUSES = {
GuestVisitStatus.APPROVED,
}
EDITABLE_BY_AUTHOR_STATUSES = {
GuestVisitStatus.DRAFT,
GuestVisitStatus.NEEDS_INFO,
}

View File

@ -0,0 +1,3 @@
from .orm_models import LdapGuestUser
__all__ = ["LdapGuestUser"]

View File

@ -0,0 +1,250 @@
"""LDAP ORM models for guest Active Directory accounts.
Guest LDAP accounts are separate from employees:
- LdapSyncState.model is always ``guest``.
- employeeNumber stores ``Guest.id`` as a string.
- userAccountControl always remains disabled; access state is represented by
OU placement under LDAP_GUESTS_ACTIVE_BASE/LDAP_GUESTS_DEACTIVATED_BASE.
"""
import datetime as _dt
import secrets
import string
from dataclasses import dataclass
from django.conf import settings
from django.utils import timezone as _dj_tz
from ldap3 import SUBTREE
from ldapdb.models import Model as LdapModel
from ldapdb.models.fields import CharField, DateTimeField, IntegerField, ListField
from employees.ldap.infrastructure.connections import _ldap
from employees.ldap.mixins import LdapSyncStateMixin, ModifyDnMixin
from employees.ldap.repositories.ldap_repository import ensure_container_exists
from employees.ldap.services.constants import UserAccountControl
from employees.ldap.services.user_password_service import UserPasswordService
from employees.ldap.utils.ldap_utils import cn_candidates
from employees.ldap.utils.text_utils import esc_filter, esc_rdn
class _UtcCompat(_dt.tzinfo):
def localize(self, dt):
return dt.replace(tzinfo=_dt.timezone.utc)
def normalize(self, dt):
return dt.replace(tzinfo=_dt.timezone.utc)
def utcoffset(self, dt):
return _dt.timedelta(0)
def tzname(self, dt):
return "UTC"
def dst(self, dt):
return _dt.timedelta(0)
def __repr__(self):
return "UTC"
if not hasattr(_dj_tz, "utc") or not all(
hasattr(_dj_tz.utc, attr) for attr in ("localize", "normalize")
):
_dj_tz.utc = _UtcCompat()
def get_guests_base():
return getattr(settings, "LDAP_GUESTS_BASE", "OU=Guests,DC=eusrr,DC=local")
@dataclass(frozen=True)
class LdapGuestSyncResult:
dn: str
sam_account_name: str
user_principal_name: str
user_account_control: int
class LdapGuestUser(LdapSyncStateMixin, ModifyDnMixin, LdapModel):
"""LDAP ORM model for guest AD users.
Creation still uses ldap3 internally because django-ldapdb cannot create a
user with dynamic DN and CN collision handling. The low-level details are
kept behind class methods so GuestLdapService can stay workflow-focused.
"""
_sync_model_name = "guest"
_sync_pk_field = "employee_number"
base_dn = get_guests_base()
object_classes = ["top", "person", "organizationalPerson", "user"]
rdn_attributes = ["cn"]
cn = CharField(db_column="cn")
sam_account_name = CharField(db_column="sAMAccountName")
user_principal_name = CharField(db_column="userPrincipalName")
given_name = CharField(db_column="givenName", blank=True)
sn = CharField(db_column="sn")
display_name = CharField(db_column="displayName")
mail = CharField(db_column="mail", blank=True)
telephone_number = CharField(db_column="telephoneNumber", blank=True)
employee_number = CharField(db_column="employeeNumber", blank=True)
description = CharField(db_column="description", blank=True)
user_account_control = IntegerField(db_column="userAccountControl")
member_of = ListField(db_column="memberOf", blank=True)
when_created = DateTimeField(db_column="whenCreated")
when_changed = DateTimeField(db_column="whenChanged")
class Meta:
managed = False
def __str__(self):
return f"{self.display_name} ({self.sam_account_name})"
@classmethod
def guest_bases(cls) -> tuple[str, str]:
base = getattr(settings, "LDAP_GUESTS_BASE", "")
active = getattr(settings, "LDAP_GUESTS_ACTIVE_BASE", "")
deactivated = getattr(settings, "LDAP_GUESTS_DEACTIVATED_BASE", "")
if not active and base:
active = f"OU=Active,{base}"
if not deactivated and base:
deactivated = f"OU=Deactivated,{base}"
return active, deactivated
@classmethod
def target_base(cls, in_active_ou: bool) -> str:
active_base, deactivated_base = cls.guest_bases()
target_base = active_base if in_active_ou else deactivated_base
if not target_base:
raise RuntimeError("Целевой LDAP контейнер гостей не задан.")
return target_base
@staticmethod
def sam_for_guest(guest) -> str:
return f"g{guest.id}"
@staticmethod
def upn_for_guest(guest) -> str:
suffix = getattr(settings, "LDAP_UPN_SUFFIX", "") or "guest.local"
return f"g{guest.id}@{suffix}"
@staticmethod
def display_name_for_guest(guest) -> str:
return guest.full_name or f"Guest {guest.id}"
@staticmethod
def description_for_guest(guest) -> str:
return f"Guest account for {guest.organization}".strip()
@staticmethod
def random_guest_password() -> str:
alphabet = string.ascii_letters + string.digits + "!@#$%^&*()-_=+"
return "".join(secrets.choice(alphabet) for _ in range(24))
@classmethod
def _guest_attrs(cls, guest) -> dict:
display_name = cls.display_name_for_guest(guest)
attrs = {
"cn": display_name,
"sAMAccountName": cls.sam_for_guest(guest),
"userPrincipalName": cls.upn_for_guest(guest),
"userAccountControl": UserAccountControl.DISABLED,
"givenName": guest.first_name or None,
"sn": guest.last_name or ".",
"displayName": display_name,
"mail": guest.email or None,
"telephoneNumber": guest.phone or None,
"employeeNumber": str(guest.id),
"description": cls.description_for_guest(guest),
}
return {k: v for k, v in attrs.items() if v not in (None, "", [])}
@classmethod
def _sync_result(cls, dn: str, guest) -> LdapGuestSyncResult:
return LdapGuestSyncResult(
dn=str(dn),
sam_account_name=cls.sam_for_guest(guest),
user_principal_name=cls.upn_for_guest(guest),
user_account_control=int(UserAccountControl.DISABLED),
)
@classmethod
def _ensure_unique_employee_number(cls, conn, guest, current_dn: str = "") -> None:
base = getattr(settings, "LDAP_BASE_DN", "") or getattr(
settings,
"LDAP_GUESTS_BASE",
"",
)
if not base:
return
conn.search(
search_base=base,
search_filter=f"(employeeNumber={esc_filter(str(guest.id))})",
search_scope=SUBTREE,
attributes=["distinguishedName"],
size_limit=2,
)
for entry in conn.entries:
dn = str(getattr(entry, "entry_dn", ""))
if dn and dn.lower() != (current_dn or "").lower():
raise RuntimeError(
f"employeeNumber {guest.id} already exists in LDAP: {dn}"
)
@classmethod
def create_for_guest(cls, guest, in_active_ou: bool) -> LdapGuestSyncResult:
base_dn = cls.target_base(in_active_ou)
with _ldap() as conn:
ensure_container_exists(conn, base_dn)
cls._ensure_unique_employee_number(conn, guest)
attrs = cls._guest_attrs(guest)
display_name = cls.display_name_for_guest(guest)
safe_cn = cls.sam_for_guest(guest)
dn = None
for cn_text in cn_candidates(display_name, safe_cn):
dn_try = f"CN={esc_rdn(cn_text)},{base_dn}"
if conn.add(dn_try, cls.object_classes, attrs):
dn = dn_try
break
if not dn:
raise RuntimeError(f"LDAP add guest failed: {conn.result}")
UserPasswordService().set_password(
conn,
dn,
cls.random_guest_password(),
)
return cls._sync_result(dn, guest)
@classmethod
def sync_existing_for_guest(
cls,
guest,
dn: str,
in_active_ou: bool,
) -> LdapGuestSyncResult:
target_base = cls.target_base(in_active_ou)
with _ldap() as conn:
ensure_container_exists(conn, target_base)
cls._ensure_unique_employee_number(conn, guest, dn)
ldap_guest = cls.objects.get(dn=dn)
ldap_guest.move_to(target_base)
ldap_guest.apply_guest(guest)
ldap_guest.save()
return cls._sync_result(str(ldap_guest.dn), guest)
def apply_guest(self, guest) -> None:
self.sam_account_name = self.sam_for_guest(guest)
self.user_principal_name = self.upn_for_guest(guest)
self.given_name = guest.first_name or ""
self.sn = guest.last_name or "."
self.display_name = self.display_name_for_guest(guest)
self.mail = guest.email or ""
self.telephone_number = guest.phone or ""
self.employee_number = str(guest.id)
self.description = self.description_for_guest(guest)
self.user_account_control = UserAccountControl.DISABLED

View File

@ -0,0 +1,145 @@
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
GUEST_ID_START = 900_000_000_000_001
GUEST_ID_MAX = 999_999_999_999_999
def create_guest_id_sequence(apps, schema_editor):
if schema_editor.connection.vendor != "postgresql":
return
with schema_editor.connection.cursor() as cursor:
cursor.execute(
"""
CREATE SEQUENCE IF NOT EXISTS guests_guest_id_seq
START WITH 900000000000001
INCREMENT BY 1
MINVALUE 900000000000001
MAXVALUE 999999999999999
NO CYCLE
"""
)
def drop_guest_id_sequence(apps, schema_editor):
if schema_editor.connection.vendor != "postgresql":
return
with schema_editor.connection.cursor() as cursor:
cursor.execute("DROP SEQUENCE IF EXISTS guests_guest_id_seq")
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
("documents", "0015_delete_documentcomment"),
("employees", "0053_cleanup_department_role_permissions"),
]
operations = [
migrations.RunPython(create_guest_id_sequence, drop_guest_id_sequence),
migrations.CreateModel(
name="Guest",
fields=[
("id", models.BigIntegerField(editable=False, primary_key=True, serialize=False)),
("last_name", models.CharField(max_length=150, verbose_name="Фамилия")),
("first_name", models.CharField(max_length=150, verbose_name="Имя")),
("patronymic", models.CharField(blank=True, max_length=150, verbose_name="Отчество")),
("birth_date", models.DateField(blank=True, null=True, verbose_name="Дата рождения")),
("phone", models.CharField(blank=True, max_length=64, verbose_name="Телефон")),
("email", models.EmailField(blank=True, max_length=254, verbose_name="Email")),
("organization", models.CharField(blank=True, max_length=255, verbose_name="Организация")),
("position", models.CharField(blank=True, max_length=255, verbose_name="Должность")),
("comment", models.TextField(blank=True, verbose_name="Комментарий")),
("is_active", models.BooleanField(default=True, verbose_name="Активен")),
("ldap_enabled", models.BooleanField(default=False, verbose_name="LDAP в Active OU")),
("ldap_username", models.CharField(blank=True, max_length=150, verbose_name="LDAP username")),
("ldap_upn", models.CharField(blank=True, max_length=255, verbose_name="LDAP UPN")),
("ldap_last_synced_at", models.DateTimeField(blank=True, null=True, verbose_name="Последняя LDAP синхронизация")),
("ldap_last_error", models.TextField(blank=True, verbose_name="Последняя LDAP ошибка")),
("created_at", models.DateTimeField(auto_now_add=True, verbose_name="Создано")),
("updated_at", models.DateTimeField(auto_now=True, verbose_name="Обновлено")),
("created_by", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="created_guests", to=settings.AUTH_USER_MODEL, verbose_name="Создал")),
],
options={
"verbose_name": "Гость",
"verbose_name_plural": "Гости",
"ordering": ["last_name", "first_name", "id"],
},
),
migrations.CreateModel(
name="GuestVisit",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("inviter_snapshot_name", models.CharField(blank=True, max_length=255, verbose_name="ФИО приглашающего на момент создания")),
("inviter_snapshot_email", models.EmailField(blank=True, max_length=254, verbose_name="Email приглашающего на момент создания")),
("purpose", models.TextField(verbose_name="Цель приглашения")),
("visit_comment", models.TextField(blank=True, verbose_name="Комментарий заявителя")),
("admin_comment", models.TextField(blank=True, verbose_name="Комментарий администратора")),
("status", models.CharField(choices=[("draft", "Черновик"), ("pending", "На рассмотрении"), ("needs_info", "Требуется информация"), ("approved", "Одобрено"), ("rejected", "Отклонено"), ("cancelled", "Отменено"), ("expired", "Истекло"), ("revoked", "Отозвано")], db_index=True, default="draft", max_length=24, verbose_name="Статус")),
("access_starts_at", models.DateTimeField(blank=True, null=True, verbose_name="Начало доступа")),
("access_expires_at", models.DateTimeField(blank=True, null=True, verbose_name="Окончание доступа")),
("all_day", models.BooleanField(default=True, verbose_name="Полные сутки")),
("unlimited", models.BooleanField(default=False, verbose_name="Бессрочно")),
("submitted_at", models.DateTimeField(blank=True, null=True, verbose_name="Отправлено на рассмотрение")),
("decided_at", models.DateTimeField(blank=True, null=True, verbose_name="Решение принято")),
("decision_comment", models.TextField(blank=True, verbose_name="Комментарий к решению")),
("cancelled_at", models.DateTimeField(blank=True, null=True)),
("cancel_reason", models.TextField(blank=True)),
("revoked_at", models.DateTimeField(blank=True, null=True)),
("revoke_reason", models.TextField(blank=True)),
("expired_at", models.DateTimeField(blank=True, null=True)),
("inviter_inactive", models.BooleanField(db_index=True, default=False)),
("created_at", models.DateTimeField(auto_now_add=True, verbose_name="Создано")),
("updated_at", models.DateTimeField(auto_now=True, verbose_name="Обновлено")),
("cancelled_by", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="cancelled_guest_visits", to=settings.AUTH_USER_MODEL)),
("decided_by", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="decided_guest_visits", to=settings.AUTH_USER_MODEL, verbose_name="Решение принял")),
("documents", models.ManyToManyField(blank=True, related_name="guest_visits", to="documents.document", verbose_name="Документы")),
("guest", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name="visits", to="guests.guest", verbose_name="Гость")),
("host_department", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="guest_visits", to="employees.department", verbose_name="Отдел приглашающего")),
("inviter", models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name="guest_visits", to=settings.AUTH_USER_MODEL, verbose_name="Приглашающий")),
("revoked_by", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="revoked_guest_visits", to=settings.AUTH_USER_MODEL)),
],
options={
"verbose_name": "Гостевой визит",
"verbose_name_plural": "Гостевые визиты",
"ordering": ["-created_at"],
"permissions": [("view_all_guestvisit", "Может просматривать все гостевые визиты"), ("decide_guestvisit", "Может принимать решения по гостевым визитам"), ("manage_guestaccount", "Может управлять гостевыми учетками")],
},
),
migrations.CreateModel(
name="GuestVisitEvent",
fields=[
("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
("event_type", models.CharField(choices=[("created", "Создано"), ("submitted", "Отправлено"), ("needs_info_requested", "Запрошена информация"), ("info_provided", "Информация предоставлена"), ("approved", "Одобрено"), ("rejected", "Отклонено"), ("decision_changed", "Решение изменено"), ("cancelled", "Отменено"), ("revoked", "Отозвано"), ("expired", "Истекло"), ("ldap_created", "LDAP создан"), ("ldap_updated", "LDAP обновлен"), ("ldap_enabled", "LDAP перемещен в Active OU"), ("ldap_disabled", "LDAP перемещен в Deactivated OU"), ("ldap_failed", "Ошибка LDAP"), ("ldap_skipped", "LDAP пропущен"), ("document_attached", "Документ прикреплен"), ("document_removed", "Документ удален"), ("inviter_inactive_detected", "Приглашающий неактивен")], db_index=True, max_length=48, verbose_name="Тип события")),
("from_status", models.CharField(blank=True, max_length=24)),
("to_status", models.CharField(blank=True, max_length=24)),
("comment", models.TextField(blank=True)),
("metadata", models.JSONField(blank=True, default=dict)),
("created_at", models.DateTimeField(auto_now_add=True)),
("actor", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name="guest_visit_events", to=settings.AUTH_USER_MODEL, verbose_name="Автор события")),
("visit", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="events", to="guests.guestvisit", verbose_name="Визит")),
],
options={
"verbose_name": "Событие гостевого визита",
"verbose_name_plural": "События гостевых визитов",
"ordering": ["-created_at", "-id"],
},
),
migrations.AddIndex(model_name="guest", index=models.Index(fields=["last_name", "first_name"], name="guest_name_idx")),
migrations.AddIndex(model_name="guest", index=models.Index(fields=["email"], name="guest_email_idx")),
migrations.AddIndex(model_name="guest", index=models.Index(fields=["phone"], name="guest_phone_idx")),
migrations.AddIndex(model_name="guest", index=models.Index(fields=["ldap_enabled"], name="guest_ldap_enabled_idx")),
migrations.AddConstraint(model_name="guest", constraint=models.CheckConstraint(condition=models.Q(("id__gte", GUEST_ID_START), ("id__lte", GUEST_ID_MAX)), name="guest_id_in_guest_range")),
migrations.AddIndex(model_name="guestvisit", index=models.Index(fields=["status", "created_at"], name="guestvisit_status_created_idx")),
migrations.AddIndex(model_name="guestvisit", index=models.Index(fields=["inviter", "created_at"], name="guestvisit_inviter_created_idx")),
migrations.AddIndex(model_name="guestvisit", index=models.Index(fields=["access_starts_at", "access_expires_at"], name="guestvisit_access_idx")),
migrations.AddIndex(model_name="guestvisit", index=models.Index(fields=["guest", "status"], name="guestvisit_guest_status_idx")),
migrations.AddConstraint(model_name="guestvisit", constraint=models.CheckConstraint(condition=models.Q(("unlimited", True), ("access_expires_at__isnull", True), ("access_starts_at__isnull", True), ("access_starts_at__lt", models.F("access_expires_at")), _connector="OR"), name="guest_visit_access_range_valid")),
migrations.AddIndex(model_name="guestvisitevent", index=models.Index(fields=["visit", "-created_at"], name="guestevent_visit_created_idx")),
migrations.AddIndex(model_name="guestvisitevent", index=models.Index(fields=["event_type", "-created_at"], name="guestevent_type_created_idx")),
]

View File

@ -0,0 +1,115 @@
import employees.ldap.mixins
import ldapdb.models.fields
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("guests", "0001_initial"),
]
operations = [
migrations.CreateModel(
name="LdapGuestUser",
fields=[
(
"dn",
ldapdb.models.fields.CharField(
max_length=200,
primary_key=True,
serialize=False,
),
),
("cn", ldapdb.models.fields.CharField(db_column="cn", max_length=200)),
(
"sam_account_name",
ldapdb.models.fields.CharField(
db_column="sAMAccountName",
max_length=200,
),
),
(
"user_principal_name",
ldapdb.models.fields.CharField(
db_column="userPrincipalName",
max_length=200,
),
),
(
"given_name",
ldapdb.models.fields.CharField(
blank=True,
db_column="givenName",
max_length=200,
),
),
("sn", ldapdb.models.fields.CharField(db_column="sn", max_length=200)),
(
"display_name",
ldapdb.models.fields.CharField(
db_column="displayName",
max_length=200,
),
),
(
"mail",
ldapdb.models.fields.CharField(
blank=True,
db_column="mail",
max_length=200,
),
),
(
"telephone_number",
ldapdb.models.fields.CharField(
blank=True,
db_column="telephoneNumber",
max_length=200,
),
),
(
"employee_number",
ldapdb.models.fields.CharField(
blank=True,
db_column="employeeNumber",
max_length=200,
),
),
(
"description",
ldapdb.models.fields.CharField(
blank=True,
db_column="description",
max_length=200,
),
),
(
"user_account_control",
ldapdb.models.fields.IntegerField(
db_column="userAccountControl",
),
),
(
"member_of",
ldapdb.models.fields.ListField(blank=True, db_column="memberOf"),
),
(
"when_created",
ldapdb.models.fields.DateTimeField(db_column="whenCreated"),
),
(
"when_changed",
ldapdb.models.fields.DateTimeField(db_column="whenChanged"),
),
],
options={
"managed": False,
},
bases=(
employees.ldap.mixins.LdapSyncStateMixin,
employees.ldap.mixins.ModifyDnMixin,
models.Model,
),
),
]

View File

@ -0,0 +1 @@

319
backend/guests/models.py Normal file
View File

@ -0,0 +1,319 @@
from __future__ import annotations
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import connection, models
from django.db.models import Max, Q
from django.utils import timezone
from .constants import (
EDITABLE_BY_AUTHOR_STATUSES,
GUEST_ID_MAX,
GUEST_ID_MIN,
GUEST_ID_START,
GuestVisitEventType,
GuestVisitStatus,
)
def _setting_int(name: str, default: int) -> int:
return int(getattr(settings, name, default))
class Guest(models.Model):
id = models.BigIntegerField(primary_key=True, editable=False)
last_name = models.CharField("Фамилия", max_length=150)
first_name = models.CharField("Имя", max_length=150)
patronymic = models.CharField("Отчество", max_length=150, blank=True)
birth_date = models.DateField("Дата рождения", null=True, blank=True)
phone = models.CharField("Телефон", max_length=64, blank=True)
email = models.EmailField("Email", blank=True)
organization = models.CharField("Организация", max_length=255, blank=True)
position = models.CharField("Должность", max_length=255, blank=True)
comment = models.TextField("Комментарий", blank=True)
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="created_guests",
verbose_name="Создал",
)
is_active = models.BooleanField("Активен", default=True)
ldap_enabled = models.BooleanField("LDAP в Active OU", default=False)
ldap_username = models.CharField("LDAP username", max_length=150, blank=True)
ldap_upn = models.CharField("LDAP UPN", max_length=255, blank=True)
ldap_last_synced_at = models.DateTimeField(
"Последняя LDAP синхронизация",
null=True,
blank=True,
)
ldap_last_error = models.TextField("Последняя LDAP ошибка", blank=True)
created_at = models.DateTimeField("Создано", auto_now_add=True)
updated_at = models.DateTimeField("Обновлено", auto_now=True)
class Meta:
verbose_name = "Гость"
verbose_name_plural = "Гости"
ordering = ["last_name", "first_name", "id"]
indexes = [
models.Index(fields=["last_name", "first_name"], name="guest_name_idx"),
models.Index(fields=["email"], name="guest_email_idx"),
models.Index(fields=["phone"], name="guest_phone_idx"),
models.Index(fields=["ldap_enabled"], name="guest_ldap_enabled_idx"),
]
constraints = [
models.CheckConstraint(
name="guest_id_in_guest_range",
condition=Q(id__gte=GUEST_ID_START) & Q(id__lte=GUEST_ID_MAX),
),
]
def __str__(self) -> str:
return self.full_name
@property
def full_name(self) -> str:
return " ".join(
part
for part in [self.last_name, self.first_name, self.patronymic]
if part
).strip()
@classmethod
def next_guest_id(cls) -> int:
start = _setting_int("GUESTS_ID_START", GUEST_ID_START)
max_value = _setting_int("GUESTS_ID_MAX", GUEST_ID_MAX)
if connection.vendor == "postgresql":
with connection.cursor() as cursor:
cursor.execute("SELECT nextval('guests_guest_id_seq')")
value = int(cursor.fetchone()[0])
else:
current = cls.objects.aggregate(max_id=Max("id"))["max_id"]
value = start if current is None or current < start else int(current) + 1
if value < start or value > max_value:
raise ValidationError("Диапазон идентификаторов гостей исчерпан.")
return value
def clean(self):
super().clean()
if self.id is not None and not (GUEST_ID_START <= int(self.id) <= GUEST_ID_MAX):
raise ValidationError(
{"id": "ID гостя должен находиться в гостевом диапазоне."}
)
def save(self, *args, **kwargs):
if self.pk is None:
self.pk = self.next_guest_id()
self.clean()
super().save(*args, **kwargs)
class GuestVisit(models.Model):
guest = models.ForeignKey(
Guest,
on_delete=models.PROTECT,
related_name="visits",
verbose_name="Гость",
)
inviter = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.PROTECT,
related_name="guest_visits",
verbose_name="Приглашающий",
)
inviter_snapshot_name = models.CharField(
"ФИО приглашающего на момент создания",
max_length=255,
blank=True,
)
inviter_snapshot_email = models.EmailField(
"Email приглашающего на момент создания",
blank=True,
)
host_department = models.ForeignKey(
"employees.Department",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="guest_visits",
verbose_name="Отдел приглашающего",
)
purpose = models.TextField("Цель приглашения")
visit_comment = models.TextField("Комментарий заявителя", blank=True)
admin_comment = models.TextField("Комментарий администратора", blank=True)
status = models.CharField(
"Статус",
max_length=24,
choices=GuestVisitStatus.choices,
default=GuestVisitStatus.DRAFT,
db_index=True,
)
access_starts_at = models.DateTimeField(
"Начало доступа",
null=True,
blank=True,
)
access_expires_at = models.DateTimeField(
"Окончание доступа",
null=True,
blank=True,
)
all_day = models.BooleanField("Полные сутки", default=True)
unlimited = models.BooleanField("Бессрочно", default=False)
documents = models.ManyToManyField(
"documents.Document",
blank=True,
related_name="guest_visits",
verbose_name="Документы",
)
submitted_at = models.DateTimeField(
"Отправлено на рассмотрение",
null=True,
blank=True,
)
decided_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="decided_guest_visits",
verbose_name="Решение принял",
)
decided_at = models.DateTimeField("Решение принято", null=True, blank=True)
decision_comment = models.TextField("Комментарий к решению", blank=True)
cancelled_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="cancelled_guest_visits",
)
cancelled_at = models.DateTimeField(null=True, blank=True)
cancel_reason = models.TextField(blank=True)
revoked_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="revoked_guest_visits",
)
revoked_at = models.DateTimeField(null=True, blank=True)
revoke_reason = models.TextField(blank=True)
expired_at = models.DateTimeField(null=True, blank=True)
inviter_inactive = models.BooleanField(default=False, db_index=True)
created_at = models.DateTimeField("Создано", auto_now_add=True)
updated_at = models.DateTimeField("Обновлено", auto_now=True)
class Meta:
verbose_name = "Гостевой визит"
verbose_name_plural = "Гостевые визиты"
ordering = ["-created_at"]
permissions = [
("view_all_guestvisit", "Может просматривать все гостевые визиты"),
("decide_guestvisit", "Может принимать решения по гостевым визитам"),
("manage_guestaccount", "Может управлять гостевыми учетками"),
]
indexes = [
models.Index(
fields=["status", "created_at"],
name="guestvisit_status_created_idx",
),
models.Index(
fields=["inviter", "created_at"],
name="guestvisit_inviter_created_idx",
),
models.Index(
fields=["access_starts_at", "access_expires_at"],
name="guestvisit_access_idx",
),
models.Index(fields=["guest", "status"], name="guestvisit_guest_status_idx"),
]
constraints = [
models.CheckConstraint(
name="guest_visit_access_range_valid",
condition=Q(unlimited=True)
| Q(access_expires_at__isnull=True)
| Q(access_starts_at__isnull=True)
| Q(access_starts_at__lt=models.F("access_expires_at")),
),
]
def __str__(self) -> str:
return f"{self.guest} — {self.get_status_display()}"
@property
def is_active_now(self) -> bool:
if self.status != GuestVisitStatus.APPROVED:
return False
if self.unlimited:
return True
now = timezone.now()
if self.access_starts_at and self.access_starts_at > now:
return False
if self.access_expires_at and self.access_expires_at <= now:
return False
return True
@property
def is_expired(self) -> bool:
return bool(
not self.unlimited
and self.access_expires_at
and self.access_expires_at <= timezone.now()
)
def can_edit_by(self, user) -> bool:
if not user or not user.is_authenticated:
return False
if getattr(user, "is_staff", False) or getattr(user, "is_superuser", False):
return True
return (
self.inviter_id == user.id
and self.status in EDITABLE_BY_AUTHOR_STATUSES
)
class GuestVisitEvent(models.Model):
visit = models.ForeignKey(
GuestVisit,
on_delete=models.CASCADE,
related_name="events",
verbose_name="Визит",
)
actor = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="guest_visit_events",
verbose_name="Автор события",
)
event_type = models.CharField(
"Тип события",
max_length=48,
choices=GuestVisitEventType.choices,
db_index=True,
)
from_status = models.CharField(max_length=24, blank=True)
to_status = models.CharField(max_length=24, blank=True)
comment = models.TextField(blank=True)
metadata = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
verbose_name = "Событие гостевого визита"
verbose_name_plural = "События гостевых визитов"
ordering = ["-created_at", "-id"]
indexes = [
models.Index(fields=["visit", "-created_at"], name="guestevent_visit_created_idx"),
models.Index(
fields=["event_type", "-created_at"],
name="guestevent_type_created_idx",
),
]
def __str__(self) -> str:
return f"{self.visit_id}: {self.event_type}"

View File

@ -0,0 +1,32 @@
from __future__ import annotations
GUEST_ADMIN_PERMS = (
"guests.view_all_guestvisit",
"guests.decide_guestvisit",
"guests.manage_guestaccount",
)
def is_guest_admin(user) -> bool:
if not user or not getattr(user, "is_authenticated", False):
return False
if getattr(user, "is_staff", False) or getattr(user, "is_superuser", False):
return True
return any(user.has_perm(perm) for perm in GUEST_ADMIN_PERMS)
def can_decide_guest_visit(user) -> bool:
if not user or not getattr(user, "is_authenticated", False):
return False
if getattr(user, "is_staff", False) or getattr(user, "is_superuser", False):
return True
return bool(user.has_perm("guests.decide_guestvisit"))
def can_manage_guest_account(user) -> bool:
if not user or not getattr(user, "is_authenticated", False):
return False
if getattr(user, "is_staff", False) or getattr(user, "is_superuser", False):
return True
return bool(user.has_perm("guests.manage_guestaccount"))

533
backend/guests/services.py Normal file
View File

@ -0,0 +1,533 @@
from __future__ import annotations
import logging
from datetime import datetime, time, timedelta
from typing import Iterable
from django.conf import settings
from django.contrib.auth import get_user_model
from django.db import transaction
from django.db.models import Q
from django.utils import timezone
from employees.models import LdapSyncQueue, LdapSyncState
from guests.ldap.orm_models import LdapGuestUser
from notifications.signals import notify
from .constants import GuestVisitEventType, GuestVisitStatus
from .models import Guest, GuestVisit, GuestVisitEvent
from .permissions import GUEST_ADMIN_PERMS
logger = logging.getLogger(__name__)
User = get_user_model()
def get_guest_admin_users():
return (
User.objects.filter(
Q(is_staff=True)
| Q(is_superuser=True)
| Q(user_permissions__codename__in=[p.split(".")[1] for p in GUEST_ADMIN_PERMS])
| Q(groups__permissions__codename__in=[p.split(".")[1] for p in GUEST_ADMIN_PERMS])
)
.filter(is_active=True)
.distinct()
)
def record_guest_event(
visit: GuestVisit,
event_type: str,
*,
actor=None,
from_status: str = "",
to_status: str = "",
comment: str = "",
metadata: dict | None = None,
) -> GuestVisitEvent:
return GuestVisitEvent.objects.create(
visit=visit,
actor=actor if actor and getattr(actor, "is_authenticated", False) else None,
event_type=event_type,
from_status=from_status or "",
to_status=to_status or "",
comment=comment or "",
metadata=metadata or {},
)
def notify_guest_visit(
visit: GuestVisit,
verb: str,
recipients: Iterable,
*,
actor=None,
title: str,
message: str,
extra: dict | None = None,
) -> None:
recipient_list = [u for u in recipients if u and getattr(u, "is_active", False)]
if not recipient_list:
return
notify.send(
sender=actor or visit.inviter,
recipient=recipient_list,
verb=verb,
action_object=visit,
target=visit.guest,
description=message,
action_url=f"/guests?visit={visit.pk}",
data={
"title": title,
"visit_id": visit.pk,
"guest_id": str(visit.guest_id),
"guest_name": visit.guest.full_name,
**(extra or {}),
},
)
def resolve_inviter_department(inviter):
departments = getattr(inviter, "departments", None)
try:
return departments.first() if departments is not None else None
except Exception:
return None
def normalize_all_day_range(date_from, date_to):
tz = timezone.get_current_timezone()
starts_at = timezone.make_aware(datetime.combine(date_from, time.min), tz)
expires_at = timezone.make_aware(
datetime.combine(date_to + timedelta(days=1), time.min),
tz,
)
return starts_at, expires_at
class GuestVisitWorkflow:
@staticmethod
def create_visit(*, actor, guest: Guest, **data) -> GuestVisit:
with transaction.atomic():
department = resolve_inviter_department(actor)
visit = GuestVisit.objects.create(
guest=guest,
inviter=actor,
inviter_snapshot_name=(actor.get_full_name() or str(actor)),
inviter_snapshot_email=getattr(actor, "email", "") or "",
host_department=department,
**data,
)
record_guest_event(
visit,
GuestVisitEventType.CREATED,
actor=actor,
to_status=visit.status,
)
return visit
@staticmethod
def _set_status(
visit: GuestVisit,
new_status: str,
*,
actor,
event_type: str,
comment: str = "",
update_fields: list[str] | None = None,
) -> GuestVisit:
old_status = visit.status
visit.status = new_status
fields = ["status", "updated_at", *(update_fields or [])]
visit.save(update_fields=fields)
record_guest_event(
visit,
event_type,
actor=actor,
from_status=old_status,
to_status=new_status,
comment=comment,
)
return visit
@classmethod
def submit(cls, visit: GuestVisit, *, actor) -> GuestVisit:
if visit.status not in {GuestVisitStatus.DRAFT, GuestVisitStatus.NEEDS_INFO}:
raise ValueError("Отправить можно только черновик или заявку с уточнениями.")
if not visit.purpose.strip():
raise ValueError("Укажите цель приглашения.")
if not visit.unlimited and not (visit.access_starts_at and visit.access_expires_at):
raise ValueError("Укажите период доступа или бессрочный доступ.")
if (
getattr(settings, "GUESTS_REQUIRE_ID_DOCUMENT", False)
and not visit.documents.exists()
):
raise ValueError("Прикрепите документ гостя.")
visit.submitted_at = timezone.now()
event = (
GuestVisitEventType.INFO_PROVIDED
if visit.status == GuestVisitStatus.NEEDS_INFO
else GuestVisitEventType.SUBMITTED
)
cls._set_status(
visit,
GuestVisitStatus.PENDING,
actor=actor,
event_type=event,
update_fields=["submitted_at"],
)
notify_guest_visit(
visit,
"guest_visit_submitted" if event == GuestVisitEventType.SUBMITTED else "guest_visit_info_provided",
get_guest_admin_users(),
actor=actor,
title="Гостевой визит на рассмотрении",
message=f"{actor} отправил гостевой визит для {visit.guest.full_name}.",
)
return visit
@classmethod
def request_info(cls, visit: GuestVisit, *, actor, comment: str) -> GuestVisit:
if visit.status != GuestVisitStatus.PENDING:
raise ValueError("Запросить уточнения можно только по заявке на рассмотрении.")
if not comment.strip():
raise ValueError("Комментарий обязателен.")
visit.admin_comment = comment
cls._set_status(
visit,
GuestVisitStatus.NEEDS_INFO,
actor=actor,
event_type=GuestVisitEventType.NEEDS_INFO_REQUESTED,
comment=comment,
update_fields=["admin_comment"],
)
notify_guest_visit(
visit,
"guest_visit_needs_info",
[visit.inviter],
actor=actor,
title="Нужна информация по гостевому визиту",
message=comment,
)
return visit
@classmethod
def approve(cls, visit: GuestVisit, *, actor, comment: str = "") -> GuestVisit:
if visit.status not in {GuestVisitStatus.PENDING, GuestVisitStatus.REJECTED}:
raise ValueError("Одобрить можно только заявку на рассмотрении или отклоненную заявку.")
old_status = visit.status
visit.decided_by = actor
visit.decided_at = timezone.now()
visit.decision_comment = comment
event_type = (
GuestVisitEventType.DECISION_CHANGED
if old_status == GuestVisitStatus.REJECTED
else GuestVisitEventType.APPROVED
)
cls._set_status(
visit,
GuestVisitStatus.APPROVED,
actor=actor,
event_type=event_type,
comment=comment,
update_fields=["decided_by", "decided_at", "decision_comment"],
)
notify_guest_visit(
visit,
"guest_visit_approved",
[visit.inviter],
actor=actor,
title="Гостевой визит одобрен",
message=f"Гостевой визит для {visit.guest.full_name} одобрен.",
)
GuestLdapService().sync_guest_for_visit(visit)
return visit
@classmethod
def reject(cls, visit: GuestVisit, *, actor, comment: str = "") -> GuestVisit:
if visit.status not in {
GuestVisitStatus.PENDING,
GuestVisitStatus.NEEDS_INFO,
GuestVisitStatus.APPROVED,
}:
raise ValueError("Отклонить можно только рассматриваемую или одобренную заявку.")
old_status = visit.status
visit.decided_by = actor
visit.decided_at = timezone.now()
visit.decision_comment = comment
event_type = (
GuestVisitEventType.DECISION_CHANGED
if old_status == GuestVisitStatus.APPROVED
else GuestVisitEventType.REJECTED
)
cls._set_status(
visit,
GuestVisitStatus.REJECTED,
actor=actor,
event_type=event_type,
comment=comment,
update_fields=["decided_by", "decided_at", "decision_comment"],
)
notify_guest_visit(
visit,
"guest_visit_rejected",
[visit.inviter],
actor=actor,
title="Гостевой визит отклонен",
message=f"Гостевой визит для {visit.guest.full_name} отклонен.",
)
GuestLdapService().disable_guest(visit.guest, visit=visit)
return visit
@classmethod
def cancel(cls, visit: GuestVisit, *, actor, comment: str = "") -> GuestVisit:
if visit.status in {
GuestVisitStatus.CANCELLED,
GuestVisitStatus.EXPIRED,
GuestVisitStatus.REVOKED,
}:
raise ValueError("Эту заявку уже нельзя отменить.")
visit.cancelled_by = actor
visit.cancelled_at = timezone.now()
visit.cancel_reason = comment
cls._set_status(
visit,
GuestVisitStatus.CANCELLED,
actor=actor,
event_type=GuestVisitEventType.CANCELLED,
comment=comment,
update_fields=["cancelled_by", "cancelled_at", "cancel_reason"],
)
GuestLdapService().disable_guest_if_no_active_visits(visit.guest, visit=visit)
return visit
@classmethod
def revoke(cls, visit: GuestVisit, *, actor, comment: str = "") -> GuestVisit:
if visit.status != GuestVisitStatus.APPROVED:
raise ValueError("Отозвать можно только одобренный гостевой доступ.")
visit.revoked_by = actor
visit.revoked_at = timezone.now()
visit.revoke_reason = comment
cls._set_status(
visit,
GuestVisitStatus.REVOKED,
actor=actor,
event_type=GuestVisitEventType.REVOKED,
comment=comment,
update_fields=["revoked_by", "revoked_at", "revoke_reason"],
)
notify_guest_visit(
visit,
"guest_visit_revoked",
[visit.inviter],
actor=actor,
title="Гостевой доступ отозван",
message=f"Доступ гостя {visit.guest.full_name} отозван.",
)
GuestLdapService().disable_guest_if_no_active_visits(visit.guest, visit=visit)
return visit
@classmethod
def expire(cls, visit: GuestVisit) -> GuestVisit:
if visit.status != GuestVisitStatus.APPROVED:
raise ValueError("Истечь может только одобренный гостевой доступ.")
visit.expired_at = timezone.now()
cls._set_status(
visit,
GuestVisitStatus.EXPIRED,
actor=None,
event_type=GuestVisitEventType.EXPIRED,
update_fields=["expired_at"],
)
if getattr(settings, "GUESTS_NOTIFY_ON_EXPIRATION", True):
notify_guest_visit(
visit,
"guest_visit_expired",
[visit.inviter],
title="Гостевой доступ истек",
message=f"Доступ гостя {visit.guest.full_name} истек.",
)
GuestLdapService().disable_guest_if_no_active_visits(visit.guest, visit=visit)
return visit
class GuestLdapService:
@staticmethod
def _is_ldap_write_enabled() -> bool:
return bool(
getattr(settings, "LDAP_ENABLED", False)
and getattr(settings, "LDAP_WRITE_ENABLED", False)
)
def _should_be_in_active_ou(self, guest: Guest) -> bool:
if not guest.is_active:
return False
return guest.visits.filter(status=GuestVisitStatus.APPROVED).filter(
Q(unlimited=True)
| (
Q(access_starts_at__lte=timezone.now())
& (
Q(access_expires_at__isnull=True)
| Q(access_expires_at__gt=timezone.now())
)
)
).exists()
def _get_state_dn(self, guest: Guest) -> str:
return (
LdapSyncState.objects.filter(model="guest", object_pk=str(guest.pk))
.values_list("ldap_dn", flat=True)
.first()
or ""
)
def _touch_state(self, guest: Guest, dn: str) -> None:
state, _ = LdapSyncState.objects.get_or_create(
model="guest",
object_pk=str(guest.pk),
)
state.touch(ldap_dn=dn, sync_dir="django")
def _record_skip(self, guest: Guest, visit: GuestVisit | None = None) -> None:
guest.ldap_last_error = ""
guest.ldap_last_synced_at = timezone.now()
guest.save(update_fields=["ldap_last_error", "ldap_last_synced_at", "updated_at"])
if visit:
record_guest_event(
visit,
GuestVisitEventType.LDAP_SKIPPED,
metadata={"reason": "LDAP write is disabled"},
)
def sync_guest_for_visit(
self,
visit: GuestVisit,
*,
enqueue_on_error: bool = True,
raise_on_error: bool = False,
) -> None:
self.sync_guest(
visit.guest,
visit=visit,
enqueue_on_error=enqueue_on_error,
raise_on_error=raise_on_error,
)
def sync_guest(
self,
guest: Guest,
visit: GuestVisit | None = None,
*,
enqueue_on_error: bool = True,
raise_on_error: bool = False,
) -> None:
if not self._is_ldap_write_enabled():
self._record_skip(guest, visit)
return
in_active_ou = self._should_be_in_active_ou(guest)
current_dn = self._get_state_dn(guest)
try:
if current_dn:
result = LdapGuestUser.sync_existing_for_guest(
guest,
current_dn,
in_active_ou,
)
event_type = (
GuestVisitEventType.LDAP_ENABLED
if in_active_ou
else GuestVisitEventType.LDAP_DISABLED
)
else:
result = LdapGuestUser.create_for_guest(guest, in_active_ou)
event_type = GuestVisitEventType.LDAP_CREATED
self._touch_state(guest, result.dn)
guest.ldap_enabled = in_active_ou
guest.ldap_username = result.sam_account_name
guest.ldap_upn = result.user_principal_name
guest.ldap_last_error = ""
guest.ldap_last_synced_at = timezone.now()
guest.save(
update_fields=[
"ldap_enabled",
"ldap_username",
"ldap_upn",
"ldap_last_error",
"ldap_last_synced_at",
"updated_at",
]
)
if visit:
record_guest_event(
visit,
event_type,
metadata={
"ldap_dn": result.dn,
"in_active_ou": in_active_ou,
"userAccountControl": str(result.user_account_control),
},
)
except Exception as exc:
guest.ldap_last_error = str(exc)[:2000]
guest.save(update_fields=["ldap_last_error", "updated_at"])
if visit:
record_guest_event(
visit,
GuestVisitEventType.LDAP_FAILED,
comment=str(exc),
)
if enqueue_on_error:
self.enqueue_guest_sync(guest, operation="guest_sync")
notify_visit = visit or guest.visits.order_by("-created_at").first()
if notify_visit:
notify_guest_visit(
notify_visit,
"guest_ldap_failed",
get_guest_admin_users(),
title="Ошибка LDAP для гостя",
message=str(exc),
)
if raise_on_error:
raise
def disable_guest(
self,
guest: Guest,
visit: GuestVisit | None = None,
*,
enqueue_on_error: bool = True,
raise_on_error: bool = False,
) -> None:
self.sync_guest(
guest,
visit=visit,
enqueue_on_error=enqueue_on_error,
raise_on_error=raise_on_error,
)
def disable_guest_if_no_active_visits(
self,
guest: Guest,
visit: GuestVisit | None = None,
*,
enqueue_on_error: bool = True,
raise_on_error: bool = False,
) -> None:
if not self._should_be_in_active_ou(guest):
self.disable_guest(
guest,
visit=visit,
enqueue_on_error=enqueue_on_error,
raise_on_error=raise_on_error,
)
@staticmethod
def enqueue_guest_sync(guest: Guest, *, operation: str = "guest_sync") -> None:
LdapSyncQueue.objects.create(
operation=operation,
model_name="guest",
object_pk=str(guest.pk),
payload={"guest_id": str(guest.pk), "_operation": operation},
)

94
backend/guests/tasks.py Normal file
View File

@ -0,0 +1,94 @@
import logging
from celery import shared_task
from django.db.models import Q
from django.utils import timezone
from .constants import GuestVisitEventType, GuestVisitStatus
from .models import Guest, GuestVisit
from .services import (
GuestLdapService,
GuestVisitWorkflow,
get_guest_admin_users,
notify_guest_visit,
record_guest_event,
)
logger = logging.getLogger(__name__)
@shared_task(ignore_result=True)
def activate_due_guest_visits():
now = timezone.now()
visits = GuestVisit.objects.filter(
status=GuestVisitStatus.APPROVED,
unlimited=False,
access_starts_at__lte=now,
).filter(Q(access_expires_at__isnull=True) | Q(access_expires_at__gt=now))
for visit in visits.select_related("guest", "inviter"):
try:
GuestLdapService().sync_guest_for_visit(visit)
except Exception:
logger.exception("Failed to activate guest visit %s", visit.pk)
@shared_task(ignore_result=True)
def expire_guest_visits():
now = timezone.now()
visits = GuestVisit.objects.filter(
status=GuestVisitStatus.APPROVED,
unlimited=False,
access_expires_at__lte=now,
)
for visit in visits.select_related("guest", "inviter"):
try:
GuestVisitWorkflow.expire(visit)
except Exception:
logger.exception("Failed to expire guest visit %s", visit.pk)
@shared_task(ignore_result=True)
def detect_inactive_inviters():
visits = GuestVisit.objects.filter(
status=GuestVisitStatus.APPROVED,
inviter_inactive=False,
).select_related("guest", "inviter")
admins = list(get_guest_admin_users())
for visit in visits:
inviter = visit.inviter
if inviter.is_active and getattr(inviter, "is_actually_active", True):
continue
visit.inviter_inactive = True
visit.save(update_fields=["inviter_inactive", "updated_at"])
record_guest_event(
visit,
GuestVisitEventType.INVITER_INACTIVE_DETECTED,
metadata={"inviter_id": inviter.id},
)
notify_guest_visit(
visit,
"guest_inviter_inactive",
admins,
title="Приглашающий гостя неактивен",
message=f"Приглашающий для гостя {visit.guest.full_name} неактивен.",
)
def execute_guest_queue_operation(operation: str, payload: dict) -> None:
guest_id = payload.get("guest_id") or payload.get("object_pk")
guest = Guest.objects.get(pk=guest_id)
service = GuestLdapService()
if operation in {"guest_sync", "guest_disable"}:
service.sync_guest(
guest,
enqueue_on_error=False,
raise_on_error=True,
)
elif operation == "guest_delete":
service.disable_guest(
guest,
enqueue_on_error=False,
raise_on_error=True,
)
else:
raise ValueError(f"Unknown guest LDAP operation: {operation}")

View File

@ -169,6 +169,7 @@
Обязательные поля:
- `id`: 15-значный числовой первичный ключ гостя;
- `last_name`: фамилия;
- `first_name`: имя;
- `patronymic`: отчество, опционально;
@ -184,7 +185,7 @@
LDAP-поля и состояние:
- `is_active`: активен ли гость в бизнес-смысле;
- `ldap_enabled`: должна ли LDAP-учетка быть включена сейчас;
- `ldap_enabled`: находится ли LDAP-учетка в `LDAP_GUESTS_ACTIVE_BASE`;
- `ldap_username`: кэш `sAMAccountName`;
- `ldap_upn`: кэш `userPrincipalName`;
- `ldap_last_synced_at`;
@ -286,18 +287,93 @@ LDAP-поля и состояние:
Гостевые учетные записи должны создаваться в отдельном контейнере AD/LDAP, отличном от сотрудников.
В Active Directory внутри гостевого контейнера должны быть две OU:
- `Active`: активные гостевые учетные записи;
- `Deactivated`: деактивированные гостевые учетные записи.
Нужно добавить настройку:
- `LDAP_GUESTS_BASE`, например `OU=Guests,OU=company,DC=robotail,DC=local`.
- `LDAP_GUESTS_ACTIVE_BASE`, по умолчанию `OU=Active,<LDAP_GUESTS_BASE>`.
- `LDAP_GUESTS_DEACTIVATED_BASE`, по умолчанию `OU=Deactivated,<LDAP_GUESTS_BASE>`.
- `GUESTS_ID_PREFIX`, по умолчанию `9`.
- `GUESTS_ID_LENGTH`, по умолчанию `15`.
- `GUESTS_ID_MIN`, по умолчанию `900000000000000`.
- `GUESTS_ID_START`, по умолчанию `900000000000001`.
- `GUESTS_ID_MAX`, по умолчанию `999999999999999`.
Нельзя создавать для гостя `employees.Employee`. Иначе гость сможет попасть в существующую логику аутентификации, прав, сотрудников и отделов.
### 10.1.1. `Guest.id` и `employeeNumber`
Так как внешние сервисы читают идентификатор только из LDAP-поля `employeeNumber`, первичный ключ `Guest.id` должен сразу соответствовать значению, которое будет записано в LDAP.
Формат `Guest.id`:
- только цифры;
- ровно 15 знаков;
- начинается с гостевого префикса `9`;
- генерируется один раз и не меняется;
- не переиспользуется после удаления или деактивации гостя.
В LDAP `employeeNumber` записывается как строковое представление `Guest.id`:
```text
employeeNumber = str(guest.id)
```
Рекомендуемая формула:
```text
guest.id = next_guest_id_from_sequence()
```
Примеры:
```text
900000000000001
900000000000025
900000000123456
```
Рекомендуемая реализация:
- использовать `BigIntegerField(primary_key=True)` или `BigAutoField` с PostgreSQL sequence;
- старт sequence: `900000000000001`;
- верхняя граница: `999999999999999`;
- значение входит в JavaScript safe integer, поэтому frontend может безопасно передавать его как number, но API может сериализовать его строкой, если потребуется совместимость.
Перед записью в LDAP система должна проверить уникальность `str(Guest.id)` в `employeeNumber` по всему AD, включая сотрудников, активных гостей и деактивированных гостей.
### 10.1.2. Разделение диапазонов `Employee.id` и `Guest.id`
Так как сотрудники и гости используют одно LDAP-поле `employeeNumber`, диапазоны их первичных ключей не должны пересекаться.
Диапазоны:
- сотрудники: `Employee.id < GUESTS_ID_MIN`;
- гости: `GUESTS_ID_START <= Guest.id <= GUESTS_ID_MAX`;
- значение `GUESTS_ID_MIN` резервируется как граница диапазона и не используется для новых гостей.
Требования к реализации:
- добавить DB-level `CheckConstraint` для `Employee`, запрещающий `id >= GUESTS_ID_MIN`;
- настроить PostgreSQL sequence для `employees.Employee.id` так, чтобы она не могла выдать значение из гостевого диапазона;
- добавить DB-level `CheckConstraint` для `Guest`, разрешающий только гостевой диапазон;
- настроить PostgreSQL sequence для `Guest.id` со стартом `GUESTS_ID_START` и верхней границей `GUESTS_ID_MAX`;
- при ручном импорте/создании сотрудника с заданным `id` backend должен возвращать ошибку валидации, если `id >= GUESTS_ID_MIN`;
- перед записью в LDAP проверять уникальность `employeeNumber` по всему AD, даже при наличии DB constraints.
Такой запрет возможен, потому что в проекте используется `BigAutoField`, а диапазон 15-значных гостевых ID помещается в PostgreSQL `bigint`.
### 10.2. Модель синхронизации
Использовать существующий подход:
- `LdapSyncState(model="guest", object_pk=str(guest.pk))`;
- `LdapSyncQueue` для retry LDAP-операций;
- `guests.ldap.orm_models.LdapGuestUser` для ORM-обновления и перемещения существующих гостевых AD-учеток;
- отдельные операции очереди: `guest_save`, `guest_disable`, `guest_delete` или `guest_sync_access`;
- отдельный сервис `guests.ldap.services.GuestLdapService`.
@ -310,13 +386,14 @@ LDAP-поля и состояние:
1. Проверить, что визит в статусе `approved`.
2. Проверить, что текущая дата попадает в период доступа или доступ должен быть подготовлен заранее.
3. Найти существующий LDAP DN гостя через `LdapSyncState`.
4. Если DN нет, создать LDAP `user` в `LDAP_GUESTS_BASE`.
5. Записать атрибут связи, например `employeeNumber` или отдельный настраиваемый атрибут со значением `guest:<id>`.
4. Если DN нет, создать LDAP `user` в `LDAP_GUESTS_ACTIVE_BASE` или `LDAP_GUESTS_DEACTIVATED_BASE` в зависимости от текущего состояния доступа.
5. Записать в LDAP `employeeNumber` значение `str(Guest.id)`.
6. Установить `sAMAccountName`, `userPrincipalName`, `displayName`, `givenName`, `sn`, `mail`, `telephoneNumber`, `description`.
7. Установить состояние учетной записи:
- enabled, если доступ уже активен;
- disabled, если визит одобрен заранее, но дата начала еще не наступила.
8. Обновить `LdapSyncState` и поля кэша в `Guest`.
7. Всегда устанавливать AD-учетку disabled через `userAccountControl`, чтобы гость не мог входить в систему под этой учеткой.
8. Состояние гостевого доступа отражать только расположением учетной записи:
- `LDAP_GUESTS_ACTIVE_BASE`, если доступ уже активен;
- `LDAP_GUESTS_DEACTIVATED_BASE`, если доступ еще не начался или уже неактуален.
9. Обновить `LdapSyncState` и поля кэша в `Guest`.
### 10.4. Запрет входа гостя в EUSRR
@ -342,12 +419,14 @@ LDAP-учетка гостя должна быть disabled, если:
Если у гостя несколько одобренных визитов, деактивировать LDAP-учетку можно только когда нет ни одного актуального визита, требующего активного доступа.
`userAccountControl` для гостевой учетки всегда должен оставаться disabled. При деактивации учетку нужно переместить в `LDAP_GUESTS_DEACTIVATED_BASE`. При повторной активации учетку нужно переместить обратно в `LDAP_GUESTS_ACTIVE_BASE`, не включая саму AD-учетку.
### 10.6. Планировщик
Нужны Celery-задачи:
- `guests.tasks.activate_due_guest_visits`: включает учетки, у которых наступило начало доступа;
- `guests.tasks.expire_guest_visits`: переводит истекшие визиты в `expired` и выключает учетки;
- `guests.tasks.activate_due_guest_visits`: переносит учетки, у которых наступило начало доступа, в `LDAP_GUESTS_ACTIVE_BASE`;
- `guests.tasks.expire_guest_visits`: переводит истекшие визиты в `expired` и переносит учетки в `LDAP_GUESTS_DEACTIVATED_BASE`;
- `guests.tasks.process_guest_ldap_queue` или расширение существующей `employees.tasks.process_ldap_queue`;
- периодичность: минимум раз в минуту, аналогично существующей LDAP retry-задаче.
@ -493,7 +572,7 @@ LDAP-учетка гостя должна быть disabled, если:
- `created_from`, `created_to`;
- `access_from`, `access_to`;
- `unlimited=true|false`;
- `ldap_enabled=true|false`;
- `ldap_enabled=true|false`: фильтр по нахождению гостевой LDAP-учетки в Active OU;
- `q`: поиск по ФИО гостя, организации, цели, email/phone;
- `ordering=-created_at|created_at|access_starts_at|-access_starts_at|access_expires_at|-access_expires_at`.
@ -658,10 +737,16 @@ Write serializer должен принимать:
- модели `Guest`, `GuestVisit`, `GuestVisitEvent`;
- настройки:
- `LDAP_GUESTS_BASE`;
- `LDAP_GUESTS_ACTIVE_BASE`;
- `LDAP_GUESTS_DEACTIVATED_BASE`;
- `GUESTS_ID_PREFIX`;
- `GUESTS_ID_LENGTH`;
- `GUESTS_ID_MIN`;
- `GUESTS_ID_START`;
- `GUESTS_ID_MAX`;
- `GUESTS_REQUIRE_ID_DOCUMENT`;
- `GUESTS_NOTIFY_ON_EXPIRATION`;
- `GUESTS_AUTO_DISABLE_ON_INVITER_INACTIVE`;
- `GUESTS_LDAP_LINK_ATTR`;
- регистрацию app в `INSTALLED_APPS`;
- регистрацию URL в `api/v1/urls.py`;
- периодические Celery-задачи в `CELERY_BEAT_SCHEDULE`;

View File

@ -0,0 +1,627 @@
import pytest
from datetime import timedelta
from django.contrib.auth.models import Permission
from django.contrib.contenttypes.models import ContentType
from django.utils import timezone
from documents.models import Document
from guests.constants import GuestVisitStatus
from guests.models import Guest, GuestVisit
def _results(payload):
return payload["results"] if isinstance(payload, dict) and "results" in payload else payload
@pytest.mark.django_db
def test_user_can_create_and_submit_guest_visit(auth_client_factory, user_factory):
user = user_factory()
client = auth_client_factory(user)
response = client.post(
"/api/v1/guests/visits/",
{
"guest": {
"first_name": "Maria",
"last_name": "Guest",
"organization": "Vendor",
},
"purpose": "Wi-Fi access",
"date_from": "2026-06-15",
"date_to": "2026-06-18",
"all_day": True,
},
format="json",
)
assert response.status_code == 201
visit_id = response.json()["id"]
visit = GuestVisit.objects.get(pk=visit_id)
assert visit.inviter == user
assert timezone.localtime(visit.access_expires_at).date().isoformat() == "2026-06-19"
submit = client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
assert submit.status_code == 200
assert submit.json()["status"] == GuestVisitStatus.PENDING
@pytest.mark.django_db
def test_admin_can_approve_guest_visit(auth_client_factory, user_factory):
user = user_factory()
admin = user_factory(staff=True)
user_client = auth_client_factory(user)
admin_client = auth_client_factory(admin)
created = user_client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Alex", "last_name": "Guest"},
"purpose": "Directory access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
user_client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
response = admin_client.post(
f"/api/v1/guests/visits/{visit_id}/approve/",
{"comment": "ok"},
format="json",
)
assert response.status_code == 200
assert response.json()["status"] == GuestVisitStatus.APPROVED
@pytest.mark.django_db
def test_admin_can_reject_approved_guest_visit(auth_client_factory, user_factory):
user = user_factory()
admin = user_factory(staff=True)
user_client = auth_client_factory(user)
admin_client = auth_client_factory(admin)
created = user_client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Reject", "last_name": "Approved"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
user_client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
admin_client.post(f"/api/v1/guests/visits/{visit_id}/approve/")
response = admin_client.post(
f"/api/v1/guests/visits/{visit_id}/reject/",
{"comment": "Stop access"},
format="json",
)
assert response.status_code == 200
assert response.json()["status"] == GuestVisitStatus.REJECTED
assert GuestVisit.objects.get(pk=visit_id).events.filter(
event_type="decision_changed"
).exists()
@pytest.mark.django_db
def test_admin_can_request_info_author_can_provide_info(
auth_client_factory,
user_factory,
):
user = user_factory()
admin = user_factory(staff=True)
user_client = auth_client_factory(user)
admin_client = auth_client_factory(admin)
created = user_client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Need", "last_name": "Info"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
user_client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
request_info = admin_client.post(
f"/api/v1/guests/visits/{visit_id}/request-info/",
{"comment": "Add sponsor details"},
format="json",
)
assert request_info.status_code == 200
assert request_info.json()["status"] == GuestVisitStatus.NEEDS_INFO
provide_info = user_client.post(
f"/api/v1/guests/visits/{visit_id}/provide-info/",
{"comment": "Sponsor added"},
format="json",
)
assert provide_info.status_code == 200
assert provide_info.json()["status"] == GuestVisitStatus.PENDING
visit = GuestVisit.objects.get(pk=visit_id)
assert visit.events.filter(event_type="needs_info_requested").exists()
assert visit.events.filter(event_type="info_provided").exists()
@pytest.mark.django_db
def test_admin_can_revoke_guest_visit(auth_client_factory, user_factory):
user = user_factory()
admin = user_factory(staff=True)
user_client = auth_client_factory(user)
admin_client = auth_client_factory(admin)
created = user_client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Revoke", "last_name": "Guest"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
user_client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
admin_client.post(f"/api/v1/guests/visits/{visit_id}/approve/")
response = admin_client.post(
f"/api/v1/guests/visits/{visit_id}/revoke/",
{"comment": "No longer needed"},
format="json",
)
assert response.status_code == 200
assert response.json()["status"] == GuestVisitStatus.REVOKED
assert GuestVisit.objects.get(pk=visit_id).events.filter(event_type="revoked").exists()
@pytest.mark.django_db
def test_author_can_cancel_guest_visit(auth_client_factory, user_factory):
user = user_factory()
client = auth_client_factory(user)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Cancel", "last_name": "Guest"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
response = client.post(
f"/api/v1/guests/visits/{visit_id}/cancel/",
{"comment": "not needed"},
format="json",
)
assert response.status_code == 200
assert response.json()["status"] == GuestVisitStatus.CANCELLED
@pytest.mark.django_db
def test_invalid_api_transitions_return_400(auth_client_factory, user_factory):
user = user_factory()
admin = user_factory(staff=True)
client = auth_client_factory(user)
admin_client = auth_client_factory(admin)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Bad", "last_name": "Transition"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
approve = admin_client.post(f"/api/v1/guests/visits/{visit_id}/approve/")
request_info = admin_client.post(
f"/api/v1/guests/visits/{visit_id}/request-info/",
{"comment": "details"},
format="json",
)
assert approve.status_code == 400
assert request_info.status_code == 400
@pytest.mark.django_db
def test_non_admin_cannot_approve_guest_visit(auth_client_factory, user_factory):
author = user_factory()
other = user_factory()
client = auth_client_factory(author)
other_client = auth_client_factory(other)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "No", "last_name": "Approve"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
response = other_client.post(f"/api/v1/guests/visits/{visit_id}/approve/")
assert response.status_code in {403, 404}
@pytest.mark.django_db
def test_view_all_permission_does_not_allow_decision(
auth_client_factory,
user_factory,
):
author = user_factory()
viewer = user_factory()
decider = user_factory()
ct = ContentType.objects.get_for_model(GuestVisit)
viewer.user_permissions.add(
Permission.objects.get(content_type=ct, codename="view_all_guestvisit")
)
decider.user_permissions.add(
Permission.objects.get(content_type=ct, codename="decide_guestvisit")
)
author_client = auth_client_factory(author)
viewer_client = auth_client_factory(viewer)
decider_client = auth_client_factory(decider)
created = author_client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Perm", "last_name": "Guest"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
author_client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
assert viewer_client.get(f"/api/v1/guests/visits/{visit_id}/").status_code == 200
denied = viewer_client.post(f"/api/v1/guests/visits/{visit_id}/approve/")
assert denied.status_code == 403
approved = decider_client.post(f"/api/v1/guests/visits/{visit_id}/approve/")
assert approved.status_code == 200
@pytest.mark.django_db
def test_guest_visit_comments(auth_client_factory, user_factory):
user = user_factory()
client = auth_client_factory(user)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Comment", "last_name": "Guest"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
response = client.post(
f"/api/v1/guests/visits/{visit_id}/comments/",
{"text": "Need badge too"},
format="json",
)
assert response.status_code == 201
listed = client.get(f"/api/v1/guests/visits/{visit_id}/comments/")
assert listed.status_code == 200
assert listed.json()[0]["text"] == "Need badge too"
@pytest.mark.django_db
def test_guest_visit_comment_delete_permissions(auth_client_factory, user_factory):
author = user_factory()
other = user_factory()
author_client = auth_client_factory(author)
other_client = auth_client_factory(other)
created = author_client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Delete", "last_name": "Comment"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
comment = author_client.post(
f"/api/v1/guests/visits/{visit_id}/comments/",
{"text": "Temporary"},
format="json",
).json()
denied = other_client.delete(
f"/api/v1/guests/visits/{visit_id}/comments/{comment['id']}/"
)
deleted = author_client.delete(
f"/api/v1/guests/visits/{visit_id}/comments/{comment['id']}/"
)
assert denied.status_code in {403, 404}
assert deleted.status_code == 204
listed = author_client.get(f"/api/v1/guests/visits/{visit_id}/comments/")
assert listed.json() == []
@pytest.mark.django_db
def test_guest_visit_documents_can_be_attached_and_removed(
auth_client_factory,
user_factory,
):
user = user_factory()
client = auth_client_factory(user)
document = Document.objects.create(title="Guest passport", uploaded_by=user)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Doc", "last_name": "Guest"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
attach = client.post(
f"/api/v1/guests/visits/{visit_id}/documents/",
{"document_id": document.id},
format="json",
)
assert attach.status_code == 200
assert attach.json()["documents"][0]["id"] == document.id
remove = client.delete(
f"/api/v1/guests/visits/{visit_id}/documents/{document.id}/"
)
assert remove.status_code == 200
assert remove.json()["documents"] == []
visit = GuestVisit.objects.get(pk=visit_id)
assert visit.events.filter(event_type="document_attached").exists()
assert visit.events.filter(event_type="document_removed").exists()
@pytest.mark.django_db
def test_guest_visit_create_and_update_document_ids(auth_client_factory, user_factory):
user = user_factory()
client = auth_client_factory(user)
first = Document.objects.create(title="First document", uploaded_by=user)
second = Document.objects.create(title="Second document", uploaded_by=user)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "DocIds", "last_name": "Guest"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
"document_ids": [first.id],
},
format="json",
)
visit_id = created.json()["id"]
assert [doc["id"] for doc in created.json()["documents"]] == [first.id]
updated = client.patch(
f"/api/v1/guests/visits/{visit_id}/",
{"document_ids": [second.id], "purpose": "Updated"},
format="json",
)
assert updated.status_code == 200
assert [doc["id"] for doc in updated.json()["documents"]] == [second.id]
assert updated.json()["purpose"] == "Updated"
@pytest.mark.django_db
def test_required_document_setting_blocks_submit_without_document(
settings,
auth_client_factory,
user_factory,
):
settings.GUESTS_REQUIRE_ID_DOCUMENT = True
user = user_factory()
client = auth_client_factory(user)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Required", "last_name": "Document"},
"purpose": "Access",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
response = client.post(f"/api/v1/guests/visits/{created.json()['id']}/submit/")
assert response.status_code == 400
@pytest.mark.django_db
def test_author_can_update_only_draft_or_needs_info(auth_client_factory, user_factory):
user = user_factory()
admin = user_factory(staff=True)
client = auth_client_factory(user)
admin_client = auth_client_factory(admin)
created = client.post(
"/api/v1/guests/visits/",
{
"guest": {"first_name": "Update", "last_name": "Guest"},
"purpose": "Original",
"date_from": "2026-06-15",
"date_to": "2026-06-15",
},
format="json",
)
visit_id = created.json()["id"]
draft_update = client.patch(
f"/api/v1/guests/visits/{visit_id}/",
{"purpose": "Draft update"},
format="json",
)
client.post(f"/api/v1/guests/visits/{visit_id}/submit/")
pending_update = client.patch(
f"/api/v1/guests/visits/{visit_id}/",
{"purpose": "Pending update"},
format="json",
)
admin_client.post(
f"/api/v1/guests/visits/{visit_id}/request-info/",
{"comment": "fix"},
format="json",
)
needs_info_update = client.patch(
f"/api/v1/guests/visits/{visit_id}/",
{"purpose": "Needs info update"},
format="json",
)
assert draft_update.status_code == 200
assert pending_update.status_code == 403
assert needs_info_update.status_code == 200
@pytest.mark.django_db
def test_guest_visit_list_scopes_filters_and_search(auth_client_factory, user_factory):
admin = user_factory(staff=True)
author = user_factory()
other = user_factory()
admin_client = auth_client_factory(admin)
author_client = auth_client_factory(author)
vendor_guest = Guest.objects.create(
first_name="Vendor",
last_name="Alpha",
organization="Acme",
email="vendor@example.test",
)
other_guest = Guest.objects.create(
first_name="Other",
last_name="Beta",
organization="OtherOrg",
)
pending = GuestVisit.objects.create(
guest=vendor_guest,
inviter=author,
purpose="Network access",
status=GuestVisitStatus.PENDING,
access_starts_at=timezone.now(),
access_expires_at=timezone.now() + timedelta(days=1),
unlimited=False,
)
GuestVisit.objects.create(
guest=other_guest,
inviter=other,
purpose="Other access",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now(),
access_expires_at=timezone.now() + timedelta(days=1),
unlimited=True,
)
own = author_client.get("/api/v1/guests/visits/?scope=all")
pending_admin = admin_client.get("/api/v1/guests/visits/?scope=pending_decision")
search_admin = admin_client.get("/api/v1/guests/visits/?q=Acme")
filtered_admin = admin_client.get(
f"/api/v1/guests/visits/?guest_id={vendor_guest.id}&"
f"inviter_id={author.id}&status=pending&unlimited=false"
)
assert [item["id"] for item in _results(own.json())] == [pending.id]
assert [item["id"] for item in _results(pending_admin.json())] == [pending.id]
assert [item["id"] for item in _results(search_admin.json())] == [pending.id]
assert [item["id"] for item in _results(filtered_admin.json())] == [pending.id]
@pytest.mark.django_db
def test_guest_admin_endpoints_permissions_and_actions(
auth_client_factory,
user_factory,
):
admin = user_factory(staff=True)
regular = user_factory()
admin_client = auth_client_factory(admin)
regular_client = auth_client_factory(regular)
guest = Guest.objects.create(
first_name="Admin",
last_name="Guest",
organization="Acme",
)
denied = regular_client.get("/api/v1/guests/")
listed = admin_client.get("/api/v1/guests/?q=Acme")
searched = admin_client.get("/api/v1/guests/search/?q=Admin")
updated = admin_client.patch(
f"/api/v1/guests/{guest.id}/",
{"comment": "checked", "organization": "Updated Org"},
format="json",
)
deactivated = admin_client.post(f"/api/v1/guests/{guest.id}/deactivate/")
assert denied.status_code == 403
assert [item["id"] for item in _results(listed.json())] == [guest.id]
assert searched.json()[0]["id"] == guest.id
assert updated.status_code == 200
assert updated.json()["organization"] == "Updated Org"
assert deactivated.status_code == 200
assert deactivated.json()["is_active"] is False
@pytest.mark.django_db
def test_manage_guestaccount_permission_can_sync_guest_ldap(
auth_client_factory,
user_factory,
):
manager = user_factory()
viewer = user_factory()
ct = ContentType.objects.get_for_model(GuestVisit)
manager.user_permissions.add(
Permission.objects.get(content_type=ct, codename="manage_guestaccount")
)
viewer.user_permissions.add(
Permission.objects.get(content_type=ct, codename="view_all_guestvisit")
)
guest = Guest.objects.create(first_name="Sync", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=user_factory(),
purpose="Sync LDAP",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now(),
access_expires_at=timezone.now() + timedelta(days=1),
)
manager_client = auth_client_factory(manager)
viewer_client = auth_client_factory(viewer)
allowed = manager_client.post(f"/api/v1/guests/visits/{visit.id}/sync-ldap/")
denied = viewer_client.post(f"/api/v1/guests/visits/{visit.id}/sync-ldap/")
assert allowed.status_code == 200
assert denied.status_code == 403

View File

@ -0,0 +1,483 @@
from contextlib import contextmanager
from datetime import timedelta
import pytest
from django.utils import timezone
from employees.ldap.services.constants import UserAccountControl
from employees.models import LdapSyncQueue, LdapSyncState
from employees.tasks import process_ldap_queue_item
from guests.constants import GuestVisitStatus
from guests.models import Guest, GuestVisit
from guests.services import GuestLdapService, GuestVisitWorkflow
from guests.tasks import execute_guest_queue_operation
class FakeLdapConnection:
def __init__(self):
self.entries = []
self.add_calls = []
self.modify_calls = []
self.result = {}
def search(self, *args, **kwargs):
self.entries = []
return True
def add(self, dn, object_classes, attrs):
self.add_calls.append((dn, object_classes, attrs))
return True
def modify(self, dn, changes):
self.modify_calls.append((dn, changes))
return True
def unbind(self):
return None
class FakeLdapEntry:
def __init__(self, dn):
self.entry_dn = dn
class ConflictingEmployeeNumberConnection(FakeLdapConnection):
def search(self, *args, **kwargs):
self.entries = [FakeLdapEntry("CN=Employee,OU=Users,DC=example,DC=test")]
return True
class FakeLdapGuestUser:
def __init__(self, dn):
self.dn = dn
self.moves = []
self.saved = False
def move_to(self, new_base_dn):
self.moves.append(new_base_dn)
rdn = self.dn.split(",", 1)[0]
self.dn = f"{rdn},{new_base_dn}"
def apply_guest(self, guest):
self.sam_account_name = f"g{guest.id}"
self.user_principal_name = f"g{guest.id}@guest.local"
self.given_name = guest.first_name or ""
self.sn = guest.last_name or "."
self.display_name = guest.full_name or f"Guest {guest.id}"
self.mail = guest.email or ""
self.telephone_number = guest.phone or ""
self.employee_number = str(guest.id)
self.description = f"Guest account for {guest.organization}".strip()
self.user_account_control = UserAccountControl.DISABLED
def save(self):
self.saved = True
@pytest.mark.django_db
def test_guest_ldap_service_writes_guest_id_to_employee_number(
settings,
monkeypatch,
user_factory,
):
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
settings.LDAP_GUESTS_ACTIVE_BASE = "OU=Active,OU=Guests,DC=example,DC=test"
settings.LDAP_GUESTS_DEACTIVATED_BASE = (
"OU=Deactivated,OU=Guests,DC=example,DC=test"
)
settings.LDAP_BASE_DN = "DC=example,DC=test"
fake_conn = FakeLdapConnection()
@contextmanager
def fake_ldap():
yield fake_conn
monkeypatch.setattr("guests.ldap.orm_models._ldap", fake_ldap)
monkeypatch.setattr(
"guests.ldap.orm_models.ensure_container_exists",
lambda *a, **k: None,
)
monkeypatch.setattr(
"guests.ldap.orm_models.UserPasswordService.set_password",
lambda *a, **k: None,
)
inviter = user_factory()
guest = Guest.objects.create(first_name="Ldap", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="LDAP test",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestLdapService().sync_guest_for_visit(visit)
assert fake_conn.add_calls
attrs = fake_conn.add_calls[0][2]
assert attrs["employeeNumber"] == str(guest.id)
assert attrs["userAccountControl"] == UserAccountControl.DISABLED
assert not fake_conn.modify_calls
guest.refresh_from_db()
assert guest.ldap_enabled is True
assert guest.ldap_username == f"g{guest.id}"
@pytest.mark.django_db
def test_guest_ldap_service_disables_and_moves_to_deactivated(
settings,
monkeypatch,
):
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
settings.LDAP_GUESTS_ACTIVE_BASE = "OU=Active,OU=Guests,DC=example,DC=test"
settings.LDAP_GUESTS_DEACTIVATED_BASE = (
"OU=Deactivated,OU=Guests,DC=example,DC=test"
)
settings.LDAP_BASE_DN = "DC=example,DC=test"
fake_conn = FakeLdapConnection()
@contextmanager
def fake_ldap():
yield fake_conn
guest = Guest.objects.create(
first_name="Disable",
last_name="Guest",
is_active=False,
ldap_enabled=True,
ldap_username="g900000000000001",
)
active_dn = (
f"CN=g{guest.id},OU=Active,OU=Guests,DC=example,DC=test"
)
LdapSyncState.objects.create(
model="guest",
object_pk=str(guest.id),
ldap_dn=active_dn,
)
fake_orm_guest = FakeLdapGuestUser(active_dn)
monkeypatch.setattr("guests.ldap.orm_models._ldap", fake_ldap)
monkeypatch.setattr(
"guests.ldap.orm_models.ensure_container_exists",
lambda *a, **k: None,
)
monkeypatch.setattr(
"guests.ldap.orm_models.LdapGuestUser.objects.get",
lambda **kwargs: fake_orm_guest,
)
GuestLdapService().disable_guest(guest)
guest.refresh_from_db()
state = LdapSyncState.objects.get(model="guest", object_pk=str(guest.id))
assert state.ldap_dn == (
f"CN=g{guest.id},OU=Deactivated,OU=Guests,DC=example,DC=test"
)
assert guest.ldap_enabled is False
assert fake_orm_guest.moves == ["OU=Deactivated,OU=Guests,DC=example,DC=test"]
assert fake_orm_guest.saved is True
assert fake_orm_guest.employee_number == str(guest.id)
assert fake_orm_guest.user_account_control == UserAccountControl.DISABLED
@pytest.mark.django_db
def test_guest_ldap_service_moves_existing_guest_to_active_ou_without_enabling(
settings,
monkeypatch,
user_factory,
):
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
settings.LDAP_GUESTS_ACTIVE_BASE = "OU=Active,OU=Guests,DC=example,DC=test"
settings.LDAP_GUESTS_DEACTIVATED_BASE = (
"OU=Deactivated,OU=Guests,DC=example,DC=test"
)
settings.LDAP_BASE_DN = "DC=example,DC=test"
fake_conn = FakeLdapConnection()
@contextmanager
def fake_ldap():
yield fake_conn
inviter = user_factory()
guest = Guest.objects.create(first_name="Active", last_name="Guest")
deactivated_dn = f"CN=g{guest.id},OU=Deactivated,OU=Guests,DC=example,DC=test"
LdapSyncState.objects.create(
model="guest",
object_pk=str(guest.id),
ldap_dn=deactivated_dn,
)
GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Active LDAP move",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
fake_orm_guest = FakeLdapGuestUser(deactivated_dn)
monkeypatch.setattr("guests.ldap.orm_models._ldap", fake_ldap)
monkeypatch.setattr(
"guests.ldap.orm_models.ensure_container_exists",
lambda *a, **k: None,
)
monkeypatch.setattr(
"guests.ldap.orm_models.LdapGuestUser.objects.get",
lambda **kwargs: fake_orm_guest,
)
GuestLdapService().sync_guest(guest)
guest.refresh_from_db()
state = LdapSyncState.objects.get(model="guest", object_pk=str(guest.id))
assert fake_orm_guest.moves == ["OU=Active,OU=Guests,DC=example,DC=test"]
assert fake_orm_guest.saved is True
assert fake_orm_guest.user_account_control == UserAccountControl.DISABLED
assert fake_conn.modify_calls == []
assert state.ldap_dn == f"CN=g{guest.id},OU=Active,OU=Guests,DC=example,DC=test"
assert guest.ldap_enabled is True
@pytest.mark.django_db
def test_guest_ldap_service_keeps_active_ou_when_other_active_visit_exists(
settings,
monkeypatch,
user_factory,
):
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
settings.LDAP_GUESTS_ACTIVE_BASE = "OU=Active,OU=Guests,DC=example,DC=test"
settings.LDAP_GUESTS_DEACTIVATED_BASE = (
"OU=Deactivated,OU=Guests,DC=example,DC=test"
)
settings.LDAP_BASE_DN = "DC=example,DC=test"
fake_conn = FakeLdapConnection()
@contextmanager
def fake_ldap():
yield fake_conn
inviter = user_factory()
guest = Guest.objects.create(
first_name="Multi",
last_name="Visit",
ldap_enabled=True,
)
active_dn = f"CN=g{guest.id},OU=Active,OU=Guests,DC=example,DC=test"
LdapSyncState.objects.create(
model="guest",
object_pk=str(guest.id),
ldap_dn=active_dn,
)
ending_visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Ending",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(days=2),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Still active",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(days=1),
access_expires_at=timezone.now() + timedelta(days=2),
)
fake_orm_guest = FakeLdapGuestUser(active_dn)
monkeypatch.setattr("guests.ldap.orm_models._ldap", fake_ldap)
monkeypatch.setattr(
"guests.ldap.orm_models.ensure_container_exists",
lambda *a, **k: None,
)
monkeypatch.setattr(
"guests.ldap.orm_models.LdapGuestUser.objects.get",
lambda **kwargs: fake_orm_guest,
)
GuestVisitWorkflow.revoke(ending_visit, actor=inviter)
guest.refresh_from_db()
assert fake_orm_guest.moves == []
assert guest.ldap_enabled is True
@pytest.mark.django_db
def test_execute_guest_queue_operation_dispatches_sync(monkeypatch):
guest = Guest.objects.create(first_name="Queue", last_name="Guest")
called = {}
def fake_sync(self, synced_guest, *, enqueue_on_error=True, raise_on_error=False):
called["guest_id"] = synced_guest.id
called["enqueue_on_error"] = enqueue_on_error
called["raise_on_error"] = raise_on_error
monkeypatch.setattr("guests.services.GuestLdapService.sync_guest", fake_sync)
execute_guest_queue_operation("guest_sync", {"guest_id": str(guest.id)})
assert called["guest_id"] == guest.id
assert called["enqueue_on_error"] is False
assert called["raise_on_error"] is True
@pytest.mark.django_db
def test_process_ldap_queue_item_completes_guest_sync(
settings,
monkeypatch,
):
settings.LDAP_ENABLED = True
guest = Guest.objects.create(first_name="Queue", last_name="Complete")
item = LdapSyncQueue.objects.create(
operation="guest_sync",
model_name="guest",
object_pk=str(guest.id),
payload={"guest_id": str(guest.id), "_operation": "guest_sync"},
)
called = {}
def fake_sync(self, synced_guest, *, enqueue_on_error=True, raise_on_error=False):
called["guest_id"] = synced_guest.id
called["raise_on_error"] = raise_on_error
monkeypatch.setattr("guests.services.GuestLdapService.sync_guest", fake_sync)
process_ldap_queue_item.run(item.id)
item.refresh_from_db()
assert item.status == LdapSyncQueue.Status.COMPLETED
assert called == {"guest_id": guest.id, "raise_on_error": True}
@pytest.mark.django_db
def test_process_ldap_queue_item_schedules_retry_on_guest_sync_error(
settings,
monkeypatch,
):
settings.LDAP_ENABLED = True
guest = Guest.objects.create(first_name="Queue", last_name="Retry")
item = LdapSyncQueue.objects.create(
operation="guest_sync",
model_name="guest",
object_pk=str(guest.id),
payload={"guest_id": str(guest.id), "_operation": "guest_sync"},
)
def failing_sync(self, synced_guest, *, enqueue_on_error=True, raise_on_error=False):
raise RuntimeError("retry me")
monkeypatch.setattr("guests.services.GuestLdapService.sync_guest", failing_sync)
process_ldap_queue_item.run(item.id)
item.refresh_from_db()
assert item.status == LdapSyncQueue.Status.PENDING
assert item.attempts == 1
assert "retry me" in item.last_error
assert item.next_retry_at is not None
@pytest.mark.django_db
def test_guest_ldap_employee_number_conflict_queues_retry(
settings,
monkeypatch,
user_factory,
):
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
settings.LDAP_GUESTS_ACTIVE_BASE = "OU=Active,OU=Guests,DC=example,DC=test"
settings.LDAP_GUESTS_DEACTIVATED_BASE = (
"OU=Deactivated,OU=Guests,DC=example,DC=test"
)
settings.LDAP_BASE_DN = "DC=example,DC=test"
conflict_conn = ConflictingEmployeeNumberConnection()
@contextmanager
def fake_ldap():
yield conflict_conn
monkeypatch.setattr("guests.ldap.orm_models._ldap", fake_ldap)
monkeypatch.setattr(
"guests.ldap.orm_models.ensure_container_exists",
lambda *a, **k: None,
)
inviter = user_factory()
admin = user_factory(staff=True)
guest = Guest.objects.create(first_name="Conflict", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Conflict",
status=GuestVisitStatus.PENDING,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestVisitWorkflow.approve(visit, actor=admin)
guest.refresh_from_db()
assert "employeeNumber" in guest.ldap_last_error
assert LdapSyncQueue.objects.filter(
operation="guest_sync",
model_name="guest",
object_pk=str(guest.id),
).exists()
@pytest.mark.django_db
def test_approve_keeps_status_and_queues_retry_on_ldap_error(
settings,
monkeypatch,
user_factory,
):
settings.LDAP_ENABLED = True
settings.LDAP_WRITE_ENABLED = True
settings.LDAP_GUESTS_ACTIVE_BASE = "OU=Active,OU=Guests,DC=example,DC=test"
settings.LDAP_GUESTS_DEACTIVATED_BASE = (
"OU=Deactivated,OU=Guests,DC=example,DC=test"
)
settings.LDAP_BASE_DN = "DC=example,DC=test"
@contextmanager
def failing_ldap():
raise RuntimeError("LDAP unavailable")
yield
monkeypatch.setattr("guests.ldap.orm_models._ldap", failing_ldap)
inviter = user_factory()
admin = user_factory(staff=True)
guest = Guest.objects.create(first_name="Retry", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Retry LDAP",
status=GuestVisitStatus.PENDING,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestVisitWorkflow.approve(visit, actor=admin)
visit.refresh_from_db()
guest.refresh_from_db()
assert visit.status == GuestVisitStatus.APPROVED
assert "LDAP unavailable" in guest.ldap_last_error
assert visit.events.filter(event_type="ldap_failed").exists()
assert LdapSyncQueue.objects.filter(
operation="guest_sync",
model_name="guest",
object_pk=str(guest.id),
).exists()

View File

@ -0,0 +1,285 @@
from datetime import timedelta
import pytest
from django.core.exceptions import ValidationError
from django.db import IntegrityError
from django.utils import timezone
from guests.constants import GUEST_ID_MIN, GUEST_ID_START, GuestVisitStatus
from guests.models import Guest, GuestVisit
from guests.services import GuestVisitWorkflow
from guests.tasks import (
activate_due_guest_visits,
detect_inactive_inviters,
expire_guest_visits,
)
@pytest.mark.django_db
def test_guest_id_generated_in_guest_range(user_factory):
user = user_factory()
guest = Guest.objects.create(
first_name="Ivan",
last_name="Guest",
created_by=user,
)
assert guest.id == GUEST_ID_START
assert len(str(guest.id)) == 15
assert str(guest.id).startswith("9")
@pytest.mark.django_db
def test_employee_id_cannot_enter_guest_range(user_factory):
with pytest.raises(IntegrityError):
user_factory(id=GUEST_ID_MIN, email="blocked@example.test")
@pytest.mark.django_db
def test_guest_visit_workflow_records_events(user_factory):
inviter = user_factory()
admin = user_factory(staff=True)
guest = Guest.objects.create(first_name="Ann", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="External integration access",
access_starts_at=timezone.now() - timedelta(minutes=5),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestVisitWorkflow.submit(visit, actor=inviter)
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.PENDING
GuestVisitWorkflow.approve(visit, actor=admin, comment="ok")
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.APPROVED
assert visit.decided_by == admin
event_types = set(visit.events.values_list("event_type", flat=True))
assert {"submitted", "approved"}.issubset(event_types)
@pytest.mark.django_db
def test_rejected_visit_can_be_approved_later(user_factory):
inviter = user_factory()
admin = user_factory(staff=True)
guest = Guest.objects.create(first_name="Ben", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Temporary access",
status=GuestVisitStatus.PENDING,
access_starts_at=timezone.now(),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestVisitWorkflow.reject(visit, actor=admin, comment="missing info")
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.REJECTED
GuestVisitWorkflow.approve(visit, actor=admin, comment="updated")
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.APPROVED
assert visit.events.filter(event_type="decision_changed").exists()
@pytest.mark.django_db
def test_unlimited_guest_visit_can_be_submitted_without_dates(user_factory):
inviter = user_factory()
guest = Guest.objects.create(first_name="No", last_name="Limit")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Long-term vendor access",
unlimited=True,
)
GuestVisitWorkflow.submit(visit, actor=inviter)
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.PENDING
assert visit.access_expires_at is None
@pytest.mark.django_db
def test_expire_guest_visit_records_event_and_disables_guest(settings, user_factory):
settings.GUESTS_NOTIFY_ON_EXPIRATION = False
inviter = user_factory()
guest = Guest.objects.create(first_name="Old", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Expired access",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(days=2),
access_expires_at=timezone.now() - timedelta(minutes=1),
)
GuestVisitWorkflow.expire(visit)
visit.refresh_from_db()
guest.refresh_from_db()
assert visit.status == GuestVisitStatus.EXPIRED
assert guest.ldap_last_synced_at is not None
assert visit.events.filter(event_type="expired").exists()
assert visit.events.filter(event_type="ldap_skipped").exists()
@pytest.mark.django_db
def test_detect_inactive_inviters_flags_risk_without_auto_disable(user_factory):
admin = user_factory(staff=True)
inviter = user_factory(active=False)
guest = Guest.objects.create(first_name="Risk", last_name="Guest")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Active guest with inactive inviter",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(days=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
detect_inactive_inviters.run()
visit.refresh_from_db()
guest.refresh_from_db()
assert admin.is_staff
assert visit.inviter_inactive is True
assert guest.ldap_enabled is False
assert visit.events.filter(event_type="inviter_inactive_detected").exists()
@pytest.mark.django_db
def test_guest_id_outside_guest_range_is_rejected():
guest = Guest(
id=GUEST_ID_MIN,
first_name="Bad",
last_name="Range",
)
with pytest.raises(ValidationError):
guest.full_clean()
@pytest.mark.django_db
@pytest.mark.parametrize(
("action", "status"),
[
("approve", GuestVisitStatus.DRAFT),
("request_info", GuestVisitStatus.DRAFT),
("revoke", GuestVisitStatus.PENDING),
("expire", GuestVisitStatus.PENDING),
],
)
def test_guest_visit_workflow_rejects_invalid_transitions(
action,
status,
user_factory,
):
inviter = user_factory()
admin = user_factory(staff=True)
guest = Guest.objects.create(first_name="Invalid", last_name="Transition")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Invalid transition",
status=status,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
with pytest.raises(ValueError):
if action == "approve":
GuestVisitWorkflow.approve(visit, actor=admin)
elif action == "request_info":
GuestVisitWorkflow.request_info(visit, actor=admin, comment="details")
elif action == "revoke":
GuestVisitWorkflow.revoke(visit, actor=admin)
elif action == "expire":
GuestVisitWorkflow.expire(visit)
@pytest.mark.django_db
def test_approved_visit_can_be_rejected_later(user_factory):
inviter = user_factory()
admin = user_factory(staff=True)
guest = Guest.objects.create(first_name="Decision", last_name="Change")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Decision change",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
GuestVisitWorkflow.reject(visit, actor=admin, comment="revise")
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.REJECTED
assert visit.events.filter(event_type="decision_changed").exists()
@pytest.mark.django_db
def test_cancelled_or_expired_visit_cannot_be_cancelled(user_factory):
inviter = user_factory()
guest = Guest.objects.create(first_name="Final", last_name="State")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Final state",
status=GuestVisitStatus.EXPIRED,
access_starts_at=timezone.now() - timedelta(days=2),
access_expires_at=timezone.now() - timedelta(days=1),
)
with pytest.raises(ValueError):
GuestVisitWorkflow.cancel(visit, actor=inviter)
@pytest.mark.django_db
def test_expire_guest_visits_task_expires_due_visits(settings, user_factory):
settings.GUESTS_NOTIFY_ON_EXPIRATION = False
inviter = user_factory()
guest = Guest.objects.create(first_name="Task", last_name="Expire")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Task expiration",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(days=2),
access_expires_at=timezone.now() - timedelta(minutes=1),
)
expire_guest_visits.run()
visit.refresh_from_db()
assert visit.status == GuestVisitStatus.EXPIRED
assert visit.events.filter(event_type="expired").exists()
@pytest.mark.django_db
def test_activate_due_guest_visits_task_syncs_due_visit(monkeypatch, user_factory):
inviter = user_factory()
guest = Guest.objects.create(first_name="Task", last_name="Activate")
visit = GuestVisit.objects.create(
guest=guest,
inviter=inviter,
purpose="Task activation",
status=GuestVisitStatus.APPROVED,
access_starts_at=timezone.now() - timedelta(minutes=1),
access_expires_at=timezone.now() + timedelta(days=1),
)
synced = []
def fake_sync(self, synced_visit):
synced.append(synced_visit.pk)
monkeypatch.setattr("guests.tasks.GuestLdapService.sync_guest_for_visit", fake_sync)
activate_due_guest_visits.run()
assert synced == [visit.pk]