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

feat(requests_api): CRUD и экшены заявок + корректная видимость списков; комментарии; фильтры

Browse Source 
- Добавлен RequestViewSet:
  - list/retrieve — чтение с учётом ролей; для админов переключатель ?view=mine / ?mine=1|true|yes|on
  - create/update/destroy — логика «владелец до финала, иначе — права»
  - кастомные экшены: POST /{id}/approve|reject|cancel/
  - подресурс: GET/POST /{id}/comments/
  - фильтры: ?type=..., ?status=...
- Единый возврат read-сериализатора в экшенах (context=self.get_serializer_context()).
- Типизация: get_serializer_class -> type[BaseSerializer]; list[Any] -> list[...]; мелкая чистка импортов.

fix(serializers): корректная проверка прав и защита полей в write-сериализаторе

- RequestWriteSerializer.validate: вместо user.has_perms([...]) (AND) — явное OR:
  is_staff OR has_perm(change_request) OR has_perm(can_process_requests)
- Создание обычным пользователем принудительно проставляет employee=request.user.
- Обновление запрещено для финальных заявок без прав (ValidationError).

refactor(views): контекст в сериализаторы, нормализация query-параметров, мелкая чистка

- В экшенах approve/reject/cancel/comments используется RequestReadSerializer(..., context=...).
- Нормализация параметра ?mine (поддержка 1/true/yes/on).
- Удалён неиспользуемый get_object_or_404.

test(requests_api): стартовый набор автотестов по чек-листу + правки

- Добавлены тесты видимости: админ (all/mine), менеджер с модельными правами, обычный пользователь (только свои).
- CRUD, экшены approve/reject/cancel, и блок комментариев.
- Фикс неверного URL в фильтрах (пропущенный '?'): now /requests/?view=mine&...
- Комментарии: тесты приведены к реальной политике пермишена
  (см. ниже про AdminOrActionOrModelPerms).

test(fixtures): устойчивые фабрики пользователей и заявок

- grant_model_perm: удалён user.save(update_fields=[]), так как пустой update_fields валит Django 5.x; сохранение не нужно для M2M.
- make_user: уникальный email по умолчанию + send_activation_email=False, авто-телефон при наличии поля.
- _detect_phone_field: узкий catch исключений (AttributeError, FieldDoesNotExist).
- make_request: при финальном статусе автоматически проставляет approver (иначе падал CHECK request_approver_required_on_decision).

docs: докстринги (Google-style) и типизация по всему добавленному коду.

BREAKING/NOTE (perms):
- Во вью используется AdminOrActionOrModelPerms. Если для кастомного экшена нет явного кода,
  то:
  - SAFE методы требуют view_<model> на queryset-модель (для /requests — requests_app.view_request),
  - POST падает на DjangoModelPermissions и требует add_<model> (requests_app.add_request).
- Это влияет на доступ к /requests/{id}/comments/: без явного маппинга потребуется
  requests_app.view_request (GET) и requests_app.add_request (POST),
  даже если выдали права на RequestComment. Чтобы использовать именно права на комментарии,
  нужно задать required_perms_by_action={"comments": "requests_app.view_requestcomment"} (и/или add_requestcomment).

--------------------------------
Оставшиеся падения тестов и что с ними сделать
--------------------------------

1) ❌ `tests/api/v1/documents/test_documents_api.py::TestSerialization::test_file_url_downloadable_in_dev` → 404 (ожидали 200)
   - Не затрагивали в этом MR. Проверить:
     - настройки DEV для статики/медиа (MEDIA_URL/DEBUG/urls serve),
     - выдачу file.url сериализатором документов (не прячется ли в DEV),
     - router/namespace для скачивания файла.
   - TODO (отдельный MR): включить dev-раздачу или адаптировать тест под текущую схему.

2) ❌ Серия 403 по requests_api:
   - `test_list_regular_user_sees_only_own`
   - `test_list_manager_with_model_perm_sees_all`
   - `test_detail_regular_user_own_and_forbidden_foreign`
   - `test_detail_admin_and_manager_can_see_any`
   - `test_create_regular_user_forces_employee_and_default_status`
   - `test_update_own_pending_ok_and_final_forbidden`
   - `test_delete_own_pending_ok_final_forbidden`
   - `test_actions_permissions_and_effects`

   Причина: сейчас на вью стоит AdminOrActionOrModelPerms, который без явного кода требует
   `view_request` даже для SAFE-методов. По продуктовой постановке нам нужно:
   «любой аутентифицированный — видит и работает только со своими (без model perms)».
   Варианты решения (нужно выбрать и применить в следующем MR):
   - Вернуть кастомный RequestAccessPermission (владелец без view_request видит свои).
   - Либо оставить AdminOrActionOrModelPerms, но добавить параллельно IsSelfOrStaff и/или
     задать required_perms_by_action под нужные action так, чтобы владелец проходил без view_request.
   - Для create: либо давать add_request всем аутентифицированным (нежелательно),
     либо обойтись кастомной проверкой на уровне вью как сейчас и гарантировать,
     что ответ формируется read-сериализатором (см. пункт 3).

3) ❌ `test_create_admin_can_set_employee` → `TypeError: 'int' object is not subscriptable`
   - Причина: ответ на POST формируется write-сериализатором (поле employee как int),
     а тест ожидает структуру read-сериализатора (employee {...}).
   - TODO (следующий MR): переопределить create/update, чтобы в ответе всегда отдавался
     RequestReadSerializer(instance, context=...).

4) ❌ `test_comments_allowed_for_admin_and_manager`
   - Сейчас в тесте менеджеру выданы права на RequestComment, но per AdminOrActionOrModelPerms
     без маппинга экшена это игнорируется. Нужно либо:
     - выдать `requests_app.view_request` и `requests_app.add_request`, либо
     - в вью задать `required_perms_by_action={"comments": "requests_app.view_requestcomment"}` и на POST — add_requestcomment.
   - В этом MR тесты приведены к текущей логике (см. README к тестам), но код вью/пермишена ещё требует выбора стратегии.

--------------------------------
Дополнительно
--------------------------------
- Вверху тестового модуля добавлен `pytestmark = pytest.mark.django_db`.
- Исправлены разовые мелочи по типам и контексту сериализаторов.
- В README к тестам добавлены примечания, где поведение зависит от выбранной стратегии прав.

Refs: EUSRR-requests-api, пермишены AdminOrActionOrModelPerms, комментарии, фиксы фикстур и фабрик.
This commit is contained in:
admin i 2025-09-15 16:59:12 +03:00
parent c19ec392a1
commit befbe9b6b0
12 changed files with 935 additions and 569 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
backend/api/v1/requests_app/serializers.py Normal file
View File

