Corp/backend/feed/models.py
Игорь Надеин 4631f076a4 **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).
2025-09-11 17:56:33 +03:00

279 lines
9.1 KiB
Python

# backend/feed/models.py
from __future__ import annotations
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.core.validators import FileExtensionValidator
from django.db import models, transaction
from django.db.models import Q
from django.utils.deconstruct import deconstructible
from .constants import TYPE_CHOICES, TYPE_COMPANY, TYPE_DEPARTMENT, TYPE_EMPLOYEE
Employee = get_user_model()
# --- helpers: валидаторы размера файлов ---
def validate_file_size(max_mb: int):
max_bytes = max_mb * 1024 * 1024
def _v(f):
if f and getattr(f, "size", 0) > max_bytes:
raise ValidationError(f"Файл превышает {max_mb} MB.")
return _v
IMAGE_EXTS = ["jpg", "jpeg", "png", "gif", "webp"]
FILE_EXTS = [
"pdf",
"doc",
"docx",
"xls",
"xlsx",
"txt",
"csv",
"png",
"jpg",
"jpeg",
"gif",
"zip",
"rar",
"7z",
]
@deconstructible
class FileSizeValidator:
def __init__(self, max_mb: int):
self.max_mb = int(max_mb)
def __call__(self, f):
if f and getattr(f, "size", 0) > self.max_mb * 1024 * 1024:
raise ValidationError(f"Файл превышает {self.max_mb} MB.")
def __eq__(self, other):
return isinstance(other, FileSizeValidator) and self.max_mb == other.max_mb
class Post(models.Model):
author = models.ForeignKey(
Employee,
on_delete=models.CASCADE,
related_name="posts",
verbose_name="Автор",
)
department = models.ForeignKey(
"employees.Department",
null=True,
blank=True,
on_delete=models.PROTECT,
related_name="posts",
verbose_name="Отдел",
help_text="Для новостей отдела укажите отдел. Для других типов оставьте пустым.",
)
title = models.CharField("Заголовок", max_length=200)
body = models.TextField("Текст")
created_at = models.DateTimeField("Создано", auto_now_add=True)
pinned = models.BooleanField("Закреплено", default=False)
type = models.CharField("Тип публикации", max_length=20, choices=TYPE_CHOICES)
image = models.ImageField(
"Изображение",
upload_to="feed/images/%Y/%m/",
blank=True,
null=True,
)
attachment = models.FileField(
"Вложение",
upload_to="feed/attachments/%Y/%m/",
blank=True,
null=True,
)
likes_count = models.PositiveIntegerField("Лайки", default=0, db_index=True)
class Meta:
verbose_name = "Публикация"
verbose_name_plural = "Публикации"
ordering = ["-pinned", "-created_at"]
indexes = [
models.Index(fields=["type", "created_at"]),
models.Index(fields=["pinned", "created_at"]),
models.Index(fields=["created_at"]),
models.Index(
fields=["department", "-created_at"]
),
models.Index(fields=["author", "-created_at"]),
]
constraints = [
models.CheckConstraint(
name="post_department_required_when_type_is_department",
condition=Q(type=TYPE_DEPARTMENT, department__isnull=False)
| ~Q(type=TYPE_DEPARTMENT),
),
models.CheckConstraint(
name="post_department_null_when_type_not_department",
condition=Q(type=TYPE_DEPARTMENT) | Q(department__isnull=True),
),
models.CheckConstraint(
name="post_likes_count_non_negative",
condition=Q(likes_count__gte=0),
),
]
def __str__(self):
if self.type == TYPE_COMPANY:
prefix = "Компания"
elif self.type == TYPE_DEPARTMENT:
prefix = f"Отдел: {self.department or '—'}"
elif self.type == TYPE_EMPLOYEE:
prefix = f"Сотрудник: {self.author}"
else:
prefix = "Публикация"
return f"[{prefix}] {self.title}"
class Comment(models.Model):
text = models.TextField("Комментарий", blank=True, default="")
image = models.ImageField(
"Изображение",
upload_to="feed/comments/images/%Y/%m/",
blank=True, null=True,
validators=[
FileExtensionValidator(allowed_extensions=IMAGE_EXTS),
FileSizeValidator(10),
],
)
attachment = models.FileField(
"Вложение",
upload_to="feed/comments/attachments/%Y/%m/",
blank=True, null=True,
validators=[
FileExtensionValidator(allowed_extensions=FILE_EXTS),
FileSizeValidator(25),
],
)
created_at = models.DateTimeField("Создано", auto_now_add=True)
author = models.ForeignKey(
"employees.Employee",
on_delete=models.CASCADE,
related_name="comments",
verbose_name="Автор",
db_index=True,
)
post = models.ForeignKey(
"feed.Post",
on_delete=models.CASCADE,
related_name="comments",
verbose_name="Публикация",
)
class Meta:
verbose_name = "Комментарий"
verbose_name_plural = "Комментарии"
ordering = ["created_at"]
indexes = [
models.Index(fields=["created_at"]),
models.Index(fields=["post", "created_at"]),
models.Index(fields=["author", "created_at"]),
]
constraints = [
models.CheckConstraint(
name="comment_has_any_content_v3",
condition=(
(Q(text__isnull=False) & ~Q(text=""))
| Q(image__isnull=False)
| Q(attachment__isnull=False)
),
violation_error_message="Нужно указать текст, изображение или файл.",
),
]
def __str__(self) -> str:
base = (self.text or "").strip()
return f"Комментарий #{self.pk or '∅'} от {self.author_id} к посту {self.post_id}: {base[:40]}"
# --- бизнес-валидация ---
def clean(self):
super().clean()
has_text = bool((self.text or "").strip())
if not (has_text or self.image or self.attachment):
raise ValidationError(
{"text": "Нужно указать текст, изображение или файл."}
)
def _delete_file(self, ffield: models.FileField):
f = getattr(self, ffield.attname)
if f and hasattr(f, "name"):
storage, name = f.storage, f.name
transaction.on_commit(lambda: storage.delete(name))
def delete(self, *args, **kwargs):
old_image = self.image
old_attach = self.attachment
super().delete(*args, **kwargs)
if old_image:
transaction.on_commit(lambda: old_image.storage.delete(old_image.name))
if old_attach:
transaction.on_commit(lambda: old_attach.storage.delete(old_attach.name))
def save(self, *args, **kwargs):
skip_validation = kwargs.pop("skip_validation", False)
if not skip_validation:
self.full_clean()
old_image_name = old_attach_name = None
if self.pk:
try:
old = Comment.objects.only("image", "attachment").get(pk=self.pk)
if old.image and old.image.name != (
self.image.name if self.image else None
):
old_image_name = old.image.name
if old.attachment and old.attachment.name != (
self.attachment.name if self.attachment else None
):
old_attach_name = old.attachment.name
except Comment.DoesNotExist:
pass
res = super().save(*args, **kwargs)
if old_image_name:
transaction.on_commit(lambda: self.image.storage.delete(old_image_name))
if old_attach_name:
transaction.on_commit(
lambda: self.attachment.storage.delete(old_attach_name)
)
return res
class PostLike(models.Model):
post = models.ForeignKey(
Post, on_delete=models.CASCADE, related_name="likes", verbose_name="Пост"
)
user = models.ForeignKey(
Employee,
on_delete=models.CASCADE,
related_name="post_likes",
verbose_name="Пользователь",
)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
verbose_name = "Лайк"
verbose_name_plural = "Лайки"
constraints = [
models.UniqueConstraint(
fields=["post", "user"], name="uq_postlike_post_user"
),
]
indexes = [
models.Index(fields=["user", "created_at"]),
models.Index(fields=["post", "created_at"]),
]
def __str__(self):
return f"❤ {self.user} → {self.post_id}"