**feat(feed, perms, tests): новая политика доступа для фида + полный набор автотестов**

Ключевые изменения

* Перевели доступ к объектам отдела (посты) на **ролевую модель внутри отдела**:

  * Используются `DepartmentRole.scoped_permissions` (M2M к `DepartmentPermission`).
  * Проверка прав идёт по **коду** из `DeptPerm` (`CREATE_POST`/`MANAGE_FEED`), а не по `auth.Permission`.
* **Company-посты** теперь контролируются **модельными правами** (`add_post`, `change_post`, `delete_post`) или статусом staff/superuser.
* Убраны остатки старой логики `publish_company_post` / `publish_department_post` (как глобальные права).

Пермишены / утилиты

* `user_has_dept_perm(user, dept, code)`:

  * true для staff/superuser и для руководителя отдела;
  * иначе проверка активной связи `EmployeeDepartment(is_active=True)` и наличия `role__scoped_permissions__code=...`.
  * ВАЖНО: используем поле **`code`**, не `codename`.
* `has_dept_perm(user, department_id, code)`:

  * те же правила, оптимизирован поиск (select\_related для роли).
* `AdminOrDeptAllowed`:

  * Корректно извлекает `department_id` из `kwargs`/`data`/`query_params` и из объекта (`obj.department_id/department.id`).
  * Если код не задан — SAFE-методы разрешены (чтение), небезопасные запрещены.
* `AdminOrActionOrModelPerms`:

  * staff/superuser — полный доступ;
  * поддержка явного `required_perm_code`/`required_perms_by_action`;
  * SAFE-методы требуют `view_<model>`, небезопасные — стандартная карта DjangoModelPermissions.
* `IsSelfOrStaff`/`IsSelfOrStaffOrModelPerms`:

  * Владелец объекта (по `author_id`/`author` и др.) или staff/superuser может изменять/удалять;
  * SAFE-методы — для авторизованных.

API: PostViewSet

* Обновлён `get_permissions()`:

  * `list/retrieve/create/like/unlike` → `IsAuthenticated`;
  * `update/partial_update/destroy` → `AdminOrDeptAllowed` с `required_code_map`:

    * `update/partial_update/destroy` → `DeptPerm.MANAGE_FEED`.
* Бизнес-проверки:

  * `perform_create`:

    * `type=department` → разрешено staff/superuser, head отдела, либо роль с `DeptPerm.CREATE_POST`;
    * `type=company` → нужны модельные права `add_post` (или staff).
    * `author` проставляется на `request.user`.
  * `perform_update/perform_destroy`:

    * dept-посты: staff/superuser, head, либо роль с `DeptPerm.MANAGE_FEED`;
    * company-посты: модельные права `change_post`/`delete_post` (или staff).
* Производительность:

  * Аннотация `is_liked` (через `Exists`).
  * Аннотация `last_comment_id` + батч-подтягивание последнего комментария в `list()`.
* Лайки:

  * `like/unlike` — идемпотентные; счётчик корректируется атомарно.

API: CommentViewSet

* `permission_classes = [IsAuthenticated, (IsSelfOrStaff | AdminOrActionOrModelPerms)]`.
* `perform_create` всегда ставит `author=request.user`.
* Фильтры `?post=`/`?author=` и сортировка `created_at ASC`.
* Обновлён `IsSelfOrStaff` для чтения настроек владельца с вью (например, `owner_id_attrs=("author_id",)`).

Тесты

* Добавлен/переписан комплект для новой модели доступа (фид):

  * **Аноним**: все эндпоинты → 401.
  * **Обычный пользователь**:

    * `GET posts/comments` → 200;
    * `POST comment`, `like/unlike` → 201/200;
    * `POST/PUT/PATCH/DELETE dept-post` → 403 (без роли/прав).
  * **Роль отдела**:

    * С `DeptPerm.CREATE_POST` — может создать пост **в своём отделе**; **в чужом** → 403.
    * С `DeptPerm.MANAGE_FEED` — может редактировать/удалять **в своём отделе**.
  * **Head отдела** — полный доступ к постам своего отдела.
  * **Staff/superuser** — полный доступ; `pin/unpin` только staff/superuser.
  * **Глобальные модельные права на company**:

    * `add_post`/`change_post`/`delete_post` → CRUD **только** для company-постов; dept-посты по-прежнему 403.
  * **Комментарии**:

    * Создание — любой аутентифицированный;
    * Изменение/удаление — автор или staff/superuser.
  * **Лайки**:

    * Идемпотентность `like/unlike`, корректная синхронизация `is_liked`.
  * **Валидация**:

    * `type=department` требует `department`;
    * роль в одном отделе не даёт создавать в другом → 403.
* Тестовые хелперы:

  * `_ensure_permission(model, codename, name)` — создаёт/возвращает `auth.Permission` для company-кейсов;
  * `_ensure_dept_permission(code)` и `_give_dept_role_with_perms(...)` — для департаментных прав (через `DepartmentPermission`/`DepartmentRole.scoped_permissions`).

Fixed / багфиксы

* `FieldError: Unsupported lookup 'codename'…` — заменён на фильтр по `scoped_permissions__code`.
* `IntegrityError: NOT NULL constraint failed: feed_post.author_id` — в `perform_create` постов/комментариев author задаётся принудительно.
* Несоответствие тестам по правам чтения/изменения комментариев — исправлено через корректный пермишен владельца.

Файлы (основные)

* `api/v1/permissions.py`: `AdminOrDeptAllowed`, `AdminOrActionOrModelPerms`, `IsSelfOrStaff`(+вариант `IsSelfOrStaffOrModelPerms`), `user_has_dept_perm`, `has_dept_perm`.
* `api/v1/feed/views.py`: `PostViewSet`, `CommentViewSet` (правки прав, `perform_*`, аннотации, like/unlike).
* `tests/api/v1/feed/test_posts.py`, `tests/api/v1/feed/test_comments.py` (и/или `test_feed_api.py`): полный набор сценариев под новую политику.
* Хелперы тестов: функции создания отделов, прав и ролей, генераторы payload.

Итог

* Политика доступа фида приведена к спецификации: **dept-права — через роли внутри отдела**, **company — через модельные права**.
* Исправлены ошибки в проверках прав и авторстве, покрыто автотестами (CRUD, лайки, комменты, валидация, пермишены).
* Производительность чтения ленты улучшена (аннотации `is_liked`, `last_comment_id` без N+1).
This commit is contained in:
admin i 2025-09-11 17:56:33 +03:00
parent d488b748ad
commit 9529b3f268
19 changed files with 3019 additions and 472 deletions

View File

