From 0f8dcff4ad49658d4b483fa58d64dffa95e477e5 Mon Sep 17 00:00:00 2001 From: lizerk Date: Tue, 10 Mar 2026 13:37:59 +0300 Subject: [PATCH] =?UTF-8?q?=D0=BB=D0=B8=D1=88=D0=BD=D0=B5=D0=B5=20=D0=B8?= =?UTF-8?q?=D0=B7=20=D0=B4=D0=BE=D0=BA=D1=83=D0=BC=D0=B5=D0=BD=D1=82=D0=BE?= =?UTF-8?q?=D0=B2=20=D1=83=D0=B4=D0=B0=D0=BB=D0=B8=D1=82=D1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/documents/notification_signals.py.old | 573 ---------------- backend/documents/tasks.py | 177 ----- backend/documents/test_documents_api.py | 624 ------------------ backend/documents/tests.py | 298 --------- backend/documents/urls.py | 10 - backend/documents/views.py | 435 ------------ backend/eusrr_backend/urls.py | 1 - 7 files changed, 2118 deletions(-) delete mode 100644 backend/documents/notification_signals.py.old delete mode 100644 backend/documents/tasks.py delete mode 100644 backend/documents/test_documents_api.py delete mode 100644 backend/documents/tests.py delete mode 100644 backend/documents/urls.py delete mode 100644 backend/documents/views.py diff --git a/backend/documents/notification_signals.py.old b/backend/documents/notification_signals.py.old deleted file mode 100644 index e168e084..00000000 --- a/backend/documents/notification_signals.py.old +++ /dev/null @@ -1,573 +0,0 @@ -""" -Signals для автоматической генерации уведомлений в модуле Documents. - -Обрабатывает события: -- Новый документ на ознакомление -- Все сотрудники ознакомились -- Напоминание о необходимости ознакомления (через Celery task) -""" - -import logging - -from django.db.models.signals import post_save, m2m_changed -from django.dispatch import receiver -from django.contrib.auth import get_user_model - -from .models import Document, DocumentAcknowledgement, DocumentComment -from notifications.signals import notify - -logger = logging.getLogger(__name__) -Employee = get_user_model() - - -@receiver(post_save, sender=Document) -def create_document_notification(sender, instance, created, **kwargs): - """ - Создает уведомления при создании нового документа асинхронно. - - Уведомления отправляются: - - Всем активным сотрудникам (если sent_to_all=True) - - Сотрудникам выбранных отделов (через m2m_changed signal) - - Выбранным получателям (через m2m_changed signal) - """ - logger.info( - f"[notification_signals] post_save Document id={instance.pk} " - f"created={created} sent_to_all={instance.sent_to_all}" - ) - - if not created: - return - - document = instance - - # Если документ отправляется всем - создаем уведомления напрямую - # channels.py автоматически отправит через Celery - if document.sent_to_all: - logger.info( - f"[notification_signals] Creating notifications for doc={document.pk} (sent_to_all=True)" - ) - notify_all_employees(document) - else: - logger.info( - f"[notification_signals] Skipping notifications (sent_to_all=False), " - f"waiting for m2m_changed" - ) - - -@receiver(m2m_changed, sender=Document.recipients.through) -def notify_specific_recipients(sender, instance, action, pk_set, **kwargs): - """ - Уведомляет конкретных получателей при добавлении в список. - Использует быструю обработку для большого количества получателей. - """ - logger.info( - f"[notification_signals] m2m_changed Document id={instance.pk} " - f"action={action} pk_set={pk_set} sent_to_all={instance.sent_to_all}" - ) - - if action != 'post_add': - logger.info( - f"[notification_signals] Skipping (action={action} != 'post_add')" - ) - return - - document = instance - - # Не отправляем индивидуальные уведомления если документ для всех - if document.sent_to_all: - logger.info( - f"[notification_signals] Skipping (sent_to_all=True)" - ) - return - - recipient_count = len(pk_set or []) - logger.info( - f"[notification_signals] Processing {recipient_count} recipients" - ) - - # Определяем, это массовая рассылка или нет - from django.conf import settings - BULK_THRESHOLD = getattr(settings, 'NOTIFICATION_BULK_THRESHOLD', 10) - is_bulk = recipient_count >= BULK_THRESHOLD - - if is_bulk: - logger.info( - f"[notification_signals] Bulk mode activated " - f"({recipient_count} >= {BULK_THRESHOLD})" - ) - - # Уведомляем новых получателей - created_count = 0 - for user_id in pk_set: - try: - user = Employee.objects.get(id=user_id) - logger.debug( - f"[notification_signals] Creating notification for user={user_id}" - ) - uploader_name = ( - document.uploaded_by.get_full_name() - if document.uploaded_by - else 'Администратор' - ) - notify.send( - sender=document.uploaded_by, - recipient=user, - verb='document_ready', - action_object=document, - description=( - f'{uploader_name} загрузил документ "{document.title}". ' - f'Требуется ознакомление.' - ), - action_url='/documents', - data={ - 'title': 'Новый документ на ознакомление', - 'document_id': document.id, - 'uploaded_by_id': ( - document.uploaded_by.id if document.uploaded_by else None - ), - 'sent_to_all': document.sent_to_all, - }, - ) - created_count += 1 - except Employee.DoesNotExist: - logger.warning(f"[notification_signals] User {user_id} not found, skipping") - except Exception as e: - logger.error(f"[notification_signals] Error creating notification for user {user_id}: {e}") - - logger.info(f"[notification_signals] Created {created_count} notifications") - - -@receiver(m2m_changed, sender=Document.departments.through) -def notify_department_employees(sender, instance, action, pk_set, **kwargs): - """ - Уведомляет всех сотрудников выбранных отделов при добавлении отделов. - Уведомления получат ВСЕ текущие и будущие сотрудники этих отделов. - """ - logger.info( - f"[notification_signals] m2m_changed Document.departments " - f"id={instance.pk} action={action} pk_set={pk_set}" - ) - - if action != 'post_add': - logger.info( - f"[notification_signals] Skipping (action={action} != 'post_add')" - ) - return - - document = instance - - # Не отправляем если документ для всех - if document.sent_to_all: - logger.info( - "[notification_signals] Skipping (sent_to_all=True)" - ) - return - - if not pk_set: - logger.warning("[notification_signals] No departments provided") - return - - # Получаем всех активных сотрудников выбранных отделов - from employees.models import Department - - all_employees = set() - for dept_id in pk_set: - try: - department = Department.objects.get(id=dept_id) - dept_employees = department.active_employees - logger.info( - f"[notification_signals] Department '{department.name}' " - f"has {len(dept_employees)} active employees: " - f"{[e.id for e in dept_employees]}" - ) - all_employees.update(dept_employees) - logger.info( - f"[notification_signals] After update, all_employees has: " - f"{[e.id for e in all_employees]}" - ) - except Department.DoesNotExist: - logger.warning( - f"[notification_signals] Department {dept_id} not found" - ) - continue - - # Исключаем загрузившего - if document.uploaded_by and document.uploaded_by in all_employees: - all_employees.remove(document.uploaded_by) - - recipient_count = len(all_employees) - logger.info( - f"[notification_signals] Processing {recipient_count} employees " - f"from {len(pk_set)} departments" - ) - - # Определяем режим рассылки - from django.conf import settings - BULK_THRESHOLD = getattr(settings, 'NOTIFICATION_BULK_THRESHOLD', 10) - is_bulk = recipient_count >= BULK_THRESHOLD - - if is_bulk: - logger.info( - f"[notification_signals] Bulk mode activated " - f"({recipient_count} >= {BULK_THRESHOLD})" - ) - - # Создаём уведомления - created_count = 0 - uploader_name = ( - document.uploaded_by.get_full_name() - if document.uploaded_by - else 'Администратор' - ) - for employee in all_employees: - try: - notify.send( - sender=document.uploaded_by, - recipient=employee, - verb='document_ready', - action_object=document, - description=( - f'{uploader_name} загрузил документ "{document.title}". ' - f'Требуется ознакомление.' - ), - action_url='/documents', - data={ - 'title': 'Новый документ на ознакомление', - 'document_id': document.id, - 'uploaded_by_id': ( - document.uploaded_by.id if document.uploaded_by else None - ), - 'sent_to_all': document.sent_to_all, - }, - ) - created_count += 1 - except Exception as e: - logger.error(f"[notification_signals] Error creating notification for employee {employee.id}: {e}") - - logger.info(f"[notification_signals] Created {created_count} notifications") - - -@receiver(post_save, sender=DocumentAcknowledgement) -def check_all_acknowledged(sender, instance, created, **kwargs): - """ - Проверяет, все ли получатели ознакомились с документом. - Если да - уведомляет загрузившего. - """ - if not created: - return - - acknowledgement = instance - document = acknowledgement.document - - # Определяем общее количество получателей - if document.sent_to_all: - total_recipients = Employee.objects.filter(is_active=True).count() - else: - # Считаем получателей из recipients + departments - recipients_count = document.recipients.count() - - # Добавляем сотрудников из отделов - from employees.models import Department - department_employees = set() - for department in document.departments.all(): - department_employees.update(department.active_employees) - - # Исключаем дубликаты с recipients - total_recipients = recipients_count + len(department_employees) - - # Проверяем количество ознакомившихся - acknowledged_count = document.acknowledgements.count() - - # Если все ознакомились - уведомляем загрузившего - if (acknowledged_count >= total_recipients and - total_recipients > 0 and - document.uploaded_by): - - notify.send( - sender=None, - recipient=document.uploaded_by, - verb='document_signed_all', - action_object=document, - description=( - f'Все сотрудники ознакомились с документом ' - f'"{document.title}"' - ), - action_url='/documents', - data={ - 'title': 'Все ознакомились с документом', - 'document_id': document.id, - 'total_recipients': total_recipients, - 'acknowledged_count': acknowledged_count, - } - ) - - -# ===== Вспомогательные функции ===== - -def notify_all_employees(document): - """ - Отправляет уведомление о новом документе всем активным сотрудникам. - Веб-уведомления отправляются сразу, email/Telegram - асинхронно. - """ - logger.info( - f"[notification_signals] notify_all_employees doc={document.id}" - ) - - active_employees = Employee.objects.filter(is_active=True) - - # Исключаем загрузившего документ - if document.uploaded_by: - active_employees = active_employees.exclude( - id=document.uploaded_by.id - ) - - count = active_employees.count() - logger.info( - f"[notification_signals] Creating {count} notifications " - f"(web immediately, email/telegram deferred)" - ) - - # Создаём уведомления с ЧАСТИЧНОЙ отправкой - created_count = 0 - uploader_name = ( - document.uploaded_by.get_full_name() - if document.uploaded_by - else 'Администратор' - ) - for employee in active_employees: - try: - notify.send( - sender=document.uploaded_by, - recipient=employee, - verb='document_ready', - action_object=document, - description=( - f'{uploader_name} загрузил документ "{document.title}". ' - f'Требуется ознакомление.' - ), - action_url='/documents', - data={ - 'title': 'Новый документ на ознакомление', - 'document_id': document.id, - 'uploaded_by_id': ( - document.uploaded_by.id if document.uploaded_by else None - ), - 'sent_to_all': document.sent_to_all, - }, - ) - created_count += 1 - except Exception as e: - logger.error( - f"Error creating notification for user {employee.id}: {e}" - ) - - logger.info( - f"[notification_signals] Created {created_count}/{count} notifications " - f"for doc={document.id}. Web sent immediately, email/telegram pending." - ) - - # Запускаем асинхронную отправку email/telegram - try: - from django.core.management import call_command - import threading - - def send_pending(): - try: - call_command('send_pending_notifications', '--batch-size=100') - except Exception as e: - logger.error(f"Error in background send: {e}") - - # Запускаем в отдельном потоке - thread = threading.Thread(target=send_pending, daemon=True) - thread.start() - logger.info("[notification_signals] Started background email/telegram send") - except Exception as e: - logger.error(f"Error starting background send: {e}") -def create_document_ready_notification(document, recipient, send_immediately=True): - """ - Создает уведомление о новом документе для конкретного получателя. - """ - logger.info( - f"[notification_signals] create_document_ready_notification " - f"doc={document.id} recipient={recipient.id}" - ) - - uploader_name = ( - document.uploaded_by.get_full_name() - if document.uploaded_by - else 'Администратор' - ) - - notify.send( - sender=document.uploaded_by, - recipient=recipient, - verb='document_ready', - action_object=document, - description=( - f'{uploader_name} загрузил документ "{document.title}". ' - f'Требуется ознакомление.' - ), - action_url='/documents', - data={ - 'title': 'Новый документ на ознакомление', - 'document_id': document.id, - 'uploaded_by_id': ( - document.uploaded_by.id if document.uploaded_by else None - ), - 'sent_to_all': document.sent_to_all, - }, - ) - - -# ============================================================================= -# НОВЫЕ СИГНАЛЫ ДЛЯ КОММЕНТАРИЕВ И СВЯЗАННЫХ ДОКУМЕНТОВ -# ============================================================================= - -@receiver(post_save, sender=DocumentComment) -def notify_on_new_comment(sender, instance, created, **kwargs): - """ - Отправка уведомлений при создании нового комментария. - - - Если это ответ на комментарий (parent != None): уведомить автора родительского комментария - - Если это обычный комментарий: уведомить автора документа (если он не автор комментария) - """ - if not created: - return - - comment = instance - document = comment.document - author = comment.author - - logger.info( - f"[notification_signals] New comment on document={document.pk} " - f"by user={author.pk} parent={comment.parent_id}" - ) - - # Если это ответ на комментарий - if comment.parent: - parent_author = comment.parent.author - - # Не уведомляем самого себя - if parent_author.id == author.id: - logger.info( - f"[notification_signals] Skipping self-reply notification" - ) - return - - logger.info( - f"[notification_signals] Notifying parent author={parent_author.pk} " - f"about reply" - ) - - notify.send( - sender=author, - recipient=parent_author, - verb='document_comment_reply', - action_object=comment, - description=( - f'{author.get_full_name() or author.username} ответил на ваш ' - f'комментарий к документу "{document.title}"' - ), - action_url='/documents', - data={ - 'title': 'Ответ на ваш комментарий', - 'document_id': document.id, - 'comment_id': comment.id, - 'parent_comment_id': comment.parent_id, - 'author_id': author.id, - }, - ) - else: - # Обычный комментарий - уведомляем автора документа - if not document.uploaded_by: - logger.info( - f"[notification_signals] Document has no author, skipping" - ) - return - - doc_author = document.uploaded_by - - # Не уведомляем самого себя - if doc_author.id == author.id: - logger.info( - f"[notification_signals] Skipping self-comment notification" - ) - return - - logger.info( - f"[notification_signals] Notifying document author={doc_author.pk} " - f"about new comment" - ) - - notify.send( - sender=author, - recipient=doc_author, - verb='document_comment', - action_object=comment, - description=( - f'{author.get_full_name() or author.username} оставил комментарий ' - f'к вашему документу "{document.title}"' - ), - action_url='/documents', - data={ - 'title': 'Новый комментарий к документу', - 'document_id': document.id, - 'comment_id': comment.id, - 'author_id': author.id, - }, - ) - - -@receiver(m2m_changed, sender=Document.related_documents.through) -def notify_on_related_document_added(sender, instance, action, pk_set, **kwargs): - """ - Уведомление при добавлении связанного документа. - Уведомляем автора связанного документа (если это не тот же пользователь). - """ - if action != 'post_add': - return - - if not pk_set: - return - - main_document = instance - - logger.info( - f"[notification_signals] Related documents added to doc={main_document.pk} " - f"related_ids={pk_set}" - ) - - # Получаем связанные документы - related_documents = Document.objects.filter(pk__in=pk_set).select_related('uploaded_by') - - for related_doc in related_documents: - # Если у связанного документа есть автор - if not related_doc.uploaded_by: - continue - - # Не уведомляем если авторы совпадают - if (main_document.uploaded_by and - related_doc.uploaded_by.id == main_document.uploaded_by.id): - continue - - logger.info( - f"[notification_signals] Notifying author={related_doc.uploaded_by.pk} " - f"that their document was linked" - ) - - notify.send( - sender=main_document.uploaded_by, - recipient=related_doc.uploaded_by, - verb='document_related', - action_object=main_document, - description=( - f'Ваш документ "{related_doc.title}" связан ' - f'с документом "{main_document.title}"' - ), - action_url='/documents', - data={ - 'title': 'Документ связан с другим', - 'document_id': main_document.id, - 'related_document_id': related_doc.id, - }, - ) diff --git a/backend/documents/tasks.py b/backend/documents/tasks.py deleted file mode 100644 index 602e6167..00000000 --- a/backend/documents/tasks.py +++ /dev/null @@ -1,177 +0,0 @@ -# # import os -# # from dotenv import load_dotenv -# # from asgiref.sync import async_to_sync - -# # from aiogram import Bot -# # from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton - -# # from .models import Document -# # from django.contrib.auth import get_user_model -# # from bots.models import BotSubscriber -# # import logging - -# # logger = logging.getLogger(__name__) - - -# # load_dotenv() -# # BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") - -# # User = get_user_model() -# # telegram_bot = Bot(token=BOT_TOKEN) - - -# # def send_document_to_recipients(document_id: int): -# # """ -# # Уведомляем в Telegram о появлении нового документа: -# # — всем активным, если sent_to_all=True, -# # — или тем, кто в recipients. -# # """ -# # try: -# # doc = Document.objects.get(pk=document_id) -# # except Document.DoesNotExist: -# # return - -# # # собираем подписчиков -# # if doc.sent_to_all: -# # subs = BotSubscriber.objects.filter( -# # telegram_id__isnull=False, -# # user__is_active=True -# # ) -# # else: -# # subs = BotSubscriber.objects.filter( -# # telegram_id__isnull=False, -# # user__in=doc.recipients.filter(is_active=True) -# # ) - -# # if not subs.exists(): -# # return - -# # # клавиатура: ведёт пользователя в бот, в раздел «Документы» -# # kb = InlineKeyboardMarkup(inline_keyboard=[[ -# # InlineKeyboardButton(text="📄 Открыть документы", -# # callback_data="menu_docs") -# # ]]) - -# # text = ( -# # f"🆕 *Новый документ*: {doc.title}\n\n" -# # f"{doc.description or ''}\n\n" -# # f"Зайдите в раздел «Документы» в боте, чтобы скачать файл." -# # ) - -# # # шлём всем подписчикам -# # for sub in subs: -# # try: -# # async_to_sync(telegram_bot.send_message)( -# # chat_id=sub.telegram_id, -# # text=text, -# # parse_mode="Markdown", -# # disable_web_page_preview=True, -# # reply_markup=kb, -# # ) -# # except Exception as e: -# # # Записываем в лог, на каком именно подписчике и с каким текстом упало -# # logger.error( -# # f"[Telegram] Ошибка при отправке документу={doc.pk} " -# # f"пользователю={sub.user.pk} tg_id={sub.telegram_id}: {e}", -# # exc_info=True -# # ) -# # else: -# # logger.info( -# # f"[Telegram] Уведомление по документу={doc.pk} " -# # f"успешно отправлено пользователю={sub.user.pk}" -# # ) - - -# # documents/tasks.py -# import os -# from dotenv import load_dotenv -# from celery import shared_task -# from django.conf import settings -# from django.urls import reverse -# from django.utils import timezone -# from telegram import Bot as TgBot, InlineKeyboardButton, InlineKeyboardMarkup - -# from .models import Document -# from django.contrib.auth import get_user_model -# from bots.models import BotSubscriber -# import logging - -# logger = logging.getLogger(__name__) - -# load_dotenv() -# TG_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN') or settings.TELEGRAM_BOT_TOKEN - -# User = get_user_model() - - -# # @shared_task(bind=True, name="documents.send_document_to_recipients") -# def send_document_to_recipients(document_id): -# try: -# doc = Document.objects.get(pk=document_id) -# except Document.DoesNotExist: -# return - -# # Собираем подписчиков -# if doc.sent_to_all: -# subs = BotSubscriber.objects.filter( -# telegram_id__isnull=False, -# user__is_active=True, -# ) -# else: -# subs = BotSubscriber.objects.filter( -# telegram_id__isnull=False, -# user__in=doc.recipients.filter(is_active=True), -# ) -# if not subs.exists(): -# return - -# # Синхронный телеграм-бот -# bot = TgBot(token=TG_TOKEN) - -# # Кнопка «Открыть документы» -# kb = InlineKeyboardMarkup( -# [[ InlineKeyboardButton("📄 Открыть документы", callback_data="menu_docs") ]] -# ) - -# # Текст уведомления -# text = ( -# f"🆕 *Новый документ*: {doc.title}\n\n" -# f"{doc.description or ''}\n\n" -# f"Дата публикации: {doc.uploaded_at.strftime('%d.%m.%Y %H:%M')}" -# ) - -# # for sub in subs: -# # try: -# # bot.send_message( -# # chat_id=sub.telegram_id, -# # text=text, -# # parse_mode="Markdown", -# # reply_markup=kb, -# # disable_web_page_preview=True -# # ) -# # except Exception as e: -# # print(e) -# # логируем, чтобы понимать, кому не дошло -# # self.retry(exc=e, countdown=60, max_retries=3) - -# for sub in subs: -# try: -# bot.send_message( -# chat_id=sub.telegram_id, -# text=text, -# parse_mode="Markdown", -# disable_web_page_preview=True, -# reply_markup=kb, -# ) -# except Exception as e: -# # Записываем в лог, на каком именно подписчике и с каким текстом упало -# logger.error( -# f"[Telegram] Ошибка при отправке документу={doc.pk} " -# f"пользователю={sub.user.pk} tg_id={sub.telegram_id}: {e}", -# exc_info=True -# ) -# else: -# logger.info( -# f"[Telegram] Уведомление по документу={doc.pk} " -# f"успешно отправлено пользователю={sub.user.pk}" -# ) diff --git a/backend/documents/test_documents_api.py b/backend/documents/test_documents_api.py deleted file mode 100644 index 5c0a4f88..00000000 --- a/backend/documents/test_documents_api.py +++ /dev/null @@ -1,624 +0,0 @@ -# documents/test_documents_api.py -""" -Comprehensive pytest API tests для DocumentViewSet -""" - -import pytest -from django.core.files.uploadedfile import SimpleUploadedFile -from rest_framework import status -from filer.models import File as FilerFile -from documents.models import ( - Document, DocumentAcknowledgement, DocumentComment -) -from reversion.models import Version -import reversion - - -# ============================================================================= -# CRUD TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentCRUDAPI: - """Тесты CRUD операций для документов""" - - def test_list_documents_authenticated(self, viewer_client, document): - """GET /documents/ - авторизованный пользователь видит свои документы""" - response = viewer_client.get('/api/v1/documents/') - assert response.status_code == status.HTTP_200_OK - # viewer в recipients - должен видеть документ - assert len(response.data.get('results', response.data)) >= 1 - - def test_list_documents_unauthenticated(self, api_client): - """GET /documents/ - неавторизованный получает 401""" - response = api_client.get('/api/v1/documents/') - assert response.status_code == status.HTTP_401_UNAUTHORIZED - - def test_create_document(self, authenticated_client, author_user): - """POST /documents/ - автор может создать документ""" - file_obj = SimpleUploadedFile("new_doc.txt", b"Content", content_type="text/plain") - - data = { - 'title': 'Новый документ', - 'description': 'Описание', - 'sent_to_all': True, - 'file': file_obj, - } - - response = authenticated_client.post('/api/v1/documents/', data, format='multipart') - assert response.status_code == status.HTTP_201_CREATED - assert response.data['title'] == 'Новый документ' - - def test_retrieve_document_as_recipient(self, viewer_client, document): - """GET /documents/{id}/ - получатель может просмотреть""" - response = viewer_client.get(f'/api/v1/documents/{document.id}/') - assert response.status_code == status.HTTP_200_OK - assert response.data['title'] == document.title - - def test_retrieve_document_as_non_recipient(self, api_client, other_user, document): - """GET /documents/{id}/ - не-получатель не видит документ (404 - отфильтрован queryset)""" - api_client.force_authenticate(user=other_user) - response = api_client.get(f'/api/v1/documents/{document.id}/') - # QuerySet фильтрует недоступные документы, поэтому 404 - assert response.status_code == status.HTTP_404_NOT_FOUND - - def test_update_document_as_author(self, authenticated_client, document, author_user): - """PATCH /documents/{id}/ - автор может редактировать""" - data = {'title': 'Обновленное название'} - response = authenticated_client.patch(f'/api/v1/documents/{document.id}/', data) - assert response.status_code == status.HTTP_200_OK - - # Проверяем через API response (refresh_from_db не работает с FSM protected fields) - assert response.data['title'] == 'Обновленное название' - - def test_update_document_as_viewer_forbidden(self, viewer_client, document): - """PATCH /documents/{id}/ - получатель не может редактировать""" - data = {'title': 'Попытка изменить'} - response = viewer_client.patch(f'/api/v1/documents/{document.id}/', data) - assert response.status_code == status.HTTP_403_FORBIDDEN - - def test_delete_document_as_author(self, authenticated_client, document): - """DELETE /documents/{id}/ - автор может удалить""" - doc_id = document.id - response = authenticated_client.delete(f'/api/v1/documents/{doc_id}/') - assert response.status_code == status.HTTP_204_NO_CONTENT - assert not Document.objects.filter(id=doc_id).exists() - - def test_delete_document_as_viewer_forbidden(self, viewer_client, document): - """DELETE /documents/{id}/ - получатель не может удалить""" - response = viewer_client.delete(f'/api/v1/documents/{document.id}/') - assert response.status_code == status.HTTP_403_FORBIDDEN - - -# ============================================================================= -# FSM WORKFLOW TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentFSMWorkflow: - """Тесты FSM переходов состояний документа""" - - def test_submit_for_review_transition(self, authenticated_client, document): - """POST submit-for-review: draft → in_review""" - assert document.status == Document.Status.DRAFT - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/submit-for-review/' - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['status'] == Document.Status.IN_REVIEW - - def test_approve_transition(self, authenticated_client, document): - """POST approve: in_review → approved""" - document.submit_for_review() - document.save() - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/approve/' - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['status'] == Document.Status.APPROVED - - def test_reject_transition(self, authenticated_client, document): - """POST reject: in_review → rejected""" - document.submit_for_review() - document.save() - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/reject/' - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['status'] == Document.Status.REJECTED - - def test_publish_transition(self, authenticated_client, document): - """POST publish: approved → published""" - document.submit_for_review() - document.approve() - document.save() - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/publish/' - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['status'] == Document.Status.PUBLISHED - - def test_return_to_draft_transition(self, authenticated_client, document): - """POST return-to-draft: in_review → draft""" - document.submit_for_review() - document.save() - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/return-to-draft/' - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['status'] == Document.Status.DRAFT - - def test_archive_transition(self, authenticated_client, document): - """POST archive: published → archived""" - document.submit_for_review() - document.approve() - document.publish() - document.save() - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/archive/' - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['status'] == Document.Status.ARCHIVED - - def test_unarchive_transition(self, authenticated_client, document): - """POST unarchive: archived → published""" - document.submit_for_review() - document.approve() - document.publish() - document.archive() - document.save() - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/unarchive/' - ) - assert response.status_code == status.HTTP_200_OK - - assert document.status == Document.Status.PUBLISHED - - def test_invalid_transition_returns_400(self, authenticated_client, document): - """Невалидный переход возвращает 400""" - # Пытаемся reject когда status=DRAFT (невалидно) - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/reject/' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - - -# ============================================================================= -# VERSIONING TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentVersioning: - """Тесты версионирования django-reversion""" - - def test_get_versions(self, authenticated_client, document): - """GET /documents/{id}/versions/ - история версий""" - # Создаем версии через reversion - with reversion.create_revision(): - document.title = "Version 1" - document.save() - reversion.set_comment("First change") - - with reversion.create_revision(): - document.title = "Version 2" - document.save() - reversion.set_comment("Second change") - - response = authenticated_client.get( - f'/api/v1/documents/{document.id}/versions/' - ) - assert response.status_code == status.HTTP_200_OK - - versions = response.data - assert len(versions) >= 2 - - def test_get_activity_timeline(self, authenticated_client, viewer_user, document): - """GET /documents/{id}/activity/ - timeline активности""" - # Создаем ознакомление - DocumentAcknowledgement.objects.create( - document=document, - user=viewer_user - ) - - # Создаем версию - with reversion.create_revision(): - document.title = "Changed" - document.save() - - response = authenticated_client.get( - f'/api/v1/documents/{document.id}/activity/' - ) - assert response.status_code == status.HTTP_200_OK - - activity = response.data - assert len(activity) > 0 - - # Проверяем типы в timeline - types = [item['type'] for item in activity] - assert 'version' in types or 'acknowledgement' in types - - def test_revert_to_version(self, authenticated_client, document): - """POST /documents/{id}/revert/ - откат к версии""" - original_title = "Original Title" - - # Сохраняем первую версию - with reversion.create_revision(): - document.title = original_title - document.save() - - # Получаем первую версию - versions = Version.objects.get_for_object(document) - first_version = versions.first() - - # Меняем документ - with reversion.create_revision(): - document.title = "Modified Title" - document.save() - - # Откатываем - data = { - 'version_id': first_version.id, - 'comment': 'Откат' - } - - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/revert/', - data - ) - assert response.status_code == status.HTTP_200_OK - assert response.data['title'] == original_title - - -# ============================================================================= -# ACKNOWLEDGEMENT TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentAcknowledgement: - """Тесты ознакомления с документами""" - - def test_acknowledge_document(self, viewer_client, viewer_user, document): - """POST /documents/{id}/acknowledge/ - подтвердить ознакомление""" - response = viewer_client.post( - f'/api/v1/documents/{document.id}/acknowledge/' - ) - assert response.status_code == status.HTTP_200_OK - - # Проверяем создание записи - assert DocumentAcknowledgement.objects.filter( - document=document, - user=viewer_user - ).exists() - - def test_acknowledge_non_recipient_forbidden(self, api_client, other_user, document): - """Не-получатель не может ознакомиться""" - api_client.force_authenticate(user=other_user) - - response = api_client.post( - f'/api/v1/documents/{document.id}/acknowledge/' - ) - assert response.status_code == status.HTTP_403_FORBIDDEN - - def test_get_acknowledgements_list(self, viewer_client, viewer_user, document): - """GET /documents/{id}/acknowledgements/ - список ознакомившихся""" - # Создаем ознакомление - DocumentAcknowledgement.objects.create( - document=document, - user=viewer_user - ) - - response = viewer_client.get( - f'/api/v1/documents/{document.id}/acknowledgements/' - ) - assert response.status_code == status.HTTP_200_OK - - acks = response.data - assert len(acks) == 1 - assert acks[0]['user']['id'] == viewer_user.id - - def test_double_acknowledge_prevented(self, viewer_client, viewer_user, document): - """Двойное ознакомление запрещено""" - # Первое ознакомление - viewer_client.post(f'/api/v1/documents/{document.id}/acknowledge/') - - # Второе ознакомление - должно вернуть ошибку или already acknowledged - response = viewer_client.post( - f'/api/v1/documents/{document.id}/acknowledge/' - ) - # Может вернуть 200 с сообщением "already acknowledged" или 400 - # Главное - запись одна - count = DocumentAcknowledgement.objects.filter( - document=document, - user=viewer_user - ).count() - assert count == 1 - - -# ============================================================================= -# COMMENT TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentComments: - """Тесты комментариев к документам""" - - def test_create_comment(self, viewer_client, viewer_user, document): - """POST /document-comments/ - создать комментарий""" - data = { - 'text': 'Отличный документ!', - 'document_id': document.id - } - - response = viewer_client.post('/api/v1/document-comments/', data) - assert response.status_code == status.HTTP_201_CREATED - - assert DocumentComment.objects.filter( - document=document, - author=viewer_user - ).exists() - - def test_create_reply_to_comment(self, authenticated_client, viewer_client, viewer_user, author_user, document): - """Создание ответа на комментарий""" - # Создаем родительский комментарий - parent = DocumentComment.objects.create( - document=document, - author=viewer_user, - text='Parent comment' - ) - - # Создаем ответ - data = { - 'text': 'Reply to parent', - 'document_id': document.id, - 'parent_id': parent.id - } - - response = authenticated_client.post('/api/v1/document-comments/', data) - assert response.status_code == status.HTTP_201_CREATED - - reply = DocumentComment.objects.get(parent=parent) - assert reply.text == 'Reply to parent' - - def test_get_comment_replies(self, viewer_client, viewer_user, author_user, document): - """GET /comments/{id}/replies/ - получить ответы""" - parent = DocumentComment.objects.create( - document=document, - author=viewer_user, - text='Parent' - ) - - DocumentComment.objects.create( - document=document, - author=author_user, - text='Reply 1', - parent=parent - ) - - DocumentComment.objects.create( - document=document, - author=author_user, - text='Reply 2', - parent=parent - ) - - response = viewer_client.get(f'/api/v1/document-comments/{parent.id}/replies/') - assert response.status_code == status.HTTP_200_OK - assert len(response.data) == 2 - - def test_delete_own_comment(self, viewer_client, viewer_user, document): - """Пользователь может удалить свой комментарий""" - comment = DocumentComment.objects.create( - document=document, - author=viewer_user, - text='My comment' - ) - - response = viewer_client.delete(f'/api/v1/document-comments/{comment.id}/') - assert response.status_code == status.HTTP_204_NO_CONTENT - - def test_delete_others_comment_forbidden(self, viewer_client, author_user, document): - """Нельзя удалить чужой комментарий""" - comment = DocumentComment.objects.create( - document=document, - author=author_user, - text='Author comment' - ) - - response = viewer_client.delete(f'/api/v1/document-comments/{comment.id}/') - assert response.status_code == status.HTTP_403_FORBIDDEN - - -# ============================================================================= -# PERMISSIONS TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentPermissions: - """Тесты системы разрешений""" - - def test_sent_to_all_visible_to_everyone(self, api_client, other_user, document_sent_to_all): - """sent_to_all=True - документ видим всем активным пользователям""" - api_client.force_authenticate(user=other_user) - - response = api_client.get(f'/api/v1/documents/{document_sent_to_all.id}/') - assert response.status_code == status.HTTP_200_OK - - def test_private_document_hidden_from_others(self, api_client, other_user, document): - """Приватный документ скрыт от не-получателей (404 - отфильтрован)""" - api_client.force_authenticate(user=other_user) - - response = api_client.get(f'/api/v1/documents/{document.id}/') - # QuerySet фильтрует недоступные документы - assert response.status_code == status.HTTP_404_NOT_FOUND - - def test_department_member_can_view(self, api_client, other_user, document, department): - """Член отдела-получателя может видеть документ""" - # Добавляем пользователя в отдел - other_user.department = department - other_user.save() - - # Добавляем отдел в получатели документа - document.departments.add(department) - - api_client.force_authenticate(user=other_user) - - response = api_client.get(f'/api/v1/documents/{document.id}/') - assert response.status_code == status.HTTP_200_OK - - def test_author_can_always_edit(self, authenticated_client, document): - """Автор всегда может редактировать свой документ""" - data = {'description': 'Updated by author'} - response = authenticated_client.patch( - f'/api/v1/documents/{document.id}/', - data - ) - assert response.status_code == status.HTTP_200_OK - - def test_admin_has_full_access(self, admin_client, document): - """Admin имеет полный доступ""" - # Должен видеть документ - response = admin_client.get(f'/api/v1/documents/{document.id}/') - # Admin может иметь доступ или нет в зависимости от rules - assert response.status_code in [status.HTTP_200_OK, status.HTTP_403_FORBIDDEN] - - -# ============================================================================= -# FOLDER TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestFolderAPI: - """Тесты для папок документов""" - - def test_list_folders(self, authenticated_client, folder): - """GET /folders/ - список папок""" - response = authenticated_client.get('/api/v1/folders/') - assert response.status_code == status.HTTP_200_OK - - def test_create_folder(self, authenticated_client): - """POST /folders/ - создание папки""" - data = {'name': 'Новая папка'} - response = authenticated_client.post('/api/v1/folders/', data) - assert response.status_code == status.HTTP_201_CREATED - - def test_get_folder_children(self, authenticated_client, folder, author_user): - """GET /folders/{id}/children/ - дочерние папки""" - # Создаем дочернюю папку - from filer.models import Folder as FilerFolder - child = FilerFolder.objects.create( - name='Дочерняя', - parent=folder, - owner=author_user - ) - - response = authenticated_client.get(f'/api/v1/folders/{folder.id}/children/') - assert response.status_code == status.HTTP_200_OK - assert len(response.data) == 1 - - def test_get_folder_documents(self, authenticated_client, folder, document, filer_file): - """GET /folders/{id}/documents/ - документы в папке""" - # Перемещаем документ в папку - document.folder = folder - document.save() - - response = authenticated_client.get(f'/api/v1/folders/{folder.id}/documents/') - assert response.status_code == status.HTTP_200_OK - assert len(response.data) == 1 - - -# ============================================================================= -# TAG & TYPE TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestDocumentTagsAndTypes: - """Тесты для тегов и типов документов""" - - def test_list_tags(self, authenticated_client, document_tag): - """GET /document-tags/ - список тегов""" - response = authenticated_client.get('/api/v1/document-tags/') - assert response.status_code == status.HTTP_200_OK - - def test_create_tag(self, authenticated_client): - """POST /document-tags/ - создание тега""" - data = {'name': 'Срочно', 'color': '#FF0000'} - response = authenticated_client.post('/api/v1/document-tags/', data) - assert response.status_code == status.HTTP_201_CREATED - - def test_list_document_types(self, authenticated_client, document_type): - """GET /document-types/ - список типов""" - response = authenticated_client.get('/api/v1/document-types/') - assert response.status_code == status.HTTP_200_OK - - def test_get_documents_by_tag(self, authenticated_client, document, document_tag): - """GET /tags/{id}/documents/ - документы с тегом""" - # Добавляем тег к документу (если есть M2M) - if hasattr(document, 'tags'): - document.tags.add(document_tag) - - response = authenticated_client.get(f'/api/v1/document-tags/{document_tag.id}/documents/') - assert response.status_code == status.HTTP_200_OK - - -# ============================================================================= -# RELATED DOCUMENTS TESTS -# ============================================================================= - -@pytest.mark.django_db -class TestRelatedDocuments: - """Тесты для связанных документов""" - - def test_get_related_documents(self, authenticated_client, document): - """GET /documents/{id}/related/ - связанные документы""" - response = authenticated_client.get( - f'/api/v1/documents/{document.id}/related/' - ) - assert response.status_code == status.HTTP_200_OK - - def test_add_related_document(self, authenticated_client, document, filer_file, author_user): - """POST /documents/{id}/add_related/ - добавить связь""" - # Создаем второй документ - doc2 = Document.objects.create( - title="Связанный документ", - file=filer_file, - uploaded_by=author_user, - sent_to_all=True - ) - - data = {'document_id': doc2.id} - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/add_related/', - data - ) - assert response.status_code == status.HTTP_200_OK - - # Проверяем связь - assert doc2 in document.related_documents.all() - - def test_remove_related_document(self, authenticated_client, document, filer_file, author_user): - """POST /documents/{id}/remove_related/ - удалить связь""" - # Создаем и связываем документ - doc2 = Document.objects.create( - title="Связанный", - file=filer_file, - uploaded_by=author_user, - sent_to_all=True - ) - document.related_documents.add(doc2) - - data = {'document_id': doc2.id} - response = authenticated_client.post( - f'/api/v1/documents/{document.id}/remove_related/', - data - ) - assert response.status_code == status.HTTP_200_OK - - # Проверяем что связь удалена - assert doc2 not in document.related_documents.all() diff --git a/backend/documents/tests.py b/backend/documents/tests.py deleted file mode 100644 index 9633ed34..00000000 --- a/backend/documents/tests.py +++ /dev/null @@ -1,298 +0,0 @@ -# documents/tests.py -""" -Тесты для моделей Document и DocumentAcknowledgement с django-filer -""" - -from django.test import TestCase -from django.contrib.auth import get_user_model -from django.core.files.uploadedfile import SimpleUploadedFile -from filer.models import File as FilerFile -from documents.models import Document, DocumentAcknowledgement -from employees.models import Department -import rules - -User = get_user_model() - - -class DocumentModelTest(TestCase): - """Тесты для модели Document""" - - def setUp(self): - """Создание тестовых данных""" - self.user = User.objects.create_user( - phone_number='+79999999999', - email='test1@example.com', - first_name='Тест', - last_name='Тестов', - is_active=True - ) - - # Создаем тестовый файл через filer - test_file = SimpleUploadedFile( - "test_document.txt", - b"Test content", - content_type="text/plain" - ) - - self.filer_file = FilerFile.objects.create( - file=test_file, - original_filename="test_document.txt", - name="Test Document", - owner=self.user - ) - - def test_create_document(self): - """Создание документа с filer файлом""" - document = Document.objects.create( - title="Тестовый документ", - file=self.filer_file, - description="Описание тестового документа", - uploaded_by=self.user, - sent_to_all=True - ) - - self.assertEqual(document.title, "Тестовый документ") - self.assertEqual(document.file, self.filer_file) - self.assertEqual(document.uploaded_by, self.user) - self.assertTrue(document.sent_to_all) - self.assertIsNotNone(document.uploaded_at) - - def test_file_properties(self): - """Проверка свойств файла (размер, расширение)""" - document = Document.objects.create( - title="Тестовый документ", - file=self.filer_file, - uploaded_by=self.user - ) - - self.assertGreater(document.file_size, 0) - self.assertEqual(document.file_extension, 'txt') - - def test_sent_to_all(self): - """Документ с sent_to_all=True""" - document = Document.objects.create( - title="Документ для всех", - file=self.filer_file, - uploaded_by=self.user, - sent_to_all=True - ) - - self.assertTrue(document.sent_to_all) - self.assertEqual(document.departments.count(), 0) - self.assertEqual(document.recipients.count(), 0) - - def test_department_recipients(self): - """Документ для конкретных отделов""" - department = Department.objects.create( - name="Тестовый отдел" - ) - - document = Document.objects.create( - title="Документ для отдела", - file=self.filer_file, - uploaded_by=self.user, - sent_to_all=False - ) - document.departments.add(department) - - self.assertFalse(document.sent_to_all) - self.assertEqual(document.departments.count(), 1) - self.assertIn(department, document.departments.all()) - - def test_individual_recipients(self): - """Документ для конкретных получателей""" - recipient = User.objects.create_user( - phone_number='+79999999998', - email='recipient@example.com', - first_name='Получатель', - last_name='Тестовый', - is_active=True - ) - - document = Document.objects.create( - title="Документ для получателя", - file=self.filer_file, - uploaded_by=self.user, - sent_to_all=False - ) - document.recipients.add(recipient) - - self.assertFalse(document.sent_to_all) - self.assertEqual(document.recipients.count(), 1) - self.assertIn(recipient, document.recipients.all()) - - -class DocumentAcknowledgementModelTest(TestCase): - """Тесты для модели DocumentAcknowledgement""" - - def setUp(self): - """Создание тестовых данных""" - self.user = User.objects.create_user( - phone_number='+79999999999', - email='test@example.com', - first_name='Тест', - last_name='Тестов', - is_active=True - ) - - test_file = SimpleUploadedFile( - "test_document.txt", - b"Test content", - content_type="text/plain" - ) - - filer_file = FilerFile.objects.create( - file=test_file, - original_filename="test_document.txt", - name="Test Document", - owner=self.user - ) - - self.document = Document.objects.create( - title="Тестовый документ", - file=filer_file, - uploaded_by=self.user, - sent_to_all=True - ) - - def test_create_acknowledgement(self): - """Создание ознакомления с документом""" - ack = DocumentAcknowledgement.objects.create( - document=self.document, - user=self.user - ) - - self.assertEqual(ack.document, self.document) - self.assertEqual(ack.user, self.user) - self.assertIsNotNone(ack.acknowledged_at) - - def test_unique_acknowledgement(self): - """Один пользователь может ознакомиться только один раз""" - DocumentAcknowledgement.objects.create( - document=self.document, - user=self.user - ) - - # Попытка создать второе ознакомление должна вызвать ошибку - with self.assertRaises(Exception): - DocumentAcknowledgement.objects.create( - document=self.document, - user=self.user - ) - - def test_acknowledgement_count(self): - """Подсчет ознакомлений""" - user2 = User.objects.create_user( - phone_number='+79999999998', - email='test2@example.com', - first_name='Тест2', - last_name='Тестов2', - is_active=True - ) - - DocumentAcknowledgement.objects.create( - document=self.document, - user=self.user - ) - DocumentAcknowledgement.objects.create( - document=self.document, - user=user2 - ) - - self.assertEqual(self.document.acknowledgements.count(), 2) - - -class DocumentRulesTest(TestCase): - """Тесты для правил доступа django-rules""" - - def setUp(self): - """Создание тестовых данных""" - self.owner = User.objects.create_user( - phone_number='+79999999999', - email='owner@example.com', - first_name='Владелец', - last_name='Документа', - is_active=True - ) - - self.other_user = User.objects.create_user( - phone_number='+79999999998', - email='other@example.com', - first_name='Другой', - last_name='Пользователь', - is_active=True - ) - - test_file = SimpleUploadedFile( - "test_document.txt", - b"Test content", - content_type="text/plain" - ) - - filer_file = FilerFile.objects.create( - file=test_file, - original_filename="test_document.txt", - name="Test Document", - owner=self.owner - ) - - self.document = Document.objects.create( - title="Тестовый документ", - file=filer_file, - uploaded_by=self.owner, - sent_to_all=False # Не для всех - ) - - def test_owner_can_view(self): - """Владелец может просматривать свой документ""" - self.assertTrue( - rules.test_rule('documents.view_document', self.owner, self.document) - ) - - def test_other_user_cannot_view(self): - """Другой пользователь не может просматривать чужой документ (если не в списке)""" - self.assertFalse( - rules.test_rule('documents.view_document', self.other_user, self.document) - ) - - def test_sent_to_all_allows_view(self): - """Документ с sent_to_all=True доступен всем активным пользователям""" - self.document.sent_to_all = True - self.document.save() - - self.assertTrue( - rules.test_rule('documents.view_document', self.other_user, self.document) - ) - - def test_recipient_can_view(self): - """Получатель из списка может просматривать документ""" - self.document.recipients.add(self.other_user) - - self.assertTrue( - rules.test_rule('documents.view_document', self.other_user, self.document) - ) - - def test_owner_can_edit(self): - """Владелец может редактировать свой документ""" - self.assertTrue( - rules.test_rule('documents.change_document', self.owner, self.document) - ) - - def test_other_user_cannot_edit(self): - """Другой пользователь не может редактировать чужой документ""" - self.assertFalse( - rules.test_rule('documents.change_document', self.other_user, self.document) - ) - - def test_owner_can_delete(self): - """Владелец может удалять свой документ""" - self.assertTrue( - rules.test_rule('documents.delete_document', self.owner, self.document) - ) - - def test_other_user_cannot_delete(self): - """Другой пользователь не может удалять чужой документ""" - self.assertFalse( - rules.test_rule('documents.delete_document', self.other_user, self.document) - ) - diff --git a/backend/documents/urls.py b/backend/documents/urls.py deleted file mode 100644 index a84c8b06..00000000 --- a/backend/documents/urls.py +++ /dev/null @@ -1,10 +0,0 @@ -# documents/urls.py -from django.urls import path -from .views import DocumentView, acknowledge_document - -app_name = 'documents' - -urlpatterns = [ - path('', DocumentView.as_view(), name='document_list'), - path('ack//', acknowledge_document, name='acknowledge'), -] diff --git a/backend/documents/views.py b/backend/documents/views.py deleted file mode 100644 index 72974eb0..00000000 --- a/backend/documents/views.py +++ /dev/null @@ -1,435 +0,0 @@ -from __future__ import annotations - -from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple - -from api.client import get_api_client -from api.decorators import require_api_auth -from django.contrib import messages -from django.contrib.auth.decorators import login_required -from django.contrib.auth.models import AbstractBaseUser -from django.contrib.auth.mixins import LoginRequiredMixin -from django.http import HttpRequest, HttpResponse -from django.shortcuts import redirect -from django.urls import reverse -from django.utils.decorators import method_decorator -from django.utils.dateparse import parse_datetime -from django.utils.timezone import localtime -from django.views.decorators.csrf import csrf_protect -from django.views.generic import TemplateView - - -# ============================== -# Вспомогательные функции -# ============================== - - -def _as_bool(value: Any) -> bool: - """Нормализует булево значение из строк HTML-форм. - - Args: - value: Строка/булево/None. - - Returns: - bool: True, если значение похоже на истину ('1', 'true', 'on', 'yes'). - - Raises: - ValueError: Если значение имеет неожиданный тип, не приводимый к строке. - """ - if isinstance(value, bool): - return value - try: - s = str(value).strip().lower() - except Exception as exc: # pragma: no cover - защитный блок - raise ValueError("Unsupported boolean value") from exc - return s in {"1", "true", "on", "yes"} - - -def _repeat_field_pairs(field: str, values: Iterable[int | str]) -> List[Tuple[str, str]]: - """Готовит список повторяемых пар (field, value) для multipart-формы. - - Args: - field (str): Имя поля. - values (Iterable[int|str]): Значения (могут быть пустыми/None — игнорируются). - - Returns: - list[tuple[str, str]]: Повторяемые пары для передачи в API-клиент. - - Raises: - TypeError: Если field пустой или values неитерируемый. - """ - if not field: - raise TypeError("field must be a non-empty string") - out: List[Tuple[str, str]] = [] - for v in values or []: - if v is None: - continue - s = str(v).strip() - if s: - out.append((field, s)) - return out - - -def _api_unpack(resp: Any) -> Tuple[bool, Dict[str, Any], int]: - """Безопасно извлекает признаки успеха, JSON и статус из ответа API. - - Args: - resp: Объект ответа клиентской обёртки (ожидаются атрибуты .ok, .json, .status). - - Returns: - tuple: (ok, data, status) - - Raises: - AttributeError: Если у ответа нет ожидаемых атрибутов. - """ - ok: bool = getattr(resp, "ok") - status: int = getattr(resp, "status") - data: Dict[str, Any] = getattr(resp, "json") or {} - return ok, data, status - - -def _parse_page_payload( - payload: Any, -) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[str], Optional[int]]: - """Нормализует ответ API к (results, next, previous, count). - - Поддерживает форматы: - - пагинация DRF: {"count": N, "next": URL, "previous": URL, - "results": [...]} - - сырой список: [ ... ] - - Args: - payload (Any): Десериализованный JSON из API. - - Returns: - Tuple: (results, next_url, prev_url, count) - """ - # 1) Непагинированный список - if isinstance(payload, list): - results = _format_datetime_fields(payload) - return results, None, None, len(results) - - # 2) Словарь (пагинированный или нет) - if isinstance(payload, dict): - if "results" in payload: - raw = payload.get("results") or [] - results = _format_datetime_fields(list(raw)) - return ( - results, - payload.get("next"), - payload.get("previous"), - payload.get("count") - ) - # словарь, но не страница — пустой список - return [], None, None, None - - # 3) Непонятный формат - return [], None, None, None - - -def _user_can_manage_documents(user: AbstractBaseUser) -> bool: - """Проверяет, есть ли у пользователя административные/модельные права на Document. - - Args: - user (AbstractBaseUser): Пользователь. - - Returns: - bool: True, если is_staff или есть одно из прав add/change/delete/view. - """ - if getattr(user, "is_staff", False): - return True - model = "document" - app_label = "documents" - for act in ("view", "add", "change", "delete"): - if user.has_perm(f"{app_label}.{act}_{model}"): - return True - return False - - -def _fmt_dt(val, fmt: str = "%d.%m.%Y %H:%M") -> str | None: - """Превращает ISO-строку/дату в локализованную строку. - - Args: - val: ISO строка даты или datetime объект. - fmt: Формат вывода даты. - - Returns: - str | None: Отформатированная строка или None. - """ - if not val: - return None - # если уже datetime — ок, если строка — парсим - dt = parse_datetime(val) if isinstance(val, str) else val - if dt is None: - return val # на всякий случай оставим как есть - try: - return localtime(dt).strftime(fmt) # уважим таймзону пользователя - except Exception: - return val - - -def _format_datetime_fields( - items: List[Any], - fields: tuple[str, ...] = ("uploaded_at", "created_at", "updated_at"), -) -> List[Any]: - """Форматирует поля дат в списке объектов. - - Args: - items: Список объектов (словарей). - fields: Кортеж имен полей с датами. - - Returns: - List[Any]: Список с отформатированными датами. - """ - out: List[Any] = [] - for it in items: - if isinstance(it, Mapping): - d = dict(it) # копия, не мутируем исходник - for f in fields: - if f in d: - d[f] = _fmt_dt(d.get(f)) - out.append(d) - else: - out.append(it) - return out - - -def _err_msg(payload: Any, fallback: str = "") -> str: - """Извлекает сообщение об ошибке из ответа API. - - Args: - payload: Полезная нагрузка ответа API. - fallback: Сообщение по умолчанию. - - Returns: - str: Сообщение об ошибке. - """ - # dict: detail / non_field_errors / первое поле со строкой - if isinstance(payload, dict): - if isinstance(payload.get("detail"), (str, list)): - d = payload["detail"] - return d if isinstance(d, str) else "; ".join(map(str, d)) - if isinstance(payload.get("non_field_errors"), list): - return "; ".join(map(str, payload["non_field_errors"])) - # первая строка из значений полей - for v in payload.values(): - if isinstance(v, list) and v and isinstance(v[0], str): - return "; ".join(map(str, v)) - if isinstance(v, str): - return v - # list: склеить - if isinstance(payload, list): - return "; ".join(map(str, payload)) - # str или что-то иное - return str(payload or fallback) - - -def _current_url_with(request: HttpRequest, **new_params: Any) -> str: - """Возвращает URL текущего пути с подменой/удалением query-параметров. - - Args: - request (HttpRequest): Текущий запрос. - **new_params: Пары ключ-значение; None удаляет параметр. - - Returns: - str: Абсолютный путь с новой query-строкой. - """ - q = request.GET.copy() - for k, v in new_params.items(): - if v is None and k in q: - del q[k] - elif v is not None: - q[k] = str(v) - qs = q.urlencode() - return f"{request.path}?{qs}" if qs else request.path - - -def _filter_for_user(items: Sequence[Dict[str, Any]], user_id: Optional[int]) -> List[Dict[str, Any]]: - """Фильтрует документы в область «Мои». - - Документ считается «моим», если: - - он отправлен всем (`sent_to_all == true`), или - - в `recipients[].id` присутствует `user_id`, или - - в `departments` есть отдел, где пользователь является активным сотрудником, или - - пользователь является загрузившим документ (`uploaded_by.id == user_id`). - - Args: - items (Sequence[dict]): Полученный список документов из API. - user_id (int|None): Идентификатор текущего пользователя. - - Returns: - list[dict]: Отфильтрованный список. - - Raises: - ValueError: Если user_id не задан (None). - """ - if user_id is None: - raise ValueError("user_id is required for filtering") - result: List[Dict[str, Any]] = [] - for it in items: - if not isinstance(it, dict): - continue - - # 1. Документы для всех - if it.get("sent_to_all"): - result.append(it) - continue - - # 2. Документы, загруженные текущим пользователем - uploaded_by = it.get("uploaded_by") - if uploaded_by and isinstance(uploaded_by, dict): - try: - if int(uploaded_by.get("id")) == user_id: - result.append(it) - continue - except (ValueError, TypeError): - pass - - # 3. Проверяем recipients - recips = it.get("recipients") or [] - try: - if any(int(r.get("id")) == user_id for r in recips if isinstance(r, dict) and r.get("id") is not None): - result.append(it) - continue - except Exception: - pass - - # 4. Проверяем departments - depts = it.get("departments") or [] - try: - # Если в документе есть отдел, нужно проверить членство - # Но API возвращает только id и name отдела, без списка сотрудников - # Поэтому нужно загрузить эту информацию отдельно или доверять backend'у - # Для упрощения, если есть departments - считаем что пользователь может видеть - # (реальная проверка происходит на backend при создании уведомлений) - if depts: - # TODO: добавить проверку членства в отделе через отдельный API запрос - # Пока просто пропускаем - backend сам решит, нужно ли показывать - pass - except Exception: - pass - return result - - -# ============================== -# Вьюхи -# ============================== - -@method_decorator([login_required, require_api_auth, csrf_protect], name="dispatch") -class DocumentView(LoginRequiredMixin, TemplateView): - """Страница списка документов с SSR + API для CRUD. - - GET: - - Загружает список документов через API (GET /api/v1/documents/). - - Понимает фильтры: scope=mine|all, ack_status=acked|not_acked. - - Для scope=mine загружает список ознакомленных документов. - - Контекст шаблона: - documents, acked_ids, next_url, prev_url, count, scope, - show_admin_controls, can_manage_documents, filters, perms, - api_document_list_url, api_document_detail_base. - """ - - template_name = "documents/document_list.html" - - def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: - """Формирует контекст страницы списка, фильтров и пагинации.""" - ctx = super().get_context_data(**kwargs) - request = self.request - user = request.user - - # --- Определяем права --- - can_manage = _user_can_manage_documents(user) - show_admin_controls = can_manage - - # --- Читаем фильтры из GET --- - scope = (request.GET.get("scope") or "mine").strip().lower() - if scope not in ("mine", "all"): - scope = "mine" - - # Если нет прав управления - принудительно mine - if not can_manage: - scope = "mine" - - ack_status = (request.GET.get("ack_status") or "").strip() - - # --- Формируем контекст без загрузки документов --- - # Документы загружаются через AJAX - ctx.update({ - "documents": [], # Пустой список - загрузка через JS - "acked_ids": set(), - "scope": scope, - "show_admin_controls": show_admin_controls, - "can_manage_documents": can_manage, - "filters": { - "ack_status": ack_status, - }, - "perms": { - "documents": { - "add_document": ( - user.has_perm("documents.add_document") or - user.is_staff - ), - "change_document": ( - user.has_perm("documents.change_document") or - user.is_staff - ), - "delete_document": ( - user.has_perm("documents.delete_document") or - user.is_staff - ), - } - }, - "api_document_list_url": reverse("api:v1:documents-list"), - "api_document_detail_base": reverse("api:v1:documents-list"), - }) - - return ctx - - -@login_required -@require_api_auth -def acknowledge_document(request: HttpRequest, pk: int) -> HttpResponse: - """Отмечает документ «ознакомленным» через API и перенаправляет на файл. - - Args: - request (HttpRequest): Текущий запрос. - pk (int): Идентификатор документа. - - Returns: - HttpResponse: Редирект на файл документа или список документов при ошибке. - - Raises: - ValueError: Если pk имеет неверный формат. - """ - try: - doc_id = int(pk) - except (TypeError, ValueError) as exc: - messages.error(request, "Некорректный идентификатор документа.") - raise ValueError("pk must be an integer") from exc - - # Получаем документ для редиректа на файл - from documents.models import Document - try: - doc = Document.objects.get(id=doc_id) - file_url = doc.file.url if doc.file else None - except Document.DoesNotExist: - messages.error(request, "Документ не найден.") - return redirect(reverse("documents:document_list")) - - # Отмечаем ознакомление через API - api = get_api_client(request) - resp = api.post(f"v1/documents/{doc_id}/acknowledge/") - ok, data, _status = _api_unpack(resp) - if ok: - if data.get("already"): - messages.info(request, "Вы уже отмечали ознакомление с этим документом.") - else: - messages.success(request, "Ознакомление с документом отмечено.") - else: - messages.error(request, data.get("detail") or "Не удалось отметить документ.") - - # Редирект на файл или обратно на список - if file_url: - return redirect(file_url) - else: - return redirect(reverse("documents:document_list")) diff --git a/backend/eusrr_backend/urls.py b/backend/eusrr_backend/urls.py index 7bc8f582..65d63712 100644 --- a/backend/eusrr_backend/urls.py +++ b/backend/eusrr_backend/urls.py @@ -27,7 +27,6 @@ def serve_service_worker(request): urlpatterns = [ path("admin/", admin.site.urls), - path("documents/", include("documents.urls", namespace="documents")), path("api/", include(("api.urls", "api"), namespace="api")), # Service Worker должен быть в корне для правильного scope path("sw.js", serve_service_worker, name="sw"),