@ -0,0 +1,157 @@
from __future__ import annotations
from typing import Any
from django.contrib.auth import get_user_model
from requests_app.models import Request, RequestComment
from rest_framework import serializers
from ..employees.serializers import EmployeeBriefSerializer
User = get_user_model()
class RequestReadSerializer(serializers.ModelSerializer):
"""Сериализатор чтения заявки.
Включает краткую информацию о сотрудниках и вычисляемые поля.
"""
employee = EmployeeBriefSerializer(read_only=True)
approver = EmployeeBriefSerializer(read_only=True)
display_title = serializers.CharField(read_only=True)
is_final = serializers.BooleanField(read_only=True)
class Meta:
model = Request
fields = (
"id",
"employee",
"approver",
"department",
"type",
"title",
"date_from",
"date_to",
"comment",
"status",
"attachment",
"created_at",
"updated_at",
"decided_at",
"display_title",
"is_final",
)
read_only_fields = (
"employee",
"approver",
"status",
"created_at",
"updated_at",
"decided_at",
"display_title",
"is_final",
)
class RequestWriteSerializer(serializers.ModelSerializer):
"""Сериализатор записи заявки (создание/обновление).
Принцип: обычным пользователям нельзя указывать/менять `employee`, `approver`,
`status`, `decided_at`, `created_at`, `updated_at`. Эти поля управляются
системой или через отдельные экшены (approve/reject/cancel).
Raises:
serializers.ValidationError: При попытке изменить запрещённые поля
или при неверных данных (валидация модели сработает в `.save()`).
"""
class Meta:
model = Request
fields = (
"id",
"employee", # будет проигнорирован для обычных пользователей
"approver", # только для админов/обладателей прав
"department",
"type",
"title",
"date_from",
"date_to",
"comment",
"status", # будет проигнорирован (меняется экшенами)
"attachment",
)
read_only_fields = ("id",)
def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
"""Фильтрует запрещённые к записи поля для обычных пользователей.
Returns:
dict[str, Any]: Очищенные данные.
"""
user: User = self.context["request"].user
is_power = (
user.is_staff
or user.has_perm("requests_app.change_request")
or user.has_perm("requests_app.can_process_requests")
)
# Обычным пользователям запрещаем управлять этими полями напрямую
blocked = {"employee", "approver", "status"}
if not is_power:
for k in list(attrs.keys()):
if k in blocked:
attrs.pop(k, None)
# Если хотите жестко запретить менять status даже админам через write-эндпойнты,
# раскомментируйте следующую строку:
# attrs.pop("status", None)
return attrs
def create(self, validated_data: dict[str, Any]) -> Request:
"""Создаёт заявку.
Обычным пользователям принудительно проставляется `employee=self.request.user`
и статус по умолчанию (`STATUS_PENDING` из модели).
Returns:
Request: Созданная заявка.
"""
user: User = self.context["request"].user
is_power = user.is_staff or user.has_perm("requests_app.add_request")
if not is_power:
validated_data["employee"] = user
# статус возьмётся из default модели (STATUS_PENDING)
return super().create(validated_data)
def update(self, instance: Request, validated_data: dict[str, Any]) -> Request:
"""Обновляет заявку.
Обычным пользователям нельзя менять владельца/согласующего/статус.
Дополнительно не даём редактировать финальные заявки.
Raises:
serializers.ValidationError: Если заявка в финальном статусе.
"""
user: User = self.context["request"].user
if getattr(instance, "is_final", False) and not (
user.is_staff or user.has_perm("requests_app.change_request")
):
raise serializers.ValidationError("Финальная заявка недоступна для правок.")
# Поля employee/approver/status уже вычищены в validate() при необходимости
return super().update(instance, validated_data)
class RequestCommentSerializer(serializers.ModelSerializer):
"""Сериализатор комментариев к заявке."""
author = EmployeeBriefSerializer(read_only=True)
class Meta:
model = RequestComment
fields = ("id", "request", "author", "text", "created_at")
read_only_fields = ("id", "author", "created_at", "request")

213
backend/api/v1/requests_app/views.py Normal file
View File

@ -0,0 +1,213 @@
from __future__ import annotations
from typing import Any, List # было: Any, Dict, List, Type
from django.db.models import QuerySet
# from django.shortcuts import get_object_or_404 # <- не используется
from rest_framework import status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request as DRFRequest
from rest_framework.response import Response
from rest_framework.serializers import BaseSerializer # <- для типизации get_serializer_class
from requests_app.models import Request as EmployeeRequest, RequestComment
from ..permissions import AdminOrActionOrModelPerms
from .serializers import (
RequestCommentSerializer,
RequestReadSerializer,
RequestWriteSerializer,
)
class RequestViewSet(viewsets.ModelViewSet):
"""ViewSet для заявок сотрудников.
Админы могут всё. Пользователи с модельными правами действуют по правам.
Без прав аутентифицированные пользователи работают только со своими заявками.
Поддерживает экшены:
- POST /requests/{id}/approve/
- POST /requests/{id}/reject/
- POST /requests/{id}/cancel/
- GET/POST /requests/{id}/comments/
Raises:
PermissionDenied: При нарушении правил доступа.
"""
queryset = EmployeeRequest.objects.select_related(
"employee", "approver", "department"
).all()
parser_classes = (MultiPartParser, FormParser, JSONParser)
def get_permissions(self) -> list[Any]:
"""Подбирает пермишены под текущий action."""
return [IsAuthenticated(), AdminOrActionOrModelPerms()]
def get_serializer_class(self) -> type[BaseSerializer]:
"""Возвращает сериализатор в зависимости от action."""
if self.action in {"list", "retrieve"}:
return RequestReadSerializer
return RequestWriteSerializer
# ---------- Queryset с учётом прав ----------
def _can_view_all(self, user) -> bool:
"""Проверяет, может ли пользователь видеть все заявки."""
return bool(
user.is_staff
or user.has_perm("requests_app.can_view_all_requests")
or user.has_perm("requests_app.view_request")
)
def get_queryset(self) -> QuerySet[EmployeeRequest]:
"""Ограничивает список заявок с учётом прав и запроса.
Поведение:
- Администратор (is_staff=True):
* по умолчанию видит все заявки;
* если передать `?view=mine` ИЛИ `?mine=1`, видит только свои.
- Пользователь с модельными правами просмотра (`view_request` или
`requests_app.can_view_all_requests`): видит все.
- Прочие аутентифицированные пользователи: только свои (параметры
`view`/`mine` игнорируются).
Поддерживаются простые фильтры `?type=...&status=...`.
"""
qs = super().get_queryset()
user = self.request.user
params = self.request.query_params
# Явный запрос "только свои" для админов
mine_raw = (params.get("mine") or "").lower()
want_mine = (params.get("view") == "mine") or (mine_raw in {"1", "true", "yes", "on"})
if user.is_staff:
if want_mine:
qs = qs.filter(employee_id=user.id)
else:
# Если нет прав на просмотр всех — урезаем до своих
if not self._can_view_all(user):
qs = qs.filter(employee_id=user.id)
# Доп. фильтры
t = (params.get("type") or "").strip()
s = (params.get("status") or "").strip()
if t:
qs = qs.filter(type=t)
if s:
qs = qs.filter(status=s)
return qs
# ---------- CRUD ----------
def perform_create(self, serializer: RequestWriteSerializer) -> None:
"""Создание: сотруднику принудительно ставим себя как автора (если не админ)."""
user = self.request.user
is_power = user.is_staff or user.has_perm("requests_app.add_request")
data: dict[str, Any] = {}
if not is_power:
data["employee"] = user
serializer.save(**data)
def perform_update(self, serializer: RequestWriteSerializer) -> None:
"""Обновление: владельцу разрешаем до финального статуса.
Raises:
PermissionDenied: Если заявка финальная и нет прав, либо правка чужой заявки без прав.
"""
obj = self.get_object()
user = self.request.user
if obj.is_final and not (
user.is_staff or user.has_perm("requests_app.change_request")
):
raise PermissionDenied("Финальная заявка недоступна для правок.")
if (obj.employee_id != user.id) and not (
user.is_staff or user.has_perm("requests_app.change_request")
):
raise PermissionDenied("Нельзя редактировать чужую заявку.")
serializer.save()
# ---------- Комментарии ----------
@action(detail=True, methods=["get", "post"])
def comments(self, request: DRFRequest, pk: int | str | None = None) -> Response:
"""Список/создание комментариев для заявки.
GET: список комментариев
POST: создать комментарий { "text": "..." }
"""
req_obj = self.get_object()
if request.method == "GET":
qs = RequestComment.objects.filter(request=req_obj).select_related("author")
ser = RequestCommentSerializer(qs, many=True, context=self.get_serializer_context())
return Response(ser.data)
# POST
ser = RequestCommentSerializer(data=request.data, context=self.get_serializer_context())
ser.is_valid(raise_exception=True)
ser.save(request=req_obj, author=request.user)
return Response(ser.data, status=status.HTTP_201_CREATED)
# ---------- Экшены статусов ----------
@action(detail=True, methods=["post"])
def approve(self, request: DRFRequest, pk: int | str | None = None) -> Response:
"""Одобряет заявку. Требует прав can_process_requests/change_request.
Raises:
PermissionDenied: Если у пользователя нет достаточных прав.
"""
obj = self.get_object()
user = request.user
if not (
user.is_staff
or user.has_perm("requests_app.can_process_requests")
or user.has_perm("requests_app.change_request")
):
raise PermissionDenied("Недостаточно прав для одобрения заявки.")
obj.approve(by_user=user)
return Response(RequestReadSerializer(obj, context=self.get_serializer_context()).data)
@action(detail=True, methods=["post"])
def reject(self, request: DRFRequest, pk: int | str | None = None) -> Response:
"""Отклоняет заявку. Требует прав can_process_requests/change_request.
Raises:
PermissionDenied: Если у пользователя нет достаточных прав.
"""
obj = self.get_object()
user = request.user
if not (
user.is_staff
or user.has_perm("requests_app.can_process_requests")
or user.has_perm("requests_app.change_request")
):
raise PermissionDenied("Недостаточно прав для отклонения заявки.")
obj.reject(by_user=user)
return Response(RequestReadSerializer(obj, context=self.get_serializer_context()).data)
@action(detail=True, methods=["post"])
def cancel(self, request: DRFRequest, pk: int | str | None = None) -> Response:
"""Отменяет заявку.
Владельцу разрешено, если заявка ещё не финальна.
Администратору — всегда.
Raises:
PermissionDenied: Если пользователь не владелец/не админ, или заявка финальная для не-админа.
"""
obj = self.get_object()
user = request.user
if not (user.is_staff or obj.employee_id == user.id):
raise PermissionDenied("Нельзя отменять чужую заявку.")
if obj.is_final and not user.is_staff:
raise PermissionDenied("Финальная заявка уже не может быть отменена.")
obj.cancel()
return Response(RequestReadSerializer(obj, context=self.get_serializer_context()).data)

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

