fix(requests): Исправлена ошибка с employee_departments и добавлены тесты

- Заменено employee_departments на departments_links в notification_signals.py
- Добавлен импорт Department в serializers.py
- Создан комплексный тестовый скрипт test_recipients.py (7 тестов)
- Создан скрипт test_recipients_api.py для API тестирования
- Добавлены утилиты для создания тестовых данных

Результаты тестирования:
✓ 7/7 тестов модели прошли успешно (100%)
✓ Проверены все методы: add_recipient, is_recipient, all_recipients
✓ Проверена работа с ManyToMany полями
✓ Проверена сериализация через API
✓ Проверен флаг sent_to_all_department
This commit is contained in:
admin i 2025-12-02 17:39:09 +03:00
parent 60fc506ef9
commit 771b93feed
6 changed files with 787 additions and 10 deletions

View File

@ -4,6 +4,7 @@ from typing import Any, Mapping
from django.contrib.auth import get_user_model
from django.db import models
from employees.models import Department
from requests_app.enums import RequestStatus, RequestType
from requests_app.models import Request, RequestComment
from rest_framework import serializers
@ -184,7 +185,7 @@ class RequestWriteSerializer(serializers.ModelSerializer):
# Новые поля для получателей
department_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=None, # Установим в __init__
queryset=Department.objects.all(),
source='departments',
required=False,
allow_empty=True

176
backend/create_test_data.py Normal file
View File

@ -0,0 +1,176 @@
#!/usr/bin/env python
"""
Создание тестовых пользователей и отделов
"""
import os
import django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
django.setup()
from django.contrib.auth import get_user_model
from employees.models import Department, EmployeeDepartment, DepartmentRole
User = get_user_model()
def create_test_users():
"""Создание тестовых пользователей"""
print("\n" + "="*60)
print(" Создание тестовых пользователей")
print("="*60)
users_data = [
{
'username': 'recipient1',
'email': 'recipient1@test.local',
'first_name': 'Получатель',
'last_name': 'Первый',
'password': 'test123'
},
{
'username': 'recipient2',
'email': 'recipient2@test.local',
'first_name': 'Получатель',
'last_name': 'Второй',
'password': 'test123'
},
{
'username': 'recipient3',
'email': 'recipient3@test.local',
'first_name': 'Получатель',
'last_name': 'Третий',
'password': 'test123'
},
{
'username': 'cc_user1',
'email': 'cc_user1@test.local',
'first_name': 'Копия',
'last_name': 'Первый',
'password': 'test123'
},
{
'username': 'cc_user2',
'email': 'cc_user2@test.local',
'first_name': 'Копия',
'last_name': 'Второй',
'password': 'test123'
},
]
created_users = []
for user_data in users_data:
username = user_data['username']
# Проверяем, существует ли
if User.objects.filter(username=username).exists():
user = User.objects.get(username=username)
print(f" ⚠ Пользователь {username} уже существует")
else:
user = User.objects.create_user(**user_data)
user.is_active = True
user.save()
print(f" ✓ Создан {username} - {user.get_full_name()}")
created_users.append(user)
return created_users
def create_test_departments():
"""Создание тестовых отделов"""
print("\n" + "="*60)
print(" Создание тестовых отделов")
print("="*60)
departments_data = [
{'name': 'Отдел тестирования 1', 'code': 'TEST_DEPT_1'},
{'name': 'Отдел тестирования 2', 'code': 'TEST_DEPT_2'},
]
created_depts = []
for dept_data in departments_data:
code = dept_data['code']
if Department.objects.filter(code=code).exists():
dept = Department.objects.get(code=code)
print(f" ⚠ Отдел {code} уже существует")
else:
dept = Department.objects.create(**dept_data)
print(f" ✓ Создан отдел {dept.name}")
created_depts.append(dept)
return created_depts
def link_users_to_departments(users, departments):
"""Привязка пользователей к отделам"""
print("\n" + "="*60)
print(" Привязка пользователей к отделам")
print("="*60)
if not departments:
print(" ⚠ Нет отделов для привязки")
return
# Получаем или создаем роль
role, created = DepartmentRole.objects.get_or_create(
name='Сотрудник',
defaults={'description': 'Обычный сотрудник отдела'}
)
if created:
print(f" ✓ Создана роль: {role.name}")
# Привязываем пользователей
for i, user in enumerate(users):
dept = departments[i % len(departments)]
link, created = EmployeeDepartment.objects.get_or_create(
employee=user,
department=dept,
defaults={'role': role, 'is_active': True}
)
if created:
print(f" ✓ {user.username} → {dept.name}")
else:
print(f" ⚠ {user.username} уже в {dept.name}")
def show_summary():
"""Показать итоги"""
print("\n" + "="*60)
print(" ИТОГИ")
print("="*60)
total_users = User.objects.filter(is_active=True).count()
total_depts = Department.objects.count()
total_links = EmployeeDepartment.objects.filter(is_active=True).count()
print(f"\nАктивных пользователей: {total_users}")
print(f"Отделов: {total_depts}")
print(f"Связей пользователь-отдел: {total_links}")
print("\nПользователи для тестирования:")
test_users = User.objects.filter(
username__in=['recipient1', 'recipient2', 'recipient3', 'cc_user1', 'cc_user2']
)
for user in test_users:
depts = user.departments_links.filter(is_active=True)
dept_names = ', '.join([d.department.name for d in depts])
print(f" - {user.username}: {user.get_full_name()} ({dept_names})")
if __name__ == '__main__':
print("\n" + "="*60)
print(" 🔧 ПОДГОТОВКА ТЕСТОВЫХ ДАННЫХ")
print("="*60)
users = create_test_users()
departments = create_test_departments()
link_users_to_departments(users, departments)
show_summary()
print("\n✓ Готово! Теперь можно запускать тесты API.\n")

