diff --git a/backend/Dockerfile b/backend/Dockerfile index 6279f9c2..2236917f 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -1,7 +1,16 @@ -FROM python:3.13.5 +# Multi‑app Django (ASGI) — EUSRR +FROM python:3.13-slim + WORKDIR /app -COPY requirements.txt . -RUN pip install -r requirements.txt --no-cache-dir -RUN apt-get update && apt-get install -y postgresql-client -COPY . . -CMD ["gunicorn", "--bind", "0.0.0.0:8000", "foodgram_backend.wsgi"] +ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 + +RUN apt-get update && apt-get install -y --no-install-recommends build-essential libpq-dev curl && rm -rf /var/lib/apt/lists/* + +COPY backend/requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +COPY backend /app + +# Use ASGI server because project uses Django Channels +EXPOSE 8000 +CMD ["daphne", "-b", "0.0.0.0", "-p", "8000", "eusrr_backend.asgi:application"] diff --git a/backend/README_EUSRR.md b/backend/README_EUSRR.md new file mode 100644 index 00000000..c0fea691 --- /dev/null +++ b/backend/README_EUSRR.md @@ -0,0 +1,41 @@ +# EUSRR Backend — Quickstart + +## What’s inside +- Django 5.2, DRF, SimpleJWT +- Apps: employees, requests_app (HR requests), calendar_app, feed (news), communications (chat via Channels), bots (Telegram via aiogram), documents, search, hikcentral (read-only models to external DB) +- ASGI via Channels; Redis required for websockets + +## Local run (SQLite, no websockets) +```bash +python -m venv .venv && source .venv/bin/activate # Windows: .venv\Scripts\activate +pip install -r backend/requirements.txt +cp backend/.env backend/.env.local || true +# edit backend/.env.local if needed (USE_SQLITE=True by default) +python backend/manage.py migrate +python backend/manage.py createsuperuser +python backend/manage.py runserver 0.0.0.0:9000 +``` +Open http://127.0.0.1:9000/ + +## Full stack with Docker (Postgres + Redis) +1. Put your secrets into `backend/.env` (use `.env.example` as a template). **Never commit real secrets.** +2. Fix Dockerfile to use `eusrr_backend.asgi` (or use the provided `Dockerfile.fixed`). +3. Launch: +```bash +docker compose up --build +``` +The app will be available on http://127.0.0.1:9000/ + +## Useful commands +```bash +# Telegram bot +python backend/manage.py start_telegram_bot + +# Run tests +pytest -q backend +``` + +## Notes +- Set `API_BASE_URL` to match where the app is served (default http://127.0.0.1:9000/api). +- For Channels websockets, keep Redis reachable at `${REDIS_HOST}:${REDIS_PORT}`. +- The `hikcentral` DB is configured as a secondary Postgres. Externalize its DSN via envs if you plan to use it. diff --git a/backend/api/v1/employees/serializers.py b/backend/api/v1/employees/serializers.py index 09afa6c9..82ea0df9 100644 --- a/backend/api/v1/employees/serializers.py +++ b/backend/api/v1/employees/serializers.py @@ -6,7 +6,7 @@ from io import BytesIO from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.core.files.base import ContentFile -from employees.constants import ACTION_CHOICES +from employees.constants import ACTION_CHOICES, DeptPerm from employees.models import ( Department, DepartmentRole, @@ -14,7 +14,9 @@ from employees.models import ( EmployeeDepartment, Position, Skill, + DepartmentPermission, ) + from PIL import Image from rest_framework import serializers @@ -371,8 +373,38 @@ class EmployeeSerializer(serializers.ModelSerializer): } +class EmployeeBriefSerializer(serializers.ModelSerializer): + display_name = serializers.SerializerMethodField() + + class Meta: + model = Employee + fields = ( + "id", + "first_name", + "last_name", + "patronymic", + "email", + "phone_number", + "display_name", + ) + + def get_display_name(self, obj) -> str: + """ + Читабельное имя сотрудника: full name -> email -> phone -> 'Сотрудник #id' + """ + parts = [obj.last_name or "", obj.first_name or "", obj.patronymic or ""] + fio = " ".join(p.strip() for p in parts if p) + if fio: + return fio + if getattr(obj, "email", None): + return obj.email + if getattr(obj, "phone_number", None): + return obj.phone_number + return f"Сотрудник #{obj.id}" + + class DepartmentSerializer(serializers.ModelSerializer): - head = EmployeeSerializer(read_only=True) + head = EmployeeBriefSerializer(read_only=True) head_id = serializers.PrimaryKeyRelatedField( source="head", queryset=Employee.objects.all(), @@ -388,25 +420,19 @@ class DepartmentSerializer(serializers.ModelSerializer): "id", "name", "description", - "head", - "head_id", - "head_appointed_at", - "created_at", + "head", # nested read + "head_id", # write "employees_count", ) - # head_appointed_at и created_at только для чтения - read_only_fields = ("head_appointed_at", "created_at") - - def validate_head_id(self, value: Employee | None): - if value and not value.is_actually_active: - raise serializers.ValidationError( - "Назначать можно только реально активного сотрудника" - ) - return value - class EmployeeListSerializer(serializers.ModelSerializer): + """ + Списочное представление сотрудника в /employees/ (листинги, выборы и т.п.). + Добавляет поле display_name по тем же правилам, что и в кратком сериализаторе. + """ + + display_name = serializers.SerializerMethodField() email = serializers.EmailField(read_only=True) phone_number = serializers.CharField(read_only=True) avatar = Base64ImageField(read_only=True) @@ -430,6 +456,7 @@ class EmployeeListSerializer(serializers.ModelSerializer): "departments", "skills", "created_at", + "display_name", ) def get_departments(self, obj): @@ -459,6 +486,18 @@ class EmployeeListSerializer(serializers.ModelSerializer): nm = f"{ln} {fn} {pt}".strip() return nm + def get_display_name(self, obj: Employee) -> str: + """См. логику в EmployeeBriefSerializer.get_display_name().""" + parts = [obj.last_name or "", obj.first_name or "", obj.patronymic or ""] + fio = " ".join(p.strip() for p in parts if p).strip() + if fio: + return fio + if getattr(obj, "email", None): + return obj.email + if getattr(obj, "phone_number", None): + return obj.phone_number + return f"Сотрудник #{obj.id}" + class PositionSerializer(serializers.ModelSerializer): # для записи — массив id групп, для чтения — подробная инфа @@ -476,23 +515,92 @@ class PositionSerializer(serializers.ModelSerializer): class DepartmentRoleSerializer(serializers.ModelSerializer): - permissions = serializers.PrimaryKeyRelatedField( - queryset=Permission.objects.all(), many=True, required=False + # Новая схема: скоуп-права внутри отдела + scoped_permissions = serializers.PrimaryKeyRelatedField( + queryset=DepartmentPermission.objects.all(), many=True, required=False ) + scoped_permission_codes = serializers.ListField( + child=serializers.CharField(), required=False, write_only=True + ) + + # Backward-compat поля (read-only), чтобы фронт/старые тесты не ломались + permissions = serializers.SerializerMethodField(read_only=True) permissions_verbose = serializers.SerializerMethodField(read_only=True) class Meta: model = DepartmentRole - fields = ["id", "department", "name", "permissions", "permissions_verbose"] + fields = ( + "id", + "department", + "name", + # новые write-поля + "scoped_permissions", + "scoped_permission_codes", + # совместимость (read-only зеркала) + "permissions", + "permissions_verbose", + ) + read_only_fields = ("id",) - def get_permissions_verbose(self, obj): + # Поддержка записи codes вместо ids + def _apply_codes_if_present(self, instance: DepartmentRole, validated_data: dict): + codes = validated_data.pop("scoped_permission_codes", None) + if codes is not None: + qs = DepartmentPermission.objects.filter(code__in=codes) + if qs.count() != len(set(codes)): + raise serializers.ValidationError( + {"scoped_permission_codes": "Некоторые коды не существуют"} + ) + instance.scoped_permissions.set(list(qs)) + + def create(self, validated_data): + # вытаскиваем write-only/внешние поля до super().create() + codes = validated_data.pop("scoped_permission_codes", None) + perms = validated_data.pop("scoped_permissions", None) + + role = super().create(validated_data) + + if perms is not None: + role.scoped_permissions.set(perms) + + if codes is not None: + qs = DepartmentPermission.objects.filter(code__in=codes) + if qs.count() != len(set(codes)): + raise serializers.ValidationError( + {"scoped_permission_codes": "Некоторые коды не существуют"} + ) + role.scoped_permissions.set(list(qs)) + + return role + + def update(self, instance, validated_data): + # аналогично: вынимаем до super().update() + codes = validated_data.pop("scoped_permission_codes", None) + perms = validated_data.pop("scoped_permissions", None) + + role = super().update(instance, validated_data) + + if perms is not None: + role.scoped_permissions.set(perms) + + if codes is not None: + qs = DepartmentPermission.objects.filter(code__in=codes) + if qs.count() != len(set(codes)): + raise serializers.ValidationError( + {"scoped_permission_codes": "Некоторые коды не существуют"} + ) + role.scoped_permissions.set(list(qs)) + + return role + + # Backward-compat read + def get_permissions(self, obj: DepartmentRole): + return list(obj.scoped_permissions.values_list("id", flat=True)) + + def get_permissions_verbose(self, obj: DepartmentRole): return [ - { - "id": p.id, - "codename": f"{p.content_type.app_label}.{p.codename}", - "name": p.name, - } - for p in obj.permissions.select_related("content_type").all() + {"id": p.id, "code": p.code, "name": p.name} + for p in obj.scoped_permissions.order_by("code").all() ] @@ -518,3 +626,39 @@ class GroupSerializer(serializers.ModelSerializer): } for p in qs ] + + +# --- payload-сериалайзеры для кастомных действий --- + + +class SetHeadInput(serializers.Serializer): + head_id = serializers.IntegerField( + allow_null=True, required=False + ) # None -> снять руководителя + + +class SetMemberRoleInput(serializers.Serializer): + # поддерживаем алиасы: employee / role + employee_id = serializers.IntegerField(required=False) + role_id = serializers.IntegerField(allow_null=True, required=False) + employee = serializers.IntegerField(required=False) + role = serializers.IntegerField(allow_null=True, required=False) + is_active = serializers.BooleanField(required=False) # 👈 новый флаг (опционально) + + def validate(self, attrs): + emp = attrs.get("employee_id") or attrs.get("employee") + role = attrs.get("role_id") if "role_id" in attrs else attrs.get("role") + if emp is None: + raise serializers.ValidationError( + {"employee_id": "This field is required."} + ) + attrs["employee_id"] = emp + attrs["role_id"] = role # может быть None (снять роль) + # по умолчанию считаем активным, если поле не передали + if "is_active" not in attrs: + attrs["is_active"] = True + return attrs + + +class RemoveMemberInput(serializers.Serializer): + employee_id = serializers.IntegerField() diff --git a/backend/api/v1/employees/views.py b/backend/api/v1/employees/views.py index 30ea3e57..3519b567 100644 --- a/backend/api/v1/employees/views.py +++ b/backend/api/v1/employees/views.py @@ -1,39 +1,82 @@ # backend/api/v1/employees/views.py from __future__ import annotations -from collections import defaultdict from datetime import timedelta import phonenumbers from common.emails import send_templated_mail from django.conf import settings +from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.db import IntegrityError, transaction -from django.db.models import (Case, Count, Exists, F, IntegerField, OuterRef, - Prefetch, Q, Subquery, Value, When) +from django.db.models import ( + Case, + Count, + Exists, + F, + IntegerField, + OuterRef, + Prefetch, + Q, + Subquery, + Value, + When, +) +from django.db.models.functions import Coalesce +from django.shortcuts import get_object_or_404 from django.utils import timezone from django.utils.crypto import get_random_string from employees.constants import ACTION_DISMISSED -from employees.models import (Department, DepartmentRole, Employee, - EmployeeAction, EmployeeDepartment, Position, - Skill) +from employees.models import ( + Department, + DepartmentPermission, + DepartmentRole, + DeptPerm, + EmployeeAction, + EmployeeDepartment, + Position, + Skill, +) from phonenumbers import PhoneNumberFormat from rest_framework import filters, status, viewsets from rest_framework.decorators import action -from rest_framework.permissions import (AllowAny, DjangoModelPermissions, - IsAdminUser, IsAuthenticated) +from rest_framework.permissions import ( + AllowAny, + DjangoModelPermissions, + IsAdminUser, + IsAuthenticated, +) from rest_framework.response import Response from rest_framework.views import APIView -from ..permissions import (ASSIGN_ROLE_PERM, MANAGE_PERM, - AdminOrActionOrModelPerms, IsDeptManagerForWrite, - IsSelfOrStaff) -from .serializers import (DepartmentRoleSerializer, DepartmentSerializer, - EmailSerializer, EmailVerifySerializer, - EmployeeActionSerializer, EmployeeListSerializer, - EmployeeSerializer, GroupSerializer, - PositionSerializer, RegisterSerializer, - SkillSerializer) +from ..permissions import ( + ASSIGN_ROLE_PERM, + MANAGE_PERM, + AdminOrActionOrModelPerms, + AdminOrDeptAllowed, + IsDeptManagerForWrite, + IsSelfOrStaff, + has_dept_perm, +) +from .serializers import ( + DepartmentRoleSerializer, + DepartmentSerializer, + EmailSerializer, + EmailVerifySerializer, + EmployeeActionSerializer, + EmployeeBriefSerializer, + EmployeeListSerializer, + EmployeeSerializer, + GroupSerializer, + PositionSerializer, + RegisterSerializer, + RemoveMemberInput, + SetHeadInput, + SetMemberRoleInput, + SkillSerializer, +) + +Employee = get_user_model() def _to_bool(val: str | None) -> bool | None: @@ -69,6 +112,139 @@ def _detect_phone_field() -> str | None: return None +def _validate_head_active( + dept: Department, + employee_id: int, + require_email_verified: bool = True, +) -> tuple[bool, dict | None]: + """ + Проверить, можно ли назначить сотрудника руководителем отдела. + + Parameters + ---------- + dept : Department + Отдел, в который назначается руководитель. + employee_id : int + Кандидат на роль руководителя. + require_email_verified : bool, optional + Требовать ли подтверждение email. Для действий, совершаемых не-руководителем, + оставляем True; для действий текущего руководителя можно ослабить до False. + + Returns + ------- + (True, None) | (False, {"head_id": [..]}) + True — назначать можно; False — нельзя (причина в словаре). + """ + employee_model = Department._meta.get_field("head").remote_field.model + emp = employee_model.objects.filter(id=employee_id).first() + if not emp: + return False, {"head_id": ["Employee not found."]} + + # Должен быть включён аккаунт + if getattr(emp, "is_active", True) is False: + return False, {"head_id": ["Employee is inactive."]} + + # Требование верификации email — опционально + if require_email_verified and getattr(emp, "email_verified", True) is False: + return False, {"head_id": ["Employee is inactive."]} + + return True, None + + +def _ensure_department_permissions() -> list[dict]: + """ + Гарантирует наличие записей DepartmentPermission на основе DeptPerm.CHOICES. + Возвращает список словарей {id, code, name} в порядке CHOICES. + """ + items: list[dict] = [] + # пробежимся по CHOICES, создадим/обновим имя, соберём выдачу + for code, label in DeptPerm.CHOICES: + obj, _ = DepartmentPermission.objects.get_or_create( + code=code, defaults={"name": label} + ) + # если имя в БД отстаёт от CHOICES — мягко синхронизируем + if obj.name != label: + obj.name = label + obj.save(update_fields=["name"]) + items.append({"id": obj.id, "code": obj.code, "name": obj.name}) + return items + + +def _build_links_for_dept(dept: Department) -> list[dict]: + """ + Возвращает список линков отдела: + [{ "employee": , "role": {"id","name"}|None, "is_active": bool }, ...] + """ + links: list[dict] = [] + qs = ( + EmployeeDepartment.objects.filter(department_id=dept.id) + .select_related("employee", "role") + .order_by( + "employee__last_name", + "employee__first_name", + "employee__patronymic", + "employee_id", + ) + ) + for link in qs: + emp_data = EmployeeBriefSerializer(link.employee).data # содержит display_name + role = {"id": link.role_id, "name": link.role.name} if link.role_id else None + links.append( + {"employee": emp_data, "role": role, "is_active": bool(link.is_active)} + ) + + # гарантируем присутствие head + if dept.head_id and all(item["employee"]["id"] != dept.head_id for item in links): + head_data = EmployeeBriefSerializer(dept.head).data + links.insert(0, {"employee": head_data, "role": None, "is_active": True}) + + return links + + +def _head_choices_for_dept(dept: Department) -> list[dict]: + """ + Возвращает [{ "id": , "name": }] для селекта руководителя. + """ + choices, seen = [], set() + qs = ( + EmployeeDepartment.objects.filter(department_id=dept.id) + .select_related("employee") + .order_by( + "employee__last_name", + "employee__first_name", + "employee__patronymic", + "employee_id", + ) + ) + for link in qs: + data = EmployeeBriefSerializer(link.employee).data + if data["id"] not in seen: + choices.append({"id": data["id"], "name": data["display_name"]}) + seen.add(data["id"]) + + if dept.head_id and dept.head_id not in seen: + head_data = EmployeeBriefSerializer(dept.head).data + choices.insert(0, {"id": head_data["id"], "name": head_data["display_name"]}) + + return choices + + +def _perm_choices_synced() -> list[dict]: + """ + Возвращает справочник прав для ролей отдела, синхронизируя записи с DeptPerm.CHOICES. + """ + items = [] + for code, label in DeptPerm.CHOICES: + obj, _ = DepartmentPermission.objects.get_or_create( + code=code, defaults={"name": label} + ) + if obj.name != label: + obj.name = label + obj.save(update_fields=["name"]) + items.append({"id": obj.id, "code": obj.code, "name": obj.name}) + return items + + PHONE_FIELD = _detect_phone_field() @@ -299,204 +475,330 @@ class RegisterAPIView(APIView): ) -class DepartmentViewSet(HistoryActionMixin, viewsets.ModelViewSet): +class DepartmentViewSet(viewsets.ModelViewSet): """ - /api/departments/ - - GET (list/detail) — только аутентифицированные пользователи - - POST/DELETE — только staff/superuser или пользователи с правами - - PATCH/PUT — руководитель отдела ИЛИ роль с perm "manage_department" - - Смена head — руководитель отдела ИЛИ роль с perm "change_department_head" - - Управление ролями — руководитель отдела ИЛИ роль с perm "assign_department_role" + CRUD отделов + действия: + - POST /departments/{id}/set_head + - POST /departments/{id}/set_member_role + - POST /departments/{id}/remove_member + Права: + - update/partial_update/destroy → manage_department + - set_head → change_department_head + - set_member_role/remove_member → assign_department_role + - create → staff/superuser + - чтение → аутентифицированным """ + queryset = Department.objects.select_related("head").prefetch_related("roles").all() serializer_class = DepartmentSerializer - permission_classes = [IsAuthenticated] - search_fields = ["name", "description"] - ordering = ["name"] - filter_backends = [filters.SearchFilter, filters.OrderingFilter] - history_diff_fields = ["name", "description", "head_id", "head_appointed_at"] - def get_queryset(self): - # подзапрос: есть ли head среди связей отдела - head_in_links_qs = EmployeeDepartment.objects.filter( - department=OuterRef("pk"), - employee_id=OuterRef("head_id"), - is_active=True, - ) + # пермишены (скоуп-право по отделу) + class ManagePerm(AdminOrDeptAllowed): + """Право на общее управление отделом.""" - return ( - Department.objects.select_related("head") - .annotate( - active_links_count=Count( - "employeedepartment", - filter=Q(employeedepartment__is_active=True), - distinct=True, - ), - # +1, если head задан и не присутствует среди активных связей - head_extra=Case( - When( - Q(head__isnull=False) & ~Exists(head_in_links_qs), - then=Value(1), - ), - default=Value(0), - output_field=IntegerField(), - ), - ) - .annotate(employees_count=F("active_links_count") + F("head_extra")) - ) + required_code = DeptPerm.MANAGE + + class ChangeHeadPerm(AdminOrDeptAllowed): + """Право на смену руководителя отдела.""" + + required_code = DeptPerm.CHANGE_HEAD + + class AssignRolePerm(AdminOrDeptAllowed): + """Право на назначение ролей участникам отдела.""" + + required_code = DeptPerm.ASSIGN_ROLE def get_permissions(self): - # Создание/удаление — только админы - if self.action in ["create", "destroy"]: + if self.action in {"update", "partial_update", "destroy"}: + return [self.ManagePerm()] + if self.action == "set_head": + return [self.ChangeHeadPerm()] + if self.action in {"set_member_role", "remove_member"}: + return [self.AssignRolePerm()] + if self.action == "create": return [AdminOrActionOrModelPerms()] - # Общие правки отдела — право manage_department на объекте - if self.action in ["update", "partial_update"]: - self.required_perm_code = "manage_department" - return [IsAuthenticated(), IsDeptManagerForWrite()] - return super().get_permissions() + if self.action in {"members", "user_perms", "ui_context"}: + return [IsAuthenticated()] + return [IsAuthenticated()] - def perform_create(self, serializer): - # head_appointed_at у тебя auto_now — обновится сам - serializer.save() + # --- поиск и сортировка, как в старой реализации/тестах --- + ordering = ["name"] + ordering_fields = ["name", "id"] - def _check_head_change_perm(self, request, dept: Department) -> Response | None: - if "head_id" not in request.data: - return None - raw = request.data.get("head_id", None) - if raw is None: - new_head_id = None + def get_queryset(self): + qs = super().get_queryset() + + # ---------- аннотации для employees_count ---------- + # Активные линкы данного отдела + active_links = EmployeeDepartment.objects.filter( + department_id=OuterRef("pk"), is_active=True + ) + + # Подсчёт distinct сотрудников по активным линкам + active_count_subq = ( + active_links.values("department_id") + .annotate(c=Count("employee_id", distinct=True)) + .values("c")[:1] + ) + + qs = qs.annotate( + active_count=Coalesce( + Subquery(active_count_subq, output_field=IntegerField()), + Value(0), + ), + head_in_active=Exists(active_links.filter(employee_id=OuterRef("head_id"))), + ).annotate( + employees_count=F("active_count") + + Case( + When( + Q(head_id__isnull=False) & Q(head_in_active=False), + then=Value(1), + ), + default=Value(0), + output_field=IntegerField(), + ) + ) + # ---------- /аннотации ---------- + + # Поиск и сортировка — как и раньше + search = ( + self.request.query_params.get("search") + or self.request.query_params.get("q") + or "" + ).strip() + if search: + qs = qs.filter( + Q(name__icontains=search) + | Q(description__icontains=search) + | Q(head__first_name__icontains=search) + | Q(head__last_name__icontains=search) + | Q(head__patronymic__icontains=search) + ).distinct() + + ordering = self.request.query_params.get("ordering") + if ordering in {"name", "-name", "id", "-id"}: + qs = qs.order_by(ordering) else: - try: - new_head_id = int(raw) - except (TypeError, ValueError): - new_head_id = None - if new_head_id != dept.head_id: - self.required_perm_code = "change_department_head" - perm = IsDeptManagerForWrite() - if not perm.has_object_permission(request, self, dept): - return Response( - {"detail": "Недостаточно прав для смены руководителя отдела"}, - status=status.HTTP_403_FORBIDDEN, - ) - return None + qs = qs.order_by("name") + + return qs + + # --- частичное изменение: изменение head требует отдельного кода --- def partial_update(self, request, *args, **kwargs): - dept = self.get_object() - resp = self._check_head_change_perm(request, dept) - if resp is not None: - return resp + """ + Если меняется head, проверяем права и валидацию. + Для действующего руководителя ослабляем требование email_verified. + """ + instance = self.get_object() + data = request.data + + wants_head_change = any(k in data for k in ("head", "head_id")) + if wants_head_change: + desired = data.get("head_id", data.get("head", None)) + if desired in ("", "null", "None"): + desired = None + + current = instance.head_id if instance.head_id is not None else None + desired_norm = int(desired) if desired is not None else None + + if current != desired_norm: + perm = self.ChangeHeadPerm() + if not ( + perm.has_permission(request, self) + and perm.has_object_permission(request, self, instance) + ): + return Response({"detail": "Forbidden"}, status=403) + + if desired_norm is None: + instance.head = None + instance.save(update_fields=["head"]) + else: + # ослабляем verified, если действие делает текущий head + require_verified = not ( + instance.head_id and instance.head_id == request.user.id + ) + ok, errs = _validate_head_active( + instance, desired_norm, require_email_verified=require_verified + ) + if not ok: + return Response(errs, status=400) + + instance.head_id = desired_norm + instance.save(update_fields=["head"]) + + # подчистим поля + mutable = getattr(request.data, "_mutable", None) + try: + if hasattr(request.data, "_mutable"): + request.data._mutable = True + request.data.pop("head", None) + request.data.pop("head_id", None) + finally: + if hasattr(request.data, "_mutable") and mutable is not None: + request.data._mutable = mutable + return super().partial_update(request, *args, **kwargs) - def update(self, request, *args, **kwargs): - dept = self.get_object() - resp = self._check_head_change_perm(request, dept) - if resp is not None: - return resp - return super().update(request, *args, **kwargs) + # -------- actions -------- @action(detail=True, methods=["post"]) + @transaction.atomic def set_head(self, request, pk=None): - dept: Department = self.get_object() - self.required_perm_code = "change_department_head" - perm = IsDeptManagerForWrite() - if not perm.has_object_permission(request, self, dept): - return Response( - {"detail": "Недостаточно прав для смены руководителя отдела"}, - status=status.HTTP_403_FORBIDDEN, - ) + """ + Назначение/снятие руководителя отдела. + Текущий руководитель может назначить нового head без email_verified, + если у кандидата is_active=True. + """ + dept = self.get_object() + payload = SetHeadInput(data=request.data) + payload.is_valid(raise_exception=True) + head_id = payload.validated_data.get("head_id") - if "head_id" not in request.data: - return Response( - { - "detail": "Параметр 'head_id' обязателен (null — чтобы снять руководителя)." - }, - status=400, - ) - - head_id = request.data.get("head_id", None) - - # снять руководителя (явно передан null) if head_id is None: - if dept.head_id is not None: - dept.head = None - dept.save(update_fields=["head"]) # auto_now обновит head_appointed_at + dept.head = None + dept.save(update_fields=["head"]) return Response(status=status.HTTP_204_NO_CONTENT) - try: - emp = Employee.objects.get(pk=head_id) - except Employee.DoesNotExist: - return Response({"detail": "Сотрудник не найден"}, status=404) + # ослабляем требование verified, если действие делает текущий head + require_verified = not (dept.head_id and dept.head_id == request.user.id) - if not emp.is_actually_active: - return Response({"detail": "Сотрудник неактивен"}, status=400) + ok, errs = _validate_head_active( + dept, head_id, require_email_verified=require_verified + ) + if not ok: + return Response(errs, status=400) - if dept.head_id != emp.id: - dept.head = emp - dept.save(update_fields=["head"]) - return Response({"head_id": dept.head_id}, status=200) + dept.head_id = head_id + dept.save(update_fields=["head"]) + return Response(self.get_serializer(dept).data, status=200) @action(detail=True, methods=["post"]) + @transaction.atomic def set_member_role(self, request, pk=None): - """ - Body: - { - "employee_id": 123, - "role_id": 7 | null, # null = снять роль (оставив сотрудника в отделе) - "is_active": true|false (необязательно) - } - """ - dept: Department = self.get_object() + dept = self.get_object() + payload = SetMemberRoleInput(data=request.data) + if not payload.is_valid(): + return Response(payload.errors, status=400) - self.required_perm_code = "assign_department_role" - perm = IsDeptManagerForWrite() - if not perm.has_object_permission(request, self, dept): - return Response( - {"detail": "Недостаточно прав для управления ролями отдела"}, - status=status.HTTP_403_FORBIDDEN, - ) + emp_id = payload.validated_data["employee_id"] + role_id = payload.validated_data.get("role_id") + make_active = payload.validated_data.get("is_active", True) - employee_id = request.data.get("employee_id") - role_id = request.data.get("role_id", None) - is_active = request.data.get("is_active", None) - - if not employee_id: - return Response({"detail": "employee_id обязателен"}, status=400) - - try: - emp = Employee.objects.get(pk=employee_id) - except Employee.DoesNotExist: - return Response({"detail": "Сотрудник не найден"}, status=404) + # модель сотрудника берём из FK Department.head (без жёсткого User) + employee_model = Department._meta.get_field("head").remote_field.model + get_object_or_404(employee_model, id=emp_id) + # проверяем роль, если указана, что она принадлежит этому отделу role = None if role_id is not None: - try: - role = DepartmentRole.objects.get(pk=role_id) - except DepartmentRole.DoesNotExist: - return Response({"detail": "Роль не найдена"}, status=404) + role = get_object_or_404(DepartmentRole, id=role_id) if role.department_id != dept.id: return Response( - {"detail": "Роль принадлежит другому отделу"}, status=400 + {"role_id": ["Role does not belong to this department."]}, + status=400, ) - with transaction.atomic(): - link, _ = EmployeeDepartment.objects.select_for_update().get_or_create( - employee=emp, department=dept, defaults={"is_active": True} - ) - link.role = role - if is_active is not None: - link.is_active = bool(is_active) - link.full_clean() - link.save() + # создаём/обновляем линк + link, _ = EmployeeDepartment.objects.get_or_create( + employee_id=emp_id, + department_id=dept.id, + defaults={"is_active": make_active, "role": role}, + ) + link.is_active = bool(make_active) + link.role = role # None => снять роль + link.save(update_fields=["is_active", "role"]) return Response( { - "employee_id": emp.id, - "role_id": role.id if role else None, + "employee_id": emp_id, + "role_id": (role.id if role else None), "is_active": link.is_active, }, - status=200, + status=status.HTTP_200_OK, ) + @action(detail=True, methods=["get"], url_path="members") + def members(self, request, pk=None): + """ + GET /api/v1/departments/{id}/members/ + Список участников отдела в формате, удобном для фронта-шаблона. + """ + dept = self.get_object() + links = _build_links_for_dept(dept) + return Response({"count": len(links), "results": links}, status=200) + + @action(detail=True, methods=["get"], url_path="user-perms") + def user_perms(self, request, pk=None): + """ + GET /api/v1/departments/{id}/user-perms/ + Флаги прав текущего пользователя в рамках отдела. + """ + dept = self.get_object() + uid = getattr(request.user, "id", None) + data = { + "is_head": bool(uid and uid == dept.head_id), + "can_manage": has_dept_perm(request.user, dept.id, DeptPerm.MANAGE), + "can_change_head": has_dept_perm( + request.user, dept.id, DeptPerm.CHANGE_HEAD + ), + "can_assign_roles": has_dept_perm( + request.user, dept.id, DeptPerm.ASSIGN_ROLE + ), + } + return Response(data, status=200) + + @action(detail=True, methods=["get"], url_path="ui-context") + def ui_context(self, request, pk=None): + """ + GET /api/v1/departments/{id}/ui-context/ + BFF-агрегатор для страницы отдела: возвращает все необходимые данные одной пачкой. + Поля: + - dept: DepartmentSerializer + - roles: DepartmentRoleSerializer[] + - links: см. /members + - head_choices: [{id,name}] + - dept_perm_choices: [{id,code,name}] + - user_perms: {is_head, can_manage, can_change_head, can_assign_roles} + """ + dept = self.get_object() + dept_data = self.get_serializer(dept).data + + roles_qs = ( + DepartmentRole.objects.filter(department_id=dept.id) + .prefetch_related("scoped_permissions") + .order_by("name", "id") + ) + roles_data = DepartmentRoleSerializer(roles_qs, many=True).data + + links = _build_links_for_dept(dept) + perm_choices = _perm_choices_synced() + head_choices = _head_choices_for_dept(dept) + user_perms = { + "is_head": ( + request.user.id == dept.head_id + if getattr(request.user, "id", None) + else False + ), + "can_manage": has_dept_perm(request.user, dept.id, DeptPerm.MANAGE), + "can_change_head": has_dept_perm( + request.user, dept.id, DeptPerm.CHANGE_HEAD + ), + "can_assign_roles": has_dept_perm( + request.user, dept.id, DeptPerm.ASSIGN_ROLE + ), + } + + payload = { + "dept": dept_data, + "roles": roles_data, + "links": links, + "head_choices": head_choices, + "dept_perm_choices": perm_choices, + "user_perms": user_perms, + } + return Response(payload, status=200) + class EmployeeViewSet(viewsets.ModelViewSet): """ @@ -904,132 +1206,116 @@ class PositionViewSet(HistoryActionMixin, viewsets.ModelViewSet): class DepartmentRoleViewSet(viewsets.ModelViewSet): """ - /api/v1/positions/ - GET list/retrieve — IsAuthenticated - POST/PUT/PATCH/DELETE — Admin OR Django model perms (add/change/delete_position) - POST /{id}/set-groups — Admin OR employees.assign_position_groups - POST /{id}/add-groups — Admin OR employees.assign_position_groups - POST /{id}/remove-groups — Admin OR employees.assign_position_groups - GET /{id}/permissions — IsAuthenticated + Роли отдела: + - list/retrieve с фильтром ?department= + - create/update/destroy → требуется право DeptPerm.ASSIGN_ROLE в рамках отдела роли + - GET /department-roles/perm_choices/ + - GET /department-roles/{id}/perms/ + - POST /department-roles/{id}/set_perms (ids или codes; полная замена) """ + queryset = ( + DepartmentRole.objects.select_related("department") + .prefetch_related("scoped_permissions") + .all() + ) serializer_class = DepartmentRoleSerializer - permission_classes = [IsAuthenticated, IsDeptManagerForWrite] - filter_backends = [filters.SearchFilter, filters.OrderingFilter] - search_fields = ["name"] - ordering_fields = ["name", "id"] - ordering = ["name"] - pagination_class = None + + class AssignRolePerm(AdminOrDeptAllowed): + """Право на назначение ролей участникам отдела.""" + + required_code = DeptPerm.ASSIGN_ROLE + + # стабильная сортировка для листинга + ordering_fields = ("name", "id") + ordering = ("name", "id") + + def get_permissions(self): + if self.action in { + "create", + "update", + "partial_update", + "destroy", + "set_perms", + }: + return [self.AssignRolePerm()] + return [IsAuthenticated()] def get_queryset(self): - qs = DepartmentRole.objects.select_related("department").prefetch_related( - "permissions__content_type" - ) - dep = self.request.query_params.get("department") - if dep: - try: - dep_id = int(dep) - except (TypeError, ValueError): - # нечисловой параметр -> пустой список без ошибки - return DepartmentRole.objects.none() - qs = qs.filter(department_id=dep_id) + """ + Фильтрация по отделу и стабильный порядок. + """ + qs = super().get_queryset() + dept = self.request.query_params.get("department") + if dept: + qs = qs.filter(department_id=dept) + + ord_param = self.request.query_params.get("ordering") + if ord_param in {"name", "-name", "id", "-id"}: + # стабильный тай-брейк по id + qs = qs.order_by( + ord_param, "id" if not ord_param.startswith("-") else "-id" + ) + else: + qs = qs.order_by(*self.ordering) return qs - # --- аккуратно задаём требуемые права под экшены --- - def get_permissions(self): - # по умолчанию чтение — только аутентификация - if self.action in ("create", "update", "partial_update", "destroy"): - self.required_perm_code = MANAGE_PERM - elif self.action in ("set_perms", "add_perms", "remove_perms"): - self.required_perm_code = ASSIGN_ROLE_PERM - else: - self.required_perm_code = None - return super().get_permissions() - - # create — объект ещё не существует, делаем явную проверку на отдел - def perform_create(self, serializer): - dept: Department = serializer.validated_data["department"] - checker = IsDeptManagerForWrite() - if not checker.has_object_permission(self.request, self, dept): - from rest_framework.exceptions import PermissionDenied - - raise PermissionDenied( - "Недостаточно прав для управления ролями этого отдела" - ) - serializer.save() - - # update/destroy — has_object_permission отработает, т.к. мы добавили IsDeptManagerForWrite - # и он умеет извлекать department из объекта роли (см. патч выше) - - # ---- экшены управления правами роли ---- - def _validate_perm_ids(self, request): - ids = request.data.get("permission_ids") - if not isinstance(ids, list): - return None, Response( - {"detail": "Поле 'permission_ids' должно быть списком id"}, status=400 - ) - perms = Permission.objects.filter(id__in=ids).select_related("content_type") - if perms.count() != len(set(ids)): - return None, Response( - {"detail": "Некоторые permissions не найдены"}, status=400 - ) - return perms, None + @action(detail=False, methods=["get"]) + def perm_choices(self, request): + """ + Возвращает справочник скоуп-прав для ролей отдела из DeptPerm.CHOICES. + Если каких-то записей нет в БД, создаёт их (идемпотентно). + """ + data = _ensure_department_permissions() + return Response({"count": len(data), "results": data}, status=200) @action(detail=True, methods=["get"]) def perms(self, request, pk=None): - role: DepartmentRole = self.get_object() + """ + Возвращает список прав, назначенных данной роли (отсортирован по коду). + """ + role = self.get_object() data = [ - { - "id": p.id, - "codename": f"{p.content_type.app_label}.{p.codename}", - "name": p.name, - } - for p in role.permissions.select_related("content_type").all() + {"id": p.id, "code": p.code, "name": p.name} + for p in role.scoped_permissions.order_by("code") ] - return Response({"count": len(data), "results": data}) + return Response({"count": len(data), "results": data}, status=200) @action(detail=True, methods=["post"]) def set_perms(self, request, pk=None): - role: DepartmentRole = self.get_object() - # required_perm_code уже выставлен -> IsDeptManagerForWrite проверит доступ по role.department - perms, err = self._validate_perm_ids(request) - if err: - return err - role.permissions.set(perms) - return Response( - { - "ok": True, - "permission_ids": list(role.permissions.values_list("id", flat=True)), - } - ) + """ + Полностью заменяет набор прав у роли. + Тело запроса: + - либо "permission_ids": [1,2,3] + - либо "permission_codes": ["manage_department", ...] + """ + role = self.get_object() - @action(detail=True, methods=["post"]) - def add_perms(self, request, pk=None): - role: DepartmentRole = self.get_object() - perms, err = self._validate_perm_ids(request) - if err: - return err - role.permissions.add(*perms) - return Response( - { - "ok": True, - "permission_ids": list(role.permissions.values_list("id", flat=True)), - } - ) + ids = request.data.get("permission_ids") or [] + codes = request.data.get("permission_codes") or [] - @action(detail=True, methods=["post"]) - def remove_perms(self, request, pk=None): - role: DepartmentRole = self.get_object() - perms, err = self._validate_perm_ids(request) - if err: - return err - role.permissions.remove(*perms) - return Response( - { - "ok": True, - "permission_ids": list(role.permissions.values_list("id", flat=True)), - } - ) + if isinstance(ids, list) and ids: + # валидация набора id + ids_int = {int(i) for i in ids if str(i).isdigit()} + qs = DepartmentPermission.objects.filter(id__in=ids_int) + if qs.count() != len(ids_int): + return Response( + {"detail": "Некоторые permission_ids не найдены."}, status=400 + ) + elif isinstance(codes, list) and codes: + # валидация набора кодов + codes_set = set(codes) + qs = DepartmentPermission.objects.filter(code__in=codes_set) + if qs.count() != len(codes_set): + return Response( + {"detail": "Некоторые permission_codes не найдены."}, status=400 + ) + else: + qs = DepartmentPermission.objects.none() + + role.scoped_permissions.set(list(qs)) + ser = self.get_serializer(role) + return Response(ser.data, status=200) class SkillViewSet(viewsets.ModelViewSet): diff --git a/backend/api/v1/permissions.py b/backend/api/v1/permissions.py index e749193d..f6ff2a0e 100644 --- a/backend/api/v1/permissions.py +++ b/backend/api/v1/permissions.py @@ -1,13 +1,11 @@ # backend\api\v1\permissions.py from __future__ import annotations -from rest_framework.permissions import ( - SAFE_METHODS, - BasePermission, - DjangoModelPermissions, -) - +from django.contrib.auth.models import AbstractBaseUser +from employees.constants import DeptPerm from employees.models import Department, EmployeeDepartment +from rest_framework.permissions import (SAFE_METHODS, BasePermission, + DjangoModelPermissions) MANAGE_PERM = "manage_department" CHANGE_HEAD_PERM = "change_department_head" @@ -38,10 +36,42 @@ def user_has_dept_perm(user, dept: Department, perm_code: str) -> bool: employee_id=user.id, department_id=dept.id, is_active=True, - role__permissions__codename=perm_code, + role__scoped_permissions__codename=perm_code, ).exists() +def has_dept_perm(user: AbstractBaseUser, department_id: int, code: str) -> bool: + """ + Возвращает True, если пользователь имеет право `code` в рамках отдела `department_id`. + + Логика: + 1) staff/superuser → всегда True + 2) пользователь является текущим руководителем отдела → True для любых «управленческих» кодов + 3) пользователь состоит в отделе и его роль содержит `code` в scoped_permissions → True + 4) иначе False + """ + if not getattr(user, "is_authenticated", False): + return False + if getattr(user, "is_superuser", False) or getattr(user, "is_staff", False): + return True + + # 2) Руководитель отдела имеет все управленческие права в своём отделе + if Department.objects.filter(id=department_id, head_id=user.id).exists(): + return True + + # 3) Проверка роли в отделе + link = ( + EmployeeDepartment.objects.select_related("role") + .filter(employee_id=user.id, department_id=department_id, is_active=True) + .first() + ) + if not link or not link.role_id: + return False + + # роль хранит «скоуп-права» только для этого отдела + return link.role.scoped_permissions.filter(code=code).exists() + + class IsDeptManagerForWrite(BasePermission): """ Для DepartmentViewSet. @@ -82,6 +112,101 @@ class IsDeptManagerForWrite(BasePermission): return user_has_dept_perm(user, dept, perm_code) +class AdminOrDeptAllowed(BasePermission): + required_code: str | None = None + required_code_map: dict[str, str] = {} + allow_safe_without_code: bool = True + + def get_required_code(self, request, view): + return ( + self.required_code_map.get(getattr(view, "action", None)) + or self.required_code + ) + + def _extract_dept_id_from_request(self, request, view): + for k in ("department_pk", "department_id"): + v = view.kwargs.get(k) + if v is not None: + return v + for k in ("department", "department_id"): + v = request.data.get(k) + if v is not None: + return v + for k in ("department", "department_id"): + v = request.query_params.get(k) + if v is not None: + return v + return None + + def _extract_dept_id_from_obj(self, obj): + if obj is None: + return None + # сам отдел + if obj.__class__.__name__ == "Department" and hasattr(obj, "id"): + return obj.id + # прямое поле FK + for attr in ("department_id", "dept_id"): + if hasattr(obj, attr): + return getattr(obj, attr) + dep = getattr(obj, "department", None) + if dep is not None: + return getattr(dep, "id", None) + return None + + def _to_int_or_none(self, v): + try: + return int(str(v)) + except Exception: + return None + + def has_permission(self, request, view): + user = getattr(request, "user", None) + if not user or not user.is_authenticated: + return False + if getattr(user, "is_superuser", False) or getattr(user, "is_staff", False): + return True + + code = self.get_required_code(request, view) + if not code: + return ( + request.method in SAFE_METHODS + if self.allow_safe_without_code + else False + ) + + dept_id = self._to_int_or_none( + self._extract_dept_id_from_request(request, view) + ) + if dept_id is None: + # detail-экшены проверим на объекте + return True if getattr(view, "detail", False) else False + + return has_dept_perm(user, dept_id, code) + + def has_object_permission(self, request, view, obj): + user = getattr(request, "user", None) + if getattr(user, "is_superuser", False) or getattr(user, "is_staff", False): + return True + + code = self.get_required_code(request, view) + if not code: + return ( + request.method in SAFE_METHODS + if self.allow_safe_without_code + else False + ) + + dept_id = self._to_int_or_none(self._extract_dept_id_from_obj(obj)) + if dept_id is None: + dept_id = self._to_int_or_none( + self._extract_dept_id_from_request(request, view) + ) + if dept_id is None: + return False + + return has_dept_perm(request.user, dept_id, code) + + class IsSelfOrStaff(BasePermission): # На class-level пускаем любого аутентифицированного. def has_permission(self, request, view): @@ -97,9 +222,8 @@ class IsSelfOrStaff(BasePermission): # Основной случай — объект сотрудника. try: - from employees.models import ( - Employee, - ) # локальный импорт, чтобы избежать циклов + from employees.models import \ + Employee # локальный импорт, чтобы избежать циклов except Exception: Employee = None @@ -141,6 +265,7 @@ class AdminOrActionOrModelPerms(DjangoModelPermissions): Если задан код права (view.required_perm_code или required_perms_by_action[action]), проверяем его через user.has_perm(...). Иначе — стандартные DjangoModelPermissions. """ + def has_permission(self, request, view): user = request.user if not user or not user.is_authenticated: diff --git a/backend/employees/admin.py b/backend/employees/admin.py index 02102377..330755d9 100644 --- a/backend/employees/admin.py +++ b/backend/employees/admin.py @@ -255,8 +255,7 @@ class DepartmentRoleAdmin(admin.ModelAdmin): list_filter = ("department",) search_fields = ("name", "department__name") autocomplete_fields = ("department",) - filter_horizontal = ("permissions",) - + filter_horizontal = ("scoped_permissions",) @admin.register(EmployeeDepartment) class EmployeeDepartmentAdmin(admin.ModelAdmin): diff --git a/backend/employees/constants.py b/backend/employees/constants.py index 4908bdad..a8806ebb 100644 --- a/backend/employees/constants.py +++ b/backend/employees/constants.py @@ -1,27 +1,38 @@ # backend\employees\constants.py GENDER_CHOICES = ( - (1, 'Мужской'), - (2, 'Женский'), - (0, 'Не указан'), + (1, "Мужской"), + (2, "Женский"), + (0, "Не указан"), ) -ACTION_HIRED = 'hired' -ACTION_DISMISSED = 'dismissed' -ACTION_ON_LEAVE = 'on_leave' -ACTION_RETURNED_FROM_LEAVE = 'returned_from_leave' -ACTION_ON_MATERNITY = 'on_maternity' -ACTION_RETURNED_FROM_MATERNITY = 'returned_from_maternity' -ACTION_TRANSFERRED = 'transferred' -ACTION_REHIRED = 'rehired' +ACTION_HIRED = "hired" +ACTION_DISMISSED = "dismissed" +ACTION_ON_LEAVE = "on_leave" +ACTION_RETURNED_FROM_LEAVE = "returned_from_leave" +ACTION_ON_MATERNITY = "on_maternity" +ACTION_RETURNED_FROM_MATERNITY = "returned_from_maternity" +ACTION_TRANSFERRED = "transferred" +ACTION_REHIRED = "rehired" ACTION_CHOICES = [ - (ACTION_HIRED, 'Принят'), - (ACTION_DISMISSED, 'Уволен'), - (ACTION_ON_LEAVE, 'В отпуске'), - (ACTION_RETURNED_FROM_LEAVE, 'Вернулся из отпуска'), - (ACTION_ON_MATERNITY, 'В декрете'), - (ACTION_RETURNED_FROM_MATERNITY, 'Вернулся из декрета'), - (ACTION_TRANSFERRED, 'Переведен'), - (ACTION_REHIRED, 'Восстановлен'), -] \ No newline at end of file + (ACTION_HIRED, "Принят"), + (ACTION_DISMISSED, "Уволен"), + (ACTION_ON_LEAVE, "В отпуске"), + (ACTION_RETURNED_FROM_LEAVE, "Вернулся из отпуска"), + (ACTION_ON_MATERNITY, "В декрете"), + (ACTION_RETURNED_FROM_MATERNITY, "Вернулся из декрета"), + (ACTION_TRANSFERRED, "Переведен"), + (ACTION_REHIRED, "Восстановлен"), +] + + +class DeptPerm: + MANAGE = "manage_department" + CHANGE_HEAD = "change_department_head" + ASSIGN_ROLE = "assign_department_role" + CHOICES = ( + (MANAGE, "Управлять отделом"), + (CHANGE_HEAD, "Назначать руководителя"), + (ASSIGN_ROLE, "Назначать роли участникам"), + ) diff --git a/backend/employees/migrations/0022_departmentpermission_and_more.py b/backend/employees/migrations/0022_departmentpermission_and_more.py new file mode 100644 index 00000000..ef22a1f4 --- /dev/null +++ b/backend/employees/migrations/0022_departmentpermission_and_more.py @@ -0,0 +1,30 @@ +# Generated by Django 5.2.4 on 2025-09-03 07:26 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('employees', '0021_alter_employee_options'), + ] + + operations = [ + migrations.CreateModel( + name='DepartmentPermission', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('code', models.CharField(choices=[('manage_department', 'Управлять отделом'), ('change_department_head', 'Назначать руководителя'), ('assign_department_role', 'Назначать роли участникам')], max_length=64, unique=True)), + ('name', models.CharField(max_length=150)), + ], + ), + migrations.RemoveField( + model_name='departmentrole', + name='permissions', + ), + migrations.AddField( + model_name='departmentrole', + name='scoped_permissions', + field=models.ManyToManyField(blank=True, related_name='roles', to='employees.departmentpermission'), + ), + ] diff --git a/backend/employees/migrations/0023_alter_departmentrole_options_and_more.py b/backend/employees/migrations/0023_alter_departmentrole_options_and_more.py new file mode 100644 index 00000000..e74bb717 --- /dev/null +++ b/backend/employees/migrations/0023_alter_departmentrole_options_and_more.py @@ -0,0 +1,42 @@ +# Generated by Django 5.2.4 on 2025-09-03 12:02 + +import django.db.models.constraints +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('employees', '0022_departmentpermission_and_more'), + ] + + operations = [ + migrations.AlterModelOptions( + name='departmentrole', + options={'ordering': ('name', 'department', 'id'), 'verbose_name': 'Роль в отделе', 'verbose_name_plural': 'Роли в отделе'}, + ), + migrations.RemoveConstraint( + model_name='employeedepartment', + name='uniq_employee_department', + ), + migrations.RemoveIndex( + model_name='employeedepartment', + name='employees_e_departm_40dcd8_idx', + ), + migrations.RemoveIndex( + model_name='employeedepartment', + name='employees_e_employe_e4fe19_idx', + ), + migrations.AddIndex( + model_name='employeedepartment', + index=models.Index(fields=['department_id', 'is_active', 'employee_id'], name='employees_e_departm_81ade2_idx'), + ), + migrations.AddIndex( + model_name='employeedepartment', + index=models.Index(fields=['employee_id', 'department_id'], name='employees_e_employe_b4b0e7_idx'), + ), + migrations.AddConstraint( + model_name='employeedepartment', + constraint=models.UniqueConstraint(deferrable=django.db.models.constraints.Deferrable['DEFERRED'], fields=('department_id', 'employee_id'), name='uniq_employee_per_department'), + ), + ] diff --git a/backend/employees/models.py b/backend/employees/models.py index c70d944d..de340817 100644 --- a/backend/employees/models.py +++ b/backend/employees/models.py @@ -1,4 +1,5 @@ # backend\employees\models.py +from common.emails import send_templated_mail from django.contrib.auth.models import AbstractUser, BaseUserManager, Group, Permission from django.core.exceptions import ValidationError from django.db import IntegrityError, models, transaction @@ -8,8 +9,7 @@ from django.utils.crypto import get_random_string from phonenumber_field.modelfields import PhoneNumberField from simple_history.models import HistoricalRecords -from common.emails import send_templated_mail -from .constants import ACTION_CHOICES, ACTION_DISMISSED, GENDER_CHOICES +from .constants import ACTION_CHOICES, ACTION_DISMISSED, GENDER_CHOICES, DeptPerm class DateRangeMixin(models.Model): @@ -406,6 +406,14 @@ class Department(models.Model): return employees +class DepartmentPermission(models.Model): + code = models.CharField(max_length=64, choices=DeptPerm.CHOICES, unique=True) + name = models.CharField(max_length=150) + + def __str__(self): + return self.name or self.code + + class DepartmentRole(models.Model): """Роль внутри КОНКРЕТНОГО отдела, с правами, заданными начальником отдела.""" @@ -413,12 +421,15 @@ class DepartmentRole(models.Model): Department, on_delete=models.CASCADE, related_name="roles" ) name = models.CharField(max_length=150) - permissions = models.ManyToManyField(Permission, blank=True) + scoped_permissions = models.ManyToManyField( + DepartmentPermission, blank=True, related_name="roles" + ) class Meta: verbose_name = "Роль в отделе" verbose_name_plural = "Роли в отделе" unique_together = ("department", "name") + ordering = ("name", "department", "id") def __str__(self): return f"{self.department}: {self.name}" @@ -447,12 +458,14 @@ class EmployeeDepartment(DateRangeMixin, models.Model): verbose_name_plural = "Принадлежности к отделам" constraints = [ models.UniqueConstraint( - fields=["employee", "department"], name="uniq_employee_department" - ) + fields=["department_id", "employee_id"], + name="uniq_employee_per_department", + deferrable=models.Deferrable.DEFERRED, + ), ] indexes = [ - models.Index(fields=["department", "is_active"]), - models.Index(fields=["employee", "is_active"]), + models.Index(fields=["department_id", "is_active", "employee_id"]), + models.Index(fields=["employee_id", "department_id"]), ] def __str__(self): diff --git a/backend/employees/urls_front.py b/backend/employees/urls_front.py index f683bd4e..675c2743 100644 --- a/backend/employees/urls_front.py +++ b/backend/employees/urls_front.py @@ -19,7 +19,7 @@ urlpatterns = [ name="department_add", ), path( - "departments//members//role/", + "departments//roles/edit/", views_front.edit_department_role, name="edit_department_role", ), diff --git a/backend/employees/views_front.py b/backend/employees/views_front.py index cd645215..558fe68f 100644 --- a/backend/employees/views_front.py +++ b/backend/employees/views_front.py @@ -464,84 +464,24 @@ def department_list(request): # ---------- DETAIL ---------- @require_api_auth def department_detail(request, pk: int): + """ + Тонкая вьюха: берёт агрегированный контекст у API и рендерит шаблон. + """ api = get_api_client(request) - - dep = api.get(f"v1/departments/{pk}/") - if not dep.ok: - messages.error( - request, - "Отдел не найден" if dep.status == 404 else f"Ошибка API ({dep.status})", - ) + r = api.get(f"v1/departments/{pk}/ui-context/") + if not r.ok: + messages.error(request, "Отдел не найден" if r.status == 404 else f"Ошибка API ({r.status})") return redirect("employees:department_list") - dept = dep.json - roles_resp = api.get("v1/department-roles/", params={"department": pk, "page_size": 1000}) - roles = (roles_resp.json.get("results", []) - if roles_resp.ok and isinstance(roles_resp.json, dict) - else roles_resp.json or []) - - links_resp = api.get( - "v1/employee-departments/", - params={"department": pk, "is_active": True, "page_size": 1000}, - ) - links = (links_resp.json.get("results", []) - if links_resp.ok and isinstance(links_resp.json, dict) - else links_resp.json or []) - - # 👉 нормализуем участников для селекта руководителя - members = [] - for ln in links: - e = ln.get("employee") - # API может вернуть dict или id; поддержим оба варианта - if isinstance(e, dict): - emp_id = e.get("id") - last_name = e.get("last_name") or "" - first_name = e.get("first_name") or "" - patronymic = e.get("patronymic") or "" - full_name = " ".join([s for s in (last_name, first_name, patronymic) if s]) - members.append({ - "id": emp_id, - "name": full_name or f"Сотрудник #{emp_id}", - }) - elif e is not None: - # fallback, если пришёл только id - emp_id = e - members.append({ - "id": emp_id, - "name": f"Сотрудник #{emp_id}", - }) - - # Уберём дубликаты, если вдруг - seen = set() - head_choices = [] - for m in members: - mid = m.get("id") - if mid and mid not in seen: - head_choices.append(m) - seen.add(mid) - - context = { - "dept": dept, - "roles": roles, - "links": links, - "head_choices": head_choices, # 👈 добавили - "edit_form": DepartmentEditForm(initial={ - "name": dept.get("name", ""), - "description": dept.get("description", ""), - }), - "set_head_form": SetHeadForm(), - "set_member_role_form": SetMemberRoleForm(), - } - - page = request.GET.get("page") or 1 - r = api.get("v1/feed/posts/", params={"department": pk, "page": page, "ordering": "-created_at"}) - posts = [] - if r.ok: - data = r.json - posts = data.get("results", data) if isinstance(data, dict) else data - - context.update({"posts": posts, "page": page}) - return render(request, "employees/department_detail.html", context) + ctx = r.json if isinstance(r.json, dict) else {} + # Для обратной совместимости с шаблоном продублируем флаги как плоские поля: + perms = ctx.get("user_perms") or {} + ctx.update({ + "can_manage": perms.get("can_manage", False), + "can_change_head": perms.get("can_change_head", False), + "can_assign_roles": perms.get("can_assign_roles", False), + }) + return render(request, "employees/department_detail.html", ctx) # ---------- EDIT BASIC FIELDS ---------- @require_api_auth @@ -680,48 +620,78 @@ class DepartmentDeleteView(LoginRequiredMixin, StaffOnlyMixin, DeleteView): success_url = reverse_lazy("employees:department_list") -@login_required -def edit_department_role(request, dept_id, employee_id): - department = get_object_or_404(Department, pk=dept_id) +@require_api_auth +@require_POST +def edit_department_role(request, pk: int): + """ + Универсальный обработчик ролей отдела: + - op=create : POST v1/department-roles/ {department: pk, name} + - op=update : PATCH v1/department-roles// {name} + - op=delete : DELETE v1/department-roles// + """ + api = get_api_client(request) + op = (request.POST.get("op") or "").lower().strip() - # только staff или руководитель отдела - if not (request.user.is_staff or department.head_id == request.user.id): - messages.error(request, "Нет прав изменять роли в этом отделе.") - return redirect("employees:employees_list") + # Быстрая проверка прав (как и в шаблоне) + dept_resp = api.get(f"v1/departments/{pk}/") + if not dept_resp.ok: + messages.error(request, "Отдел не найден.") + return redirect("employees:department_detail", pk=pk) + dept = dept_resp.json + allowed = request.user.is_staff or (dept.get("head") and dept["head"].get("id") == request.user.id) + if not allowed: + messages.error(request, "Недостаточно прав для управления ролями отдела.") + return redirect("employees:department_detail", pk=pk) - link = get_object_or_404( - EmployeeDepartment, - department=department, - employee_id=employee_id, - ) + try: + if op == "create": + name = (request.POST.get("name") or "").strip() + if not name: + messages.error(request, "Название роли не может быть пустым.") + return redirect("employees:department_detail", pk=pk) - if request.method == "POST": - form = DepartmentMemberRoleForm(request.POST, instance=link) - if form.is_valid(): - form.save() - messages.success(request, "Роль в отделе обновлена.") - next_url = request.POST.get("next") or ( - reverse("employees:employees_list") + f"?department={dept_id}" - ) - return redirect(next_url) - else: - form = DepartmentMemberRoleForm(instance=link) + resp = api.post("v1/department-roles/", json={"department": pk, "name": name}) + if resp.ok: + messages.success(request, f"Роль «{name}» создана.") + else: + messages.error(request, f"Ошибка создания роли: {getattr(resp, 'status', '')} {resp.text or ''}") - employee = link.employee - next_url = request.GET.get("next") or ( - reverse("employees:employees_list") + f"?department={dept_id}" - ) - return render( - request, - "employees/department_role_form.html", - { - "department": department, - "employee": employee, - "form": form, - "next_url": next_url, - }, - ) + elif op == "update": + role_id = request.POST.get("role_id") + name = (request.POST.get("name") or "").strip() + # 1) Сохраняем имя + if name: + api.patch(f"v1/department-roles/{role_id}/", json={"name": name}) + + # 2) Сохраняем права роли + perm_ids = request.POST.getlist("permission_ids") # из чекбоксов + perm_ids = [int(x) for x in perm_ids if str(x).isdigit()] + resp = api.post(f"v1/department-roles/{role_id}/set_perms", json={"permission_ids": perm_ids}) + + if resp.ok: + messages.success(request, f"Роль обновлена: «{name}».") + else: + messages.error(request, f"Ошибка обновления роли: {getattr(resp, 'status', '')} {resp.text or ''}") + + elif op == "delete": + role_id = request.POST.get("role_id") + if not role_id: + messages.error(request, "Не указан идентификатор роли для удаления.") + return redirect("employees:department_detail", pk=pk) + + resp = api.delete(f"v1/department-roles/{role_id}/") + if resp.ok or getattr(resp, "status", None) == 204: + messages.success(request, "Роль удалена.") + else: + messages.error(request, f"Ошибка удаления роли: {getattr(resp, 'status', '')} {resp.text or ''}") + + else: + messages.error(request, "Неизвестная операция.") + except Exception as exc: + messages.error(request, f"Ошибка: {exc}") + + return redirect("employees:department_detail", pk=pk) # ---------- Приглашение в отдел ---------- @@ -897,7 +867,7 @@ def employee_set_position_front(request: HttpRequest, emp_id: int) -> JsonRespon if not position_id: return JsonResponse({"detail": "position_id is required"}, status=400) - + api = get_api_client(request) resp = api.patch(f"v1/employees/{emp_id}/", json={"position_id": position_id}) ok, data, status = _api_ok_or_error(resp) diff --git a/backend/env.example b/backend/env.example new file mode 100644 index 00000000..d2dd6898 --- /dev/null +++ b/backend/env.example @@ -0,0 +1,45 @@ +# --- Django --- +DJANGO_SECRET_KEY=change-me +DEBUG=True +USE_SQLITE=True +ALLOWED_HOSTS=localhost,127.0.0.1 + +# If you want to use Postgres instead of SQLite (set USE_SQLITE=False) +POSTGRES_USER=eusrr_user +POSTGRES_PASSWORD=change-me +POSTGRES_DB=eusrr +DB_HOST=postgres +DB_PORT=5432 + +# --- Secondary HikCentral DB (optional; map your own instance) --- +HIK_DB_NAME=cms +HIK_DB_USER=postgres +HIK_DB_PASSWORD=change-me +HIK_DB_HOST=postgres +HIK_DB_PORT=5432 + +# --- Redis for Channels/WebSockets --- +REDIS_HOST=redis +REDIS_PORT=6379 + +# --- Email (SMTP) --- +EMAIL_BACKEND=django.core.mail.backends.smtp.EmailBackend +EMAIL_HOST=smtp.example.com +EMAIL_PORT=465 +EMAIL_USE_SSL=true +EMAIL_USE_TLS=false +EMAIL_HOST_USER=no-reply@example.com +EMAIL_HOST_PASSWORD=change-me +DEFAULT_FROM_EMAIL=Robotail + +# --- Regional / API base --- +PHONE_DEFAULT_REGION=RU +CSRF_TRUSTED_ORIGINS=http://127.0.0.1:9000,http://localhost:9000 +API_BASE_URL=http://127.0.0.1:9000/api + +# --- Integrations (placeholders) --- +TELEGRAM_BOT_TOKEN=put-your-token-here +USERNAME_TELEGRAM_BOT=your_bot_username +SMS_API_TOKEN=... +SMS_API_URL=https://new.smsgorod.ru/apiSms/create/ +SMS_API_SENDER=EUSRR diff --git a/backend/templates/employees/_department_controls.html b/backend/templates/employees/_department_controls.html new file mode 100644 index 00000000..d6fcd765 --- /dev/null +++ b/backend/templates/employees/_department_controls.html @@ -0,0 +1,531 @@ +{# backend/templates/employees/_department_controls.html #} +{# Ожидаемый контекст: + dept: {id, name, description, head?} + head_choices: [{id, name}] # список кандидатов + roles: [{id, name, permissions: [ids]}] # для совместимости: ids выбранных прав + dept_perm_choices: [{id, code, name}] # справочник прав для ролей + can_manage: bool + can_assign_roles: bool + can_change_head: bool +#} + +{# ===== 1) Управление отделом ===== #} +{% if can_manage %} +
+
+ +
+
Управление отделом
+
Название и описание
+
+
+ +
+
+ {% csrf_token %} +
+
+ + +
+
+ + +
+
+ +
+
+
+
+
+{% endif %} + +{# ===== 2) Роли отдела ===== #} +{% if can_assign_roles %} +
+
+ +
Роли отдела
+
+ +
+
+ + {# === РЕДАКТИРОВАТЬ РОЛЬ (одна кнопка "Сохранить изменения") === #} +
+
+ {% csrf_token %} + + +
+
+ + +
+
+ + +
+ +
+
+ +
+
+ {# сюда JS при сохранении подложит скрытые permission_ids (если менялись) #} +
+
+ +
+ + +
+
+
+
+ + {# === СОЗДАТЬ РОЛЬ (панель показывается по кнопке) === #} +
+ + + +
+
+ {% csrf_token %} + +
+
+ + +
+ +
+
+ +
+
+ {# сюда JS перед сабмитом положит #} +
+
+ +
+ + +
+
+
+
+
+ +
{# /row #} +
+
+ + {# === МОДАЛ: выбор прав (редактирование) === #} + + + {# === МОДАЛ: выбор прав (создание) === #} + + + +{% endif %} + +{# ===== 3) Назначение руководителя ===== #} +{% if can_change_head %} +
+
+ +
+
Назначение руководителя
+
Выбор или снятие руководителя отдела
+
+
+ +
+
+ {% csrf_token %} +
+
+ + +
+
+ +
+
+
+
+
+{% endif %} + +{# Если нет ни одного права — информируем #} +{% if not can_manage and not can_assign_roles and not can_change_head %} +
+ У вас нет прав для управления этим отделом. +
+{% endif %} diff --git a/backend/templates/employees/department_detail.html b/backend/templates/employees/department_detail.html index 6852cab4..37ed5325 100644 --- a/backend/templates/employees/department_detail.html +++ b/backend/templates/employees/department_detail.html @@ -59,71 +59,8 @@ {# ===== Управление отделом (для staff/главы) ===== #} {% if user.is_staff or dept.head and dept.head.id == user.id %} -
-
- -
-
Управление отделом
-
Название, описание и руководитель
-
-
-
-
+ {% include "employees/_department_controls.html" with dept=dept edit_form=edit_form head_choices=head_choices %} - {# Редактирование name/description -> /employees/departments//edit/ #} -
-
- {% csrf_token %} -
- - -
-
- - -
- -
-
- - {# Назначение/снятие руководителя -> /employees/departments//set-head/ #} -
-
- {% csrf_token %} - - -
- -
-
-
- -
-
-
{% endif %} {# ===== Команда отдела (только сотрудники с назначённой ролью) ===== #} diff --git a/backend/tests/api/v1/employees/test_department_head_rights.py b/backend/tests/api/v1/employees/test_department_head_rights.py index 36ab1ae4..318c7375 100644 --- a/backend/tests/api/v1/employees/test_department_head_rights.py +++ b/backend/tests/api/v1/employees/test_department_head_rights.py @@ -1,29 +1,51 @@ +# backend/tests/api/v1/employees/test_department_head_rights.py import pytest from django.urls import reverse -from django.utils import timezone -from django.contrib.auth.models import Permission +from django.contrib.auth import get_user_model +from rest_framework.test import APIClient from rest_framework import status -from employees.models import Employee, Department, DepartmentRole, EmployeeDepartment +from employees.models import Department, DepartmentRole, EmployeeDepartment, Employee -# --- helpers --- - -_seq = 1 +User = get_user_model() -def _unique_email(prefix="user"): - global _seq - _seq += 1 - return f"{prefix}{_seq}@example.com" +# ========================= +# Helpers +# ========================= -def _unique_phone(): - global _seq - _seq += 1 - return f"+7999{_seq:07d}" +def _unique_phone() -> str: + base = 79000000000 # 79 + 9 нулей + return str(base + User.objects.count()) + + +def make_user(email: str, staff: bool = False, verified: bool = True) -> User: + + u = User.objects.create_user( + email=email, + password="pwd12345", + phone_number=_unique_phone(), + send_activation_email=False, + email_verified=True, + ) + # 👇 явная активация для тестов управленческих прав + if not u.is_active: + u.is_active = True + u.save(update_fields=["is_active"]) + return u + + +def _unique_email() -> str: + base = "asd@mail.com" + return str(Employee.objects.count()) + base def _make_user(staff=False, superuser=False) -> Employee: + """ + Создаёт активного и 'email_verified' пользователя, чтобы не споткнуться о + валидацию при назначении руководителя. + """ u = Employee.objects.create_user( email=_unique_email(), password="pass", @@ -34,149 +56,114 @@ def _make_user(staff=False, superuser=False) -> Employee: ) u.is_staff = staff u.is_superuser = superuser - u.email_verified = True - u.is_active = True + u.email_verified = True # <- критично для твоей нынешней валидации + u.is_active = True # <- тоже нужен для твоей ветки is_active u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"]) return u -def _dept_set_head(api_client, dept_id, head_id): - url = reverse("api:v1:departments-set-head", args=[dept_id]) - payload = {"head_id": head_id} if head_id is not None else {"head_id": None} - return api_client.post(url, payload, format="json") +def make_role( + dept: Department, name: str, codes: list[str] | None = None +) -> DepartmentRole: + role = DepartmentRole.objects.create(department=dept, name=name) + # в этих тестах права роли не нужны (руководитель имеет доступ сам по себе), + # но фабрика оставлена для удобства + return role -# --- tests --- - - -@pytest.mark.django_db -def test_old_head_loses_rights_after_change(api_client): - # setup: old head - old_head = _make_user() - api_client.force_authenticate(user=old_head) - dept = Department.objects.create(name="Alpha", head=old_head) - - # Как глава — может править отдел (partial_update требует manage_department) - url_detail = reverse("api:v1:departments-detail", args=[dept.id]) - resp = api_client.patch(url_detail, {"description": "init"}, format="json") - assert resp.status_code == 200 - - # Назначаем нового главу (делает текущий глава) - new_head = _make_user() - resp = _dept_set_head(api_client, dept.id, new_head.id) - assert resp.status_code == 200 - assert resp.data["head_id"] == new_head.id - - # Старый глава больше не глава -> прав на управление отделом больше нет - resp = api_client.patch(url_detail, {"description": "should fail"}, format="json") - assert resp.status_code == status.HTTP_403_FORBIDDEN - - -@pytest.mark.django_db -def test_new_head_gets_rights_and_membership_created(api_client): - # setup: без главы - creator = _make_user() # просто актор для вызовов - api_client.force_authenticate(user=creator) - dept = Department.objects.create(name="Beta") - - # Назначаем нового главу (создатель не обязан иметь права — у вас это ограничивается IsDeptManagerForWrite; - # для чистоты будем действовать от staff) - staff = _make_user(staff=True) - api_client.force_authenticate(user=staff) - new_head = _make_user() - - resp = _dept_set_head(api_client, dept.id, new_head.id) - assert resp.status_code == 200 - - # У нового главы есть права управления отделом - api_client.force_authenticate(user=new_head) - url_detail = reverse("api:v1:departments-detail", args=[dept.id]) - resp = api_client.patch(url_detail, {"description": "ok"}, format="json") - assert resp.status_code == 200 - - # И ему создана активная связь членства - link = EmployeeDepartment.objects.filter( - employee=new_head, department=dept, is_active=True - ).first() - assert link is not None - assert link.date_from is not None - - -@pytest.mark.django_db -def test_remove_head_deactivates_membership_and_revokes_rights(api_client): - # setup: есть глава - head = _make_user() - staff = _make_user(staff=True) - dept = Department.objects.create(name="Gamma", head=head) - - # как глава он может управлять - api_client.force_authenticate(user=head) - url_detail = reverse("api:v1:departments-detail", args=[dept.id]) - assert ( - api_client.patch(url_detail, {"description": "ok"}, format="json").status_code - == 200 - ) - - # staff снимает главу (head -> null) - api_client.force_authenticate(user=staff) - resp = _dept_set_head(api_client, dept.id, None) - assert resp.status_code == status.HTTP_204_NO_CONTENT - - # связь старого главы должна стать неактивной и получить date_to - link = EmployeeDepartment.objects.filter(employee=head, department=dept).first() - assert link is not None - assert link.is_active is False - assert link.date_to is not None - - # и прав больше нет - api_client.force_authenticate(user=head) - resp = api_client.patch(url_detail, {"description": "nope"}, format="json") - assert resp.status_code == status.HTTP_403_FORBIDDEN - - -@pytest.mark.django_db -def test_old_head_keeps_rights_if_role_grants_after_change(api_client): - # setup: старый глава - old_head = _make_user() - dept = Department.objects.create(name="Delta", head=old_head) - - # подготовим роль отдела с manage_department и назначим её старому главе (как обычному члену) - role = DepartmentRole.objects.create(department=dept, name="ManagerRole") - p_manage = Permission.objects.get( - content_type__app_label="employees", codename="manage_department" - ) - role.permissions.add(p_manage) +def add_member( + user: User, + dept: Department, + role: DepartmentRole | None = None, + is_active: bool = True, +): EmployeeDepartment.objects.update_or_create( - employee=old_head, + employee=user, department=dept, - defaults={"role": role, "is_active": True, "date_from": timezone.now().date()}, + defaults={"is_active": is_active, "role": role}, ) - # меняем главу - staff = _make_user(staff=True) - api_client.force_authenticate(user=staff) - new_head = _make_user() - assert _dept_set_head(api_client, dept.id, new_head.id).status_code == 200 - # старый глава уже не head, но он обладает ролью с manage_department -> должен иметь доступ - api_client.force_authenticate(user=old_head) - url_detail = reverse("api:v1:departments-detail", args=[dept.id]) - resp = api_client.patch(url_detail, {"description": "still ok"}, format="json") - assert resp.status_code == 200 +# ========================= +# Fixtures +# ========================= + + +@pytest.fixture() +def api_client() -> APIClient: + return APIClient() + + +# ========================= +# Tests +# ========================= @pytest.mark.django_db -def test_change_head_does_not_deactivate_old_membership_by_default(api_client): - # при смене head -> старый head остаётся активным членом (согласно текущей логике save()) - old_head = _make_user() - staff = _make_user(staff=True) - api_client.force_authenticate(user=staff) - dept = Department.objects.create(name="Epsilon", head=old_head) +def test_head_can_update_department_without_explicit_role_perms(api_client: APIClient): + head = make_user("head@example.com") + d = Department.objects.create(name="Dept", description="old", head=head) - new_head = _make_user() - assert _dept_set_head(api_client, dept.id, new_head.id).status_code == 200 + # руководитель аутентифицируется + api_client.force_authenticate(user=head) + url = reverse("api:v1:departments-detail", args=[d.id]) - # Проверяем, что связь старого главы осталась активной (деактивируется только когда head -> None) - link = EmployeeDepartment.objects.filter(employee=old_head, department=dept).first() - assert link is not None - assert link.is_active is True + # PATCH произвольного поля (не head) должен быть разрешён руководителю + resp = api_client.patch(url, {"description": "new"}, format="json") + assert resp.status_code == status.HTTP_200_OK + assert resp.json()["description"] == "new" + + +@pytest.mark.django_db +def test_head_can_set_member_role_and_non_head_cannot(api_client: APIClient): + head = make_user("head@example.com") + d = Department.objects.create(name="Dept", head=head) + + worker = make_user("worker@example.com") + add_member(worker, d) # сотрудник состоит в отделе + + role_worker = make_role(d, "Worker") + + # руководитель назначает роль участнику + api_client.force_authenticate(user=head) + url = reverse("api:v1:departments-set-member-role", args=[d.id]) + payload = {"employee_id": worker.id, "role_id": role_worker.id, "is_active": True} + resp = api_client.post(url, payload, format="json") + assert resp.status_code == status.HTTP_200_OK + data = resp.json() + assert data["employee_id"] == worker.id + assert data["role_id"] == role_worker.id + assert data["is_active"] is True + + # другой пользователь без статуса руководителя — не может + stranger = make_user("stranger@example.com") + api_client.force_authenticate(user=stranger) + resp = api_client.post(url, payload, format="json") + assert resp.status_code == status.HTTP_403_FORBIDDEN + + +@pytest.mark.django_db +def test_head_can_change_head_and_loses_rights_afterwards(api_client: APIClient): + old_head = make_user("old@example.com") + d = Department.objects.create(name="Dept", head=old_head) + + new_head = make_user("new@example.com") + + # старый руководитель меняет руководителя на другого + api_client.force_authenticate(user=old_head) + url_set_head = reverse("api:v1:departments-set-head", args=[d.id]) + print("DEBUG:", new_head.id, new_head.email_verified, new_head.is_active) + resp = api_client.post(url_set_head, {"head_id": new_head.id}, format="json") + assert resp.status_code == status.HTTP_200_OK + assert resp.json()["head"]["id"] == new_head.id + + # теперь старый руководитель больше не глава — у него не должно быть прав управления отделом + url_detail = reverse("api:v1:departments-detail", args=[d.id]) + resp = api_client.patch(url_detail, {"description": "after"}, format="json") + assert resp.status_code == status.HTTP_403_FORBIDDEN + + # а у нового руководителя — есть + api_client.force_authenticate(user=new_head) + resp = api_client.patch(url_detail, {"description": "ok"}, format="json") + assert resp.status_code == status.HTTP_200_OK + assert resp.json()["description"] == "ok" diff --git a/backend/tests/api/v1/employees/test_department_roles.py b/backend/tests/api/v1/employees/test_department_roles.py index fe426a75..ef268c1e 100644 --- a/backend/tests/api/v1/employees/test_department_roles.py +++ b/backend/tests/api/v1/employees/test_department_roles.py @@ -1,176 +1,305 @@ +# backend/tests/api/v1/employees/test_department_roles.py import pytest from django.urls import reverse -from django.contrib.auth.models import Permission +from django.contrib.auth import get_user_model +from rest_framework.test import APIClient from rest_framework import status -from employees.models import Employee, Department, DepartmentRole, EmployeeDepartment +from employees.models import ( + Department, + DepartmentRole, + DepartmentPermission, + EmployeeDepartment, # <-- если у тебя EmployeeDepartmentLink, замени импорт и упоминания ниже +) -# --- helpers --- +User = get_user_model() -_seq = 1 -def _unique_email(prefix="user"): - global _seq - _seq += 1 - return f"{prefix}{_seq}@example.com" -def _unique_phone(): - global _seq - _seq += 1 - return f"+7999{_seq:07d}" +# ========================= +# Helpers +# ========================= -def _make_user(staff=False, superuser=False) -> Employee: - u = Employee.objects.create_user( - email=_unique_email(), password="pass", phone_number=_unique_phone(), - send_activation_email=False, first_name="T", last_name="U", +def _unique_phone() -> str: + """ + Делает уникальный 11-значный номер: 79xxxxxxxxx + Используем текущее количество пользователей как инкремент. + """ + base = 79000000000 # 79 + 9 нулей + return str(base + User.objects.count()) + +def make_user(email: str, staff: bool = False, verified: bool = True) -> User: + """ + Создаёт пользователя с обязательным phone_number и отключённой отправкой активационного письма. + Поле 'verified' ставим только если оно существует в модели. + """ + extra = { + "phone_number": _unique_phone(), + "is_staff": staff, + "send_activation_email": False, + } + # Добавим verified, если поле есть в модели + try: + User._meta.get_field("verified") + extra["verified"] = verified + except Exception: + pass + + return User.objects.create_user( + email=email, + password="pwd12345", + **extra, ) - u.is_staff = staff - u.is_superuser = superuser - u.email_verified = True - u.is_active = True - u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"]) - return u -def _grant_perm(user: Employee, perm_code: str): - app_label, codename = perm_code.split(".", 1) - p = Permission.objects.get(content_type__app_label=app_label, codename=codename) - user.user_permissions.add(p) - user.save() - # сброс кэша прав на всякий случай - for attr in ("_perm_cache", "_user_perm_cache", "_group_perm_cache"): - if hasattr(user, attr): - try: delattr(user, attr) - except: pass - return p +def ensure_dept_perm(code: str, name: str | None = None) -> DepartmentPermission: + return DepartmentPermission.objects.get_or_create( + code=code, defaults={"name": name or code} + )[0] -# --- tests --- +def make_role(dept: Department, name: str, codes: list[str] | None = None) -> DepartmentRole: + role = DepartmentRole.objects.create(department=dept, name=name) + if codes: + perms = [ensure_dept_perm(c) for c in codes] + role.scoped_permissions.add(*perms) + return role + +def grant_assign_in_dept(user: User, dept: Department) -> DepartmentRole: + role = make_role(dept, "assigner", ["assign_department_role"]) + EmployeeDepartment.objects.update_or_create( + employee=user, department=dept, + defaults={"is_active": True, "role": role}, + ) + return role + + +# ========================= +# Fixtures +# ========================= + +@pytest.fixture() +def api_client() -> APIClient: + return APIClient() + + +# ========================= +# Tests +# ========================= @pytest.mark.django_db -def test_list_and_filter_requires_auth(api_client): +def test_list_filtered_by_department(api_client: APIClient): + d1 = Department.objects.create(name="Dept A") + d2 = Department.objects.create(name="Dept B") + + r1 = make_role(d1, "Engineer", []) + r2 = make_role(d1, "Manager", []) + r3 = make_role(d2, "Sales", []) + + user = make_user("u@example.com") + api_client.force_authenticate(user=user) + url = reverse("api:v1:department-roles-list") - # unauth -> 401/403 - resp = api_client.get(url) - assert resp.status_code in (401, 403) - - # auth -> 200, фильтр по department - user = _make_user() - api_client.force_authenticate(user=user) - d1 = Department.objects.create(name="Sales") - d2 = Department.objects.create(name="Support") - DepartmentRole.objects.create(department=d1, name="Manager") - DepartmentRole.objects.create(department=d2, name="Operator") - + # без фильтра — все 3 resp = api_client.get(url) assert resp.status_code == 200 - assert len(resp.data) == 2 + body = resp.json() + items = body.get("results", body) + ids = {it["id"] for it in items} + assert ids == {r1.id, r2.id, r3.id} + # с фильтром по d1 — только роли d1 resp = api_client.get(url, {"department": d1.id}) assert resp.status_code == 200 - assert len(resp.data) == 1 - assert resp.data[0]["name"] == "Manager" + body = resp.json() + items = body.get("results", body) + ids = {it["id"] for it in items} + assert ids == {r1.id, r2.id} + @pytest.mark.django_db -def test_head_can_crud_in_own_department(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="R&D", head=head) - # гарантируем активную связь головы с отделом - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) +def test_create_requires_assign_department_role(api_client: APIClient): + d = Department.objects.create(name="Dept") + user = make_user("m@example.com") + api_client.force_authenticate(user=user) url = reverse("api:v1:department-roles-list") - # create в своём отделе - resp = api_client.post(url, {"department": dept.id, "name": "Lead"}, format="json") - assert resp.status_code == status.HTTP_201_CREATED - role_id = resp.data["id"] - - # update - url_detail = reverse("api:v1:department-roles-detail", args=[role_id]) - resp = api_client.patch(url_detail, {"name": "Lead++"}, format="json") - assert resp.status_code == 200 - - # delete - resp = api_client.delete(url_detail) - assert resp.status_code == status.HTTP_204_NO_CONTENT - -@pytest.mark.django_db -def test_head_cannot_crud_in_foreign_department(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - own = Department.objects.create(name="Own", head=head) - foreign = Department.objects.create(name="Foreign") - EmployeeDepartment.objects.get_or_create(employee=head, department=own, defaults={"is_active": True}) - - # попытка создать роль в чужом отделе -> 403 - url = reverse("api:v1:department-roles-list") - resp = api_client.post(url, {"department": foreign.id, "name": "X"}, format="json") + # без прав — 403 + resp = api_client.post(url, {"department": d.id, "name": "Worker"}, format="json") assert resp.status_code == status.HTTP_403_FORBIDDEN + # выдаём право assign_department_role в этом отделе + grant_assign_in_dept(user, d) + + # теперь можно создать + resp = api_client.post(url, {"department": d.id, "name": "Worker"}, format="json") + assert resp.status_code == status.HTTP_201_CREATED + data = resp.json() + assert data["department"] == d.id + # back-compat поле присутствует + assert "permissions" in data and isinstance(data["permissions"], list) + + @pytest.mark.django_db -def test_non_head_needs_manage_for_crud_and_assign_for_perms(api_client): - user = _make_user() +def test_update_and_destroy_scope_enforced(api_client: APIClient): + d1 = Department.objects.create(name="Dept1") + d2 = Department.objects.create(name="Dept2") + user = make_user("m@example.com") api_client.force_authenticate(user=user) - dept = Department.objects.create(name="Ops") - # нет ни head, ни ролей — CRUD запрещён - url = reverse("api:v1:department-roles-list") - resp = api_client.post(url, {"department": dept.id, "name": "Role"}, format="json") - assert resp.status_code == 403 - # даём право manage_department -> можно создавать/редактировать/удалять - _grant_perm(user, "employees.manage_department") - resp = api_client.post(url, {"department": dept.id, "name": "Role"}, format="json") - assert resp.status_code == 201 - role_id = resp.data["id"] + # роль в другом отделе + role_other = make_role(d2, "Other", []) + url_detail_other = reverse("api:v1:department-roles-detail", args=[role_other.id]) - url_detail = reverse("api:v1:department-roles-detail", args=[role_id]) - resp = api_client.patch(url_detail, {"name": "Role2"}, format="json") - assert resp.status_code == 200 + # нет прав в d2 -> 403 + resp = api_client.patch(url_detail_other, {"name": "X"}, format="json") + assert resp.status_code == status.HTTP_403_FORBIDDEN - # но set_perms всё ещё нельзя: нужен assign_department_role - url_set = reverse("api:v1:department-roles-set-perms", args=[role_id]) - resp = api_client.post(url_set, {"permission_ids": []}, format="json") - assert resp.status_code == 403 + # выдаём права в d1 — всё ещё 403 для d2 + grant_assign_in_dept(user, d1) + resp = api_client.patch(url_detail_other, {"name": "X"}, format="json") + assert resp.status_code == status.HTTP_403_FORBIDDEN + + # выдаём права в d2 — теперь можно + grant_assign_in_dept(user, d2) + resp = api_client.patch(url_detail_other, {"name": "X"}, format="json") + assert resp.status_code == status.HTTP_200_OK + assert resp.json()["name"] == "X" + + # destroy с правами — 204/200 + resp = api_client.delete(url_detail_other) + assert resp.status_code in (status.HTTP_204_NO_CONTENT, status.HTTP_200_OK) - # выдаём assign_department_role -> теперь можно - _grant_perm(user, "employees.assign_department_role") - # выберем 2 реальных права из employees - p1 = Permission.objects.get(content_type__app_label="employees", codename="manage_department") - p2 = Permission.objects.get(content_type__app_label="employees", codename="assign_department_role") - resp = api_client.post(url_set, {"permission_ids": [p1.id, p2.id]}, format="json") - assert resp.status_code == 200 - assert set(DepartmentRole.objects.get(id=role_id).permissions.values_list("id", flat=True)) == {p1.id, p2.id} @pytest.mark.django_db -def test_perms_listing_and_add_remove(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="QA", head=head) - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) +def test_perm_choices_and_perms_and_set_perms(api_client: APIClient): + # засеем три стандартных кода + ensure_dept_perm("manage_department", "Управлять отделом") + ensure_dept_perm("change_department_head", "Назначать руководителя") + ensure_dept_perm("assign_department_role", "Назначать роли") - role = DepartmentRole.objects.create(department=dept, name="Reviewer") - p_view = Permission.objects.get(content_type__app_label="employees", codename="view_employee") - p_manage = Permission.objects.get(content_type__app_label="employees", codename="manage_department") + d = Department.objects.create(name="Dept") + user = make_user("m@example.com") + api_client.force_authenticate(user=user) + grant_assign_in_dept(user, d) - # add_perms (как head) - url_add = reverse("api:v1:department-roles-add-perms", args=[role.id]) - resp = api_client.post(url_add, {"permission_ids": [p_view.id]}, format="json") + role = make_role(d, "Engineer", []) + + # perm_choices + url_choices = reverse("api:v1:department-roles-perm-choices") + resp = api_client.get(url_choices) assert resp.status_code == 200 + codes = {row["code"] for row in resp.json().get("results", [])} + assert {"manage_department", "change_department_head", "assign_department_role"}.issubset(codes) + + # set_perms (codes) + url_set = reverse("api:v1:department-roles-set-perms", args=[role.id]) + resp = api_client.post(url_set, {"permission_codes": ["manage_department", "assign_department_role"]}, format="json") + assert resp.status_code == 200 + data = resp.json() + verbose = {p["code"] for p in data.get("permissions_verbose", [])} + assert verbose == {"manage_department", "assign_department_role"} # perms (GET) url_perms = reverse("api:v1:department-roles-perms", args=[role.id]) resp = api_client.get(url_perms) assert resp.status_code == 200 - codes = {row["codename"] for row in resp.data["results"]} - assert "employees.view_employee" in codes + codes_now = {p["code"] for p in resp.json().get("results", [])} + assert codes_now == {"manage_department", "assign_department_role"} - # remove_perms - url_rm = reverse("api:v1:department-roles-remove-perms", args=[role.id]) - resp = api_client.post(url_rm, {"permission_ids": [p_view.id]}, format="json") + # set_perms (ids) — заменим на один id + dp_manage = DepartmentPermission.objects.get(code="manage_department") + resp = api_client.post(url_set, {"permission_ids": [dp_manage.id]}, format="json") assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == set() + role.refresh_from_db() + assert set(role.scoped_permissions.values_list("code", flat=True)) == {"manage_department"} - # set_perms + # set_perms (пусто) — очистить все + resp = api_client.post(url_set, {"permission_codes": []}, format="json") + assert resp.status_code == 200 + role.refresh_from_db() + assert role.scoped_permissions.count() == 0 + + +@pytest.mark.django_db +def test_set_perms_validates_payload(api_client: APIClient): + d = Department.objects.create(name="Dept") + user = make_user("m@example.com") + api_client.force_authenticate(user=user) + grant_assign_in_dept(user, d) + + role = make_role(d, "QA", []) + + # несуществующий код -> 400 url_set = reverse("api:v1:department-roles-set-perms", args=[role.id]) - resp = api_client.post(url_set, {"permission_ids": [p_manage.id]}, format="json") + resp = api_client.post(url_set, {"permission_codes": ["no_such_code"]}, format="json") + assert resp.status_code == 400 + + # несуществующий id -> 400 + resp = api_client.post(url_set, {"permission_ids": [999999]}, format="json") + assert resp.status_code == 400 + + +@pytest.mark.django_db +def test_back_compat_permissions_field_contains_ids(api_client: APIClient): + d = Department.objects.create(name="Dept") + user = make_user("m@example.com") + api_client.force_authenticate(user=user) + grant_assign_in_dept(user, d) + + role = make_role(d, "Dev", []) + dp_manage = ensure_dept_perm("manage_department") + dp_assign = ensure_dept_perm("assign_department_role") + role.scoped_permissions.add(dp_manage, dp_assign) + + url_detail = reverse("api:v1:department-roles-detail", args=[role.id]) + resp = api_client.get(url_detail) assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == {p_manage.id} + data = resp.json() + assert set(data["permissions"]) == {dp_manage.id, dp_assign.id} + + +@pytest.mark.django_db +def test_create_with_scoped_permission_codes_and_update_name(api_client: APIClient): + d = Department.objects.create(name="Dept") + ensure_dept_perm("manage_department") + + m = make_user("m@example.com") + api_client.force_authenticate(user=m) + grant_assign_in_dept(m, d) + + url_list = reverse("api:v1:department-roles-list") + + # create с codes + resp = api_client.post( + url_list, + { + "department": d.id, + "name": "Lead", + "scoped_permission_codes": ["manage_department"], + }, + format="json", + ) + assert resp.status_code == 201 + rid = resp.json()["id"] + + # update имени (PATCH) + url_detail = reverse("api:v1:department-roles-detail", args=[rid]) + resp = api_client.patch(url_detail, {"name": "Lead+1"}, format="json") + assert resp.status_code == 200 + assert resp.json()["name"] == "Lead+1" + + +@pytest.mark.django_db +def test_cross_department_access_denied_for_set_perms(api_client: APIClient): + d1 = Department.objects.create(name="Dept1") + d2 = Department.objects.create(name="Dept2") + + user = make_user("m@example.com") + api_client.force_authenticate(user=user) + grant_assign_in_dept(user, d1) # права только в Dept1 + + role_d2 = make_role(d2, "R", []) + url_set = reverse("api:v1:department-roles-set-perms", args=[role_d2.id]) + + resp = api_client.post(url_set, {"permission_codes": ["manage_department"]}, format="json") + assert resp.status_code == 403 diff --git a/backend/tests/api/v1/employees/test_department_roles_extra.py b/backend/tests/api/v1/employees/test_department_roles_extra.py index e8a104bb..f701e1de 100644 --- a/backend/tests/api/v1/employees/test_department_roles_extra.py +++ b/backend/tests/api/v1/employees/test_department_roles_extra.py @@ -1,285 +1,303 @@ +# backend/tests/api/v1/employees/test_department_roles_extra.py import pytest from django.urls import reverse -from django.contrib.auth.models import Permission +from django.contrib.auth import get_user_model +from rest_framework.test import APIClient from rest_framework import status from employees.models import ( - Employee, Department, DepartmentRole, - EmployeeDepartment, + DepartmentPermission, + EmployeeDepartment, # если у вас EmployeeDepartmentLink, замените импорт и упоминания ниже ) -# --- helpers (локальные, чтобы тесты были самодостаточными) --- +User = get_user_model() -_seq = 1 -def _unique_email(prefix="user"): - global _seq - _seq += 1 - return f"{prefix}{_seq}@example.com" -def _unique_phone(): - global _seq - _seq += 1 - return f"+7999{_seq:07d}" +# ========================= +# Helpers +# ========================= -def _make_user(staff=False, superuser=False) -> Employee: - u = Employee.objects.create_user( - email=_unique_email(), password="pass", - phone_number=_unique_phone(), - send_activation_email=False, first_name="T", last_name="U", + +def _unique_phone() -> str: + base = 79000000000 # 79 + 9 нулей + return str(base + User.objects.count()) + + +def make_user(email: str, staff: bool = False, verified: bool = True) -> User: + extra = { + "phone_number": _unique_phone(), + "is_staff": staff, + "send_activation_email": False, + } + # добавим verified если поле есть + try: + User._meta.get_field("verified") + extra["verified"] = verified + except Exception: + pass + return User.objects.create_user(email=email, password="pwd12345", **extra) + + +def ensure_perm(code: str, name: str | None = None) -> DepartmentPermission: + return DepartmentPermission.objects.get_or_create( + code=code, defaults={"name": name or code} + )[0] + + +def make_role( + dept: Department, name: str, codes: list[str] | None = None +) -> DepartmentRole: + role = DepartmentRole.objects.create(department=dept, name=name) + if codes: + perms = [ensure_perm(c) for c in codes] + role.scoped_permissions.add(*perms) + return role + + +def grant_assign_in_dept(user: User, dept: Department) -> DepartmentRole: + role = make_role(dept, "assigner", ["assign_department_role"]) + EmployeeDepartment.objects.update_or_create( + employee=user, + department=dept, + defaults={"is_active": True, "role": role}, ) - u.is_staff = staff - u.is_superuser = superuser - u.email_verified = True - u.is_active = True - u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"]) - return u + return role -def _grant_perm(user: Employee, perm_code: str): - app_label, codename = perm_code.split(".", 1) - p = Permission.objects.get(content_type__app_label=app_label, codename=codename) - user.user_permissions.add(p) - user.save() - # сброс кэша прав — иначе has_perm может брать старый кеш - for attr in ("_perm_cache", "_user_perm_cache", "_group_perm_cache"): - if hasattr(user, attr): - try: - delattr(user, attr) - except Exception: - pass - return p -# --- tests --- +# ========================= +# Fixtures +# ========================= + + +@pytest.fixture() +def api_client() -> APIClient: + return APIClient() + + +# ========================= +# Tests +# ========================= + @pytest.mark.django_db -def test_perms_endpoint_requires_auth(api_client): - # prepare - d = Department.objects.create(name="A") - role = DepartmentRole.objects.create(department=d, name="R") +def test_unauth_cannot_list_or_get(): + client = APIClient() + url_list = reverse("api:v1:department-roles-list") + assert client.get(url_list).status_code in (401, 403) - url = reverse("api:v1:department-roles-perms", args=[role.id]) - # unauth -> 401/403 - resp = api_client.get(url) - assert resp.status_code in (401, 403) +@pytest.mark.django_db +def test_read_only_allowed_without_assign(api_client: APIClient): + d = Department.objects.create(name="Dept") + r = make_role(d, "Worker", []) + u = make_user("u@example.com") + api_client.force_authenticate(user=u) - # auth (любой) -> 200 - user = _make_user() - api_client.force_authenticate(user=user) + # list + url_list = reverse("api:v1:department-roles-list") + resp = api_client.get(url_list, {"department": d.id}) + assert resp.status_code == 200 + items = resp.json().get("results", resp.json()) + assert any(x["id"] == r.id for x in items) + + # retrieve + url_detail = reverse("api:v1:department-roles-detail", args=[r.id]) + assert api_client.get(url_detail).status_code == 200 + + +@pytest.mark.django_db +def test_staff_override_for_write(api_client: APIClient): + d = Department.objects.create(name="Dept") + staff = make_user("s@example.com", staff=True) + api_client.force_authenticate(user=staff) + + # create + url_list = reverse("api:v1:department-roles-list") + resp = api_client.post( + url_list, {"department": d.id, "name": "SRole"}, format="json" + ) + assert resp.status_code == 201 + rid = resp.json()["id"] + + # update + url_detail = reverse("api:v1:department-roles-detail", args=[rid]) + assert ( + api_client.patch(url_detail, {"name": "SRole+"}, format="json").status_code + == 200 + ) + + # set_perms + ensure_perm("manage_department") + url_set = reverse("api:v1:department-roles-set-perms", args=[rid]) + assert ( + api_client.post( + url_set, {"permission_codes": ["manage_department"]}, format="json" + ).status_code + == 200 + ) + + +@pytest.mark.django_db +def test_ordering_stable_by_name_then_id(api_client: APIClient): + d1 = Department.objects.create(name="Dept1") + d2 = Department.objects.create(name="Dept2") + u = make_user("u@example.com") + api_client.force_authenticate(user=u) + + # одинаковые имена в РАЗНЫХ отделах — не нарушаем уникальность (department, name) + r1 = make_role(d1, "A", []) + r2 = make_role(d2, "A", []) + r3 = make_role(d1, "B", []) + + url = reverse("api:v1:department-roles-list") + resp = api_client.get(url, {"ordering": "name"}) # без фильтра по department + assert resp.status_code == 200 + items = resp.json().get("results", resp.json()) + + picked = [x["id"] for x in items if x["id"] in {r1.id, r2.id, r3.id}] + # Ожидаем: сначала две "A" по возрастанию id, затем "B" + assert len(picked) == 3 + assert picked[0] in (r1.id, r2.id) and picked[1] in (r1.id, r2.id) + assert picked[0] < picked[1] # тай-брейк по id для одинаковых name + assert picked[2] == r3.id + + +@pytest.mark.django_db +def test_set_perms_ids_take_precedence_over_codes(api_client: APIClient): + d = Department.objects.create(name="Dept") + u = make_user("u@example.com") + api_client.force_authenticate(user=u) + grant_assign_in_dept(u, d) + + role = make_role(d, "R", []) + dp_manage = ensure_perm("manage_department") + dp_assign = ensure_perm("assign_department_role") + + url = reverse("api:v1:department-roles-set-perms", args=[role.id]) + # передаём одновременно ids и codes — по вьюхе приоритет у ids + resp = api_client.post( + url, + { + "permission_ids": [dp_manage.id], + "permission_codes": ["assign_department_role"], + }, + format="json", + ) + assert resp.status_code == 200 + role.refresh_from_db() + codes = set(role.scoped_permissions.values_list("code", flat=True)) + assert codes == {"manage_department"} # ids приоритетнее + + +@pytest.mark.django_db +def test_serializer_codes_override_scoped_permissions_on_create(api_client: APIClient): + d = Department.objects.create(name="Dept") + ensure_perm("manage_department") + ensure_perm("assign_department_role") + + u = make_user("u@example.com") + api_client.force_authenticate(user=u) + grant_assign_in_dept(u, d) + + # создадим роль, передав и ids, и codes — в сериализаторе codes должны перезаписать ids + dp_manage = DepartmentPermission.objects.get(code="manage_department") + url = reverse("api:v1:department-roles-list") + resp = api_client.post( + url, + { + "department": d.id, + "name": "Lead", + "scoped_permissions": [dp_manage.id], # сначала ids + "scoped_permission_codes": [ + "assign_department_role" + ], # потом codes — должны победить + }, + format="json", + ) + assert resp.status_code == 201 + rid = resp.json()["id"] + role = DepartmentRole.objects.get(id=rid) + codes = set(role.scoped_permissions.values_list("code", flat=True)) + assert codes == {"assign_department_role"} + + +@pytest.mark.django_db +def test_perm_choices_fields_and_nonempty(api_client: APIClient): + ensure_perm("manage_department", "Управлять") + ensure_perm("change_department_head", "Глава") + ensure_perm("assign_department_role", "Назначение") + + u = make_user("u@example.com") + api_client.force_authenticate(user=u) + url = reverse("api:v1:department-roles-perm-choices") resp = api_client.get(url) assert resp.status_code == 200 - assert isinstance(resp.data, dict) - assert "results" in resp.data + results = resp.json().get("results", []) + assert results and {"id", "code", "name"}.issubset(results[0].keys()) @pytest.mark.django_db -def test_filter_invalid_department_param(api_client): - user = _make_user() - api_client.force_authenticate(user=user) +def test_destroy_requires_assign_in_same_department(api_client: APIClient): + d1 = Department.objects.create(name="D1") + d2 = Department.objects.create(name="D2") + u = make_user("u@example.com") + api_client.force_authenticate(user=u) - d = Department.objects.create(name="D") - DepartmentRole.objects.create(department=d, name="R1") + role = make_role(d2, "R", []) - url = reverse("api:v1:department-roles-list") + # нет прав в d2 — 403 + url = reverse("api:v1:department-roles-detail", args=[role.id]) + assert api_client.delete(url).status_code == 403 - # нечисловой фильтр department -> просто пустой список (ошибки быть не должно) - resp = api_client.get(url, {"department": "oops"}) + # права в d1 не помогают — всё ещё 403 + grant_assign_in_dept(u, d1) + assert api_client.delete(url).status_code == 403 + + # выдаём права в d2 — можно удалять + grant_assign_in_dept(u, d2) + assert api_client.delete(url).status_code in (204, 200) + + +@pytest.mark.django_db +def test_set_perms_dedup_and_replace_semantics(api_client: APIClient): + d = Department.objects.create(name="Dept") + u = make_user("u@example.com") + api_client.force_authenticate(user=u) + grant_assign_in_dept(u, d) + + role = make_role(d, "R", []) + ensure_perm("manage_department") + ensure_perm("assign_department_role") + + url = reverse("api:v1:department-roles-set-perms", args=[role.id]) + + # сначала ставим один код + resp = api_client.post( + url, {"permission_codes": ["manage_department"]}, format="json" + ) assert resp.status_code == 200 - assert isinstance(resp.data, list) - # В худшем случае фильтр не сработает — тогда >=1. - # Основная проверка — отсутствие исключений. + role.refresh_from_db() + assert set(role.scoped_permissions.values_list("code", flat=True)) == { + "manage_department" + } - -@pytest.mark.django_db -def test_member_role_with_manage_can_crud_in_own_department(api_client): - # пользователь не head, но имеет роль отдела с правом manage_department - user = _make_user() - api_client.force_authenticate(user=user) - dept = Department.objects.create(name="Ops") - - role_mgr = DepartmentRole.objects.create(department=dept, name="Manager") - p_manage = Permission.objects.get(content_type__app_label="employees", codename="manage_department") - role_mgr.permissions.add(p_manage) - - # делаем пользователя членом отдела с этой ролью - EmployeeDepartment.objects.create(employee=user, department=dept, role=role_mgr, is_active=True) - - url = reverse("api:v1:department-roles-list") - - # create в своём отделе — должен пройти - resp = api_client.post(url, {"department": dept.id, "name": "NewRole"}, format="json") - assert resp.status_code == 201 - role_id = resp.data["id"] - - # update — тоже должен пройти - url_detail = reverse("api:v1:department-roles-detail", args=[role_id]) - resp = api_client.patch(url_detail, {"name": "NewRole++"}, format="json") + # потом отправим дубликаты и другой код — должно быть два уникальных, и ПОЛНАЯ замена + resp = api_client.post( + url, + { + "permission_codes": [ + "manage_department", + "manage_department", + "assign_department_role", + ] + }, + format="json", + ) assert resp.status_code == 200 - - # delete — тоже должен пройти - resp = api_client.delete(url_detail) - assert resp.status_code == 204 - - -@pytest.mark.django_db -def test_member_role_assign_can_set_perms_without_manage(api_client): - # пользователь не head, но имеет роль с правом assign_department_role — может управлять правами роли - user = _make_user() - api_client.force_authenticate(user=user) - dept = Department.objects.create(name="Ops2") - - role_assigner = DepartmentRole.objects.create(department=dept, name="Assigner") - p_assign = Permission.objects.get(content_type__app_label="employees", codename="assign_department_role") - role_assigner.permissions.add(p_assign) - - EmployeeDepartment.objects.create(employee=user, department=dept, role=role_assigner, is_active=True) - - # создадим цель-роль, которой будем управлять правами - target_role = DepartmentRole.objects.create(department=dept, name="Target") - p_view_emp = Permission.objects.get(content_type__app_label="employees", codename="view_employee") - - # set_perms должен позволяться с одним правом assign_department_role - url_set = reverse("api:v1:department-roles-set-perms", args=[target_role.id]) - resp = api_client.post(url_set, {"permission_ids": [p_view_emp.id]}, format="json") - assert resp.status_code == 200 - assert set(target_role.permissions.values_list("id", flat=True)) == {p_view_emp.id} - - -@pytest.mark.django_db -def test_global_manage_grants_crud_in_any_department(api_client): - # Глобальный пермишен employees.manage_department должен открыть CRUD в любом отделе - user = _make_user() - api_client.force_authenticate(user=user) - _grant_perm(user, "employees.manage_department") - - foreign = Department.objects.create(name="Foreign-Dept") - - url = reverse("api:v1:department-roles-list") - resp = api_client.post(url, {"department": foreign.id, "name": "GloballyAllowed"}, format="json") - assert resp.status_code == 201 - - role_id = resp.data["id"] - url_detail = reverse("api:v1:department-roles-detail", args=[role_id]) - resp = api_client.patch(url_detail, {"name": "EditOK"}, format="json") - assert resp.status_code == 200 - - resp = api_client.delete(url_detail) - assert resp.status_code == 204 - - -@pytest.mark.django_db -def test_role_in_other_dept_does_not_grant_cross_access(api_client): - # Роль с правом manage_department в отделе A не даёт права управлять отделом B - user = _make_user() - api_client.force_authenticate(user=user) - dept_a = Department.objects.create(name="A") - dept_b = Department.objects.create(name="B") - - role_mgr_a = DepartmentRole.objects.create(department=dept_a, name="MgrA") - p_manage = Permission.objects.get(content_type__app_label="employees", codename="manage_department") - role_mgr_a.permissions.add(p_manage) - EmployeeDepartment.objects.create(employee=user, department=dept_a, role=role_mgr_a, is_active=True) - - # create в отделе B -> 403 - url = reverse("api:v1:department-roles-list") - resp = api_client.post(url, {"department": dept_b.id, "name": "NoAccess"}, format="json") - assert resp.status_code == 403 - - -@pytest.mark.django_db -def test_set_perms_replaces_completely_and_dedup(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="QA", head=head) - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) - - role = DepartmentRole.objects.create(department=dept, name="Reviewer") - p_view = Permission.objects.get(content_type__app_label="employees", codename="view_employee") - p_manage = Permission.objects.get(content_type__app_label="employees", codename="manage_department") - - # сначала добавим одно право - role.permissions.add(p_view) - - url_set = reverse("api:v1:department-roles-set-perms", args=[role.id]) - # передаём список с дублями -> в итоге только p_manage - resp = api_client.post(url_set, {"permission_ids": [p_manage.id, p_manage.id]}, format="json") - assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == {p_manage.id} - - -@pytest.mark.django_db -def test_add_and_remove_idempotent(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="Prod", head=head) - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) - - role = DepartmentRole.objects.create(department=dept, name="Operator") - p1 = Permission.objects.get(content_type__app_label="employees", codename="view_employee") - p2 = Permission.objects.get(content_type__app_label="employees", codename="assign_department_role") - - # add_perms с дублями - url_add = reverse("api:v1:department-roles-add-perms", args=[role.id]) - resp = api_client.post(url_add, {"permission_ids": [p1.id, p1.id, p2.id]}, format="json") - assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == {p1.id, p2.id} - - # повторный add_perms — без изменений - resp = api_client.post(url_add, {"permission_ids": [p1.id]}, format="json") - assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == {p1.id, p2.id} - - # remove_perms частичный и повторный (идемпотентность) - url_rm = reverse("api:v1:department-roles-remove-perms", args=[role.id]) - resp = api_client.post(url_rm, {"permission_ids": [p1.id]}, format="json") - assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == {p2.id} - - # повторный remove для уже удалённого — состояние не меняется - resp = api_client.post(url_rm, {"permission_ids": [p1.id]}, format="json") - assert resp.status_code == 200 - assert set(role.permissions.values_list("id", flat=True)) == {p2.id} - - -@pytest.mark.django_db -def test_missing_department_on_create_returns_400(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="Doc", head=head) - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) - - url = reverse("api:v1:department-roles-list") - # забыли передать department -> 400 (валидация сериализатора) - resp = api_client.post(url, {"name": "RoleX"}, format="json") - assert resp.status_code == 400 - - -@pytest.mark.django_db -def test_permissions_payload_validation_errors(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="Check", head=head) - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) - role = DepartmentRole.objects.create(department=dept, name="R") - - # не список -> 400 - url_set = reverse("api:v1:department-roles-set-perms", args=[role.id]) - resp = api_client.post(url_set, {"permission_ids": "oops"}, format="json") - assert resp.status_code == 400 - - # несуществующие id -> 400 - resp = api_client.post(url_set, {"permission_ids": [999999]}, format="json") - assert resp.status_code == 400 - - -@pytest.mark.django_db -def test_delete_nonexistent_returns_404(api_client): - head = _make_user() - api_client.force_authenticate(user=head) - dept = Department.objects.create(name="Del", head=head) - EmployeeDepartment.objects.get_or_create(employee=head, department=dept, defaults={"is_active": True}) - - url_detail = reverse("api:v1:department-roles-detail", args=[999999]) - resp = api_client.delete(url_detail) - assert resp.status_code == 404 + role.refresh_from_db() + assert set(role.scoped_permissions.values_list("code", flat=True)) == { + "manage_department", + "assign_department_role", + } diff --git a/backend/tests/api/v1/employees/test_departments.py b/backend/tests/api/v1/employees/test_departments.py index 499323c4..bf88ea97 100644 --- a/backend/tests/api/v1/employees/test_departments.py +++ b/backend/tests/api/v1/employees/test_departments.py @@ -46,16 +46,19 @@ # * `extract_results` — вытаскивает `results` из пагинированного ответа DRF или возвращает список как есть. import itertools + import pytest from django.contrib.auth import get_user_model -from django.contrib.auth.models import Permission -from django.contrib.contenttypes.models import ContentType from django.urls import reverse +from employees.models import ( + Department, + DepartmentPermission, + DepartmentRole, + EmployeeDepartment, +) from rest_framework import status from rest_framework.test import APIClient -from employees.models import Department, EmployeeDepartment, DepartmentRole - pytestmark = pytest.mark.django_db User = get_user_model() @@ -98,15 +101,10 @@ def make_user( return u -def perm_for_department(codename: str) -> Permission: - """ - Безопасно получаем/создаём Permission именно для employees.Department. - """ - ct = ContentType.objects.get_for_model(Department) - p, _ = Permission.objects.get_or_create( - content_type=ct, - codename=codename, - defaults={"name": codename}, +def perm_for_department(code: str) -> DepartmentPermission: + """Безопасно получаем/создаём скоуп-право отдела по коду.""" + p, _ = DepartmentPermission.objects.get_or_create( + code=code, defaults={"name": code} ) return p @@ -116,7 +114,7 @@ def make_role( ) -> DepartmentRole: r = DepartmentRole.objects.create(department=dept, name=name) if codes: - r.permissions.add(*[perm_for_department(c) for c in codes]) + r.scoped_permissions.add(*[perm_for_department(c) for c in codes]) return r @@ -177,7 +175,6 @@ def test_search_and_ordering(api_client: APIClient): assert names == ["Alpha", "Beta", "Gamma"] - # ---------- tests: employees_count annotation ----------