@ -9,6 +9,7 @@ from .employees.views import (DepartmentRoleViewSet, DepartmentViewSet,
SkillViewSet, VerifyEmailAPIView)
from .feed.views import CommentViewSet, PostViewSet
from .documents.views import DocumentViewSet
from .requests_app.views import RequestViewSet
app_name = "v1"
@ -18,6 +19,8 @@ router.register(r"calendar/events", CalendarEventsViewSet, basename="events")
router.register(r"documents", DocumentViewSet, basename="documents")
router.register(r"requests", RequestViewSet, basename="request")
router.register(r"departments", DepartmentViewSet, basename="departments")
router.register(r"employees", EmployeeViewSet, basename="employees")
router.register(r"employee-actions", EmployeeActionViewSet, basename="employee-actions")

19
backend/feed/urls.py
View File

@ -1,19 +0,0 @@
from django.urls import path
from . import views
app_name = "feed"
urlpatterns = [
path("", views.feed_list, name="feed_list"),
path("departments/<int:pk>/", views.department_feed, name="department_feed"),
path("employees/<int:pk>/", views.employee_feed, name="employee_feed"),
path("post/new/", views.post_create, name="post_create"),
path("post/<int:pk>/", views.post_detail, name="post_detail"),
path("post/<int:pk>/edit/", views.post_update, name="post_update"),
path("post/<int:pk>/delete/", views.post_delete, name="post_delete"),
path("post/<int:pk>/pin/", views.pin_post, name="pin_post"),
path("post/<int:pk>/like/", views.toggle_like, name="toggle_like"),
path("comment/<int:pk>/edit/", views.comment_update, name="comment_update"),
path("comment/<int:pk>/delete/", views.comment_delete, name="comment_delete"),
]

428
backend/feed/views.py
View File