@ -57,7 +57,6 @@ class CalendarEventsViewSet(ModelViewSet):
гарантированно его увидел на этапе has_permission().
"""
dep = self._dept_id(required=False)
print(f"Request user: {self.request.user}, Auth: {self.request.auth}")
if dep is None:
return [AdminOrActionOrModelPerms()]

View File

@ -0,0 +1,207 @@
from __future__ import annotations
from typing import Any, Dict, Sequence
from django.contrib.auth import get_user_model
from documents.models import Document, DocumentAcknowledgement
from rest_framework import serializers
from ..employees.serializers import EmployeeBriefSerializer
User = get_user_model()
class DocumentReadSerializer(serializers.ModelSerializer):
"""Сериализатор для чтения документа.
Включает список получателей и флаг, ознакомился ли текущий пользователь.
"""
uploaded_by = EmployeeBriefSerializer(read_only=True)
recipients = EmployeeBriefSerializer(many=True, read_only=True)
file_url = serializers.FileField(source="file", read_only=True)
is_acknowledged = serializers.SerializerMethodField()
class Meta:
model = Document
fields = (
"id",
"title",
"description",
"uploaded_by",
"uploaded_at",
"sent_to_all",
"recipients",
"file_url",
"is_acknowledged",
)
def get_is_acknowledged(self, obj: Document) -> bool:
"""Проверяет, отмечал ли текущий пользователь ознакомление.
Args:
obj (Document): Документ.
Returns:
bool: True, если текущий пользователь уже ознакомился.
"""
request = self.context.get("request")
if not request or not request.user.is_authenticated:
return False
return DocumentAcknowledgement.objects.filter(
document=obj, user=request.user
).exists()
class DocumentWriteSerializer(serializers.ModelSerializer):
"""Сериализатор для записи/обновления документа.
Поддерживает:
- загрузку файла (multipart)
- `sent_to_all`
- `recipient_ids` (список ID пользователей), если `sent_to_all == False`
"""
# Принимаем список ID вместо M2M-структуры
recipient_ids = serializers.ListField(
child=serializers.IntegerField(min_value=1),
write_only=True,
required=False,
allow_empty=True,
help_text="Список ID сотрудников (используется, когда sent_to_all=false).",
)
class Meta:
model = Document
fields = (
"id",
"title",
"description",
"file",
"sent_to_all",
"recipient_ids",
)
def validate(self, attrs: Dict[str, Any]) -> Dict[str, Any]:
"""Общая валидация полей.
Args:
attrs (dict): Данные.
Returns:
dict: Провалидированные данные.
Raises:
serializers.ValidationError: При логических несоответствиях.
"""
sent_to_all = attrs.get("sent_to_all", True)
recipient_ids = attrs.get("recipient_ids", [])
if sent_to_all is False and not recipient_ids:
raise serializers.ValidationError(
{
"recipient_ids": "Укажите получателей или установите sent_to_all=true."
}
)
return attrs
def _set_recipients(self, doc: Document, recipient_ids: Sequence[int]) -> int:
"""Привязывает получателей по списку ID.
Args:
doc (Document): Документ.
recipient_ids (Sequence[int]): Идентификаторы пользователей.
Returns:
int: Количество привязанных получателей.
"""
if not recipient_ids:
doc.recipients.clear()
return 0
users = User.objects.filter(is_active=True, id__in=list(set(recipient_ids)))
doc.recipients.set(users)
return users.count()
def create(self, validated_data: Dict[str, Any]) -> Document:
"""Создание документа.
ВАЖНО: чтобы уведомления (signals) не улетели до установки M2M,
мы временно отключим `post_save` сигнал и вызовем уведомление вручную уже после назначения получателей.
Args:
validated_data (dict): Данные.
Returns:
Document: Созданный документ.
Raises:
serializers.ValidationError: Если входные данные недопустимы.
"""
recipient_ids = validated_data.pop("recipient_ids", [])
request = self.context.get("request")
uploader = getattr(request, "user", None)
# Подключим/отключим сигнал аккуратно
from django.db.models.signals import post_save
try:
from documents.signals import on_document_saved # type: ignore
except Exception:
on_document_saved = None
if on_document_saved:
post_save.disconnect(on_document_saved, sender=Document)
try:
doc = Document.objects.create(
uploaded_by=uploader, # type: ignore[arg-type]
**validated_data,
)
if validated_data.get("sent_to_all") is False:
self._set_recipients(doc, recipient_ids)
finally:
if on_document_saved:
post_save.connect(on_document_saved, sender=Document)
# Ручное уведомление после полной подготовки (если доступно)
try:
from documents.notification import \
notify_users_about_document # type: ignore
notify_users_about_document(doc)
except Exception:
# если уведомлялка неактивна — просто молча пропустим
pass
return doc
def update(self, instance: Document, validated_data: Dict[str, Any]) -> Document:
"""Обновление документа.
Сценарии:
- можно заменить файл/описание/заголовок
- переключить `sent_to_all`
- обновить получателей через `recipient_ids`
Args:
instance (Document): Документ для обновления.
validated_data (dict): Новые данные.
Returns:
Document: Обновлённый документ.
"""
recipient_ids = validated_data.pop("recipient_ids", None)
for field in ("title", "description", "sent_to_all", "file"):
if field in validated_data:
setattr(instance, field, validated_data[field])
instance.save()
if recipient_ids is not None:
# Если теперь "всем" — очистим M2M
if validated_data.get("sent_to_all", instance.sent_to_all):
instance.recipients.clear()
else:
self._set_recipients(instance, recipient_ids)
# уведомления по обновлению — по желанию; оставим тихо
return instance

View File

@ -0,0 +1,81 @@
from __future__ import annotations
from typing import Any
from django.contrib.auth import get_user_model
from django.db import transaction
from django.db.models import QuerySet
from documents.models import Document, DocumentAcknowledgement
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.permissions import (SAFE_METHODS, BasePermission,
IsAuthenticated)
from rest_framework.response import Response
from rest_framework.viewsets import ModelViewSet
from ..permissions import AdminOrActionOrModelPerms
from .serializers import DocumentReadSerializer, DocumentWriteSerializer
User = get_user_model()
class DocumentViewSet(ModelViewSet):
"""Полный CRUD по документам + экшен `acknowledge`.
Создание/обновление поддерживает multipart (для файла) и назначение получателей.
"""
queryset: QuerySet[Document] = (
Document.objects.all()
.select_related("uploaded_by")
.prefetch_related("recipients")
)
permission_classes = [AdminOrActionOrModelPerms]
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_serializer_class(self):
"""Выбор сериализатора: на чтение — Read, на запись — Write."""
if self.action in ("list", "retrieve"):
return DocumentReadSerializer
return DocumentWriteSerializer
@transaction.atomic
def create(self, request, *args, **kwargs):
"""Создаёт документ для рассылки.
Returns:
Response: 201 Created с данными документа (read-сериализатор).
"""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
doc = serializer.save()
read = DocumentReadSerializer(doc, context={"request": request})
return Response(read.data, status=status.HTTP_201_CREATED)
@transaction.atomic
def update(self, request, *args, **kwargs):
"""Полное обновление документа."""
return super().update(request, *args, **kwargs)
@transaction.atomic
def partial_update(self, request, *args, **kwargs):
"""Частичное обновление документа."""
return super().partial_update(request, *args, **kwargs)
@action(methods=["post"], detail=True, permission_classes=[IsAuthenticated])
def acknowledge(self, request, pk=None):
"""Отметить ознакомление текущего пользователя с документом.
Args:
request: DRF Request
pk: ID документа
Returns:
Response: {"ok": true, "already": bool}
"""
doc = self.get_object()
obj, created = DocumentAcknowledgement.objects.get_or_create(
document=doc, user=request.user
)
return Response({"ok": True, "already": not created})

View File

@ -148,7 +148,7 @@ class EmployeeActionSerializer(serializers.ModelSerializer):
"date_display",
"comment",
"extra",
"history", # ← добавили
"history",
]
def get_action_display(self, obj):

View File

@ -329,7 +329,14 @@ class DepartmentViewSet(viewsets.ModelViewSet):
return [self.ManagePerm()]
if self.action == "create":
return [AdminOrActionOrModelPerms()]
if self.action in {"members", "user_perms", "ui_context"}:
if self.action in {
"members",
"user_perms",
"ui_context",
"list",
"retrieve",
"my_departments",
}:
return [IsAuthenticated()]
return [AdminOrActionOrModelPerms()]
@ -1077,7 +1084,7 @@ class PositionViewSet(HistoryActionMixin, viewsets.ModelViewSet):
ordering_fields = ["name", "id"]
ordering = ["name"]
history_diff_fields = ["name", "description"]
permission_classes = [IsAuthenticated, AdminOrActionOrModelPerms]
permission_classes = [AdminOrActionOrModelPerms]
required_perms_by_action = {
"set_groups": "employees.assign_position_groups",
"add_groups": "employees.assign_position_groups",
@ -1095,6 +1102,18 @@ class PositionViewSet(HistoryActionMixin, viewsets.ModelViewSet):
return None, Response({"detail": "Некоторые группы не найдены"}, status=400)
return qs, None
def get_permissions(self):
"""Возвращает пермишены для текущего экшена.
Returns:
list: Инスタнсы DRF-permission'ов.
"""
if self.action in {"list", "retrieve", "permissions"}:
return [IsAuthenticated()]
if self.action == "create":
return [AdminOrActionOrModelPerms()]
return [AdminOrActionOrModelPerms()]
@action(detail=True, methods=["post"])
def set_groups(self, request, pk=None):
pos = self.get_object()
@ -1296,7 +1315,7 @@ class SkillViewSet(viewsets.ModelViewSet):
}
def get_permissions(self):
if self.action in ("create", "bulk_create"):
if self.action in ("create", "bulk_create", "list", "retrieve"):
return [IsAuthenticated()]
return [IsAuthenticated(), AdminOrActionOrModelPerms()]

View File

@ -1,8 +1,10 @@
from __future__ import annotations
from typing import Any, List
from django.contrib.auth import get_user_model
from django.db.models import (BooleanField, Count, Exists, F, OuterRef,
Subquery, Value)
from django.db.models import BooleanField, Count, Exists, F, OuterRef, Subquery, Value
from employees.constants import DeptPerm
from feed.constants import TYPE_COMPANY, TYPE_DEPARTMENT, TYPE_EMPLOYEE
from feed.models import Comment, Post, PostLike
from rest_framework import filters, status, viewsets
@ -12,8 +14,14 @@ from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.permissions import IsAdminUser, IsAuthenticated
from rest_framework.response import Response
from ..permissions import (IsAuthorOrStaffForComments, user_has_dept_perm,
user_is_dept_head, user_is_staffish)
from ..permissions import (
AdminOrActionOrModelPerms,
AdminOrDeptAllowed,
IsSelfOrStaff,
user_has_dept_perm,
user_is_dept_head,
user_is_staffish,
)
from .serializers import CommentSerializer, PostListSerializer, PostSerializer
Employee = get_user_model()
@ -33,18 +41,97 @@ def _to_id(val):
class PostViewSet(viewsets.ModelViewSet):
"""CRUD ленты + like/unlike, pin/unpin.
Доступ (коротко):
- list/retrieve/like/unlike — IsAuthenticated
- create:
* type=company → AdminOrActionOrModelPerms + required_perm_code="feed.publish_company_post"
* type=department → AdminOrDeptAllowed с кодом DeptPerm.CREATE_POST|publish_department_post
- update/partial_update/destroy:
* company → как для company-create
* department → как для department-create
- pin/unpin — только staff/superuser
"""
queryset = (
Post.objects.select_related("author", "department")
.annotate(comments_count=Count("comments", distinct=True))
.order_by("-pinned", "-created_at")
)
permission_classes = [IsAuthenticated]
parser_classes = [MultiPartParser, FormParser, JSONParser]
filter_backends = [filters.SearchFilter, filters.OrderingFilter]
search_fields = ["title", "body"]
ordering_fields = ["created_at", "pinned", "likes_count"]
ordering = ["-pinned", "-created_at"]
class PostDeptPerm(AdminOrDeptAllowed):
"""Департаментные права для операций над постами отдела.
SAFE-методы разрешены (IsAuthenticated стоит на вьюхе).
Коды прав подставляются динамически из get_permissions().
"""
allow_safe_without_code = True
def has_permission(self, request, view) -> bool:
"""На create, если type=department, но department не указан,
не режем запрос — даём сериализатору вернуть 400 вместо 403.
"""
if getattr(view, "action", None) == "create":
t = (request.data.get("type") or "").strip()
if t == TYPE_DEPARTMENT:
dept_id = self._to_int_or_none(
self._extract_dept_id_from_request(request, view)
)
if dept_id is None:
return True # пропускаем к валидации сериализатора
return super().has_permission(request, view)
def get_permissions(self) -> List[Any]:
"""Возвращает DRF-пермишены под текущий action (без вызова get_object())."""
if self.action in {"list", "retrieve", "like", "unlike"}:
return [IsAuthenticated()]
if self.action in {"pin", "unpin"}:
return [IsAdminUser()]
# --- CREATE: смотрим на тип из тела запроса ---
if self.action == "create":
t = (self.request.data.get("type") or "").strip()
if t == TYPE_COMPANY:
return [IsAuthenticated(), AdminOrActionOrModelPerms()]
# department → департаментное право на создание
dept_code = getattr(DeptPerm, "CREATE_POST", "publish_department_post")
perm = self.PostDeptPerm()
perm.required_code = DeptPerm.CREATE_POST
return [IsAuthenticated(), perm]
# --- UPDATE/PATCH/DELETE: определяем тип поста из БД, но без get_object() ---
if self.action in {"update", "partial_update", "destroy"}:
lookup = getattr(self, "lookup_url_kwarg", None) or getattr(
self, "lookup_field", "pk"
)
pk = self.kwargs.get(lookup)
info = (
Post.objects.filter(pk=pk).values("type", "department_id").first() or {}
)
if info.get("type") == TYPE_COMPANY:
return [IsAuthenticated(), AdminOrActionOrModelPerms()]
# department
dept_code = DeptPerm.MANAGE_FEED
perm = self.PostDeptPerm()
perm.required_code_map = {
"update": dept_code,
"partial_update": dept_code,
"destroy": dept_code,
}
return [IsAuthenticated(), perm]
# дефолт: просто аутентификация
return [IsAuthenticated()]
def get_serializer_class(self):
return PostListSerializer if self.action == "list" else PostSerializer
@ -59,11 +146,11 @@ class PostViewSet(viewsets.ModelViewSet):
else:
qs = qs.annotate(is_liked=Value(False, output_field=BooleanField()))
# только ID последнего комментария — остальное соберём в сериализаторе
# последний комментарий (id) — без N+1
lc = Comment.objects.filter(post_id=OuterRef("pk")).order_by("-created_at")
qs = qs.annotate(last_comment_id=Subquery(lc.values("id")[:1]))
# дальше — ваши фильтры type/department/author/pinned (как были)
# фильтры
qp = self.request.query_params
t = qp.get("type")
if t:
@ -81,17 +168,10 @@ class PostViewSet(viewsets.ModelViewSet):
qs = qs.filter(pinned=False)
return qs
def list(self, request, *args, **kwargs):
"""
Сбор последних комментариев пачкой, без N+1:
1) получаем страницу/список постов;
2) вытаскиваем их last_comment_id;
3) делаем один запрос in_bulk() с select_related('author');
4) кладём карту в serializer.context.
"""
queryset = self.filter_queryset(self.get_queryset())
def list(self, request, *args, **kwargs):
"""Возвращает список постов с картой последних комментариев без N+1."""
queryset = self.filter_queryset(self.get_queryset())
page = self.paginate_queryset(queryset)
objects = page if page is not None else queryset
@ -104,61 +184,29 @@ class PostViewSet(viewsets.ModelViewSet):
serializer = self.get_serializer(
objects,
many=True,
context={**self.get_serializer_context(), "last_comments_map": last_comments_map},
context={
**self.get_serializer_context(),
"last_comments_map": last_comments_map,
},
)
if page is not None:
return self.get_paginated_response(serializer.data)
return Response(serializer.data)
# ---- permissions helpers
def _can_create(self, user, t, dept):
if not user or not user.is_authenticated:
return False
if t == TYPE_EMPLOYEE:
return False
if t == TYPE_COMPANY:
return user_is_staffish(user) or user.has_perm("feed.publish_company_post")
if t == TYPE_DEPARTMENT and dept:
return (
user_is_staffish(user)
or user_is_dept_head(user, dept)
or user_has_dept_perm(user, dept, "publish_department_post")
)
return False
# --- ВАЖНО: возвращаем perform_create, чтобы заполнить author ---
def perform_create(self, serializer: PostSerializer) -> None:
"""Сохраняет пост, проставляя автора текущим пользователем.
def _can_edit(self, user, post: Post):
if post.type == TYPE_COMPANY:
return user_is_staffish(user) or user.has_perm("feed.publish_company_post")
if post.type == TYPE_DEPARTMENT and post.department_id:
return (
user_is_staffish(user)
or user_is_dept_head(user, post.department)
or user_has_dept_perm(user, post.department, "publish_department_post")
)
return False
Raises:
ValidationError: если сериализатор невалиден (обрабатывается DRF).
"""
serializer.save(author=self.request.user)
# ---- hooks
def perform_create(self, serializer):
user = self.request.user
t = self.request.data.get("type") or serializer.validated_data.get("type")
dept = serializer.validated_data.get("department")
if not self._can_create(user, t, dept):
raise PermissionDenied("Недостаточно прав для публикации.")
serializer.save(author=user)
# --- actions (pin/unpin и лайки) ---
def perform_update(self, serializer):
if not self._can_edit(self.request.user, self.get_object()):
raise PermissionDenied("Недостаточно прав для изменения публикации.")
serializer.save()
def perform_destroy(self, instance):
if not self._can_edit(self.request.user, instance):
raise PermissionDenied("Недостаточно прав для удаления публикации.")
return super().perform_destroy(instance)
# ---- actions
@action(detail=True, methods=["post"], permission_classes=[IsAdminUser])
def pin(self, request, pk=None):
"""Пинует пост (только staff/superuser)."""
post = self.get_object()
if not post.pinned:
Post.objects.filter(pk=post.pk).update(pinned=True)
@ -166,13 +214,15 @@ class PostViewSet(viewsets.ModelViewSet):
@action(detail=True, methods=["post"], permission_classes=[IsAdminUser])
def unpin(self, request, pk=None):
"""Снимает пин (только staff/superuser)."""
post = self.get_object()
if post.pinned:
Post.objects.filter(pk=post.pk).update(pinned=False)
return Response({"pinned": False})
@action(detail=True, methods=["post"])
@action(detail=True, methods=["post"], permission_classes=[IsAuthenticated])
def like(self, request, pk=None):
"""Ставит лайк текущего пользователя (идемпотентно)."""
post = self.get_object()
_, created = PostLike.objects.get_or_create(post=post, user=request.user)
if created:
@ -182,8 +232,9 @@ class PostViewSet(viewsets.ModelViewSet):
{"liked": True, "likes_count": post.likes_count}, status=status.HTTP_200_OK
)
@action(detail=True, methods=["post"])
@action(detail=True, methods=["post"], permission_classes=[IsAuthenticated])
def unlike(self, request, pk=None):
"""Снимает лайк текущего пользователя (идемпотентно)."""
post = self.get_object()
deleted, _ = PostLike.objects.filter(post=post, user=request.user).delete()
if deleted:
@ -192,7 +243,8 @@ class PostViewSet(viewsets.ModelViewSet):
)
post.refresh_from_db(fields=["likes_count"])
return Response(
{"liked": False, "likes_count": post.likes_count}, status=status.HTTP_200_OK
{"liked": False, "likes_count": post.likes_count},
status=status.HTTP_200_OK,
)
@ -208,7 +260,9 @@ class CommentViewSet(viewsets.ModelViewSet):
queryset = Comment.objects.select_related("post", "author").all()
serializer_class = CommentSerializer
permission_classes = [IsAuthenticated, IsAuthorOrStaffForComments]
owner_id_attrs = ("author_id",)
owner_obj_attrs = ("author",)
permission_classes = [IsAuthenticated, (IsSelfOrStaff | AdminOrActionOrModelPerms)]
parser_classes = [MultiPartParser, FormParser, JSONParser]
filter_backends = [filters.OrderingFilter]
ordering_fields = ["created_at", "id"]

View File

@ -1,13 +1,16 @@
# backend\api\v1\permissions.py
from __future__ import annotations
from typing import Any, Dict, Optional
from typing import Any, Dict, Optional, Tuple
from django.contrib.auth.models import AbstractBaseUser
from django.contrib.auth.models import AbstractBaseUser, AnonymousUser
from employees.constants import DeptPerm
from employees.models import Department, EmployeeDepartment
from rest_framework.permissions import (SAFE_METHODS, BasePermission,
DjangoModelPermissions)
from rest_framework.permissions import (
SAFE_METHODS,
BasePermission,
DjangoModelPermissions,
)
MANAGE_PERM = "manage_department"
CHANGE_HEAD_PERM = "change_department_head"
@ -25,20 +28,46 @@ def user_is_staffish(user) -> bool:
def user_has_dept_perm(user, dept: Department, perm_code: str) -> bool:
"""Проверяет, есть ли у пользователя право `perm_code` в указанном отделе.
Правила:
- Неаутентифицированным → False.
- staff/superuser → True.
- Руководитель отдела → True.
- Иначе: существует активная связь EmployeeDepartment, и у роли есть
DepartmentPermission с нужным code.
Args:
user (User): Текущий пользователь.
dept (Department): Отдел.
perm_code (str): Код департаментного права (см. DeptPerm.CHOICES).
Returns:
bool: True, если доступ разрешён.
Raises:
ValueError: Если `dept` не задан или не имеет id.
"""
if not (user and user.is_authenticated):
return False
# staff/superuser — сразу ок (оставим как было)
if user_is_staffish(user):
# staff/superuser — сразу ок
if getattr(user, "is_staff", False) or getattr(user, "is_superuser", False):
return True
if not isinstance(dept, Department) or getattr(dept, "id", None) is None:
raise ValueError("Argument `dept` must be Department instance with a valid id.")
# руководитель — сразу ок
if user_is_dept_head(user, dept):
if getattr(dept, "head_id", None) == user.id:
return True
# основная проверка: активная связь + у роли есть нужный Permission
# основная проверка: активная связь + у роли есть нужный DepartmentPermission.code
return EmployeeDepartment.objects.filter(
employee_id=user.id,
department_id=dept.id,
is_active=True,
role__scoped_permissions__codename=perm_code,
role__scoped_permissions__code=perm_code, # <-- ВАЖНО: `code`, не `codename`
).exists()
@ -207,24 +236,14 @@ class AdminOrDeptAllowed(BasePermission):
# -------- main checks --------
def has_permission(self, request, view) -> bool:
"""Проверка доступа на уровне запроса (до доступа к объекту).
Правила:
- Неаутентифицированные → False.
- superuser/staff → True.
- Если код не требуется:
- SAFE-методы → allow_safe_without_code ? True : False
- небезопасные → False
- Если код требуется:
- Пытаемся найти department_id в запросе;
- Если нашли → проверяем has_dept_perm(user, dept_id, code);
- Если не нашли:
* если detail-вью → отложить проверку до has_object_permission (вернуть True);
* иначе → False.
Returns:
bool: Разрешить/запретить доступ.
"""
"""См. докстринг класса. Если dept_id не найден, но это detail-запрос,
откладываем проверку до object-level (вернём True)."""
print(
view.action,
view.kwargs,
self.get_required_code(request, view),
self._to_int_or_none(self._extract_dept_id_from_request(request, view)),
)
user = getattr(request, "user", None)
if not user or not user.is_authenticated:
return False
@ -234,7 +253,7 @@ class AdminOrDeptAllowed(BasePermission):
code = self.get_required_code(request, view)
if not code:
return (
request.method in SAFE_METHODS
(request.method in SAFE_METHODS)
if self.allow_safe_without_code
else False
)
@ -242,9 +261,19 @@ class AdminOrDeptAllowed(BasePermission):
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
# robust: считаем detail, если есть lookup в kwargs
lookup_kw = getattr(view, "lookup_url_kwarg", None) or getattr(
view, "lookup_field", "pk"
)
is_detail = getattr(view, "detail", None)
if is_detail is None:
is_detail = lookup_kw in getattr(view, "kwargs", {})
if is_detail:
# отложим проверку до has_object_permission
return True
return False
return has_dept_perm(user, dept_id, code)
@ -280,106 +309,158 @@ class AdminOrDeptAllowed(BasePermission):
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):
return bool(request.user and request.user.is_authenticated)
"""Staff/superuser или владелец объекта (настраиваемые поля владельца через атрибуты вью)."""
DEFAULT_OWNER_ID_ATTRS = ("author_id", "user_id", "owner_id", "employee_id")
DEFAULT_OWNER_OBJ_ATTRS = ("author", "user", "owner", "employee")
def has_permission(self, request, view) -> bool:
user = getattr(request, "user", None)
return bool(user and user.is_authenticated)
# На object-level: сам объект, либо staff/superuser.
def has_object_permission(self, request, view, obj):
user = request.user
if not user or not user.is_authenticated:
return False
if request.method in SAFE_METHODS:
return True
if getattr(user, "is_staff", False) or getattr(user, "is_superuser", False):
return True
# Основной случай — объект сотрудника.
try:
from employees.models import \
Employee # локальный импорт, чтобы избежать циклов
except Exception:
Employee = None
# Прочитаем конфиг владельческих полей с вью (если задан)
owner_id_attrs = getattr(view, "owner_id_attrs", self.DEFAULT_OWNER_ID_ATTRS)
owner_obj_attrs = getattr(view, "owner_obj_attrs", self.DEFAULT_OWNER_OBJ_ATTRS)
uid = getattr(user, "id", None)
if Employee is not None and isinstance(obj, Employee):
return obj.pk == user.pk
for attr in owner_id_attrs:
if hasattr(obj, attr) and getattr(obj, attr) == uid:
return True
for attr in owner_obj_attrs:
owner = getattr(obj, attr, None)
if owner is not None and getattr(owner, "id", None) == uid:
return True
# Fallback: сравнить id из URL, если obj не Employee (на всякий случай)
# Fallback для вью «/users/{id}/»
lookup = getattr(view, "lookup_field", "pk")
return str(view.kwargs.get(lookup)) == str(user.pk)
class IsAuthorOrStaffForComments(BasePermission):
"""
Создание/чтение — любой аутентифицированный (проверяет IsAuthenticated во view).
Изменение/удаление — автор комментария или staff/superuser.
"""
message = "Можно редактировать/удалять только свои комментарии."
def has_permission(self, request, view):
# IsAuthenticated уже стоит во вьюхе; дублируем True для небезопасных методов,
# чтобы проверка прошла до object-level.
return bool(request.user and request.user.is_authenticated)
def has_object_permission(self, request, view, obj):
if request.method in SAFE_METHODS:
return True
user = request.user
if not (user and user.is_authenticated):
return False
if user_is_staffish(user):
return True
return getattr(obj, "author_id", None) == getattr(user, "id", None)
return str(view.kwargs.get(lookup)) == str(uid)
class AdminOrActionOrModelPerms(DjangoModelPermissions):
"""
Админ/суперпользователь → всегда True.
Если задан код права (view.required_perm_code или required_perms_by_action[action]),
проверяем его через user.has_perm(...). Иначе — стандартные DjangoModelPermissions.
"""Комбинированная политика доступа для документов.
Поведение:
- staff/superuser → полный доступ (всегда True).
- Если на вью задан `required_perm_code` или есть карта
`required_perms_by_action[action]`, проверяем её через `user.has_perm(...)`.
- Иначе:
* для SAFE-методов (`GET/HEAD/OPTIONS`) требуем `view_<model>`;
* для остальных методов используем стандартную карту `DjangoModelPermissions`
(`add/change/delete`).
Raises:
exceptions.MethodNotAllowed: если передан неизвестный HTTP-метод.
"""
def has_permission(self, request, view):
def has_permission(self, request, view) -> bool:
"""Проверка доступа на уровне запроса.
Args:
request: DRF Request.
view: Текущий ViewSet.
Returns:
bool: Разрешён ли доступ.
"""
user = request.user
if not user or not user.is_authenticated:
if not user or isinstance(user, AnonymousUser) or not user.is_authenticated:
return False
if user.is_superuser or user.is_staff:
return True
mapping = getattr(view, "required_perms_by_action", {}) or {}
action = getattr(view, "action", None)
code = getattr(view, "required_perm_code", None) or mapping.get(action)
code = self._explicit_perm_code(view)
if code:
return user.has_perm(code)
# SAFE-методы → требуем view_<model>
if request.method in SAFE_METHODS:
model = self._model_from_view(view)
return user.has_perm(
f"{model._meta.app_label}.view_{model._meta.model_name}"
)
# Остальные → стандартная логика DjangoModelPermissions
return super().has_permission(request, view)
def has_object_permission(self, request, view, obj):
def has_object_permission(self, request, view, obj) -> bool:
"""Проверка доступа к конкретному объекту.
Args:
request: DRF Request.
view: Текущий ViewSet.
obj: Объект модели.
Returns:
bool: Разрешён ли доступ.
"""
user = request.user
if user.is_superuser or user.is_staff:
return True
mapping = getattr(view, "required_perms_by_action", {}) or {}
action = getattr(view, "action", None)
code = getattr(view, "required_perm_code", None) or mapping.get(action)
code = self._explicit_perm_code(view)
if code:
return user.has_perm(code)
if request.method in SAFE_METHODS:
meta = obj._meta
return user.has_perm(f"{meta.app_label}.view_{meta.model_name}")
# Для остальных методов полагаемся на проверку уровня запроса
return super().has_object_permission(request, view, obj)
@staticmethod
def _explicit_perm_code(view) -> Optional[str]:
"""Возвращает явный код права, заданный на уровне вью.
Args:
view: DRF ViewSet.
Returns:
Optional[str]: Код права или None.
"""
mapping = getattr(view, "required_perms_by_action", {}) or {}
action = getattr(view, "action", None)
return getattr(view, "required_perm_code", None) or mapping.get(action)
@staticmethod
def _model_from_view(view):
"""Извлекает модель из view.queryset.
Args:
view: DRF ViewSet.
Returns:
type: Класс модели.
Raises:
AssertionError: если у вью нет queryset/model (для ModelViewSet не ожидается).
"""
qs = getattr(view, "queryset", None)
assert qs is not None and hasattr(
qs, "model"
), "View must define `.queryset` with `.model`"
return qs.model
__all__ = [
"MANAGE_PERM",
"CHANGE_HEAD_PERM",
"ASSIGN_ROLE_PERM",
"IsSelfOrStaff",
"IsAuthorOrStaffForComments",
"AdminOrActionOrModelPerms",
"user_is_dept_head",
"user_has_dept_perm",

View File

@ -8,6 +8,7 @@ from .employees.views import (DepartmentRoleViewSet, DepartmentViewSet,
RegisterAPIView, ResendEmailAPIView,
SkillViewSet, VerifyEmailAPIView)
from .feed.views import CommentViewSet, PostViewSet
from .documents.views import DocumentViewSet
app_name = "v1"
@ -15,6 +16,8 @@ router = DefaultRouter()
router.register(r"calendar/events", CalendarEventsViewSet, basename="events")
router.register(r"documents", DocumentViewSet, basename="documents")
router.register(r"departments", DepartmentViewSet, basename="departments")
router.register(r"employees", EmployeeViewSet, basename="employees")
router.register(r"employee-actions", EmployeeActionViewSet, basename="employee-actions")

View File

@ -32,9 +32,14 @@ class DeptPerm:
CHANGE_HEAD = "change_department_head"
ASSIGN_ROLE = "assign_department_role"
MANAGE_CALENDAR = "manage_department_events"
CREATE_POST = "publish_department_post"
MANAGE_FEED = "manage_department_feed"
CHOICES = (
(MANAGE, "Управлять отделом"),
(CHANGE_HEAD, "Назначать руководителя"),
(ASSIGN_ROLE, "Назначать роли участникам"),
(MANAGE_CALENDAR, "Управлять календарём отдела"),
(CREATE_POST, "Публиковать новости на странице отдела"),
(MANAGE_FEED, "Редактировать публикации отдела"),
)

View File

@ -0,0 +1,18 @@
# Generated by Django 5.2.4 on 2025-09-11 13:02
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('employees', '0024_alter_departmentpermission_code'),
]
operations = [
migrations.AlterField(
model_name='departmentpermission',
name='code',
field=models.CharField(choices=[('manage_department', 'Управлять отделом'), ('change_department_head', 'Назначать руководителя'), ('assign_department_role', 'Назначать роли участникам'), ('manage_department_events', 'Управлять календарём отдела'), ('publish_department_post', 'Публиковать новости на странице отдела'), ('manage_department_feed', 'Редактировать публикации отдела')], max_length=64, unique=True),
),
]