View File

@ -101,8 +101,8 @@ def create_comment_notification(sender, instance, created, **kwargs):
# Если sent_to_all_department - все сотрудники отделов
if request_obj.sent_to_all_department:
dept_employees = Employee.objects.filter(
employee_departments__department__in=request_obj.departments.all(),
employee_departments__is_active=True,
departments_links__department__in=request_obj.departments.all(),
departments_links__is_active=True,
is_active=True
).exclude(id__in=[author.id, request_obj.employee.id]).distinct()
@ -155,8 +155,8 @@ def notify_new_request(request_obj):
# 3. Если sent_to_all_department - все сотрудники отделов
if request_obj.sent_to_all_department:
dept_employees = Employee.objects.filter(
employee_departments__department__in=request_obj.departments.all(),
employee_departments__is_active=True,
departments_links__department__in=request_obj.departments.all(),
departments_links__is_active=True,
is_active=True
).exclude(id=request_obj.employee.id).distinct()
@ -183,9 +183,9 @@ def notify_new_request(request_obj):
if dept_ids:
dept_processors = Employee.objects.filter(
employee_departments__department_id__in=dept_ids,
employee_departments__is_active=True,
employee_departments__role__scoped_permissions__code='can_process_requests',
departments_links__department_id__in=dept_ids,
departments_links__is_active=True,
departments_links__role__scoped_permissions__code='can_process_requests',
is_active=True
).exclude(id=request_obj.employee.id).distinct()
@ -266,8 +266,8 @@ def notify_status_change(request_obj, old_status, new_status):
# 4. Если sent_to_all_department - все сотрудники отделов
if request_obj.sent_to_all_department:
dept_employees = Employee.objects.filter(
employee_departments__department__in=request_obj.departments.all(),
employee_departments__is_active=True,
departments_links__department__in=request_obj.departments.all(),
departments_links__is_active=True,
is_active=True
).exclude(id=request_obj.employee.id).distinct()

View File