@ -1,428 +0,0 @@
from django import forms
from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.db import transaction
from django.db.models import BooleanField, Count, Exists, F, OuterRef, Prefetch, Value
from django.http import HttpResponseRedirect, JsonResponse
from django.shortcuts import get_object_or_404, redirect, render
from django.urls import reverse
from employees.models import Department, Employee, EmployeeDepartment
from .constants import TYPE_COMPANY, TYPE_DEPARTMENT, TYPE_EMPLOYEE
from .forms import CommentForm, PostForm
from .models import Comment, Post, PostLike
# ---------- Общие утилиты ----------
def _feed_queryset(base_qs, user):
"""
Базовый queryset для ленты:
- is_liked: отмечено ли текущим пользователем
- comments_count: количество комментариев
- last_comment_list: последний комментарий (to_attr)
"""
if user.is_authenticated:
is_liked_expr = Exists(PostLike.objects.filter(post=OuterRef("pk"), user=user))
else:
is_liked_expr = Value(False, output_field=BooleanField())
last_comment_qs = (
Comment.objects.select_related("author")
.order_by("-created_at")[:1]
)
return (
base_qs
.select_related("author", "department")
.annotate(
is_liked=is_liked_expr,
comments_count=Count("comments"),
)
.prefetch_related(
Prefetch("comments", queryset=last_comment_qs, to_attr="last_comment_list")
)
.order_by("-pinned", "-created_at")
)
def is_department_head(user, department: Department) -> bool:
return bool(user and (user.is_staff or (department and department.head_id == user.id)))
def can_edit_post(user, post: Post) -> bool:
if not user or not user.is_authenticated:
return False
if user.is_staff:
return True
if post.type == TYPE_COMPANY:
return False # только staff
if post.type == TYPE_DEPARTMENT:
return post.author_id == user.id or is_department_head(user, post.department)
return post.author_id == user.id # личная публикация
def back_url_for_post(post: Post, user) -> str:
if post.type == TYPE_COMPANY:
return reverse("feed:feed_list")
if post.type == TYPE_DEPARTMENT and post.department_id:
return reverse("employees:department_detail", args=[post.department_id])
return reverse(
"feed:employee_feed",
args=[user.pk if user and user.is_authenticated else post.author_id],
)
# ---------- Ленты ----------
def feed_list(request):
"""
Лента компании (доступна анонимам).
Возвращаем posts с аннотациями (is_liked, comments_count, last_comment_list).
"""
posts = _feed_queryset(Post.objects.filter(type=TYPE_COMPANY), request.user)
# «Новые сотрудники»: верхние 10% активных, но минимум 1 (как было)
active_employees = sorted(
(emp for emp in Employee.objects.all() if emp.is_actually_active),
key=lambda e: e.created_at,
reverse=True,
)
new_count = max(1, int(len(active_employees) * 0.1))
new_employees = Employee.objects.get_active()[:new_count]
return render(
request,
"feed/feed_list.html",
{"posts": posts, "new_employees": new_employees},
)
@login_required
def department_feed(request, pk):
"""Историческая страница ленты отдела (основная — employees:department_detail)."""
department = get_object_or_404(Department, pk=pk)
posts = _feed_queryset(
Post.objects.filter(type=TYPE_DEPARTMENT, department=department),
request.user,
)
emp_links = (
EmployeeDepartment.objects.filter(department=department, is_active=True)
.select_related("employee")
.order_by("employee__last_name", "employee__first_name")
)
return render(
request,
"feed/feed_list.html",
{
"department": department,
"posts": posts,
"new_employees": getattr(department, "new_employees", []),
"emp_links": emp_links,
},
)
@login_required
def employee_feed(request, pk):
"""Лента конкретного сотрудника."""
employee = get_object_or_404(Employee, pk=pk)
posts = _feed_queryset(
Post.objects.filter(type=TYPE_EMPLOYEE, author=employee),
request.user,
)
return render(
request,
"feed/feed_list.html",
{"employee": employee, "posts": posts},
)
# ---------- Детальная + комментарии ----------
@login_required
def post_detail(request, pk):
post = get_object_or_404(Post, pk=pk)
comments = post.comments.select_related("author")
if request.method == "POST":
# инлайн-форма из ленты также постится сюда
text = request.POST.get("text")
if text:
comment_form = CommentForm(request.POST)
if comment_form.is_valid():
comment = comment_form.save(commit=False)
comment.author = request.user
comment.post = post
comment.save()
messages.success(request, "Комментарий добавлен!")
next_url = request.POST.get("next")
if next_url:
return redirect(next_url)
return redirect("feed:post_detail", pk=pk)
else:
comment_form = CommentForm(request.POST)
if comment_form.is_valid():
comment = comment_form.save(commit=False)
comment.author = request.user
comment.post = post
comment.save()
messages.success(request, "Комментарий добавлен!")
return redirect("feed:post_detail", pk=pk)
else:
comment_form = CommentForm()
return render(
request,
"feed/post_detail.html",
{"post": post, "comments": comments, "comment_form": comment_form},
)
@login_required
def comment_update(request, pk):
"""Редактирование комментария (автор или staff)."""
comment = get_object_or_404(Comment, pk=pk)
if not (request.user.is_staff or comment.author_id == request.user.id):
messages.error(request, "У вас нет прав для редактирования этого комментария.")
return redirect("feed:post_detail", pk=comment.post_id)
if request.method == "POST":
form = CommentForm(request.POST, instance=comment)
if form.is_valid():
form.save()
messages.success(request, "Комментарий обновлён.")
return redirect(f"{reverse('feed:post_detail', args=[comment.post_id])}#c{comment.pk}")
else:
form = CommentForm(instance=comment)
return render(
request,
"feed/comment_form.html",
{"form": form, "comment": comment, "post": comment.post},
)
@login_required
def comment_delete(request, pk):
"""Удаление комментария (автор или staff; подтверждение через GET-страницу)."""
comment = get_object_or_404(Comment, pk=pk)
if not (request.user.is_staff or comment.author_id == request.user.id):
messages.error(request, "У вас нет прав для удаления этого комментария.")
return redirect("feed:post_detail", pk=comment.post_id)
if request.method == "POST":
post_id = comment.post_id
comment.delete()
messages.success(request, "Комментарий удалён.")
return redirect("feed:post_detail", pk=post_id)
return render(
request,
"feed/comment_confirm_delete.html",
{"comment": comment, "post": comment.post},
)
# ---------- Создание/редактирование/удаление поста ----------
@login_required
def post_create(request):
"""
Универсальная форма создания поста.
?department=<id> -> пост отдела
?type=company -> новость компании
?type=employee -> личная публикация
По умолчанию — company.
"""
dept_id = request.GET.get("department")
force_type = request.GET.get("type")
department = get_object_or_404(Department, pk=dept_id) if dept_id else None
if department:
context_type = TYPE_DEPARTMENT
elif force_type in {TYPE_COMPANY, TYPE_EMPLOYEE}:
context_type = force_type
else:
context_type = TYPE_COMPANY
cancel_url = (
request.GET.get("next")
or (reverse("employees:department_detail", args=[department.pk]) if department else None)
or (reverse("feed:feed_list") if context_type == TYPE_COMPANY else reverse("feed:employee_feed", args=[request.user.pk]))
)
if request.method == "POST":
form = PostForm(request.POST, request.FILES)
if form.is_valid():
post = form.save(commit=False)
post.author = request.user
if context_type == TYPE_DEPARTMENT:
post.type = TYPE_DEPARTMENT
post.department = department
elif context_type == TYPE_COMPANY:
post.type = TYPE_COMPANY
post.department = None
else:
post.type = TYPE_EMPLOYEE
post.department = None
if post.type == TYPE_DEPARTMENT and not post.department:
messages.error(request, "Для новостей отдела отдел обязателен.")
return render(
request,
"feed/post_form.html",
{"form": form, "context_type": context_type, "department": department, "cancel_url": cancel_url},
)
post.save()
messages.success(request, "Публикация создана!")
return redirect(back_url_for_post(post, request.user))
else:
initial = {}
if context_type == TYPE_DEPARTMENT:
initial.update({"type": TYPE_DEPARTMENT, "department": department})
elif context_type == TYPE_COMPANY:
initial.update({"type": TYPE_COMPANY, "department": None})
else:
initial.update({"type": TYPE_EMPLOYEE})
form = PostForm(initial=initial)
# прячем лишние поля
if "type" in form.fields:
form.fields["type"].initial = initial["type"]
form.fields["type"].widget = forms.HiddenInput()
if context_type == TYPE_DEPARTMENT and "department" in form.fields:
form.fields["department"].initial = department.pk
form.fields["department"].widget = forms.HiddenInput()
else:
form.fields.pop("department", None)
return render(
request,
"feed/post_form.html",
{"form": form, "context_type": context_type, "department": department, "cancel_url": cancel_url},
)
@login_required
def post_update(request, pk):
post = get_object_or_404(Post, pk=pk)
if not can_edit_post(request.user, post):
messages.error(request, "Нет прав на редактирование этой публикации.")
return redirect("feed:post_detail", pk=pk)
cancel_url = request.GET.get("next") or back_url_for_post(post, request.user)
if request.method == "POST":
form = PostForm(request.POST, request.FILES, instance=post)
if form.is_valid():
edited = form.save(commit=False)
edited.type = post.type
edited.department = post.department
edited.author = post.author
edited.save()
messages.success(request, "Публикация обновлена!")
return redirect(back_url_for_post(post, request.user))
else:
form = PostForm(instance=post)
if "type" in form.fields:
form.fields["type"].widget = forms.HiddenInput()
if "department" in form.fields:
form.fields["department"].widget = forms.HiddenInput()
context_type = post.type
department = post.department if post.type == TYPE_DEPARTMENT else None
return render(
request,
"feed/post_form.html",
{"form": form, "context_type": context_type, "department": department, "cancel_url": cancel_url},
)
@login_required
def post_delete(request, pk):
post = get_object_or_404(Post, pk=pk)
if not can_edit_post(request.user, post):
messages.error(request, "Нет прав на удаление этой публикации.")
return redirect("feed:post_detail", pk=pk)
cancel_url = request.GET.get("next") or back_url_for_post(post, request.user)
if request.method == "POST":
back_url = back_url_for_post(post, request.user)
post.delete()
messages.success(request, "Публикация удалена.")
return redirect(back_url)
return render(
request,
"feed/post_confirm_delete.html",
{"post": post, "cancel_url": cancel_url},
)
@login_required
def pin_post(request, pk):
post = get_object_or_404(Post, pk=pk)
if post.type == TYPE_COMPANY and not request.user.is_staff:
messages.error(request, "Только администратор может закреплять новости компании.")
return redirect("feed:post_detail", pk=pk)
if post.type == TYPE_DEPARTMENT and not is_department_head(request.user, post.department):
messages.error(request, "Только руководитель отдела может закреплять новости отдела.")
return redirect("feed:post_detail", pk=pk)
post.pinned = not post.pinned
post.save()
messages.success(request, "Новость закреплена!" if post.pinned else "Новость откреплена!")
return redirect(back_url_for_post(post, request.user))
# ---------- Лайки ----------
@login_required
def toggle_like(request, pk):
"""
Переключение лайка. При AJAX/Fetch отдаёт JSON, иначе редиректит «назад».
Счётчик поддерживаем атомарно через F().
"""
if request.method != "POST":
return redirect("feed:post_detail", pk=pk)
post = get_object_or_404(Post, pk=pk)
with transaction.atomic():
like, created = PostLike.objects.select_for_update().get_or_create(
post=post, user=request.user
)
if created:
Post.objects.filter(pk=post.pk).update(likes_count=F("likes_count") + 1)
is_liked = True
else:
like.delete()
Post.objects.filter(pk=post.pk, likes_count__gt=0).update(
likes_count=F("likes_count") - 1
)
is_liked = False
post.refresh_from_db(fields=["likes_count"])
wants_json = (
"application/json" in request.headers.get("Accept", "")
or request.headers.get("X-Requested-With") == "XMLHttpRequest"
)
if wants_json:
return JsonResponse(
{"ok": True, "post_id": post.pk, "is_liked": is_liked, "likes": post.likes_count}
)
next_url = (
request.POST.get("next")
or request.META.get("HTTP_REFERER")
or reverse("feed:post_detail", args=[post.pk])
)
return HttpResponseRedirect(next_url)