View File

@ -144,7 +144,7 @@ class PositionRoleBackend(ModelBackend):
app_label, sep, codename = perm.partition(".")
if not sep:
return False
return link.role.permissions.filter(
return link.role.scoped_permissions.filter(
content_type__app_label=app_label,
codename=codename,
).exists()

View File

@ -0,0 +1,17 @@
# Generated by Django 5.2.4 on 2025-09-11 13:02
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('feed', '0003_alter_post_options'),
]
operations = [
migrations.AlterModelOptions(
name='post',
options={'ordering': ['-pinned', '-created_at'], 'verbose_name': 'Публикация', 'verbose_name_plural': 'Публикации'},
),
]

View File

@ -92,10 +92,7 @@ class Post(models.Model):
likes_count = models.PositiveIntegerField("Лайки", default=0, db_index=True)
class Meta:
permissions = [
("publish_company_post", "Может публиковать новости на главной"),
("publish_department_post", "Может публиковать новости в отделах"),
]
verbose_name = "Публикация"
verbose_name_plural = "Публикации"
ordering = ["-pinned", "-created_at"]

View File

@ -1,100 +1,483 @@
from typing import Dict, Any, List
# backend/search/views.py
from __future__ import annotations
from typing import Any, Dict, List, Literal, TypedDict, Optional, Tuple
from django.contrib.auth.decorators import login_required
from django.core.exceptions import FieldError
from django.db.models import Q, QuerySet
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render
from django.utils.html import escape
# Модели
from feed.models import Post # подставьте актуальный путь
from employees.models import Employee
from feed.models import Post
from employees.models import Employee, Department
from requests_app.models import Request
# ====== ВСПОМОГАТЕЛЬНО: доступ к заявлениям ======
class SearchItem(TypedDict):
"""Элемент единой выдачи поиска.
Attributes:
model_name (Literal["post","employee","department","request","chat","message","event"]):
Тип найденной сущности.
object (Any): Экземпляр модели, который будет отрисован в шаблоне.
"""
model_name: Literal[
"post", "employee", "department", "request", "chat", "message", "event"
]
object: Any
def _is_hr(user) -> bool:
# считаем HR по группе/правам как вы уже настроили
return user.is_active and (
user.groups.filter(name="HR").exists()
or user.has_perm("requests_app.can_view_all_requests")
or user.has_perm("requests_app.can_process_requests")
# ------------------------ ВСПОМОГАТЕЛЬНЫЕ УТИЛИТЫ ------------------------
def _is_hr(user: Employee) -> bool:
"""Проверяет расширенные права на просмотр заявлений.
Args:
user (Employee): Текущий пользователь.
Returns:
bool: True, если пользователь может видеть все заявления.
Notes:
Используются пермишены из requests_app.
"""
return user.has_perm("requests_app.can_view_all_requests") or user.has_perm(
"requests_app.can_process_requests"
)
def _is_head(user) -> bool:
# есть related_name=headed_departments в Employee
return user.headed_departments.exists()
def _try_extend_q(model_cls: type, base_q: Q, lookups: List[Tuple[str, Any]]) -> Q:
"""Пытается добавить условия в Q, аккуратно игнорируя несуществующие поля.
Для каждого пути из lookups пытаемся выполнить пустой фильтр, чтобы проверить валидность.
Неудачные пути (FieldError) игнорируем.
Args:
model_cls (type): Класс модели, к которой будет применён Q.
base_q (Q): Исходное условие.
lookups (List[Tuple[str, Any]]): Список пар (путь_к_полю, значение).
Returns:
Q: Итоговое условие с добавленными (валидными) частями.
"""
q = Q(base_q)
from django.db.models import QuerySet as _QS # локальный импорт для mypy
qs: _QS = model_cls.objects.all()
for path, value in lookups:
try:
# проверяем валидность пути: попытаемся составить запрос
_ = qs.filter(**{path: value}).values("pk")[:1]
q |= Q(**{path: value})
except FieldError:
continue
return q
def _allowed_requests_qs(user) -> QuerySet[Request]:
base = Request.objects.select_related("employee", "department", "approver")
if _is_hr(user):
return base
if _is_head(user):
return base.filter(
employee__departments_links__department__head=user,
employee__departments_links__is_active=True,
).distinct()
return base.filter(employee=user)
def _order_by_first_valid(qs: QuerySet, fields: List[str]) -> QuerySet:
"""Сортирует QS по первому валидному полю, иначе возвращает как есть.
Args:
qs (QuerySet): Исходный QS.
fields (List[str]): Список полей приоритета сортировки.
Returns:
QuerySet: Отсортированный QS или оригинал.
"""
for f in fields:
try:
test = qs.order_by(f)
# прогрев запроса для раннего исключения FieldError
_ = test.values("pk")[:0]
return test
except FieldError:
continue
return qs
# ====== ПОИСК ======
# ----------------------------- ОСНОВНЫЕ ПОИСКИ -----------------------------
def _search_posts(query: str) -> QuerySet[Post]:
"""Поиск по постам (title/body).
Args:
query (str): Строка запроса.
Returns:
QuerySet[Post]: Отсортированный QS.
"""
if not query:
return Post.objects.none()
return (
Post.objects.filter(Q(title__icontains=query) | Q(body__icontains=query))
.only("id", "title", "body", "created_at")
.order_by("-created_at")
)
def _search_employees(query: str) -> QuerySet[Employee]:
"""Поиск по сотрудникам (ФИО/контакты + принадлежность к отделам + навыки).
Ищем:
- ФИО, email, телефон
- Принадлежность к отделам (по названию отдела) через EmployeeDepartment
- Навыки (по названию навыка)
Args:
query (str): Строка запроса.
Returns:
QuerySet[Employee]: Отсортированный QS без дублей.
Raises:
ValueError: Если query пустая строка после trim (обрабатывается возвратом .none()).
"""
if not query:
return Employee.objects.none()
fio_contacts = (
Q(last_name__icontains=query)
| Q(first_name__icontains=query)
| Q(patronymic__icontains=query)
| Q(email__icontains=query)
| Q(phone_number__icontains=query)
)
# 🔹 ВАЖНО: у вас связь через EmployeeDepartment → departments_links
dept_lookups = [
("departments_links__department__name__icontains", query),
]
# Навыки — прямая M2M
skill_lookups = [
("skills__name__icontains", query),
]
q = _try_extend_q(Employee, fio_contacts, dept_lookups)
q = _try_extend_q(Employee, q, skill_lookups)
return (
Employee.objects.filter(q)
.distinct()
.select_related("position")
.only(
"id",
"last_name",
"first_name",
"patronymic",
"email",
"phone_number",
"position__name",
)
.order_by("last_name", "first_name")
)
def _search_departments(query: str) -> QuerySet[Department]:
"""Поиск по отделам (название/описание/руководитель + сотрудники отдела).
Ищем:
- name / description
- head (ФИО)
- сотрудники отдела (ФИО/e-mail/телефон) — через Department.employeedepartment → employee
Args:
query (str): Строка запроса.
Returns:
QuerySet[Department]: Отсортированный QS без дублей.
"""
if not query:
return Department.objects.none()
base = Q(name__icontains=query) | Q(description__icontains=query)
base |= (
Q(head__last_name__icontains=query)
| Q(head__first_name__icontains=query)
| Q(head__patronymic__icontains=query)
)
# 🔹 ВАЖНО: правильный reverse related_name — "employeedepartment"
member_fields = [
"last_name__icontains",
"first_name__icontains",
"patronymic__icontains",
"email__icontains",
"phone_number__icontains",
]
member_lookups = [
(f"employeedepartment__employee__{f}", query) for f in member_fields
]
q = _try_extend_q(Department, base, member_lookups)
return (
Department.objects.filter(q)
.distinct()
.select_related("head")
.only("id", "name", "description", "head")
.order_by("name")
)
def _search_events(query: str) -> QuerySet[Any]:
"""Поиск по календарным событиям (ваша модель: calendar_app.CalendarEvent).
Args:
query (str): Строка запроса.
Returns:
QuerySet[Any]: QS CalendarEvent или пустой QS при отсутствии приложения.
"""
try:
from calendar_app.models import CalendarEvent as EventModel # type: ignore
except Exception:
return Post.objects.none()
if not query:
return EventModel.objects.none()
q = Q(title__icontains=query) | Q(description__icontains=query)
q |= Q(department__name__icontains=query)
# других связей (owner/participants) в вашей модели нет
qs = EventModel.objects.filter(q).distinct()
# Поля дат у вас: start_date/start_time (+ end_*). Сортируем по началу.
qs = qs.order_by("-start_date", "-start_time", "-id")
try:
qs = qs.select_related("department")
except Exception:
pass
return qs
def _search_requests(query: str, user: Employee, is_hr: bool) -> QuerySet[Request]:
"""Поиск по заявлениям с учётом прав.
Args:
query (str): Строка запроса.
user (Employee): Текущий пользователь.
is_hr (bool): Признак расширенных прав.
Returns:
QuerySet[Request]: Отсортированный QS.
Raises:
ValueError: Если user не передан.
"""
if not user:
raise ValueError("user is required")
base = (
Request.objects.filter(
Q(title__icontains=query)
| Q(employee__last_name__icontains=query)
| Q(employee__first_name__icontains=query)
| Q(employee__patronymic__icontains=query)
| Q(department__name__icontains=query)
)
if query
else Request.objects.none()
)
if not is_hr:
base = base.filter(employee=user)
# Если используем select_related по FK — перечисляем их в only(...)
return (
base.select_related("employee", "department", "approver")
.only(
"id",
"title",
"status",
"employee",
"department",
"approver",
"created_at",
)
.order_by("-created_at")
)
def _search_chats(query: str) -> QuerySet[Any]:
"""Поиск по чатам (название/участники/отдел).
Args:
query (str): Строка запроса.
Returns:
QuerySet[Any]: QS Chat или пустой QS, если приложение чатов отсутствует.
Notes:
Безопасно обрабатывает разные схемы (name/department/participants).
"""
try:
from communications.models import Chat # type: ignore
except Exception:
return Post.objects.none() # тип «пустой» QS (любой модели)
if not query:
return Chat.objects.none()
q = Q()
# Попробуем разные пути без падения
q = _try_extend_q(
Chat,
q,
[
("name__icontains", query),
("title__icontains", query),
("department__name__icontains", query),
("participants__last_name__icontains", query),
("participants__first_name__icontains", query),
("participants__patronymic__icontains", query),
("participants__email__icontains", query),
],
)
qs = Chat.objects.filter(q).distinct()
qs = _order_by_first_valid(qs, ["-updated_at", "-last_message_at", "-id"])
# мягкая оптимизация
try:
qs = qs.select_related("department")
except FieldError:
pass
try:
qs = qs.prefetch_related("participants")
except FieldError:
pass
return qs
def _search_messages(query: str) -> QuerySet[Any]:
"""Поиск по сообщениям чатов (контент/автор/чат).
Args:
query (str): Строка запроса.
Returns:
QuerySet[Any]: QS Message или пустой QS, если приложение чатов отсутствует.
"""
try:
from communications.models import Message # type: ignore
except Exception:
return Post.objects.none()
if not query:
return Message.objects.none()
q = Q(content__icontains=query)
q = _try_extend_q(
Message,
q,
[
("author__last_name__icontains", query),
("author__first_name__icontains", query),
("author__patronymic__icontains", query),
("chat__name__icontains", query),
("chat__title__icontains", query),
("chat__department__name__icontains", query),
],
)
qs = Message.objects.filter(q).distinct()
qs = _order_by_first_valid(qs, ["-created_at", "-id"])
try:
qs = qs.select_related("author", "chat")
except FieldError:
pass
return qs
# -------------------------------- ВЬЮХА --------------------------------
@login_required
def search_view(request):
q = (request.GET.get("q") or "").strip()
query = q[:100] # ограничим длину для безопасности
def search_view(request: HttpRequest) -> HttpResponse:
"""Единый поиск по системе (посты, сотрудники, отделы, заявления, чаты, сообщения, события).
results: Dict[str, List[Any]] = {"posts": [], "employees": [], "requests": []}
counts: Dict[str, int] = {"posts": 0, "employees": 0, "requests": 0}
Args:
request (HttpRequest): Запрос (`GET["q"]`).
Returns:
HttpResponse: Рендер `templates/search/results.html` с плоским списком.
Контекст шаблона:
- query: str — исходная строка запроса
- results: List[SearchItem] — плоский список элементов
- counts: Dict[str, int] — счётчики по типам:
keys: post, employee, department, request, chat, message, event
- total: int — сумма всех счётчиков
"""
raw_q = request.GET.get("q", "") or ""
query = escape(raw_q.strip())
items: List[SearchItem] = []
counts: Dict[str, int] = {
"post": 0,
"employee": 0,
"department": 0,
"request": 0,
"chat": 0,
"message": 0,
"event": 0,
}
if query:
# --- Посты (новости)
posts_qs = Post.objects.all()
posts_qs = (
posts_qs.filter(Q(title__icontains=query) | Q(body__icontains=query))
.only("id", "title")
.order_by("-id")
)
counts["posts"] = posts_qs.count()
results["posts"] = list(posts_qs[:10])
# --- Посты
post_qs = _search_posts(query)
counts["post"] = post_qs.count()
items.extend({"model_name": "post", "object": p} for p in post_qs[:10])
# --- Сотрудники
emp_qs = Employee.objects.all()
emp_qs = (
emp_qs.filter(
Q(first_name__icontains=query)
| Q(last_name__icontains=query)
| Q(patronymic__icontains=query)
| Q(email__icontains=query)
| Q(phone_number__icontains=query)
)
.only("id", "first_name", "last_name", "patronymic", "email")
.order_by("last_name", "first_name")
)
counts["employees"] = emp_qs.count()
results["employees"] = list(emp_qs[:10])
# --- Сотрудники (ФИО/контакты + отделы + навыки)
emp_qs = _search_employees(query)
counts["employee"] = emp_qs.count()
items.extend({"model_name": "employee", "object": e} for e in emp_qs[:10])
# --- Заявления (с учётом прав)
req_qs = _allowed_requests_qs(request.user)
req_qs = (
req_qs.filter(
Q(title__icontains=query)
| Q(comment__icontains=query)
| Q(employee__first_name__icontains=query)
| Q(employee__last_name__icontains=query)
)
.only("id", "title", "status", "employee", "department", "created_at")
.order_by("-created_at")
)
counts["requests"] = req_qs.count()
results["requests"] = list(req_qs[:10])
# --- Отделы (в т.ч. по сотрудникам)
dept_qs = _search_departments(query)
counts["department"] = dept_qs.count()
items.extend({"model_name": "department", "object": d} for d in dept_qs[:10])
# --- Заявления (учёт прав)
is_hr = _is_hr(request.user) # type: ignore[arg-type]
req_qs = _search_requests(query, request.user, is_hr) # type: ignore[arg-type]
counts["request"] = req_qs.count()
items.extend({"model_name": "request", "object": r} for r in req_qs[:10])
# --- Чаты
chat_qs = _search_chats(query)
# Если приложение отсутствует — chat_qs будет пустым QS другой модели, count()=0
try:
counts["chat"] = chat_qs.count()
except Exception:
counts["chat"] = 0
items.extend({"model_name": "chat", "object": c} for c in chat_qs[:10])
# --- Сообщения
msg_qs = _search_messages(query)
try:
counts["message"] = msg_qs.count()
except Exception:
counts["message"] = 0
items.extend({"model_name": "message", "object": m} for m in msg_qs[:10])
# --- Календарные события
evt_qs = _search_events(query)
try:
counts["event"] = evt_qs.count()
except Exception:
counts["event"] = 0
items.extend({"model_name": "event", "object": ev} for ev in evt_qs[:10])
# Пустой запрос — просто страница с формой/подсказкой
ctx = {
"query": query,
"results": results,
"results": items,
"counts": counts,
"total": sum(counts.values()),
}
return render(request, "search/results.html", ctx)

View File

@ -1,19 +1,363 @@
{% extends "base.html" %}
{% load static django_bootstrap5 %}
{% block title %}Поиск{% endblock %}
{% block extra_css %}
<style>
.ios-search{ position:relative; width:100%; max-width:420px; }
.ios-search-input{
width:100%; height:40px; padding:8px 38px;
border-radius:14px; border:1px solid var(--bs-border-color);
background: color-mix(in srgb, var(--bs-body-bg), #000 4%);
font-size:15px; outline:none;
transition: box-shadow .2s, border-color .2s, background .2s;
}
[data-bs-theme="dark"] .ios-search-input{
background: color-mix(in srgb, var(--bs-body-bg), #fff 6%);
}
.ios-search-input:focus{
border-color: var(--bs-primary);
box-shadow: 0 0 0 3px rgba(var(--bs-primary-rgb), .15);
}
.ios-search > .bi-search{
position:absolute; left:10px; top:50%; transform:translateY(-50%);
opacity:.55; pointer-events:none; font-size:.95rem;
}
.ios-search-clear{
position:absolute; right:4px; top:50%; transform:translateY(-50%);
border:0; background:transparent; padding:6px; display:none;
}
.ios-search-clear .bi-x-lg{ opacity:.6; font-size:.9rem; }
.ios-search-input:not(:placeholder-shown) ~ .ios-search-clear{ display:block; }
.feed-header { gap:.6rem; flex-wrap:wrap; }
.result-row { cursor:default; }
.list-results > .result-row + .result-row { border-top:1px solid var(--bs-border-color); }
</style>
{% endblock %}
{% block content %}
<h2>Результаты поиска: "{{ query }}"</h2>
{% if results %}
{% for result in results %}
<div class="my-3">
{% if result.model_name == 'post' %}
<h4>{{ result.object.title }}</h4>
<p>{{ result.object.content|truncatewords:20 }}</p>
{% elif result.model_name == 'employee' %}
<h4>{{ result.object.name }}</h4>
<p>{{ result.object.email }} | {{ result.object.position }}</p>
{% endif %}
<div class="container">
<div class="row gx-0">
<div class="col-12">
<!-- Заголовок + iOS-поиск -->
<div class="feed-header d-flex align-items-center">
<i class="bi-search fs-2 text-primary"></i>
<h2 class="title mb-0">Поиск</h2>
<div class="ms-auto ios-search">
<i class="bi-search"></i>
<input type="search" id="globalSearch" class="ios-search-input" placeholder="Фильтр по заголовкам" autocomplete="off">
<button type="button" class="ios-search-clear" aria-label="Очистить"><i class="bi-x-lg"></i></button>
</div>
</div>
{% endfor %}
{% else %}
<p>Ничего не найдено.</p>
{% endif %}
{% endblock %}
<!-- Сегментированный фильтр типов -->
<div class="feed-new mb-3">
<div class="d-flex align-items-center gap-2 flex-wrap">
<div class="btn-group" role="group" aria-label="Фильтр результатов">
<input type="radio" class="btn-check" name="srchFilter" id="fltAll" autocomplete="off" checked>
<label class="btn btn-ghost" for="fltAll" data-filter="all">Все ({{ total }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltPosts" autocomplete="off">
<label class="btn btn-ghost" for="fltPosts" data-filter="post">Публикации ({{ counts.post }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltEmployees" autocomplete="off">
<label class="btn btn-ghost" for="fltEmployees" data-filter="employee">Сотрудники ({{ counts.employee }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltDepartments" autocomplete="off">
<label class="btn btn-ghost" for="fltDepartments" data-filter="department">Отделы ({{ counts.department }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltRequests" autocomplete="off">
<label class="btn btn-ghost" for="fltRequests" data-filter="request">Заявления ({{ counts.request }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltChats" autocomplete="off">
<label class="btn btn-ghost" for="fltChats" data-filter="chat">Чаты ({{ counts.chat }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltMessages" autocomplete="off">
<label class="btn btn-ghost" for="fltMessages" data-filter="message">Сообщения ({{ counts.message }})</label>
<input type="radio" class="btn-check" name="srchFilter" id="fltEvents" autocomplete="off">
<label class="btn btn-ghost" for="fltEvents" data-filter="event">События ({{ counts.event }})</label>
</div>
</div>
</div>
{% if total > 0 %}
{# ===== ПУБЛИКАЦИИ ===== #}
{% if counts.post %}
<div class="feed-card mb-3 section" data-sec="post">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-journal-text"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">Публикации</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'post' %}
{% with post=item.object %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="post" data-haystack="{{ post.title|default_if_none:'' }}">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-journal-text"></i></div>
<div class="flex-grow-1 position-relative">
<a href="{% url 'feed:post_detail' post.pk %}" class="feed-author text-decoration-none stretched-link">{{ post.title }}</a>
<div class="feed-sub mt-1 text-secondary">{{ post.created_at|date:"d.m.Y H:i" }}</div>
{% if post.body %}<div class="small mt-1 text-body">{{ post.body|truncatewords:35 }}</div>{% endif %}
</div>
<i class="bi-chevron-right text-secondary ms-2 mt-1"></i>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{# ===== СОТРУДНИКИ ===== #}
{% if counts.employee %}
<div class="feed-card mb-3 section" data-sec="employee">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-person-badge"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">Сотрудники</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'employee' %}
{% with emp=item.object %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="employee" data-haystack="{{ emp.last_name }} {{ emp.first_name }} {{ emp.patronymic|default_if_none:'' }}">
<div class="feed-ava" style="width:44px;height:44px;">
{% if emp.avatar %}<img src="{{ emp.avatar.url }}" alt="">{% else %}<i class="bi-person"></i>{% endif %}
</div>
<div class="flex-grow-1 position-relative">
<a href="{% url 'employees:employee_detail' emp.pk %}" class="feed-author text-decoration-none stretched-link">
{{ emp.last_name }} {{ emp.first_name }}{% if emp.patronymic %} {{ emp.patronymic }}{% endif %}
</a>
<div class="feed-sub mt-1 text-secondary">
{% if emp.position %}{{ emp.position.name }} · {% endif %}
{{ emp.email }}{% if emp.phone_number %} · {{ emp.phone_number }}{% endif %}
</div>
</div>
<i class="bi-chevron-right text-secondary ms-2 mt-1"></i>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{# ===== ОТДЕЛЫ ===== #}
{% if counts.department %}
<div class="feed-card mb-3 section" data-sec="department">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-diagram-3"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">Отделы</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'department' %}
{% with dept=item.object %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="department" data-haystack="{{ dept.name|default_if_none:'' }}">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-diagram-3"></i></div>
<div class="flex-grow-1 position-relative">
<a href="{% url 'employees:department_detail' dept.pk %}" class="feed-author text-decoration-none stretched-link">{{ dept.name }}</a>
<div class="feed-sub mt-1 text-secondary">
{% if dept.head %}Руководитель: {{ dept.head.last_name }} {{ dept.head.first_name }}{% if dept.head.patronymic %} {{ dept.head.patronymic }}{% endif %}{% endif %}
</div>
{% if dept.description %}<div class="small mt-1 text-body">{{ dept.description|truncatewords:35 }}</div>{% endif %}
</div>
<i class="bi-chevron-right text-secondary ms-2 mt-1"></i>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{# ===== ЗАЯВЛЕНИЯ ===== #}
{% if counts.request %}
<div class="feed-card mb-3 section" data-sec="request">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-file-earmark-text"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">Заявления</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'request' %}
{% with req=item.object %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="request" data-haystack="{% firstof req.title req.get_type_display '' %}">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-file-earmark-text"></i></div>
<div class="flex-grow-1 position-relative">
<a href="{% url 'requests_app:request_detail' req.pk %}" class="feed-author text-decoration-none stretched-link">
{% firstof req.title req.get_type_display "Заявление" %}
</a>
<div class="feed-sub mt-1 text-secondary">
Автор: {{ req.employee }}{% if req.department %} · Отдел: {{ req.department }}{% endif %} · {{ req.created_at|date:"d.m.Y H:i" }}
</div>
{% if req.comment %}<div class="small mt-1 text-body">{{ req.comment|truncatewords:35 }}</div>{% endif %}
</div>
<div class="ms-2 mt-1">
{% if req.status == 'pending' %}<span class="badge bg-secondary-subtle text-secondary-emphasis border border-secondary-subtle">На рассмотрении</span>
{% elif req.status == 'approved' %}<span class="badge bg-success-subtle text-success-emphasis border border-success-subtle">Одобрено</span>
{% elif req.status == 'rejected' %}<span class="badge bg-danger-subtle text-danger-emphasis border border-danger-subtle">Отклонено</span>
{% elif req.status == 'cancelled' %}<span class="badge bg-dark-subtle text-dark-emphasis border border-dark-subtle">Отменено</span>
{% elif req.status == 'draft' %}<span class="badge bg-light text-body border">Черновик</span>
{% endif %}
</div>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{# ===== ЧАТЫ ===== #}
{% if counts.chat %}
<div class="feed-card mb-3 section" data-sec="chat">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-chat-dots"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">Чаты</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'chat' %}
{% with chat=item.object %}
{% firstof chat.name chat.department.name chat.pk as chat_label %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="chat" data-haystack="{{ chat_label }}">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-chat-dots"></i></div>
<div class="flex-grow-1 position-relative">
<a href="{% url 'communications:chat_detail' chat.pk %}" class="feed-author text-decoration-none stretched-link">
{% if chat.type == 'global' %}
Глобальный чат
{% elif chat.type == 'department' and chat.department %}
Чат отдела: {{ chat.department.name }}
{% elif chat.type == 'private' %}
{% for u in chat.participants.all %}{% if u != request.user %}{% if not forloop.first %}, {% endif %}{{ u.get_full_name|default:u.username }}{% endif %}{% endfor %}
{% else %}
{{ chat_label }}
{% endif %}
</a>
</div>
<i class="bi-chevron-right text-secondary ms-2 mt-1"></i>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{# ===== СООБЩЕНИЯ ===== #}
{% if counts.message %}
<div class="feed-card mb-3 section" data-sec="message">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-chat-left-text"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">Сообщения</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'message' %}
{% with msg=item.object %}
{% firstof msg.chat.name msg.chat.department.name msg.chat.pk as msg_chat_label %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="message" data-haystack="{{ msg.content|default_if_none:'' }}">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-chat-left-text"></i></div>
<div class="flex-grow-1 position-relative">
<a href="{% url 'communications:chat_detail' msg.chat.pk %}" class="feed-author text-decoration-none stretched-link">
{% if msg.author %}{{ msg.author.get_full_name|default:msg.author.username }}: {% endif %}
{{ msg.content|default:"(вложение)"|truncatechars:120 }}
</a>
<div class="feed-sub mt-1 text-secondary">
{{ msg.created_at|date:"d.m.Y H:i" }} · Чат: {{ msg_chat_label }}
</div>
</div>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{# ===== СОБЫТИЯ ===== #}
{% if counts.event %}
<div class="feed-card mb-3 section" data-sec="event">
<div class="feed-hd">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-calendar-event"></i></div>
<div class="feed-meta flex-grow-1"><h6 class="mb-0 text-secondary">События</h6></div>
</div>
<div class="list-results">
{% for item in results %}
{% if item.model_name == 'event' %}
{% with ev=item.object %}
{% firstof ev.title ev.name "Событие" as ev_title %}
<div class="result-row d-flex align-items-start gap-2 py-2 px-2" data-type="event" data-haystack="{{ ev_title }}">
<div class="feed-ava" style="width:44px;height:44px;"><i class="bi-calendar-event"></i></div>
<div class="flex-grow-1 position-relative">
<div class="feed-author">{{ ev_title }}</div>
<div class="feed-sub mt-1 text-secondary">
{% if ev.start %}Начало: {{ ev.start|date:"d.m.Y H:i" }}
{% elif ev.starts_at %}Начало: {{ ev.starts_at|date:"d.m.Y H:i" }}{% endif %}
{% if ev.location %} · {{ ev.location }}{% endif %}
{% if ev.department %} · {% firstof ev.department.name ev.department " " %}{% endif %}
</div>
</div>
</div>
{% endwith %}
{% endif %}
{% endfor %}
</div>
</div>
{% endif %}
{% else %}
<div class="alert alert-info shadow-sm mt-3">Ничего не найдено.</div>
{% endif %}
</div>
</div>
</div>
<script>
(function(){
const input = document.getElementById('globalSearch');
const clearBtn = document.querySelector('.ios-search-clear');
const labels = document.querySelectorAll('[data-filter]');
const radios = document.querySelectorAll('input[name="srchFilter"]');
const sections = document.querySelectorAll('.section');
const norm = s => (s||'').toString().toLowerCase().trim();
function applyFilter(){
const text = norm(input?.value || '');
const activeLabel = document.querySelector('input[name="srchFilter"]:checked + label');
const active = activeLabel ? activeLabel.dataset.filter : 'all';
sections.forEach(sec => {
const type = sec.dataset.sec;
const typeAllowed = (active === 'all' || active === type);
if (!typeAllowed) { sec.style.display = 'none'; return; }
const rows = sec.querySelectorAll('.result-row');
let visible = 0;
rows.forEach(row => {
const hay = norm(row.dataset.haystack);
const show = (text === '' || hay.includes(text));
row.style.display = show ? '' : 'none';
if (show) visible++;
});
sec.style.display = (text === '' || visible > 0) ? '' : 'none';
});
}
labels.forEach(l => l.addEventListener('click', () => setTimeout(applyFilter, 0)));
radios.forEach(r => r.addEventListener('change', applyFilter));
if (input){ input.addEventListener('input', applyFilter); }
if (clearBtn && input){
clearBtn.addEventListener('click', () => { input.value=''; input.focus(); input.dispatchEvent(new Event('input')); });
}
applyFilter();
})();
</script>
{% endblock %}

View File

@ -0,0 +1,905 @@
# backend/tests/api/v1/documents/test_documents_api.py
from __future__ import annotations
"""
Тесты CRUD API документов (/api/v1/documents/ + экшен acknowledge) по чек-листу.
ВАЖНО: API ещё не реализовано — эти тесты задают ожидаемое поведение (TDD).
Все функции и классы снабжены докстрингами (Google-style).
"""
import itertools
from typing import Iterable
import pytest
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.urls import reverse
from django.utils import timezone
from django.core.files.uploadedfile import SimpleUploadedFile
from django.conf import settings
from rest_framework import status
from rest_framework.test import APIClient
from documents.models import Document, DocumentAcknowledgement
pytestmark = pytest.mark.django_db
User = get_user_model()
# -------------- helpers / factories --------------
_phone_seq = itertools.count(1000)
def _unique_phone() -> str:
"""Генерирует уникальный валидный E.164 номер в рамках сессии тестов.
Returns:
str: Телефонный номер в формате +7999000xxx.
"""
return f"+7999000{next(_phone_seq):03d}"
def _file(name: str = "doc.pdf", content: bytes = b"x") -> SimpleUploadedFile:
"""Создаёт минимальный загружаемый файл.
Args:
name (str): Имя файла.
content (bytes): Содержимое.
Returns:
SimpleUploadedFile: Файл для multipart POST/PATCH.
"""
return SimpleUploadedFile(
name=name, content=content, content_type="application/pdf"
)
def grant_perms(user: User, *codenames: str) -> None:
"""Выдаёт пользователю модельные права по кодовым именам из приложения documents.
Args:
user (User): Пользователь.
*codenames (str): Набор кодов прав (например, "view_document").
Raises:
Permission.DoesNotExist: Если указанного права не существует.
"""
for code in codenames:
perm = Permission.objects.get(
codename=code, content_type__app_label="documents"
)
user.user_permissions.add(perm)
user.save(update_fields=[])
def make_document(
*,
title: str = "Doc",
uploaded_by: User,
description: str = "desc",
sent_to_all: bool = True,
recipients: Iterable[User] | None = None,
) -> Document:
"""Создаёт Document напрямую (минует API).
Args:
title (str): Заголовок.
uploaded_by (User): Кто загрузил.
description (str): Описание.
sent_to_all (bool): Признак рассылки всем.
recipients (Iterable[User] | None): Конкретные получатели при sent_to_all=false.
Returns:
Document: Созданная модель с прикреплённым файлом.
"""
doc = Document.objects.create(
title=title,
description=description,
uploaded_by=uploaded_by,
uploaded_at=timezone.now(),
sent_to_all=sent_to_all,
file=_file(),
)
if not sent_to_all and recipients:
doc.recipients.set(recipients)
return doc
@pytest.fixture
def make_user(user_factory):
"""Фабрика пользователей для тестов API уровня документов.
Args:
user_factory (Callable): Фабрика из tests/conftest.py.
Returns:
Callable[..., User]: Функция создания Employee.
"""
def _make(email: str, **extra) -> User:
return user_factory(email=email, **extra)
return _make
@pytest.fixture
def auth_client(auth_client_factory):
"""Фабрика DRF-клиентов, аутентифицированных под заданным пользователем.
Args:
auth_client_factory: Фабрика из tests/conftest.py.
Returns:
Callable[[User|None], APIClient]: Клиент с force_authenticate.
"""
return auth_client_factory
@pytest.fixture
def api_urls():
"""Возвращает именованные URL API документов.
Returns:
dict: Словарь с ключами list, detail(id), ack(id).
"""
def _detail(pk: int) -> str:
return reverse("api:v1:documents-detail", args=[pk])
def _ack(pk: int) -> str:
return reverse("api:v1:documents-acknowledge", args=[pk])
return {
"list": reverse("api:v1:documents-list"),
"detail": _detail,
"ack": _ack,
}
# -------------------- A. Роли и доступ --------------------
class TestAuthAndPermissions:
"""A. Роли и доступ — статусы для разных ролей и прав."""
def test_anonymous_all_401(self, api_client: APIClient, api_urls):
"""Аноним: все запросы возвращают 401."""
list_url = api_urls["list"]
# list
assert api_client.get(list_url).status_code == status.HTTP_401_UNAUTHORIZED
# create
r = api_client.post(list_url, {"title": "T"}, format="multipart")
assert r.status_code == status.HTTP_401_UNAUTHORIZED
# prepare object for other methods
author = User.objects.create(
email="auth@example.com", phone_number=_unique_phone()
)
d = make_document(uploaded_by=author)
# detail
assert (
api_client.get(api_urls["detail"](d.pk)).status_code
== status.HTTP_401_UNAUTHORIZED
)
# put/patch/delete
assert (
api_client.put(
api_urls["detail"](d.pk), {"title": "X"}, format="json"
).status_code
== 401
)
assert (
api_client.patch(
api_urls["detail"](d.pk), {"title": "X"}, format="json"
).status_code
== 401
)
assert api_client.delete(api_urls["detail"](d.pk)).status_code == 401
# acknowledge action
assert api_client.post(api_urls["ack"](d.pk), {}).status_code == 401
def test_regular_user_forbidden_except_ack(self, auth_client, make_user, api_urls):
"""Обычный user без прав: CRUD → 403, ACK → 200.
По текущей спецификации экшен acknowledge пускает любого аутентифицированного.
Этот тест фиксирует такое поведение.
"""
u = make_user("u@example.com")
client = auth_client(u)
list_url = api_urls["list"]
assert client.get(list_url).status_code == 403
# POST (multipart)
resp = client.post(
list_url,
{"title": "T", "file": _file(), "sent_to_all": True},
format="multipart",
)
assert resp.status_code == 403
# object
author = make_user("a@example.com")
d = make_document(uploaded_by=author)
# detail & write ops
assert client.get(api_urls["detail"](d.pk)).status_code == 403
assert (
client.put(
api_urls["detail"](d.pk), {"title": "X"}, format="json"
).status_code
== 403
)
assert (
client.patch(
api_urls["detail"](d.pk), {"title": "X"}, format="json"
).status_code
== 403
)
assert client.delete(api_urls["detail"](d.pk)).status_code == 403
# acknowledge → 200
r = client.post(api_urls["ack"](d.pk), {})
assert r.status_code == 200
assert r.json().get("ok") is True
@pytest.mark.parametrize(
"perm,method,expected",
[
("view_document", "get_list", 200),
("view_document", "get_detail", 200),
("add_document", "post", 201),
("change_document", "put", 200),
("change_document", "patch", 200),
("delete_document", "delete", 204),
],
)
def test_users_with_individual_perms(
self, auth_client, make_user, api_urls, perm, method, expected
):
"""Пользователь с конкретным правом получает соответствующий доступ."""
author = make_user("author@example.com")
target = make_user("p@example.com")
grant_perms(target, perm)
client = auth_client(target)
list_url = api_urls["list"]
# set up object
d = make_document(uploaded_by=author)
if method == "get_list":
r = client.get(list_url)
assert r.status_code == expected
elif method == "get_detail":
r = client.get(api_urls["detail"](d.pk))
assert r.status_code == expected
elif method == "post":
data = {"title": "T", "file": _file(), "sent_to_all": True}
r = client.post(list_url, data, format="multipart")
assert r.status_code == expected, r.content
elif method == "put":
r = client.put(
api_urls["detail"](d.pk),
{"title": "X", "description": "D", "sent_to_all": True},
format="json",
)
assert r.status_code == expected
elif method == "patch":
r = client.patch(api_urls["detail"](d.pk), {"title": "X"}, format="json")
assert r.status_code == expected
elif method == "delete":
r = client.delete(api_urls["detail"](d.pk))
assert r.status_code == expected
else:
raise AssertionError("unknown method")
@pytest.mark.parametrize(
"is_staff,is_superuser", [(True, False), (True, True), (False, True)]
)
def test_staff_and_superuser_full_access(
self, auth_client, make_user, api_urls, is_staff, is_superuser
):
"""staff/superuser: полный доступ (все запросы 2xx)."""
author = make_user("author@example.com")
admin = make_user("admin@example.com", staff=is_staff, superuser=is_superuser)
client = auth_client(admin)
# create
r = client.post(
api_urls["list"],
{"title": "T", "file": _file(), "sent_to_all": True},
format="multipart",
)
assert r.status_code in (200, 201)
doc_id = r.json().get("id")
# list/detail
assert client.get(api_urls["list"]).status_code == 200
assert client.get(api_urls["detail"](doc_id)).status_code == 200
# update
assert (
client.patch(
api_urls["detail"](doc_id), {"title": "TT"}, format="json"
).status_code
== 200
)
# delete
assert client.delete(api_urls["detail"](doc_id)).status_code in (200, 204)
# -------------------- B. Создание (POST) --------------------
@pytest.mark.django_db
class TestCreate:
"""B. Создание документа multipart/form-data."""
def test_create_sent_to_all_true(
self, auth_client, make_user, api_urls, monkeypatch
):
"""Успех: sent_to_all=true → 201 и корректное тело ответа + вызов notify() один раз."""
uploader = make_user("uploader@example.com", staff=True)
client = auth_client(uploader)
called = {"n": 0, "doc": None}
def fake_notify(doc):
called["n"] += 1
called["doc"] = doc
# замокаем side-effect
monkeypatch.setattr(
"documents.notification.notify_users_about_document",
fake_notify,
raising=False,
)
data = {"title": "Title", "file": _file(), "sent_to_all": True}
r = client.post(api_urls["list"], data, format="multipart")
assert r.status_code in (200, 201), r.content
body = r.json()
assert {"id", "title", "file_url", "sent_to_all"} <= body.keys()
assert body["sent_to_all"] is True
assert body["title"] == "Title"
assert isinstance(body["id"], int)
assert body["file_url"]
# notify был вызван один раз уже c назначенными получателями
assert called["n"] == 1
assert isinstance(called["doc"], Document)
def test_create_sent_to_all_false_with_recipients_variants(
self, auth_client, make_user, api_urls
):
"""sent_to_all=false + recipient_ids (repeat / JSON / CSV) → 201; recipient_count = числу активных."""
uploader = make_user("uploader@example.com", staff=True)
client = auth_client(uploader)
# активные
r1 = make_user("r1@example.com", active=True)
r2 = make_user("r2@example.com", active=True)
# неактивный
r3 = make_user("r3@example.com", active=False)
list_url = api_urls["list"]
# 1) repeat-params
data1 = {
"title": "T1",
"file": _file(),
"sent_to_all": False,
"recipient_ids": [r1.id, r2.id, r3.id],
}
resp1 = client.post(list_url, data1, format="multipart")
assert resp1.status_code in (200, 201), resp1.content
body1 = resp1.json()
assert body1.get("recipient_count") == 2 # только активные
# 2) JSON-массив
data2 = {
"title": "T2",
"file": _file(name="f2.pdf"),
"sent_to_all": False,
"recipient_ids": f"[{r1.id},{r3.id}]",
}
resp2 = client.post(list_url, data2, format="multipart")
assert resp2.status_code in (200, 201), resp2.content
assert resp2.json().get("recipient_count") == 1
# 3) CSV-строка
data3 = {
"title": "T3",
"file": _file(name="f3.pdf"),
"sent_to_all": False,
"recipient_ids": f"{r1.id},{r2.id},{999999}",
}
resp3 = client.post(list_url, data3, format="multipart")
assert resp3.status_code in (200, 201), resp3.content
assert resp3.json().get("recipient_count") == 2
def test_create_validations(self, auth_client, make_user, api_urls):
"""Валидации обязательных полей и recipient_ids."""
uploader = make_user("uploader@example.com", staff=True)
client = auth_client(uploader)
url = api_urls["list"]
# нет title
r = client.post(url, {"file": _file(), "sent_to_all": True}, format="multipart")
assert r.status_code == 400
# нет file
r = client.post(url, {"title": "T", "sent_to_all": True}, format="multipart")
assert r.status_code == 400
# sent_to_all=false и пустые recipient_ids → 400
r = client.post(url, {"title": "T", "sent_to_all": False}, format="multipart")
assert r.status_code == 400
def test_create_skips_nonexistent_or_inactive_recipients(
self, auth_client, make_user, api_urls
):
"""recipient_ids с несуществующими/неактивными → создаётся без них; в recipients их нет."""
uploader = make_user("uploader@example.com", staff=True)
client = auth_client(uploader)
active = make_user("a@example.com", active=True)
inactive = make_user("i@example.com", active=False)
r = client.post(
api_urls["list"],
{
"title": "T",
"file": _file(),
"sent_to_all": False,
"recipient_ids": [active.id, inactive.id, 999999],
},
format="multipart",
)
assert r.status_code in (200, 201)
doc_id = r.json()["id"]
doc = Document.objects.get(pk=doc_id)
ids = set(doc.recipients.values_list("id", flat=True))
assert active.id in ids
assert inactive.id not in ids
# -------------------- C. Чтение (GET) --------------------
class TestRead:
"""C. Чтение: list/detail, структура, пагинация, file_url."""
def test_get_list_with_view_perm_and_pagination(
self, auth_client, make_user, api_urls
):
"""GET list с правом view_document → 200 + DRF пагинация и page_size=20."""
author = make_user("author@example.com")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
client = auth_client(viewer)
for i in range(25):
make_document(uploaded_by=author, title=f"D{i:02d}")
r1 = client.get(api_urls["list"])
assert r1.status_code == 200
data = r1.json()
assert {"count", "next", "previous", "results"} <= data.keys()
assert len(data["results"]) == 20
# вторая страница
r2 = client.get(api_urls["list"], {"page": 2})
assert r2.status_code == 200
assert len(r2.json()["results"]) == 5
# проверим базовые поля
sample = data["results"][0]
expected = {
"id",
"title",
"description",
"uploaded_by",
"uploaded_at",
"sent_to_all",
"recipients",
"file_url",
"is_acknowledged",
}
assert expected <= set(sample.keys())
def test_get_detail_fields_and_file_url(
self, auth_client, make_user, api_urls, settings
):
"""GET detail возвращает нужные поля; is_acknowledged=false до отметки; file_url внутри MEDIA_URL."""
author = make_user("author@example.com")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
client = auth_client(viewer)
doc = make_document(uploaded_by=author, title="T")
r = client.get(api_urls["detail"](doc.pk))
assert r.status_code == 200
body = r.json()
assert body["is_acknowledged"] is False
assert body["file_url"]
assert body["file_url"].startswith(settings.MEDIA_URL)
# -------------------- D. Обновление (PUT/PATCH) --------------------
class TestUpdate:
"""D. Обновление документа (PUT/PATCH)."""
def test_update_title_description(self, auth_client, make_user, api_urls):
"""PUT/PATCH обновляют title/description."""
author = make_user("author@example.com")
editor = make_user("editor@example.com")
grant_perms(editor, "change_document")
client = auth_client(editor)
d = make_document(uploaded_by=author, title="Old", description="old")
r = client.patch(
api_urls["detail"](d.pk),
{"title": "New", "description": "new"},
format="json",
)
assert r.status_code == 200
d.refresh_from_db()
assert d.title == "New"
assert d.description == "new"
def test_replace_file_multipart_patch(self, auth_client, make_user, api_urls):
"""Заменить file (multipart PATCH) → file_url изменился."""
author = make_user("author@example.com")
editor = make_user("editor@example.com")
grant_perms(editor, "change_document")
client = auth_client(editor)
d = make_document(uploaded_by=author)
# получим старый url через detail (этап API)
r1 = client.get(api_urls["detail"](d.pk))
old_url = r1.json()["file_url"]
r2 = client.patch(
api_urls["detail"](d.pk),
{"file": _file(name="new.pdf")},
format="multipart",
)
assert r2.status_code == 200
new_url = r2.json()["file_url"]
assert new_url != old_url
def test_toggle_sent_to_all_affects_recipients(
self, auth_client, make_user, api_urls
):
"""Переключение sent_to_all ↔ влияет на recipients."""
author = make_user("author@example.com")
editor = make_user("editor@example.com")
r1 = make_user("r1@example.com")
r2 = make_user("r2@example.com")
grant_perms(editor, "change_document")
client = auth_client(editor)
d = make_document(uploaded_by=author, sent_to_all=False, recipients=[r1, r2])
# false -> true -> recipients очищены
r = client.patch(api_urls["detail"](d.pk), {"sent_to_all": True}, format="json")
assert r.status_code == 200
d.refresh_from_db()
assert d.sent_to_all is True
assert d.recipients.count() == 0
# true -> false с recipient_ids
r = client.patch(
api_urls["detail"](d.pk),
{"sent_to_all": False, "recipient_ids": [r2.id]},
format="json",
)
assert r.status_code == 200
d.refresh_from_db()
assert d.sent_to_all is False
assert list(d.recipients.values_list("id", flat=True)) == [r2.id]
def test_patch_only_recipient_ids_replaces_list(
self, auth_client, make_user, api_urls
):
"""Частичный PATCH только recipient_ids (при sent_to_all=false) → замена списка."""
author = make_user("author@example.com")
editor = make_user("editor@example.com")
a = make_user("a@example.com")
b = make_user("b@example.com")
c = make_user("c@example.com")
grant_perms(editor, "change_document")
client = auth_client(editor)
d = make_document(uploaded_by=author, sent_to_all=False, recipients=[a, b])
r = client.patch(
api_urls["detail"](d.pk), {"recipient_ids": [c.id]}, format="json"
)
assert r.status_code == 200
d.refresh_from_db()
assert list(d.recipients.values_list("id", flat=True)) == [c.id]
# -------------------- E. Удаление (DELETE) --------------------
class TestDelete:
"""E. Удаление документа."""
def test_delete_and_cascade_acknowledgements(
self, auth_client, make_user, api_urls
):
"""DELETE → 204; последующий GET → 404; каскад: DocumentAcknowledgement удалены."""
author = make_user("author@example.com")
deleter = make_user("deleter@example.com")
grant_perms(deleter, "delete_document")
client = auth_client(deleter)
d = make_document(uploaded_by=author)
# создадим ack
ack_user = make_user("ack@example.com")
DocumentAcknowledgement.objects.create(document=d, user=ack_user)
r = client.delete(api_urls["detail"](d.pk))
assert r.status_code in (200, 204)
# detail теперь 404
r2 = client.get(api_urls["detail"](d.pk))
assert r2.status_code == 404
# каскад: записей нет
assert not DocumentAcknowledgement.objects.filter(document=d).exists()
# -------------------- F. Экшен acknowledge --------------------
class TestAcknowledge:
"""F. Экшен POST /{id}/acknowledge/ — идемпотентность и политика доступа."""
def test_ack_first_and_repeat(self, auth_client, make_user, api_urls):
"""Первый вызов создаёт запись, повторный — already=true (200 оба раза)."""
author = make_user("author@example.com")
any_user = make_user("u@example.com")
client = auth_client(any_user)
d = make_document(
uploaded_by=author, sent_to_all=False
) # умышленно не в получателях
r1 = client.post(api_urls["ack"](d.pk), {})
assert r1.status_code == 200
assert r1.json() == {"ok": True, "already": False}
assert (
DocumentAcknowledgement.objects.filter(document=d, user=any_user).count()
== 1
)
# повторный
r2 = client.post(api_urls["ack"](d.pk), {})
assert r2.status_code == 200
assert r2.json() == {"ok": True, "already": True}
assert (
DocumentAcknowledgement.objects.filter(document=d, user=any_user).count()
== 1
)
def test_detail_reflects_is_acknowledged(self, auth_client, make_user, api_urls):
"""GET detail после отметки → is_acknowledged=true."""
author = make_user("author@example.com")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
client = auth_client(viewer)
d = make_document(uploaded_by=author)
# пока false
assert client.get(api_urls["detail"](d.pk)).json()["is_acknowledged"] is False
# отметим
other = make_user("x@example.com") # не влияет на viewer
DocumentAcknowledgement.objects.create(document=d, user=viewer)
assert client.get(api_urls["detail"](d.pk)).json()["is_acknowledged"] is True
def test_policy_ack_even_if_not_recipient_current_behavior(
self, auth_client, make_user, api_urls
):
"""Пользователь вне recipients и при sent_to_all=false тоже может ACK (текущая политика)."""
author = make_user("author@example.com")
u = make_user("u@example.com")
client = auth_client(u)
r = client.post(
api_urls["ack"](make_document(uploaded_by=author, sent_to_all=False).pk), {}
)
assert r.status_code == 200
# Если это поведение нежелательно — создать тикет на ужесточение.
# -------------------- G. Сериализация и данные --------------------
class TestSerialization:
"""G. Сериализация: uploaded_by/recipients поля и скачивание файла."""
def test_uploaded_by_and_recipients_shape(self, auth_client, make_user, api_urls):
"""uploaded_by и recipients[] содержат id, full_name, email; full_name корректен."""
author = make_user("author@example.com", first_name="Иван", last_name="Иванов")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
client = auth_client(viewer)
r1 = make_user("r1@example.com", first_name="Петр", last_name="Петров")
r2 = make_user("r2@example.com", first_name="Сидор", last_name="Сидоров")
d = make_document(uploaded_by=author, sent_to_all=False, recipients=[r1, r2])
body = client.get(api_urls["detail"](d.pk)).json()
ub = body["uploaded_by"]
assert {"id", "full_name", "email"} <= ub.keys()
assert "Иван" in ub["full_name"]
rec = body["recipients"]
assert isinstance(rec, list) and rec
assert {"id", "full_name", "email"} <= rec[0].keys()
full_names = [x["full_name"] for x in rec]
assert any("Петр" in fn for fn in full_names)
def test_file_url_downloadable_in_dev(
self, client, auth_client, make_user, api_urls
):
"""Интеграционный: можно скачать файл по file_url (если DEBUG и MEDIA настроены)."""
author = make_user("author@example.com")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
api = auth_client(viewer)
d = make_document(uploaded_by=author)
file_url = api.get(api_urls["detail"](d.pk)).json()["file_url"]
# обычный Django client (без APIClient) для скачивания
resp = client.get(file_url)
assert resp.status_code == 200
assert resp.content # не пусто
# -------------------- H. Ошибки/краевые случаи --------------------
class TestErrorsAndEdgeCases:
"""H. Неудачные Content-Type, perms и размер файла."""
def test_wrong_content_type_on_create(self, auth_client, make_user, api_urls):
"""Неверный Content-Type (JSON без файла) → 400 или 415 (фиксируем фактическое поведение)."""
uploader = make_user("uploader@example.com", staff=True)
client = auth_client(uploader)
r = client.post(
api_urls["list"], {"title": "X", "sent_to_all": True}, format="json"
)
assert r.status_code in (400, 415)
def test_put_patch_delete_without_perms(self, auth_client, make_user, api_urls):
"""Попытка PUT/PATCH/DELETE без прав → 403."""
user = make_user("u@example.com")
client = auth_client(user)
author = make_user("a@example.com")
d = make_document(uploaded_by=author)
assert (
client.put(
api_urls["detail"](d.pk), {"title": "X"}, format="json"
).status_code
== 403
)
assert (
client.patch(
api_urls["detail"](d.pk), {"title": "X"}, format="json"
).status_code
== 403
)
assert client.delete(api_urls["detail"](d.pk)).status_code == 403
def test_get_without_view_perm(self, auth_client, make_user, api_urls):
"""GET list/detail без view_document → 403."""
user = make_user("u@example.com")
client = auth_client(user)
author = make_user("a@example.com")
d = make_document(uploaded_by=author)
assert client.get(api_urls["list"]).status_code == 403
assert client.get(api_urls["detail"](d.pk)).status_code == 403
def test_big_file_over_limit_if_configured(
self, auth_client, make_user, api_urls, settings
):
"""Большой файл > лимита → ожидаем 413 или 400 (если лимит включён); иначе 201."""
uploader = make_user("uploader@example.com", staff=True)
client = auth_client(uploader)
big = SimpleUploadedFile("big.bin", b"x" * (5 * 1024 * 1024)) # 5 MB
r = client.post(
api_urls["list"],
{"title": "Big", "file": big, "sent_to_all": True},
format="multipart",
)
if getattr(settings, "DATA_UPLOAD_MAX_MEMORY_SIZE", None):
assert r.status_code in (400, 413)
else:
# фиксируем текущий результат при отсутствии лимита
assert r.status_code in (200, 201)
# -------------------- I. Производительность/фильтры --------------------
class TestPerformance:
"""I. N+1 и пагинация."""
def test_no_n_plus_one_in_list(
self, django_assert_num_queries, auth_client, make_user, api_urls
):
"""Листинг с prefetch/select_related: количество SQL не растёт линейно."""
author = make_user("author@example.com")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
client = auth_client(viewer)
# создадим 1 и 10 документов
make_document(uploaded_by=author, title="one")
for i in range(10):
make_document(uploaded_by=author, title=f"doc{i:02d}")
# допуская кэш/хитрости, выставим безопасный порог
with django_assert_num_queries(10):
r = client.get(api_urls["list"])
assert r.status_code == 200
def test_pagination_navigation(self, auth_client, make_user, api_urls):
"""Навигация по страницам с глобальной DRF-пагинацией."""
author = make_user("author@example.com")
viewer = make_user("viewer@example.com")
grant_perms(viewer, "view_document")
client = auth_client(viewer)
for i in range(45):
make_document(uploaded_by=author, title=f"D{i}")
# page=1
r1 = client.get(api_urls["list"], {"page": 1})
assert r1.status_code == 200
# page=3 (последняя)
r3 = client.get(api_urls["list"], {"page": 3})
assert r3.status_code == 200
# -------------------- J. Совместимость и безопасность --------------------
class TestSecurityAndPolicy:
"""J. Базовая совместимость по авторизации и политика acknowledge."""
def test_jwt_or_session_happy_path(self, api_client: APIClient):
"""Проверка базовой настройки DRF: IsAuthenticated + DjangoModelPermissions включены глобально.
Тест не бьёт конкретный endpoint, а страхует глобальную конфигурацию.
"""
from django.conf import settings as dj_settings
assert "rest_framework.permissions.IsAuthenticated" in str(
dj_settings.REST_FRAMEWORK.get("DEFAULT_PERMISSION_CLASSES", [])
)
assert "rest_framework.permissions.DjangoModelPermissions" in str(
dj_settings.REST_FRAMEWORK.get("DEFAULT_PERMISSION_CLASSES", [])
)
def test_policy_ack_without_perms_is_allowed_but_crud_is_not(
self, auth_client, make_user, api_urls
):
"""Авторизованный без perms не видит список/деталь (403), но может acknowledge (200) — подтвердить спецификацию."""
u = make_user("u@example.com")
client = auth_client(u)
author = make_user("a@example.com")
d = make_document(uploaded_by=author, sent_to_all=False)
assert client.get(api_urls["list"]).status_code == 403
assert client.get(api_urls["detail"](d.pk)).status_code == 403
r = client.post(api_urls["ack"](d.pk), {})
assert r.status_code == 200

View File

@ -152,7 +152,6 @@ def test_head_can_change_head_and_loses_rights_afterwards(api_client: APIClient)
# старый руководитель меняет руководителя на другого
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

View File

@ -1,252 +1,656 @@
# backend/tests/api/v1/feed/test_feed_api.py
from __future__ import annotations
"""
Тесты API фида (посты, комментарии, лайки) под новую политику доступа:
- права на объекты отдела выдаются через роль внутри конкретного отдела;
- company-посты требуют глобальных модельных прав/статуса staff.
"""
from typing import Iterable, Optional, Tuple
import pytest
from django.urls import reverse
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.constants import DeptPerm
from employees.models import (
Department,
DepartmentPermission,
DepartmentRole,
EmployeeDepartment,
)
from feed.constants import TYPE_COMPANY, TYPE_DEPARTMENT
from feed.models import Comment, Post, PostLike
from rest_framework import status
from rest_framework.test import APIClient
from employees.models import Employee, Department, EmployeeDepartment, DepartmentRole
from feed.models import Post
from feed.constants import TYPE_COMPANY, TYPE_DEPARTMENT, TYPE_EMPLOYEE
User = get_user_model()
pytestmark = pytest.mark.django_db
# ----------------- helpers -----------------
# ------------------------ helpers ------------------------
_seq = 1
def _uniq_email(prefix="user"):
global _seq
_seq += 1
return f"{prefix}{_seq}@example.com"
def _uniq_phone():
global _seq
_seq += 1
return f"+7999{_seq:07d}"
def _ensure_permission(
app_label: str, model: type, codename: str, name: Optional[str] = None
) -> Permission:
"""Возвращает Permission с указанным codename, создаёт при отсутствии.
def _user(staff=False, superuser=False) -> Employee:
u = Employee.objects.create_user(
email=_uniq_email(), password="pass",
phone_number=_uniq_phone(),
send_activation_email=False, first_name="T", last_name="U",
Примечание:
app_label параметр сейчас не используется и нужен для унификации вызовов.
Args:
app_label (str): Игнорируется.
model (type): Класс модели (для ContentType).
codename (str): Код права (напр. 'publish_company_post').
name (Optional[str]): Человекочитаемое имя права.
Returns:
Permission: Объект права.
"""
ct = ContentType.objects.get_for_model(model)
perm, _ = Permission.objects.get_or_create(
content_type=ct,
codename=codename,
defaults={"name": name or codename.replace("_", " ").title()},
)
if staff:
u.is_staff = True
if superuser:
u.is_superuser = True
u.email_verified = True
u.is_active = True
u.save(update_fields=["is_staff", "is_superuser", "email_verified", "is_active"])
# сбрасываем кеши прав на всякий
for attr in ("_perm_cache", "_user_perm_cache", "_group_perm_cache"):
if hasattr(u, attr):
try: delattr(u, attr)
except: pass
return u
return perm
def _grant_user_perm(user: Employee, perm_code: str):
"""
perm_code: 'app_label.codename', напр. 'feed.publish_company_post'
"""
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 _grant_dept_role_perm(user: Employee, dept: Department, perm_code: str):
def _mk_dept(name: str = "Dept", *, head: Optional[User] = None) -> Department:
"""Создаёт отдел.
Args:
name (str): Название отдела.
head (Optional[User]): Руководитель.
Returns:
Department: Созданный отдел.
"""
Создаёт DepartmentRole с указанным Permission и выдаёт её пользователю в отделе.
"""
app_label, codename = perm_code.split(".", 1)
p = Permission.objects.get(content_type__app_label=app_label, codename=codename)
role = DepartmentRole.objects.create(department=dept, name="Publisher")
role.permissions.add(p)
EmployeeDepartment.objects.create(employee=user, department=dept, role=role, is_active=True)
return Department.objects.create(name=name, head=head)
def _ensure_dept_permission(code: str, name: str | None = None) -> DepartmentPermission:
"""Возвращает/создаёт DepartmentPermission с заданным кодом."""
perm, _ = DepartmentPermission.objects.get_or_create(
code=code,
defaults={"name": name or code.replace("_", " ").title()},
)
return perm
def _give_dept_role_with_perms(user, dept, *, perm_codenames, role_name="Role"):
"""Создаёт роль отдела, вешает scoped_permissions и назначает пользователю."""
role = DepartmentRole.objects.create(name=role_name, department=dept)
for code in perm_codenames:
dp = _ensure_dept_permission(code)
role.scoped_permissions.add(
dp
) # ⟵ ВАЖНО: scoped_permissions, не auth.Permission
EmployeeDepartment.objects.create(
employee=user, department=dept, role=role, is_active=True
)
return role
def _items(resp):
# поддержка пагинации и без
if isinstance(resp.data, dict) and "results" in resp.data:
return resp.data["results"]
return resp.data
def _post_data(
*, type_: str, title: str = "T", body: str = "B", dept: Optional[Department] = None
) -> dict:
"""Собирает payload для создания/изменения поста.
Args:
type_ (str): Тип поста (см. feed.constants).
title (str): Заголовок.
body (str): Тело.
dept (Optional[Department]): Отдел для department-поста.
Returns:
dict: Данные для POST/PUT/PATCH запроса.
"""
data = {"type": type_, "title": title, "body": body}
if type_ == TYPE_DEPARTMENT:
data["department"] = dept.id if dept else None
return data
# ----------------- tests -----------------
def _create_post(
author: User, *, type_: str, dept: Optional[Department] = None, title: str = "T"
) -> Post:
"""Утилита создания поста напрямую (минуя API), чтобы подготовить данные.
@pytest.mark.django_db
def test_list_requires_auth(api_client):
url = reverse("api:v1:posts-list")
assert api_client.get(url).status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN)
Args:
author (User): Автор.
type_ (str): Тип поста.
dept (Optional[Department]): Отдел для department.
title (str): Заголовок.
user = _user()
api_client.force_authenticate(user=user)
assert api_client.get(url).status_code == status.HTTP_200_OK
Returns:
Post: Созданный пост.
"""
return Post.objects.create(
author=author, type=type_, department=dept, title=title, body="B"
)
@pytest.mark.django_db
def test_create_company_requires_staff_or_perm(api_client):
url = reverse("api:v1:posts-list")
author = _user()
api_client.force_authenticate(user=author)
# ------------------------ URL helpers ------------------------
payload = {
"type": TYPE_COMPANY,
"title": "Company News",
"body": "hello",
# department не указываем
@pytest.fixture
def urls():
"""Именованные URL-роуты фида.
Returns:
dict: Словарь с callables для detail/action.
"""
return {
"posts_list": reverse("api:v1:posts-list"),
"post_detail": lambda pk: reverse("api:v1:posts-detail", args=[pk]),
"post_pin": lambda pk: reverse("api:v1:posts-pin", args=[pk]),
"post_unpin": lambda pk: reverse("api:v1:posts-unpin", args=[pk]),
"post_like": lambda pk: reverse("api:v1:posts-like", args=[pk]),
"post_unlike": lambda pk: reverse("api:v1:posts-unlike", args=[pk]),
"comments_list": reverse("api:v1:comments-list"),
"comment_detail": lambda pk: reverse("api:v1:comments-detail", args=[pk]),
}
# без прав -> 403
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_403_FORBIDDEN
# выдаём feed.publish_company_post -> 201
_grant_user_perm(author, "feed.publish_company_post")
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_201_CREATED
assert resp.data["type"] == TYPE_COMPANY
assert resp.data.get("department") is None
# staff тоже может -> 201
staff = _user(staff=True)
api_client.force_authenticate(user=staff)
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_201_CREATED
# ------------------------ fixtures (project-aware) ------------------------
@pytest.mark.django_db
def test_create_department_requires_head_or_role_or_staff(api_client):
url = reverse("api:v1:posts-list")
dept = Department.objects.create(name="R&D")
@pytest.fixture
def make_user(user_factory):
"""Фабрика пользователей на основе проектной фикстуры.
u = _user()
api_client.force_authenticate(user=u)
Returns:
Callable[..., User]: Создатель пользователя.
"""
payload = {
"type": TYPE_DEPARTMENT,
"title": "Dept News",
"body": "text",
"department": dept.id,
}
def _mk(email: str, **extra) -> User:
return user_factory(email=email, **extra)
# без прав -> 403
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_403_FORBIDDEN
# глава отдела -> 201
head = _user()
dept.head = head
dept.save(update_fields=["head"])
api_client.force_authenticate(user=head)
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_201_CREATED
# сотрудник с ролью, имеющей publish_department_post -> 201
member = _user()
_grant_dept_role_perm(member, dept, "feed.publish_department_post")
api_client.force_authenticate(user=member)
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_201_CREATED
# staff -> 201
staff = _user(staff=True)
api_client.force_authenticate(user=staff)
resp = api_client.post(url, payload, format="multipart")
assert resp.status_code == status.HTTP_201_CREATED
return _mk
@pytest.mark.django_db
def test_employee_type_is_forbidden_and_department_field_validation(api_client):
url = reverse("api:v1:posts-list")
staff = _user(staff=True)
api_client.force_authenticate(user=staff)
dept = Department.objects.create(name="Ops")
@pytest.fixture
def auth(api_client: APIClient):
"""Возвращает функцию для аутентификации клиента под заданным пользователем.
# TYPE_EMPLOYEE вообще нельзя
resp = api_client.post(url, {"type": TYPE_EMPLOYEE, "title": "x", "body": "b"}, format="multipart")
assert resp.status_code == status.HTTP_400_BAD_REQUEST
Args:
api_client (APIClient): DRF APIClient.
# type=department, но department не указан -> 400
resp = api_client.post(url, {"type": TYPE_DEPARTMENT, "title": "x", "body": "b"}, format="multipart")
assert resp.status_code == status.HTTP_400_BAD_REQUEST
Returns:
Callable[[User], APIClient]: Аутентифицированный клиент.
"""
# type!=department, но department указан -> 400
resp = api_client.post(
url,
{"type": TYPE_COMPANY, "title": "x", "body": "b", "department": dept.id},
format="multipart",
def _as(u: User) -> APIClient:
api_client.force_authenticate(user=u)
return api_client
return _as
# ------------------------ A. Аноним ------------------------
class TestAnonymousAccess:
"""A. Аноним: все эндпоинты → 401."""
def test_all_unauth_401(self, api_client: APIClient, make_user, urls):
"""Проверяет, что анониму всё недоступно (401)."""
author = make_user("a@example.com")
dept = _mk_dept(head=author)
p = _create_post(author, type_=TYPE_DEPARTMENT, dept=dept)
assert (
api_client.get(urls["posts_list"]).status_code
== status.HTTP_401_UNAUTHORIZED
)
assert (
api_client.get(urls["post_detail"](p.id)).status_code
== status.HTTP_401_UNAUTHORIZED
)
assert (
api_client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=dept),
format="json",
).status_code
== 401
)
assert api_client.post(urls["post_like"](p.id), {}).status_code == 401
assert api_client.get(urls["comments_list"]).status_code == 401
assert (
api_client.post(
urls["comments_list"], {"post": p.id, "text": "hi"}, format="json"
).status_code
== 401
)
# ------------------------ B. Обычный пользователь ------------------------
class TestRegularUser:
"""B. Обычный аутентифицированный пользователь: чтение/коммент/лайки 2xx, посты отдела 403."""
def test_read_and_interact_but_no_dept_post_rights(self, make_user, auth, urls):
"""GET list/detail → 200, комментарий и like/unlike → 200, CRUD dept-поста → 403."""
u = make_user("u@example.com")
a = make_user("a@example.com")
dept = _mk_dept(head=a)
client = auth(u)
# подготовим пост для чтения/лайков
post = _create_post(a, type_=TYPE_DEPARTMENT, dept=dept)
# чтение
assert client.get(urls["posts_list"]).status_code == 200
assert client.get(urls["post_detail"](post.id)).status_code == 200
# комментарий
r = client.post(
urls["comments_list"], {"post": post.id, "text": "hi"}, format="json"
)
assert r.status_code == 201
# лайк/анлайк
assert client.post(urls["post_like"](post.id), {}).status_code == 200
assert client.post(urls["post_unlike"](post.id), {}).status_code == 200
# попытки создать/менять/удалять dept-пост → 403
assert (
client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=dept),
format="json",
).status_code
== 403
)
assert client.patch(
urls["post_detail"](post.id), {"title": "X"}, format="json"
).status_code in (403, 404)
assert client.delete(urls["post_detail"](post.id)).status_code in (403, 404)
# pin/unpin недоступны
assert client.post(urls["post_pin"](post.id)).status_code == 403
assert client.post(urls["post_unpin"](post.id)).status_code == 403
# ------------------------ C. Роль в отделе ------------------------
class TestDepartmentRole:
"""C. Роль в отделе с perm 'manage_department_feed' даёт запись в своём отделе, но не в чужом."""
PUBLISH_CODE = getattr(DeptPerm, "CREATE_POST", "publish_department_post")
def test_create_update_delete_own_department(self, make_user, auth, urls):
"""В своём отделе → 201/200/204; в другом отделе → 403."""
print(
"TEST PUBLISH CODE =",
getattr(DeptPerm, "CREATE_POST", "publish_department_post"),
)
u = make_user("role@example.com")
d1 = _mk_dept("D1")
d2 = _mk_dept("D2")
_give_dept_role_with_perms(
u, d1,
perm_codenames=[DeptPerm.CREATE_POST, DeptPerm.MANAGE_FEED]
)
client = auth(u)
# create в своем отделе
r = client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=d1, title="R1"),
format="json",
)
assert r.status_code == 201, r.content
pid = r.json()["id"]
# update/destroy своего поста
assert (
client.patch(
urls["post_detail"](pid), {"title": "R1X"}, format="json"
).status_code
== 200
)
assert client.delete(urls["post_detail"](pid)).status_code in (200, 204)
# create в чужом отделе → 403
r2 = client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=d2, title="No"),
format="json",
)
assert r2.status_code == 403
# ------------------------ D. Руководитель отдела ------------------------
class TestDepartmentHead:
"""D. Head: полные права на посты своего отдела."""
def test_head_can_manage_department_posts(self, make_user, auth, urls):
"""head своего отдела может создавать/изменять/удалять dept-посты."""
head = make_user("head@example.com")
dept = _mk_dept("HD", head=head)
client = auth(head)
r = client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=dept, title="H"),
format="json",
)
assert r.status_code == 201
pid = r.json()["id"]
assert (
client.patch(
urls["post_detail"](pid), {"title": "HX"}, format="json"
).status_code
== 200
)
assert client.delete(urls["post_detail"](pid)).status_code in (200, 204)
# ------------------------ E. staff/superuser и pin ------------------------
class TestStaffAndPinning:
"""E. Staff/superuser: полный доступ; pin/unpin доступны только staff."""
def test_staff_full_access_and_pin(self, make_user, auth, urls):
"""staff: CRUD company/dept + pin/unpin."""
staff = make_user("staff@example.com", staff=True)
d = _mk_dept("PD")
client = auth(staff)
# company пост
r = client.post(
urls["posts_list"], _post_data(type_=TYPE_COMPANY, title="C"), format="json"
)
assert r.status_code == 201
pid = r.json()["id"]
# dept пост
r2 = client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=d, title="D"),
format="json",
)
assert r2.status_code == 201
pid2 = r2.json()["id"]
# pin/unpin
assert client.post(urls["post_pin"](pid)).status_code == 200
assert client.post(urls["post_unpin"](pid)).status_code == 200
# update/delete
assert (
client.patch(
urls["post_detail"](pid2), {"title": "DX"}, format="json"
).status_code
== 200
)
assert client.delete(urls["post_detail"](pid2)).status_code in (200, 204)
# ------------------------ F. Глобальные модельные права на company ------------------------
class TestCompanyModelPermsGranular:
"""Гранулярные проверки модельных прав для company-постов."""
def test_company_create_requires_add_only(self, make_user, auth, urls):
"""Наличие только feed.add_post → create=201; update/delete без change/delete → 403."""
u = make_user("only-add@example.com")
client = auth(u)
# даём только add_post
u.user_permissions.add(
_ensure_permission("feed", Post, "add_post", "Can add Post")
)
# create OK
rc = client.post(
urls["posts_list"], _post_data(type_=TYPE_COMPANY, title="C"), format="json"
)
assert rc.status_code == status.HTTP_201_CREATED, rc.content
pid = rc.json()["id"]
# update/delete запрещены (нет change/delete)
assert (
client.patch(
urls["post_detail"](pid), {"title": "CX"}, format="json"
).status_code
== 403
)
assert client.delete(urls["post_detail"](pid)).status_code == 403
def test_company_update_requires_change_only(self, make_user, auth, urls):
"""Только feed.change_post → update=200; create без add и delete без delete → 403."""
author = make_user("author@example.com")
editor = make_user("only-change@example.com")
client = auth(editor)
# готовим пост напрямую, чтобы не требовать add_post у editor
p = _create_post(author, type_=TYPE_COMPANY, title="C2")
# даём только change_post
editor.user_permissions.add(
_ensure_permission("feed", Post, "change_post", "Can change Post")
)
# create запрещён (нет add_post)
assert (
client.post(
urls["posts_list"],
_post_data(type_=TYPE_COMPANY, title="NC"),
format="json",
).status_code
== 403
)
# update OK
assert (
client.patch(
urls["post_detail"](p.id), {"title": "C2X"}, format="json"
).status_code
== 200
)
# delete запрещён (нет delete_post)
assert client.delete(urls["post_detail"](p.id)).status_code == 403
def test_company_delete_requires_delete_only(self, make_user, auth, urls):
"""Только feed.delete_post → delete=204; create/update без add/change → 403/403."""
author = make_user("author@example.com")
deleter = make_user("only-delete@example.com")
client = auth(deleter)
p = _create_post(author, type_=TYPE_COMPANY, title="C3")
# даём только delete_post
deleter.user_permissions.add(
_ensure_permission("feed", Post, "delete_post", "Can delete Post")
)
# create/update запрещены
assert (
client.post(
urls["posts_list"],
_post_data(type_=TYPE_COMPANY, title="NC"),
format="json",
).status_code
== 403
)
assert (
client.patch(
urls["post_detail"](p.id), {"title": "C3X"}, format="json"
).status_code
== 403
)
# delete OK
resp = client.delete(urls["post_detail"](p.id))
assert resp.status_code in (200, 204)
def test_company_perms_do_not_grant_department(self, make_user, auth, urls):
"""Модельные права на Post НЕ дают права создавать department-посты."""
u = make_user("model-perms@example.com")
client = auth(u)
# дадим все три модельных права — всё равно dept должен быть 403
u.user_permissions.add(
_ensure_permission("feed", Post, "add_post", "Can add Post"),
_ensure_permission("feed", Post, "change_post", "Can change Post"),
_ensure_permission("feed", Post, "delete_post", "Can delete Post"),
)
d = _mk_dept("DeptX")
assert (
client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=d, title="D"),
format="json",
).status_code
== 403
)
# ------------------------ G. Комментарии ------------------------
class TestComments:
"""G. Комментарии: создание любым аутентифицированным; изменение — автор/staff."""
def test_create_and_update_delete_policies(self, make_user, auth, urls):
"""Создание: 201; редактирование/удаление: автор/staff."""
author = make_user("a@example.com")
reader = make_user("r@example.com")
staff = make_user("s@example.com", staff=True)
dept = _mk_dept()
post = _create_post(author, type_=TYPE_DEPARTMENT, dept=dept)
# reader создаёт комментарий
client_r = auth(reader)
rc = client_r.post(
urls["comments_list"], {"post": post.id, "text": "hi"}, format="json"
)
assert rc.status_code == 201
cid = rc.json()["id"]
# чужой юзер менять не может
other = make_user("o@example.com")
client_o = auth(other)
assert (
client_o.patch(
urls["comment_detail"](cid), {"text": "hack"}, format="json"
).status_code
== 403
)
assert client_o.delete(urls["comment_detail"](cid)).status_code == 403
# автор может
resp = client_r.patch(urls["comment_detail"](cid), {"text": "ok"}, format="json")
assert resp.status_code == 200
assert resp.json().get("text") == "ok"
# staff может
client_s = auth(staff)
del_resp = client_s.delete(urls["comment_detail"](cid))
assert del_resp.status_code in (200, 204)
# после удаления — 404
get_after_del = client_r.get(urls["comment_detail"](cid))
assert get_after_del.status_code == 404
# ------------------------ H. Лайки (идемпотентность и is_liked) ------------------------
class TestLikes:
"""H. Лайки: идемпотентность и синхронизация is_liked."""
def test_like_unlike_idempotent_and_reflected(self, make_user, auth, urls):
"""Повторный like/unlike не искажает счётчик; поле is_liked корректно отражается в detail."""
a = make_user("a@example.com")
u = make_user("u@example.com")
dept = _mk_dept()
post = _create_post(a, type_=TYPE_DEPARTMENT, dept=dept)
client = auth(u)
# старт: не лайкнуто
d0 = client.get(urls["post_detail"](post.id)).json()
assert d0.get("is_liked") in (False, None)
# первый like увеличивает счётчик
r1 = client.post(urls["post_like"](post.id), {})
assert r1.status_code == 200
liked1 = r1.json().get("liked")
assert liked1 is True
# повторный like не увеличивает
r2 = client.post(urls["post_like"](post.id), {})
assert r2.status_code == 200
# detail отражает is_liked=True
d1 = client.get(urls["post_detail"](post.id)).json()
assert d1.get("is_liked") is True
# unlike уменьшает, повторный не уводит < 0
r3 = client.post(urls["post_unlike"](post.id), {})
assert r3.status_code == 200 and r3.json().get("liked") is False
r4 = client.post(urls["post_unlike"](post.id), {})
assert r4.status_code == 200
d2 = client.get(urls["post_detail"](post.id)).json()
assert d2.get("is_liked") in (False, None)
# ------------------------ I. Негативные кейсы валидации ------------------------
class TestValidation:
"""I. Негативные кейсы: отсутствие department, запрет чужого отдела и т.п."""
def test_department_type_requires_department_field(self, make_user, auth, urls):
"""type=department без department → 400."""
u = make_user("u@example.com")
client = auth(u)
r = client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=None),
format="json",
)
assert r.status_code == 400
def test_cannot_create_in_other_department_with_role(self, make_user, auth, urls):
"""Роль в одном отделе не даёт создании в другом отделе → 403."""
u = make_user("u@example.com")
d1 = _mk_dept("D1")
d2 = _mk_dept("D2")
_give_dept_role_with_perms(u, d1, perm_codenames=["publish_department_post"])
client = auth(u)
r = client.post(
urls["posts_list"],
_post_data(type_=TYPE_DEPARTMENT, dept=d2),
format="json",
)
assert r.status_code == 403
def test_company_create_forbidden_for_regular_user(make_user, auth, urls):
u = make_user("regular@example.com")
client = auth(u)
resp = client.post(
urls["posts_list"],
{"type": "company", "title": "X", "body": "B"},
format="json",
)
assert resp.status_code == status.HTTP_400_BAD_REQUEST
@pytest.mark.django_db
def test_update_and_delete_permissions_match_create_rules(api_client):
dept = Department.objects.create(name="QA")
head = _user()
dept.head = head
dept.save(update_fields=["head"])
# создаём пост отдела от лица главы
api_client.force_authenticate(user=head)
url = reverse("api:v1:posts-list")
resp = api_client.post(
url,
{"type": TYPE_DEPARTMENT, "title": "t1", "body": "b1", "department": dept.id},
format="multipart",
)
assert resp.status_code == status.HTTP_201_CREATED
post_id = resp.data["id"]
url_detail = reverse("api:v1:posts-detail", args=[post_id])
# другой обычный пользователь не может править/удалять
stranger = _user()
api_client.force_authenticate(user=stranger)
assert api_client.patch(url_detail, {"title": "nope"}, format="json").status_code == status.HTTP_403_FORBIDDEN
assert api_client.delete(url_detail).status_code == status.HTTP_403_FORBIDDEN
# выдаём этому пользователю роль с правом публикации в отделе -> теперь может править
_grant_dept_role_perm(stranger, dept, "feed.publish_department_post")
assert api_client.patch(url_detail, {"title": "ok"}, format="json").status_code == status.HTTP_200_OK
# и удалять
assert api_client.delete(url_detail).status_code == status.HTTP_204_NO_CONTENT
@pytest.mark.django_db
def test_list_by_author_shows_all_authors_posts(api_client):
# автор со всеми нужными правами, чтобы создавать через API
author = _user()
_grant_user_perm(author, "feed.publish_company_post")
dept = Department.objects.create(name="Support")
_grant_dept_role_perm(author, dept, "feed.publish_department_post")
api_client.force_authenticate(user=author)
url = reverse("api:v1:posts-list")
# company пост
r1 = api_client.post(
url, {"type": TYPE_COMPANY, "title": "c", "body": "b"}, format="multipart"
)
assert r1.status_code == status.HTTP_201_CREATED
# department пост
r2 = api_client.post(
url,
{"type": TYPE_DEPARTMENT, "title": "d", "body": "b", "department": dept.id},
format="multipart",
)
assert r2.status_code == status.HTTP_201_CREATED
# выборка по author
lurl = f"{url}?author={author.id}"
resp = api_client.get(lurl)
assert resp.status_code == status.HTTP_200_OK
items = _items(resp)
ids = {row["id"] for row in items}
assert ids == {r1.data["id"], r2.data["id"]}
# убедимся, что тип employee не проходит (на всякий)
assert all(row["type"] in (TYPE_COMPANY, TYPE_DEPARTMENT) for row in items)
assert resp.status_code == 403

View File

@ -19,6 +19,7 @@ User = get_user_model()
# Speed & isolation
# -------------------------
@pytest.fixture(autouse=True)
def _fast_hashers_and_email(settings, tmp_path):
# быстрые хеши паролей и локальная почта
@ -35,19 +36,23 @@ def _fast_hashers_and_email(settings, tmp_path):
# API clients
# -------------------------
@pytest.fixture
def api_client() -> APIClient:
"""DRF APIClient (имеет force_authenticate)."""
return APIClient()
@pytest.fixture
def auth_client_factory():
"""Фабрика аутентифицированных клиентов: client = auth_client_factory(user)."""
def _make(user=None):
client = APIClient()
if user:
client.force_authenticate(user=user)
return client
return _make
@ -57,10 +62,12 @@ def auth_client_factory():
_phone_seq = itertools.count(1000)
def _unique_phone() -> str:
# валидный E.164 и уникальный в рамках сессии
return f"+7999000{next(_phone_seq):03d}"
@pytest.fixture
def user_factory():
"""
@ -68,14 +75,17 @@ def user_factory():
verified=True, active=True, **extra)
Создаёт пользователя напрямую (без менеджера create_user, чтобы не слать письма).
"""
def _make(email: str,
*,
staff: bool = False,
superuser: bool = False,
verified: bool = True,
active: bool = True,
password: str = "pass",
**extra):
def _make(
email: str,
*,
staff: bool = False,
superuser: bool = False,
verified: bool = True,
active: bool = True,
password: str = "pass",
**extra,
):
u = User.objects.create(
email=email,
phone_number=extra.pop("phone_number", _unique_phone()),
@ -90,20 +100,25 @@ def user_factory():
u.set_password(password)
u.save()
return u
return _make
@pytest.fixture
def department_factory():
def _make(**kwargs) -> Department:
return Department.objects.create(**kwargs)
return _make
@pytest.fixture
def perm_for_department():
"""
p = perm_for_department("manage_department")
Гарантированно возвращает Permission для модели Department с нужным codename.
"""
def _make(codename: str) -> Permission:
ct = ContentType.objects.get_for_model(Department)
p, _ = Permission.objects.get_or_create(
@ -112,27 +127,35 @@ def perm_for_department():
defaults={"name": codename},
)
return p
return _make
@pytest.fixture
def role_factory(perm_for_department):
"""
role = role_factory(dept, name="mgr", codes=["manage_department", ...])
"""
def _make(dept: Department, name: str = "role", codes=None) -> DepartmentRole:
r = DepartmentRole.objects.create(department=dept, name=name)
if codes:
perms = [perm_for_department(code) for code in codes]
r.permissions.add(*perms)
return r
return _make
@pytest.fixture
def link_factory():
"""
link = link_factory(employee, department, is_active=True, role=None, **dates)
"""
def _make(employee, department, is_active=True, role=None, **dates) -> EmployeeDepartment:
def _make(
employee, department, is_active=True, role=None, **dates
) -> EmployeeDepartment:
return EmployeeDepartment.objects.create(
employee=employee,
department=department,
@ -140,6 +163,7 @@ def link_factory():
role=role,
**dates,
)
return _make
@ -147,15 +171,18 @@ def link_factory():
# Test helpers
# -------------------------
@pytest.fixture
def extract_results():
"""
items = extract_results(resp.json()) — поддерживает DRF пагинацию и «плоский» список.
"""
def _do(payload):
if isinstance(payload, dict) and "results" in payload:
return payload["results"]
return payload
return _do
@ -163,6 +190,7 @@ def extract_results():
# Small misc utilities
# -------------------------
@pytest.fixture
def data_uri_image(tmp_path):
"""
@ -170,6 +198,7 @@ def data_uri_image(tmp_path):
"""
# минимальный валидный PNG 1x1 (чёрный пиксель)
import base64
png_b64 = (
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4nGMA"
"AQAABQABDQottAAAAABJRU5ErkJggg=="
@ -183,6 +212,8 @@ def content_file_from_b64():
ContentFile <- base64 строки (без data: префикса).
"""
import base64
def _make(b64: str, name: str = "upload.bin") -> ContentFile:
return ContentFile(base64.b64decode(b64), name=name)
return _make