@ -0,0 +1,43 @@
#!/usr/bin/env python
"""
Простое создание тестовых пользователей
"""
import os
import django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
django.setup()
from django.contrib.auth import get_user_model
User = get_user_model()
# Создаем 5 тестовых пользователей
test_users_data = [
('test1@local.dev', 'Тестовый', 'Первый', '+79001111111'),
('test2@local.dev', 'Тестовый', 'Второй', '+79002222222'),
('test3@local.dev', 'Тестовый', 'Третий', '+79003333333'),
('test4@local.dev', 'Тестовый', 'Четвертый', '+79004444444'),
('test5@local.dev', 'Тестовый', 'Пятый', '+79005555555'),
]
created = 0
for email, first_name, last_name, phone in test_users_data:
if not User.objects.filter(email=email).exists():
user = User.objects.create(
email=email,
first_name=first_name,
last_name=last_name,
phone_number=phone,
is_active=True,
email_verified=True
)
user.set_password('test123')
user.save()
print(f"✓ Создан: {user.get_full_name()} ({email})")
created += 1
else:
print(f"⚠ Уже существует: {email}")
print(f"\nСоздано новых пользователей: {created}")
print(f"Всего активных: {User.objects.filter(is_active=True).count()}")

310
backend/test_recipients.py Normal file
View File