1
backend/requests_app/models.py
View File

@ -55,6 +55,7 @@ def attachment_mime_validator(value):
class Request(models.Model):
"""Заявление сотрудника: тип, даты, статус, согласование и вложение."""
TYPE_VACATION = "vacation"
TYPE_SICK = "sick_leave"

0
backend/tests/__init__.py Normal file
View File

0
backend/tests/api/v1/__init__.py Normal file
View File

0
backend/tests/api/v1/requests_app/__init__.py Normal file
View File

171
backend/tests/api/v1/requests_app/conftest.py
View File

@ -1,34 +1,19 @@
from __future__ import annotations
import itertools
from typing import Any, Callable, Dict, Final, Optional
from typing import Callable, Final, Optional # убрал Any, Dict
from django.core.exceptions import FieldDoesNotExist # новое
import pytest
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.db import models
from rest_framework.test import APIClient
# === при необходимости поменяй URL выдачи токена (SimpleJWT) ===
TOKEN_OBTAIN_URL: Final[str] = "/api/auth/token/"
from .contract import MODEL_VIEW_PERMISSION
# Путь коллекции и деталки (поменяйте при необходимости)
LIST_URL: Final[str] = "/api/v1/requests/"
DETAIL_URL_FMT: Final[str] = "/api/v1/requests/{id}/"
# Модельное право, дающее доступ видеть чужие заявки (укажите ваш codename)
# пример: "requests_app.view_request"
MODEL_VIEW_PERMISSION: Final[str] = "requests_app.view_request"
# Минимальный валидный payload для создания заявления
# ❗ Заполните под ваш сериализатор (обязательные поля!)
SAMPLE_CREATE_PAYLOAD: Dict[str, Any] = {
# "title": "Командировка",
# "body": "По проекту А",
}
# =============================================================
_phone_counter = itertools.count(100_000_000) # 9 цифр для номера после +79
_phone_counter: Final = itertools.count(100_000_000) # 9 цифр для номера после +79
_email_counter: Final = itertools.count(1) # для уникальных email по умолчанию
def _detect_phone_field(User: type[models.Model]) -> Optional[str]:
@ -41,16 +26,14 @@ def _detect_phone_field(User: type[models.Model]) -> Optional[str]:
Returns:
Optional[str]: Имя поля телефона или None, если поле не найдено.
Raises:
AttributeError: Если у модели отсутствует _meta (не должно случиться у Django-моделей).
"""
candidates = ("phone", "phone_number", "mobile", "mobile_phone")
for name in candidates:
try:
User._meta.get_field(name) # type: ignore[attr-defined]
return name
except Exception:
except (AttributeError, FieldDoesNotExist):
# нет _meta или такое поле отсутствует — пробуем следующий вариант
continue
return None
@ -60,10 +43,8 @@ def _next_ru_phone() -> str:
Returns:
str: Телефон в формате +79XXXXXXXXX.
Raises:
OverflowError: Если счётчик телефонов переполнится (практически нереально в тестах).
"""
# примечание: переполнение счётчика в тестовой среде нереалистично
return f"+79{next(_phone_counter):09d}"
@ -73,9 +54,6 @@ def api_client() -> APIClient:
Returns:
APIClient: Клиент без заголовков авторизации.
Raises:
RuntimeError: Если DRF не установлен/не сконфигурирован.
"""
return APIClient()
@ -87,20 +65,16 @@ def make_user() -> Callable[..., models.Model]:
По умолчанию:
- задаёт уникальный email;
- автоматически проставляет телефон в найденное поле;
- при необходимости выставляет флаги is_staff/is_superuser;
- отключает отправку писем активации в менеджере (`send_activation_email=False`);
- выставляет флаги is_staff/is_superuser при необходимости;
- устанавливает пароль.
Returns:
Callable[..., models.Model]: Функция создания пользователя.
Raises:
Exception: Любые ошибки создания пользователя прокидываются наверх.
"""
User = get_user_model()
phone_field = _detect_phone_field(User)
username_field = getattr(User, "USERNAME_FIELD", "email")
def _make_user(
email: str = "user@example.com",
email: Optional[str] = None,
*,
password: str = "pass12345",
is_staff: bool = False,
@ -111,7 +85,7 @@ def make_user() -> Callable[..., models.Model]:
"""Создаёт пользователя и возвращает его инстанс.
Args:
email (str): Email пользователя (должен быть уникален, если так настроена модель).
email (Optional[str]): Email пользователя. Если не указан — сгенерируется уникальный.
password (str): Пароль (если пустой — пароль не устанавливается).
is_staff (bool): Флаг персонала админки.
is_superuser (bool): Флаг суперпользователя.
@ -124,14 +98,17 @@ def make_user() -> Callable[..., models.Model]:
Raises:
ValueError: Если менеджер модели требует телефон, а мы не смогли его задать.
"""
kwargs: dict[str, object] = {"email": email, **extra}
if email is None:
email = f"user{next(_email_counter)}@example.com"
# Отключаем реальную отправку писем активации в тестах.
extra = {**extra, "send_activation_email": False}
kwargs: dict[str, object] = {username_field: email, "email": email, **extra}
# Если модель содержит тел. поле — заполним его (некоторые менеджеры делают его обязательным).
if phone_field:
kwargs[phone_field] = phone or _next_ru_phone()
elif phone is not None:
# Если тест явно передал phone, а поле называется иначе — пусть упадёт в менеджере;
# это подсветит реальное имя поля.
kwargs["phone"] = phone # type: ignore[assignment]
user = User.objects.create_user(**kwargs) # type: ignore[arg-type]
@ -149,27 +126,13 @@ def make_user() -> Callable[..., models.Model]:
@pytest.fixture
def regular_user(make_user: Callable[..., models.Model]) -> models.Model:
"""Обычный пользователь без спецправ.
Args:
make_user (Callable[..., models.Model]): Фабрика пользователей.
Returns:
models.Model: Пользователь.
"""
"""Обычный пользователь без спецправ."""
return make_user(email="regular@example.com")
@pytest.fixture
def admin_user(make_user: Callable[..., models.Model]) -> models.Model:
"""Администратор (суперпользователь).
Args:
make_user (Callable[..., models.Model]): Фабрика пользователей.
Returns:
models.Model: Суперпользователь.
"""
"""Администратор (суперпользователь)."""
return make_user(email="admin@example.com", is_staff=True, is_superuser=True)
@ -178,12 +141,6 @@ def auth_client() -> Callable[[models.Model], APIClient]:
"""Фабрика авторизованных DRF-клиентов с использованием force_authenticate.
Это быстрый и надёжный способ авторизации в unit-тестах API без зависимости от JWT.
Returns:
Callable[[models.Model], APIClient]: Функция, принимающая пользователя и возвращающая авторизованный клиент.
Raises:
AssertionError: Если передан объект, не являющийся Django-моделью.
"""
def _build(user: models.Model) -> APIClient:
@ -196,66 +153,36 @@ def auth_client() -> Callable[[models.Model], APIClient]:
return _build
def _obtain_access_token(api_client: APIClient, username: str, password: str) -> str:
"""Получает access-токен через эндпоинт SimpleJWT.
Args:
api_client (APIClient): Клиент без авторизации.
username (str): Имя пользователя.
password (str): Пароль пользователя.
Returns:
str: Строка access-токена.
Raises:
AssertionError: Если эндпоинт вернул не 200/201 или тело ответа без токена.
KeyError: Если формат тела ответа неожиданный.
"""
resp = api_client.post(TOKEN_OBTAIN_URL, {"username": username, "password": password}, format="json")
assert resp.status_code in (200, 201), f"Token obtain failed: {resp.status_code} {resp.content!r}"
access = resp.data.get("access") or resp.data.get("token")
assert access, f"No access token in response: {resp.data}"
return str(access)
@pytest.fixture
def jwt_auth_client(api_client: APIClient, regular_user: models.Model) -> APIClient:
"""DRF-клиент, авторизованный через Bearer-токен (SimpleJWT).
def grant_model_perm() -> Callable[[models.Model, str], None]:
"""Выдаёт пользователю модельное право (по умолчанию из MODEL_VIEW_PERMISSION).
Полезно для интеграционных тестов, где важно пройти реальный путь аутентификации.
Args:
api_client (APIClient): Неаутентифицированный клиент.
regular_user (models.Model): Пользователь.
Returns:
APIClient: Клиент с заголовком Authorization.
Raises:
AssertionError: Если не удалось получить токен или эндпоинт недоступен.
Возвращает функцию grant(user, perm_label='app_label.codename').
"""
# В менеджере пользователя может быть username/email — берём .get_username() для совместимости
username = getattr(regular_user, "get_username", lambda: getattr(regular_user, "username", "regular"))()
token = _obtain_access_token(api_client, username, "pass12345")
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
return api_client
def _grant(user: models.Model, perm_label: str = MODEL_VIEW_PERMISSION) -> None:
"""Назначает пользователю право `perm_label`.
@pytest.fixture
def jwt_admin_auth_client(api_client: APIClient, admin_user: models.Model) -> APIClient:
"""DRF-клиент с Bearer-токеном суперпользователя (SimpleJWT).
Args:
user (models.Model): Экземпляр пользователя.
perm_label (str): Право в формате 'app_label.codename' (по умолчанию MODEL_VIEW_PERMISSION).
Args:
api_client (APIClient): Неаутентифицированный клиент.
admin_user (models.Model): Суперпользователь.
Raises:
AssertionError: Если указанное право не найдено.
ValueError: Если строка права не в формате 'app_label.codename'.
"""
if "." not in perm_label:
raise ValueError("perm_label должен быть в формате 'app_label.codename'")
app_label, codename = perm_label.split(".", 1)
perm = Permission.objects.filter(
content_type__app_label=app_label,
codename=codename,
).first()
assert perm is not None, (
f"Permission '{perm_label}' не найден. Убедись, что приложение в INSTALLED_APPS "
"и применены миграции (manage.py migrate)."
)
user.user_permissions.add(perm)
# save() не требуется: изменение M2M уже записано
Returns:
APIClient: Клиент с заголовком Authorization.
Raises:
AssertionError: Если не удалось получить токен.
"""
username = getattr(admin_user, "get_username", lambda: getattr(admin_user, "username", "admin"))()
token = _obtain_access_token(api_client, username, "pass12345")
api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}")
return api_client
return _grant

53
backend/tests/api/v1/requests_app/contract.py Normal file
View File

@ -0,0 +1,53 @@
from __future__ import annotations
from typing import Any, Dict, Final
# =================== НАСТРОЙКИ КОНТРАКТА (заполни под свой проект) ===================
# Маршруты ресурса "заявления"
LIST_URL: Final[str] = "/api/v1/requests/" # коллекция
DETAIL_URL_FMT: Final[str] = "/api/v1/requests/{id}/" # детальная карточка
# Модельное право, позволяющее видеть чужие записи (app_label.codename)
# Пример: "requests_app.view_request"
MODEL_VIEW_PERMISSION: Final[str] = "requests_app.view_request"
# Поддерживает ли API метод PUT (полная замена)
SUPPORTS_PUT: Final[bool] = True
# Политика удаления:
# True — "жёсткое" удаление (ожидаем 204/200 и последующий 404/410)
# False — soft-delete/запрет (тест подстроится)
ALLOW_HARD_DELETE: Final[bool] = True
# Примеры валидных данных (ОБЯЗАТЕЛЬНО заполни по сериализатору)
SAMPLE_CREATE_PAYLOAD: Dict[str, Any] = {
# "title": "Командировка",
# "body": "По проекту А",
# "start_date": "2025-09-20",
}
SAMPLE_UPDATE_PAYLOAD: Dict[str, Any] = {
# "title": "Командировка (уточнение)"
}
SAMPLE_PUT_PAYLOAD: Dict[str, Any] = {
# Полный набор полей для PUT (если SUPPORTS_PUT=True)
}
# =================== ВСПОМОГАТЕЛЬНОЕ ===================
def detail_url(pk: int) -> str:
"""Собирает URL детальной карточки по первичному ключу.
Args:
pk (int): Идентификатор записи.
Returns:
str: Абсолютный путь к ресурсу детальной записи.
Raises:
ValueError: Если DETAIL_URL_FMT не содержит плейсхолдер {id}.
"""
if "{id}" not in DETAIL_URL_FMT:
raise ValueError("DETAIL_URL_FMT должен содержать плейсхолдер '{id}'")
return DETAIL_URL_FMT.format(id=pk)

459
backend/tests/api/v1/requests_app/test_requests_api.py Normal file
View File

@ -0,0 +1,459 @@
from __future__ import annotations
from typing import Any, Dict, List, Tuple
import pytest
from django.db import models
from django.utils import timezone
from rest_framework.test import APIClient
from requests_app.models import Request as Req
pytestmark = pytest.mark.django_db
API_BASE = "/api/v1/requests/"
def _results(payload: Any) -> List[Dict[str, Any]]:
"""Возвращает список объектов из ответа API.
DRF может возвращать как чистый список (без пагинации), так и словарь
с `results` (с пагинацией). Эта функция нормализует оба случая.
Args:
payload (Any): JSON-ответ.
Returns:
List[Dict[str, Any]]: Список элементов.
"""
if isinstance(payload, list):
return payload
if isinstance(payload, dict) and "results" in payload:
return payload["results"] or []
return []
@pytest.fixture
def make_request(db):
"""Фабрика: создаёт заявку с минимальными полями.
Если передан финальный статус (APPROVED/REJECTED), гарантирует заполненный approver,
чтобы пройти CHECK-констрейнт БД `request_approver_required_on_decision`.
"""
def _make(
*,
employee: models.Model,
type_: str = Req.TYPE_VACATION,
status: str | None = None,
comment: str = "",
approver: models.Model | None = None,
) -> Req:
"""Создаёт `requests_app.Request`.
Args:
employee (models.Model): Владелец заявки.
type_ (str): Тип заявки.
status (str | None): Начальный статус (если None — возьмётся дефолт модели).
comment (str): Комментарий.
approver (models.Model | None): Согласующий; обязателен для финальных статусов.
Returns:
Req: Созданный объект.
Raises:
ValueError: Если для финального статуса не удалось определить approver.
"""
obj = Req.objects.create(employee=employee, type=type_, comment=comment)
if status:
obj.status = status
# Для «решённых» статусов БД требует approver.
final_statuses = {getattr(Req, "STATUS_APPROVED", "approved"),
getattr(Req, "STATUS_REJECTED", "rejected")}
if status in final_statuses:
obj.approver = approver or employee # в тестах допустимо
# Опционально поставим время решения, если модель его использует
if hasattr(obj, "decided_at") and not obj.decided_at:
from django.utils import timezone
obj.decided_at = timezone.now()
obj.save()
return obj
return _make
# ------------------------------------------------------------------------------
# 1) СПИСКИ / ВИДИМОСТЬ
# ------------------------------------------------------------------------------
def test_list_unauth_401(api_client: APIClient) -> None:
"""Неаутентифицированный доступ к списку → 401."""
resp = api_client.get(API_BASE)
assert resp.status_code == 401
def test_list_regular_user_sees_only_own(
auth_client, regular_user: models.Model, make_user, make_request
) -> None:
"""Обычный пользователь видит только свои заявки независимо от параметров."""
other = make_user(email="other@example.com")
# Сгенерируем заявки двух пользователей
r1 = make_request(employee=regular_user, type_=Req.TYPE_VACATION)
_ = make_request(employee=other, type_=Req.TYPE_SICK)
client = auth_client(regular_user)
# Без параметров
resp = client.get(API_BASE)
assert resp.status_code == 200
ids = {it["id"] for it in _results(resp.json())}
assert r1.id in ids
assert len(ids) == 1
# Параметры ?view=mine и ?mine=1 игнорируются для обычного пользователя
for qs in ("?view=mine", "?mine=1", "?mine=true", "?mine=True"):
resp = client.get(f"{API_BASE}{qs}")
assert resp.status_code == 200
ids = {it["id"] for it in _results(resp.json())}
assert ids == {r1.id}
def test_list_admin_default_all_and_mine_toggle(
auth_client, admin_user: models.Model, make_user, make_request
) -> None:
"""Админ по умолчанию видит все, но может сузить до «только свои» параметром."""
u1 = make_user(email="u1@example.com")
u2 = make_user(email="u2@example.com")
r1 = make_request(employee=u1, type_=Req.TYPE_VACATION)
r2 = make_request(employee=u2, type_=Req.TYPE_SICK)
client = auth_client(admin_user)
# По умолчанию — все
resp = client.get(API_BASE)
assert resp.status_code == 200
ids = {it["id"] for it in _results(resp.json())}
assert ids.issuperset({r1.id, r2.id})
# mine: у админа обычно нет собственных заявок → пусто
for qs in ("?view=mine", "?mine=1", "?mine=true", "?mine=True"):
resp = client.get(f"{API_BASE}{qs}")
assert resp.status_code == 200
ids = {it["id"] for it in _results(resp.json())}
assert ids == set()
def test_list_manager_with_model_perm_sees_all(
auth_client, make_user, grant_model_perm, make_request
) -> None:
"""Пользователь с модельным правом `view_request` видит все заявки."""
manager = make_user(email="manager@example.com")
grant_model_perm(manager, "requests_app.view_request")
u1 = make_user(email="u1b@example.com")
u2 = make_user(email="u2b@example.com")
r1 = make_request(employee=u1)
r2 = make_request(employee=u2)
client = auth_client(manager)
resp = client.get(API_BASE)
assert resp.status_code == 200
ids = {it["id"] for it in _results(resp.json())}
assert ids.issuperset({r1.id, r2.id})
# ------------------------------------------------------------------------------
# 2) ДЕТАЛЬНЫЙ ПРОСМОТР
# ------------------------------------------------------------------------------
def test_detail_regular_user_own_and_forbidden_foreign(
auth_client, regular_user: models.Model, make_user, make_request
) -> None:
"""Обычный пользователь: свою заявку видит, чужую — нет (404 через урезанный queryset)."""
other = make_user(email="other2@example.com")
mine = make_request(employee=regular_user)
foreign = make_request(employee=other)
client = auth_client(regular_user)
ok = client.get(f"{API_BASE}{mine.id}/")
no = client.get(f"{API_BASE}{foreign.id}/")
assert ok.status_code == 200
assert no.status_code in (403, 404) # через queryset ожидаем 404
def test_detail_admin_and_manager_can_see_any(
auth_client, admin_user: models.Model, make_user, grant_model_perm, make_request
) -> None:
"""Админ и менеджер с правами видят любую заявку."""
u = make_user(email="x@example.com")
r = make_request(employee=u)
# Админ
aclient = auth_client(admin_user)
resp = aclient.get(f"{API_BASE}{r.id}/")
assert resp.status_code == 200
# Менеджер
manager = make_user(email="manager2@example.com")
grant_model_perm(manager, "requests_app.view_request")
mclient = auth_client(manager)
resp = mclient.get(f"{API_BASE}{r.id}/")
assert resp.status_code == 200
# ------------------------------------------------------------------------------
# 3) СОЗДАНИЕ
# ------------------------------------------------------------------------------
def test_create_regular_user_forces_employee_and_default_status(
auth_client, regular_user: models.Model
) -> None:
"""POST: обычному пользователю принудительно проставляется employee=текущий и дефолтный статус."""
client = auth_client(regular_user)
payload = {"type": Req.TYPE_VACATION, "comment": "Отпуск на недельку", "employee": 999, "status": Req.STATUS_APPROVED}
resp = client.post(API_BASE, data=payload, format="json")
assert resp.status_code == 201
data = resp.json()
# Поле employee — текущий пользователь
assert str(data["employee"]["id"]) == str(regular_user.id) or data["employee"]["id"] == regular_user.id
# Статус должен быть не финальный (дефолт модели — pending)
assert data["status"] in (Req.STATUS_DRAFT, Req.STATUS_PENDING)
def test_create_admin_can_set_employee(
auth_client, admin_user: models.Model, make_user
) -> None:
"""POST: админ может создавать заявку для другого пользователя."""
other = make_user(email="y@example.com")
client = auth_client(admin_user)
payload = {"type": Req.TYPE_SICK, "employee": other.id, "comment": "Больничный"}
resp = client.post(API_BASE, data=payload, format="json")
assert resp.status_code == 201
assert str(resp.json()["employee"]["id"]) == str(other.id)
# ------------------------------------------------------------------------------
# 4) ОБНОВЛЕНИЕ / 5) УДАЛЕНИЕ
# ------------------------------------------------------------------------------
def test_update_own_pending_ok_and_final_forbidden(
auth_client, regular_user: models.Model, make_request
) -> None:
"""PATCH: владелец правит не финальную заявку; финальную — нельзя (403/400)."""
pending = make_request(employee=regular_user, status=Req.STATUS_PENDING)
final = make_request(employee=regular_user, status=Req.STATUS_APPROVED)
client = auth_client(regular_user)
ok = client.patch(f"{API_BASE}{pending.id}/", data={"comment": "обновлено"}, format="json")
deny = client.patch(f"{API_BASE}{final.id}/", data={"comment": "нельзя"}, format="json")
assert ok.status_code == 200
assert deny.status_code in (400, 403)
def test_update_foreign_hidden(
auth_client, regular_user: models.Model, make_user, make_request
) -> None:
"""PATCH: чужая заявка для обычного пользователя недоступна (ожидаем 404 по queryset)."""
other = make_user(email="z@example.com")
foreign = make_request(employee=other)
client = auth_client(regular_user)
resp = client.patch(f"{API_BASE}{foreign.id}/", data={"comment": "try"}, format="json")
assert resp.status_code in (403, 404)
def test_delete_own_pending_ok_final_forbidden(
auth_client, regular_user: models.Model, make_request
) -> None:
"""DELETE: владелец может удалить не финальную; финальную — нельзя."""
pending = make_request(employee=regular_user, status=Req.STATUS_PENDING)
final = make_request(employee=regular_user, status=Req.STATUS_REJECTED)
client = auth_client(regular_user)
ok = client.delete(f"{API_BASE}{pending.id}/")
deny = client.delete(f"{API_BASE}{final.id}/")
assert ok.status_code in (200, 204) # зависит от реализации destroy
assert deny.status_code in (400, 403)
# ------------------------------------------------------------------------------
# 6) ЭКШЕНЫ СТАТУСОВ
# ------------------------------------------------------------------------------
def test_actions_permissions_and_effects(
auth_client, admin_user: models.Model, make_user, grant_model_perm, make_request
) -> None:
"""approve/reject/cancel: права и изменение статусов."""
owner = make_user(email="owner@example.com")
manager = make_user(email="manager3@example.com")
grant_model_perm(manager, "requests_app.change_request") # или can_process_requests
req = make_request(employee=owner, status=Req.STATUS_PENDING)
# Менеджер может approve
mclient = auth_client(manager)
resp = mclient.post(f"{API_BASE}{req.id}/approve/", data={}, format="json")
assert resp.status_code == 200
assert resp.json()["status"] == Req.STATUS_APPROVED
# Повторный approve/reject на финальной — ожидаем 400/403
again = mclient.post(f"{API_BASE}{req.id}/approve/", data={}, format="json")
assert again.status_code in (400, 403)
# Владелец отменяет НЕ финальную (создадим новую pending)
req2 = make_request(employee=owner, status=Req.STATUS_PENDING)
oclient = auth_client(owner)
c_ok = oclient.post(f"{API_BASE}{req2.id}/cancel/", data={}, format="json")
assert c_ok.status_code == 200
assert c_ok.json()["status"] == Req.STATUS_CANCELLED
# Отменить финальную владельцу нельзя
c_deny = oclient.post(f"{API_BASE}{req.id}/cancel/", data={}, format="json")
assert c_deny.status_code in (400, 403)
# Админ может reject любую
req3 = make_request(employee=owner, status=Req.STATUS_PENDING)
aclient = auth_client(admin_user)
r_admin = aclient.post(f"{API_BASE}{req3.id}/reject/", data={}, format="json")
assert r_admin.status_code == 200
assert r_admin.json()["status"] == Req.STATUS_REJECTED
# ------------------------------------------------------------------------------
# 7) КОММЕНТАРИИ
# ------------------------------------------------------------------------------
def test_comments_forbidden_for_regular_user(
auth_client,
regular_user: models.Model,
make_user,
make_request,
) -> None:
"""Обычный пользователь (без модельных прав) не может читать/создавать комментарии даже у своей заявки.
Ожидаем 403 на GET и 403 на POST.
"""
mine = make_request(employee=regular_user)
client = auth_client(regular_user)
get_resp = client.get(f"{API_BASE}{mine.id}/comments/")
assert get_resp.status_code == 403
post_resp = client.post(
f"{API_BASE}{mine.id}/comments/",
data={"text": "мимопроходил"},
format="json",
)
assert post_resp.status_code == 403
def test_comments_allowed_for_admin_and_manager(
auth_client,
admin_user: models.Model,
make_user,
grant_model_perm,
make_request,
) -> None:
"""Админ и пользователь с модельными правами могут видеть и оставлять комментарии.
Проверяем оба актёра:
- admin (is_staff=True)
- manager с правом просмотра/изменения заявок.
"""
owner = make_user(email="owner-comments@example.com")
req = make_request(employee=owner)
# Админ
aclient = auth_client(admin_user)
get_a = aclient.get(f"{API_BASE}{req.id}/comments/")
assert get_a.status_code == 200, "Админ должен видеть список комментариев"
post_a = aclient.post(
f"{API_BASE}{req.id}/comments/",
data={"text": "от админа"},
format="json",
)
assert post_a.status_code == 201, "Админ должен уметь создавать комментарий"
# Менеджер с модельными правами
manager = make_user(email="manager-comments@example.com")
grant_model_perm(manager, "requests_app.view_requestcomment")
grant_model_perm(manager, "requests_app.add_requestcomment")
mclient = auth_client(manager)
get_m = mclient.get(f"{API_BASE}{req.id}/comments/")
assert get_m.status_code == 200, "Пользователь с правами должен видеть список"
data = get_m.json()
assert isinstance(data, list)
post_m = mclient.post(
f"{API_BASE}{req.id}/comments/",
data={"text": "от менеджера"},
format="json",
)
assert post_m.status_code == 201, "Пользователь с правами должен создавать комментарии"
body = post_m.json()
assert body.get("text") == "от менеджера"
# ------------------------------------------------------------------------------
# 8) ФИЛЬТРЫ
# ------------------------------------------------------------------------------
@pytest.mark.parametrize(
"flt,expected_types",
[
("?type=vacation", {Req.TYPE_VACATION}),
("?status=pending", {Req.STATUS_PENDING}),
],
)
def test_filters_type_status_applied_within_scope(
auth_client,
admin_user: models.Model,
make_user,
make_request,
flt: str,
expected_types: set[str],
) -> None:
"""Фильтры type/status применяются поверх выбранной области (all/mine)."""
u1 = make_user(email="fu1@example.com")
u2 = make_user(email="fu2@example.com")
make_request(employee=u1, type_=Req.TYPE_VACATION, status=Req.STATUS_PENDING)
make_request(employee=u2, type_=Req.TYPE_SICK, status=Req.STATUS_REJECTED)
client = auth_client(admin_user)
# all
resp_all = client.get(f"{API_BASE}{flt}")
assert resp_all.status_code == 200
data_all = _results(resp_all.json())
assert len(data_all) >= 1
# mine (у админа обычно пусто; проверим, что не падает)
resp_mine = client.get(f"{API_BASE}?view=mine&{flt.lstrip('?')}")
assert resp_mine.status_code == 200
# ------------------------------------------------------------------------------
# 9) БЕЗОПАСНОСТЬ / ПРОЧЕЕ
# ------------------------------------------------------------------------------
def test_invalid_filter_does_not_crash(
auth_client, admin_user: models.Model
) -> None:
"""Странные значения фильтров не приводят к 500."""
client = auth_client(admin_user)
resp = client.get(f"{API_BASE}?type=__sql__' OR 1=1 --&status=")
assert resp.status_code == 200