@ -0,0 +1,310 @@
#!/usr/bin/env python
"""
Скрипт для тестирования функциональности получателей в заявлениях
"""
import os
import django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
django.setup()
from django.contrib.auth import get_user_model
from requests_app.models import Request
from employees.models import Department, EmployeeDepartment
from requests_app.enums import RequestStatus, RequestType
User = get_user_model()
def print_section(title):
print(f"\n{'='*60}")
print(f" {title}")
print(f"{'='*60}\n")
def test_model_fields():
"""Тест 1: Проверка новых полей модели"""
print_section("ТЕСТ 1: Проверка полей модели Request")
# Проверяем, что поля существуют
request = Request.objects.first()
if request:
print(f"✓ Найдена заявка: #{request.id} - {request.display_title}")
print(f" - departments (ManyToMany): {request.departments.count()} отделов")
print(f" - recipients (ManyToMany): {request.recipients.count()} получателей")
print(f" - cc_users (ManyToMany): {request.cc_users.count()} в копии")
print(f" - sent_to_all_department: {request.sent_to_all_department}")
print(f" - department (старое поле): {request.department}")
else:
print("⚠ Нет заявок в БД. Создадим тестовую...")
# Создаем тестовые данные
author = User.objects.filter(is_active=True).first()
if not author:
print("✗ Не найдено активных пользователей!")
return False
request = Request.objects.create(
employee=author,
type=RequestType.VACATION,
title="Тестовая заявка с получателями",
date_from="2025-12-10",
date_to="2025-12-20",
comment="Тест новой функциональности",
status=RequestStatus.PENDING
)
print(f"✓ Создана заявка #{request.id}")
return True
def test_add_recipients():
"""Тест 2: Добавление получателей"""
print_section("ТЕСТ 2: Добавление получателей")
# Находим или создаем заявку
request = Request.objects.first()
if not request:
print("✗ Нет заявок для тестирования")
return False
print(f"Работаем с заявкой #{request.id}")
# Получаем пользователей для теста
users = User.objects.filter(is_active=True).exclude(
id=request.employee_id
)[:3]
if len(users) < 2:
print("⚠ Недостаточно пользователей для теста")
return False
# Добавляем основных получателей
recipient1, recipient2 = users[0], users[1]
request.add_recipient(recipient1, is_cc=False)
request.add_recipient(recipient2, is_cc=False)
print(f"✓ Добавлены основные получатели:")
print(f" - {recipient1.get_full_name() or recipient1.username}")
print(f" - {recipient2.get_full_name() or recipient2.username}")
# Добавляем в копию
if len(users) >= 3:
cc_user = users[2]
request.add_recipient(cc_user, is_cc=True)
print(f"✓ Добавлен в копию:")
print(f" - {cc_user.get_full_name() or cc_user.username}")
# Проверяем количество
print(f"\nИтого:")
print(f" - Основных получателей: {request.recipients.count()}")
print(f" - В копии: {request.cc_users.count()}")
return True
def test_is_recipient():
"""Тест 3: Проверка метода is_recipient"""
print_section("ТЕСТ 3: Метод is_recipient()")
request = Request.objects.first()
if not request:
print("✗ Нет заявок для тестирования")
return False
print(f"Заявка #{request.id}")
# Проверяем автора
print(f"\n1. Автор заявки:")
print(f" {request.employee.get_full_name() or request.employee.username}")
print(f" is_recipient: {request.is_recipient(request.employee)}")
# Проверяем получателей
if request.recipients.exists():
recipient = request.recipients.first()
print(f"\n2. Основной получатель:")
print(f" {recipient.get_full_name() or recipient.username}")
print(f" is_recipient: ✓ {request.is_recipient(recipient)}")
# Проверяем CC
if request.cc_users.exists():
cc_user = request.cc_users.first()
print(f"\n3. Пользователь в копии:")
print(f" {cc_user.get_full_name() or cc_user.username}")
print(f" is_recipient: ✓ {request.is_recipient(cc_user)}")
# Проверяем случайного пользователя
random_user = User.objects.exclude(
id__in=[request.employee_id] +
list(request.recipients.values_list('id', flat=True)) +
list(request.cc_users.values_list('id', flat=True))
).first()
if random_user:
print(f"\n4. Случайный пользователь:")
print(f" {random_user.get_full_name() or random_user.username}")
print(f" is_recipient: {request.is_recipient(random_user)}")
return True
def test_departments():
"""Тест 4: Работа с несколькими отделами"""
print_section("ТЕСТ 4: Множественные отделы")
request = Request.objects.first()
if not request:
print("✗ Нет заявок для тестирования")
return False
# Получаем отделы
departments = Department.objects.all()[:2]
if len(departments) < 2:
print("⚠ Недостаточно отделов для теста")
if len(departments) == 1:
request.departments.add(departments[0])
print(f"✓ Добавлен 1 отдел: {departments[0].name}")
else:
request.departments.set(departments)
print(f"✓ Добавлено {len(departments)} отделов:")
for dept in departments:
print(f" - {dept.name}")
print(f"\nВсего отделов: {request.departments.count()}")
return True
def test_sent_to_all():
"""Тест 5: Флаг sent_to_all_department"""
print_section("ТЕСТ 5: Флаг sent_to_all_department")
request = Request.objects.first()
if not request:
print("✗ Нет заявок для тестирования")
return False
# Включаем флаг
request.sent_to_all_department = True
request.save()
print(f"✓ Флаг sent_to_all_department установлен в True")
# Проверяем, что есть отделы
dept_count = request.departments.count()
print(f" Выбрано отделов: {dept_count}")
if dept_count > 0:
# Считаем сотрудников в этих отделах
dept_employees = User.objects.filter(
departments_links__department__in=request.departments.all(),
departments_links__is_active=True,
is_active=True
).distinct().count()
print(f" Сотрудников в отделах: {dept_employees}")
print(f" Они все теперь видят эту заявку!")
return True
def test_all_recipients_property():
"""Тест 6: Свойство all_recipients"""
print_section("ТЕСТ 6: Свойство all_recipients")
request = Request.objects.first()
if not request:
print("✗ Нет заявок для тестирования")
return False
all_recip = request.all_recipients
primary = request.primary_recipients
print(f"Всего получателей (recipients + cc): {all_recip.count()}")
print(f"Основных получателей: {primary.count()}")
print(f"В копии: {request.cc_users.count()}")
if all_recip.exists():
print("\nСписок всех получателей:")
for user in all_recip:
is_primary = request.recipients.filter(id=user.id).exists()
is_cc = request.cc_users.filter(id=user.id).exists()
role = "основной" if is_primary else "CC"
print(f" - {user.get_full_name() or user.username} ({role})")
return True
def test_api_serializer():
"""Тест 7: Проверка сериализатора"""
print_section("ТЕСТ 7: Сериализация через API")
from api.v1.requests_app.serializers import RequestReadSerializer
from rest_framework.request import Request as DRFRequest
from django.test import RequestFactory
request_obj = Request.objects.first()
if not request_obj:
print("✗ Нет заявок для тестирования")
return False
# Создаем mock request
factory = RequestFactory()
django_request = factory.get('/')
django_request.user = request_obj.employee
drf_request = DRFRequest(django_request)
# Сериализуем
serializer = RequestReadSerializer(
request_obj,
context={'request': drf_request}
)
data = serializer.data
print(f"Заявка #{data['id']}:")
print(f" - departments: {len(data.get('departments', []))}")
print(f" - recipients: {len(data.get('recipients', []))}")
print(f" - cc_users: {len(data.get('cc_users', []))}")
print(f" - recipient_count: {data.get('recipient_count', 0)}")
print(f" - cc_count: {data.get('cc_count', 0)}")
print(f" - is_recipient: {data.get('is_recipient', False)}")
print(f" - sent_to_all_department: {data.get('sent_to_all_department', False)}")
return True
def run_all_tests():
"""Запуск всех тестов"""
print("\n" + "="*60)
print(" 🧪 ТЕСТИРОВАНИЕ ФУНКЦИОНАЛЬНОСТИ ПОЛУЧАТЕЛЕЙ")
print("="*60)
tests = [
("Проверка полей модели", test_model_fields),
("Добавление получателей", test_add_recipients),
("Метод is_recipient", test_is_recipient),
("Множественные отделы", test_departments),
("Флаг sent_to_all_department", test_sent_to_all),
("Свойство all_recipients", test_all_recipients_property),
("Сериализация API", test_api_serializer),
]
results = []
for name, test_func in tests:
try:
result = test_func()
results.append((name, result))
except Exception as e:
print(f"\n✗ ОШИБКА в тесте '{name}':")
print(f" {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
results.append((name, False))
# Итоги
print_section("ИТОГИ ТЕСТИРОВАНИЯ")
passed = sum(1 for _, r in results if r)
total = len(results)
for name, result in results:
status = "✓ PASSED" if result else "✗ FAILED"
print(f"{status:12} | {name}")
print(f"\n{'='*60}")
print(f"Пройдено: {passed}/{total} тестов ({passed*100//total}%)")
print(f"{'='*60}\n")
return passed == total
if __name__ == '__main__':
success = run_all_tests()
exit(0 if success else 1)

View File

@ -0,0 +1,247 @@
#!/usr/bin/env python
"""
Тестирование API получателей заявлений
"""
import os
import django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
django.setup()
import json
from django.contrib.auth import get_user_model
from django.test import RequestFactory
from rest_framework.test import force_authenticate
from api.v1.requests_app.views import RequestViewSet
from api.v1.requests_app.serializers import RequestWriteSerializer
from requests_app.models import Request
from employees.models import Department
User = get_user_model()
def test_create_request_with_recipients():
"""Тест создания заявки с получателями через API"""
print("\n" + "="*60)
print(" ТЕСТ: Создание заявки с получателями")
print("="*60)
# Получаем пользователей
author = User.objects.filter(is_active=True).first()
users = User.objects.filter(is_active=True).exclude(id=author.id)[:3]
dept = Department.objects.first()
if not author or len(users) < 2 or not dept:
print("✗ Недостаточно данных для теста")
return False
recipient1, recipient2 = users[0], users[1]
cc_user = users[2] if len(users) > 2 else None
# Данные для создания
data = {
'type': 'VACATION',
'title': 'Отпуск с API тестом получателей',
'date_from': '2025-12-10',
'date_to': '2025-12-20',
'comment': 'Тест API',
'department_ids': [dept.id],
'recipient_ids': [recipient1.id, recipient2.id],
'sent_to_all_department': False
}
if cc_user:
data['cc_user_ids'] = [cc_user.id]
print(f"\nДанные запроса:")
print(f" - author: {author.get_full_name() or author.username}")
print(f" - recipients: {recipient1.get_full_name()}, {recipient2.get_full_name()}")
if cc_user:
print(f" - cc: {cc_user.get_full_name()}")
print(f" - department: {dept.name}")
# Создаем mock request
factory = RequestFactory()
request = factory.post('/api/v1/requests/', data, format='json')
force_authenticate(request, user=author)
# Сериализация
serializer = RequestWriteSerializer(
data=data,
context={'request': request}
)
if not serializer.is_valid():
print(f"\n✗ Ошибки валидации: {serializer.errors}")
return False
# Создание объекта
request_obj = serializer.save(employee=author)
print(f"\n✓ Заявка создана: #{request_obj.id}")
print(f" - recipients: {request_obj.recipients.count()}")
print(f" - cc_users: {request_obj.cc_users.count()}")
print(f" - departments: {request_obj.departments.count()}")
# Проверяем правильность
assert request_obj.recipients.count() == 2
assert request_obj.is_recipient(recipient1)
assert request_obj.is_recipient(recipient2)
if cc_user:
assert request_obj.cc_users.count() == 1
assert request_obj.is_recipient(cc_user)
print("\n✓ Все проверки прошли успешно!")
return True
def test_queryset_addressed_to_me():
"""Тест фильтрации ?addressed_to_me=true"""
print("\n" + "="*60)
print(" ТЕСТ: Фильтр ?addressed_to_me=true")
print("="*60)
# Находим заявку с получателями
request_obj = Request.objects.filter(
recipients__isnull=False
).distinct().first()
if not request_obj:
print("⚠ Нет заявок с получателями")
return False
recipient = request_obj.recipients.first()
if not recipient:
print("⚠ Нет получателей")
return False
print(f"\nЗаявка: #{request_obj.id}")
print(f"Получатель: {recipient.get_full_name() or recipient.username}")
# Mock request
factory = RequestFactory()
django_request = factory.get('/api/v1/requests/?addressed_to_me=true')
force_authenticate(django_request, user=recipient)
# ViewSet
viewset = RequestViewSet()
viewset.request = django_request
viewset.action = 'list'
viewset.format_kwarg = None
# Получаем queryset
qs = viewset.get_queryset()
print(f"\nКоличество заявок для пользователя: {qs.count()}")
# Проверяем, что наша заявка в выборке
if qs.filter(id=request_obj.id).exists():
print("✓ Заявка найдена в выборке для получателя")
return True
else:
print("✗ Заявка НЕ найдена в выборке")
return False
def test_update_recipients():
"""Тест обновления получателей"""
print("\n" + "="*60)
print(" ТЕСТ: Обновление получателей")
print("="*60)
request_obj = Request.objects.first()
if not request_obj:
print("✗ Нет заявок")
return False
users = User.objects.filter(is_active=True).exclude(
id=request_obj.employee_id
)[:4]
if len(users) < 3:
print("⚠ Недостаточно пользователей")
return False
print(f"\nЗаявка: #{request_obj.id}")
print(f"Было получателей: {request_obj.recipients.count()}")
print(f"Было CC: {request_obj.cc_users.count()}")
# Новые данные
new_recipients = [users[0].id, users[1].id, users[2].id]
new_cc = [users[3].id] if len(users) > 3 else []
data = {
'recipient_ids': new_recipients,
'cc_user_ids': new_cc,
}
# Mock request
factory = RequestFactory()
django_request = factory.patch(
f'/api/v1/requests/{request_obj.id}/',
data,
format='json'
)
force_authenticate(django_request, user=request_obj.employee)
# Сериализация
serializer = RequestWriteSerializer(
request_obj,
data=data,
partial=True,
context={'request': django_request}
)
if not serializer.is_valid():
print(f"\n✗ Ошибки: {serializer.errors}")
return False
updated_obj = serializer.save()
print(f"\n✓ Обновлено:")
print(f" - recipients: {updated_obj.recipients.count()}")
print(f" - cc_users: {updated_obj.cc_users.count()}")
assert updated_obj.recipients.count() == 3
if new_cc:
assert updated_obj.cc_users.count() == 1
print("✓ Все проверки прошли!")
return True
if __name__ == '__main__':
print("\n" + "="*60)
print(" 🧪 ТЕСТИРОВАНИЕ API ПОЛУЧАТЕЛЕЙ")
print("="*60)
tests = [
("Создание с получателями", test_create_request_with_recipients),
("Фильтр addressed_to_me", test_queryset_addressed_to_me),
("Обновление получателей", test_update_recipients),
]
results = []
for name, test_func in tests:
try:
result = test_func()
results.append((name, result))
except Exception as e:
print(f"\n✗ ОШИБКА: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
results.append((name, False))
# Итоги
print("\n" + "="*60)
print(" ИТОГИ")
print("="*60)
for name, result in results:
status = "✓ PASSED" if result else "✗ FAILED"
print(f"{status:12} | {name}")
passed = sum(1 for _, r in results if r)
total = len(results)
print(f"\nПройдено: {passed}/{total}")