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

поудалял лишнего

Browse Source
This commit is contained in:
lizerk 2026-03-10 13:18:57 +03:00
parent 5dce32c303
commit 258ff58df1
18 changed files with 0 additions and 3170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
backend/employees/authentication.py
View File

@ -1,30 +0,0 @@
# backend\employees\authentication.py
from django.contrib.auth.backends import ModelBackend
from .models import Employee
class EmailOrPhoneBackend(ModelBackend):
def authenticate(self, request, username=None, password=None, **kwargs):
if username is None:
username = kwargs.get("username")
try:
# Проверяем, email ли это
if "@" in username:
user = Employee.objects.get(email=username)
else:
# Иначе phone
user = Employee.objects.get(phone_number=username)
except Employee.DoesNotExist:
return None
if user.check_password(password) and self.user_can_authenticate(user):
return user
return None
def get_user(self, user_id):
try:
return Employee.objects.get(pk=user_id)
except Employee.DoesNotExist:
return None

53
backend/employees/constants.py.orig
View File

@ -1,53 +0,0 @@
# backend\employees\constants.py
GENDER_CHOICES = (
(1, "Мужской"),
(2, "Женский"),
(0, "Не указан"),
)
ACTION_HIRED = "hired"
ACTION_DISMISSED = "dismissed"
ACTION_ON_LEAVE = "on_leave"
ACTION_RETURNED_FROM_LEAVE = "returned_from_leave"
ACTION_ON_MATERNITY = "on_maternity"
ACTION_RETURNED_FROM_MATERNITY = "returned_from_maternity"
ACTION_TRANSFERRED = "transferred"
ACTION_REHIRED = "rehired"
ACTION_CHOICES = [
(ACTION_HIRED, "Принят"),
(ACTION_DISMISSED, "Уволен"),
(ACTION_ON_LEAVE, "В отпуске"),
(ACTION_RETURNED_FROM_LEAVE, "Вернулся из отпуска"),
(ACTION_ON_MATERNITY, "В декрете"),
(ACTION_RETURNED_FROM_MATERNITY, "Вернулся из декрета"),
(ACTION_TRANSFERRED, "Переведен"),
(ACTION_REHIRED, "Восстановлен"),
]
class DeptPerm:
MANAGE = "manage_department"
CHANGE_HEAD = "change_department_head"
ASSIGN_ROLE = "assign_department_role"
MANAGE_CALENDAR = "manage_department_events"
CREATE_POST = "publish_department_post"
MANAGE_FEED = "manage_department_feed"
VIEW_REQUESTCOMMENT = "view_requestcomment"
ADD_REQUESTCOMMENT = "add_requestcomment"
CHOICES = (
(MANAGE, "Управлять отделом"),
(CHANGE_HEAD, "Назначать руководителя"),
(ASSIGN_ROLE, "Назначать роли участникам"),
(MANAGE_CALENDAR, "Управлять календарём отдела"),
(CREATE_POST, "Публиковать новости на странице отдела"),
(MANAGE_FEED, "Редактировать публикации отдела"),
(VIEW_REQUESTCOMMENT, "Просмотр комментариев по заявлениям"),
(ADD_REQUESTCOMMENT, "Добавление коментариев по заявлениям"),
<<<<<<< HEAD
)
=======
)
>>>>>>> origin/feat/ldap-writeback

386
backend/employees/forms_auth.py
View File

@ -1,386 +0,0 @@
# backend/employees/forms_auth.py
from __future__ import annotations
import base64
import logging
import re
from typing import Any, Dict, List, Optional
from django import forms
from django.contrib.auth.forms import AuthenticationForm, PasswordResetForm
from django.core.files.uploadedfile import UploadedFile
from PIL import Image
logger = logging.getLogger(__name__)
class EmailOrPhoneAuthenticationForm(AuthenticationForm):
username = forms.CharField(
label="Email или телефон",
widget=forms.TextInput(
attrs={
"autofocus": True,
"placeholder": "user@example.com или +7 999 123-45-67",
}
),
)
password = forms.CharField(
label="Пароль",
strip=False,
widget=forms.PasswordInput(attrs={"autocomplete": "current-password"}),
)
class RegistrationForm(forms.Form):
"""Форма регистрации для POST /api/v1/auth/register/.
Валидирует обязательные поля, совпадение паролей и наличие хотя бы одного контактного способа.
Метод :meth:`to_api_payload` готовит JSON-совместимый payload под ваш DRF-сериализатор,
включая аватар, закодированный как data URI (base64).
Notes:
Вызывайте :meth:`to_api_payload` только после успешной валидации (`form.is_valid()`).
Raises:
ValidationError: Бросается не напрямую, а через `self.add_error(...)` — форма станет невалидной.
"""
# Обязательные поля
first_name = forms.CharField(max_length=150, label="Имя")
last_name = forms.CharField(max_length=150, label="Фамилия")
phone_number = forms.CharField(
max_length=17, # + и до 15 цифр (с запасом)
label="Номер телефона",
widget=forms.TextInput(attrs={
"type": "tel",
"inputmode": "numeric",
"autocomplete": "tel",
"title": "Введите номер в формате +79991234567",
})
)
email = forms.EmailField(label="Email")
birth_date = forms.DateField(
label="Дата рождения", widget=forms.DateInput(attrs={"type": "date"})
)
# Пароли как в шаблоне (password1/password2)
password1 = forms.CharField(label="Пароль", widget=forms.PasswordInput)
password2 = forms.CharField(label="Повторите пароль", widget=forms.PasswordInput)
# Контакты: нужен хотя бы один
telegram = forms.CharField(required=False, label="Telegram")
whatsapp = forms.CharField(required=False, label="WhatsApp")
wechat = forms.CharField(required=False, label="WeChat")
# Опциональные поля модели
patronymic = forms.CharField(required=False, label="Отчество")
gender = forms.ChoiceField(
required=True,
label="Пол",
choices=[(1, "Мужской"), (2, "Женский")],
widget=forms.RadioSelect,
error_messages={
'required': 'Необходимо указать пол.',
'invalid_choice': 'Выберите Мужской или Женский.',
}
)
position = forms.IntegerField(required=False, label="ID должности")
# skills — список ID через запятую
skills = forms.CharField(required=False, label="Навыки (ID через запятую)")
avatar = forms.ImageField(required=True, label="Аватар")
def clean_first_name(self) -> str:
"""Валидация имени: только буквы, пробелы и дефисы."""
value = self.cleaned_data.get("first_name", "").strip()
if not value:
raise forms.ValidationError("Имя обязательно для заполнения.")
if re.search(r'\d', value):
raise forms.ValidationError(
"Имя не должно содержать цифры."
)
# Разрешаем буквы (любых алфавитов), пробелы, дефисы, апострофы
if not re.match(r'^[\w\s\-\']+$', value, re.UNICODE):
raise forms.ValidationError(
"Имя может содержать только буквы, пробелы, дефисы и апострофы."
)
return value
def clean_last_name(self) -> str:
"""Валидация фамилии: только буквы, пробелы и дефисы."""
value = self.cleaned_data.get("last_name", "").strip()
if not value:
raise forms.ValidationError("Фамилия обязательна для заполнения.")
if re.search(r'\d', value):
raise forms.ValidationError(
"Фамилия не должна содержать цифры."
)
if not re.match(r'^[\w\s\-\']+$', value, re.UNICODE):
raise forms.ValidationError(
"Фамилия может содержать только буквы, пробелы, дефисы и апострофы."
)
return value
def clean_patronymic(self) -> str:
"""Валидация отчества: только буквы, пробелы и дефисы."""
value = self.cleaned_data.get("patronymic", "").strip()
if not value:
return value # Отчество опциональное
if re.search(r'\d', value):
raise forms.ValidationError(
"Отчество не должно содержать цифры."
)
if not re.match(r'^[\w\s\-\']+$', value, re.UNICODE):
raise forms.ValidationError(
"Отчество может содержать только буквы, пробелы, дефисы и апострофы."
)
return value
def clean(self) -> Dict[str, Any]:
"""Глобальная валидация формы.
Проверяет:
- наличие хотя бы одного контактного поля (Telegram/WhatsApp/WeChat);
- корректность паролей (оба заполнены и совпадают).
Returns:
dict: Очищенные данные формы.
"""
data = super().clean()
# хотя бы один контакт
if not (
(data.get("telegram") or "").strip()
or (data.get("whatsapp") or "").strip()
or (data.get("wechat") or "").strip()
):
self.add_error(
None,
"Заполните хотя бы одно из полей: WhatsApp, WeChat или Telegram",
)
# пароли
p1 = (data.get("password1") or "").strip()
p2 = (data.get("password2") or "").strip()
if not p1:
self.add_error("password1", "Введите пароль.")
if not p2:
self.add_error("password2", "Подтвердите пароль.")
if p1 and p2 and p1 != p2:
self.add_error("password2", "Пароли не совпадают.")
return data
def skills_ids(self) -> List[int]:
"""Возвращает список ID навыков из текстового поля.
Returns:
list[int]: Список целых ID. Пустой, если поле пустое или состоит из мусора.
"""
raw = (self.cleaned_data.get("skills") or "").strip()
if not raw:
return []
out: List[int] = []
for chunk in raw.split(","):
s = chunk.strip()
if not s:
continue
try:
out.append(int(s))
except ValueError:
# игнорируем мусор
continue
return out
# ---------- Вспомогательные методы для аватара ----------
@staticmethod
def _guess_mime(upload: UploadedFile, content: bytes) -> str:
"""Определяет MIME-тип изображения.
Args:
upload: Загруженный файл (может содержать content_type).
content: Байты файла.
Returns:
str: MIME-тип, например 'image/jpeg'.
"""
# 1) используем content_type, если его дал браузер/клиент
ct = getattr(upload, "content_type", None)
if ct and ct.startswith("image/"):
return ct
# 2) пробуем Pillow
if Image is not None:
try:
im = Image.open(upload) # type: ignore[arg-type]
fmt = (im.format or "").upper()
mime = Image.MIME.get(fmt)
if mime:
# откатим указатель, чтобы не ломать повторные чтения
try:
upload.seek(0)
except Exception:
pass
return mime
except Exception:
try:
upload.seek(0)
except Exception:
pass
# 3) дефолт
return "image/jpeg"
@staticmethod
def _file_to_data_uri(upload: UploadedFile) -> Optional[str]:
"""Кодирует изображение в data URI (base64).
Args:
upload: Загруженный файл.
Returns:
Optional[str]: Строка вида 'data:<mime>;base64,<payload>' или None, если нет данных.
Raises:
ValueError: Если файл нельзя прочитать.
"""
if not upload:
return None
try:
content: bytes = upload.read()
try:
upload.seek(0)
except Exception:
pass
except Exception as exc:
raise ValueError("Не удалось прочитать файл аватара.") from exc
if not content:
return None
mime = RegistrationForm._guess_mime(upload, content)
b64 = base64.b64encode(content).decode("ascii")
return f"data:{mime};base64,{b64}"
# ---------- Публичные методы ----------
def to_api_payload(self) -> Dict[str, Any]:
"""Готовит JSON payload для POST `/api/v1/auth/register/`.
Включает:
- все обязательные поля;
- контакты (как пустые строки, если не указаны);
- дополнительные поля (если заданы);
- `avatar` как data URI (base64), если файл загружен.
Returns:
dict[str, Any]: Готовый словарь.
Raises:
KeyError: Если метод вызван до `is_valid()` и ключей нет в `cleaned_data`.
ValueError: При проблеме чтения файла аватара.
"""
cd = self.cleaned_data # предполагается, что форма валидна
payload: Dict[str, Any] = {
"first_name": cd["first_name"],
"last_name": cd["last_name"],
"phone_number": cd["phone_number"],
"email": cd["email"],
"birth_date": cd["birth_date"].isoformat(),
"password": cd["password1"],
"gender": int(cd["gender"]), # обязательное поле
"telegram": cd.get("telegram") or "",
"whatsapp": cd.get("whatsapp") or "",
"wechat": cd.get("wechat") or "",
}
# Доп. поля
if cd.get("patronymic"):
payload["patronymic"] = cd["patronymic"]
if cd.get("position") is not None:
payload["position"] = cd["position"]
ids = self.skills_ids()
if ids:
payload["skills"] = ids
# Аватар → base64 data URI (совместимо с Base64ImageField)
avatar_file = cd.get("avatar")
if avatar_file:
data_uri = self._file_to_data_uri(avatar_file)
if data_uri:
payload["avatar"] = data_uri
return payload
class VerifyEmailForm(forms.Form):
email = forms.EmailField(label="Email")
code = forms.CharField(max_length=32, label="Код из письма")
class CustomPasswordResetForm(PasswordResetForm):
"""Кастомная форма восстановления пароля с логирование"""
def save(self, *args, **kwargs):
"""Переопределяем save для логирования процесса отправки"""
email = self.cleaned_data["email"]
logger.info(
f"\n{'='*80}\n"
f"[CustomPasswordResetForm.save] 🔐 СОХРАНЕНИЕ ФОРМЫ ВОССТАНОВЛЕНИЯ ПАРОЛЯ\n"
f" Email: {email}\n"
f"{'='*80}"
)
logger.info(
f"[CustomPasswordResetForm.save] ➡️ Вызов super().save() для отправки emails..."
)
try:
result = super().save(*args, **kwargs)
logger.info(
f"[CustomPasswordResetForm.save] ✅ super().save() успешно завершен"
)
return result
except Exception as e:
logger.error(
f"[CustomPasswordResetForm.save] ❌ Ошибка в super().save(): {type(e).__name__}: {e}",
exc_info=True
)
raise
def send_mail(self, subject_template_name, email_template_name, context,
from_email, to_email, html_email_template_name=None, **kwargs):
"""Переопределяем send_mail для логирования отправки каждого письма"""
logger.info(
f"\n{'='*80}\n"
f"[CustomPasswordResetForm.send_mail] 📧 ОТПРАВКА EMAIL\n"
f" To: {to_email}\n"
f" Subject template: {subject_template_name}\n"
f" Email template: {email_template_name}\n"
f" HTML template: {html_email_template_name}\n"
f"{'='*80}"
)
try:
# Вызываем родительский метод
result = super().send_mail(
subject_template_name=subject_template_name,
email_template_name=email_template_name,
context=context,
from_email=from_email,
to_email=to_email,
html_email_template_name=html_email_template_name,
**kwargs
)
logger.info(
f"[CustomPasswordResetForm.send_mail] ✅ EMAIL УСПЕШНО ОТПРАВЛЕН на {to_email}"
)
return result
except Exception as e:
logger.error(
f"[CustomPasswordResetForm.send_mail] ❌ ОШИБКА ОТПРАВКИ EMAIL: {type(e).__name__}: {e}",
exc_info=True
)
raise

58
backend/employees/forms_front.py
View File

@ -1,58 +0,0 @@
# backend\employees\forms_front.py
from django import forms
from .models import Department, Employee
class DepartmentEditForm(forms.Form):
name = forms.CharField(max_length=255, label="Название")
description = forms.CharField(
required=False, widget=forms.Textarea, label="Описание"
)
class SetHeadForm(forms.Form):
# просто номер id; мы не тянем ORM
head_id = forms.IntegerField(
required=False, label="ID руководителя (пусто — снять)"
)
class SetMemberRoleForm(forms.Form):
employee_id = forms.IntegerField(label="ID сотрудника")
role_id = forms.IntegerField(required=False, label="ID роли (пусто — снять)")
is_active = forms.BooleanField(required=False, label="Активен в отделе")
class DepartmentForm(forms.ModelForm):
class Meta:
model = Department
fields = ["name", "description", "head"]
class EmployeeForm(forms.ModelForm):
class Meta:
model = Employee
fields = (
"first_name",
"last_name",
"patronymic",
"telegram",
"whatsapp",
"wechat",
)
widgets = {
"first_name": forms.TextInput(attrs={"autocomplete": "given-name"}),
"last_name": forms.TextInput(attrs={"autocomplete": "family-name"}),
"whatsapp": forms.TextInput(attrs={"inputmode": "tel"}),
"telegram": forms.TextInput(),
"wechat": forms.TextInput(),
}
labels = {
"first_name": "Имя",
"last_name": "Фамилия",
"patronymic": "Отчество",
"telegram": "Telegram",
"whatsapp": "WhatsApp",
"wechat": "WeChat",
}

3
backend/employees/tests/__init__.py
View File

@ -1,3 +0,0 @@
"""
Тесты для employees приложения.
"""

332
backend/employees/tests/test_birthday_services.py
View File

@ -1,332 +0,0 @@
"""
Тесты для сервисного слоя работы с днями рождения (Service Layer Pattern).
Используется django-service-objects для инкапсуляции бизнес-логики.
"""
import pytest
from datetime import date
from django.contrib.auth import get_user_model
from django.db.models import signals
from django.utils import timezone
from employees.models import Employee
from employees.services import (
UpsertBirthdayEventService,
DeleteBirthdayEventService,
BulkSyncBirthdaysService
)
from schedule.models import Calendar, Event, Rule
User = get_user_model()
@pytest.fixture(autouse=True)
def disable_birthday_signals(settings):
"""Отключаем сигналы дней рождений для изолированного тестирования сервисов."""
from unittest.mock import patch
# Мокируем только функции-обработчики сигналов, а не сами сервисы
with patch('employees.signals_birthday.sync_birthday_event_on_employee_save'):
with patch('employees.signals_birthday.delete_birthday_event_on_employee_delete'):
yield
@pytest.fixture
def create_test_employee():
"""Фикстура для создания тестовых сотрудников с валидными данными."""
employee_counter = 0
def _create_employee(**kwargs):
nonlocal employee_counter
employee_counter += 1
defaults = {
'username': f'testuser{employee_counter}',
'email': f'testuser{employee_counter}@example.com',
'phone_number': f'+7900000{employee_counter:04d}',
'password': 'test123',
'first_name': 'Test',
'last_name': f'User{employee_counter}'
}
defaults.update(kwargs)
# Employee наследуется от AbstractUser
employee = Employee.objects.create_user(**defaults)
return employee
return _create_employee
@pytest.mark.django_db
class TestUpsertBirthdayEventService:
"""Тесты для сервиса создания/обновления событий дня рождения."""
def test_create_birthday_event_for_new_employee(self, create_test_employee):
"""Создание нового события дня рождения."""
# Arrange
employee = create_test_employee(
username='john',
email='john@example.com',
first_name='John',
last_name='Doe',
birth_date=date(1990, 5, 15)
)
# Act
result = UpsertBirthdayEventService.execute({'employee': employee})
# Assert
assert result['success'] is True
assert result['created'] is True
assert result['event'] is not None
event = result['event']
assert event.title == '🎂 День рождения: John Doe'
assert event.creator_id == employee.pk
assert event.rule is not None
assert event.rule.frequency == 'YEARLY'
assert event.end_recurring_period is None # Бесконечное повторение
def test_update_existing_birthday_event(self, create_test_employee):
"""Обновление существующего события при изменении данных."""
# Arrange
employee = create_test_employee(
username='jane',
email='jane@example.com',
first_name='Jane',
last_name='Smith',
birth_date=date(1985, 3, 20)
)
# Создаем событие
first_result = UpsertBirthdayEventService.execute({'employee': employee})
first_event_id = first_result['event'].id
# Act - меняем имя сотрудника
employee.first_name = 'Janet'
employee.save()
second_result = UpsertBirthdayEventService.execute({'employee': employee})
# Assert
assert second_result['success'] is True
assert second_result['created'] is False # Обновлено, не создано
assert second_result['event'].id == first_event_id # Тот же event
assert second_result['event'].title == '🎂 День рождения: Janet Smith'
def test_skip_employee_without_birth_date(self, create_test_employee):
"""Пропуск сотрудника без даты рождения."""
# Arrange
employee = create_test_employee(
username='nobirthday',
email='nobirthday@example.com',
first_name='No',
last_name='Birthday',
birth_date=None
)
# Act
result = UpsertBirthdayEventService.execute({'employee': employee})
# Assert
assert result['success'] is False
assert result['reason'] == 'no_birth_date'
assert result['event'] is None
def test_creates_personal_calendar_if_not_exists(self, create_test_employee):
"""Автоматически создает личный календарь при первом событии."""
# Arrange
employee = create_test_employee(
username='newuser',
email='newuser@example.com',
first_name='New',
last_name='User',
birth_date=date(1992, 7, 10)
)
# Убедимся, что календаря нет
assert not Calendar.objects.filter(slug=f'employee-{employee.pk}').exists()
# Act
result = UpsertBirthdayEventService.execute({'employee': employee})
# Assert
assert result['success'] is True
calendar = Calendar.objects.get(slug=f'employee-{employee.pk}')
assert calendar.name == '👤 Мой календарь'
assert result['event'].calendar == calendar
def test_yearly_rule_reused_across_employees(self, create_test_employee):
"""Правило 'Ежегодно' переиспользуется для всех событий."""
# Arrange
employee1 = create_test_employee(
username='user1',
email='user1@example.com',
first_name='User',
last_name='One',
birth_date=date(1990, 1, 1)
)
employee2 = create_test_employee(
username='user2',
email='user2@example.com',
first_name='User',
last_name='Two',
birth_date=date(1991, 2, 2)
)
# Act
result1 = UpsertBirthdayEventService.execute({'employee': employee1})
result2 = UpsertBirthdayEventService.execute({'employee': employee2})
# Assert
assert result1['event'].rule.id == result2['event'].rule.id
assert Rule.objects.filter(name='Ежегодно').count() == 1
@pytest.mark.django_db
class TestDeleteBirthdayEventService:
"""Тесты для сервиса удаления событий дня рождения."""
def test_delete_existing_birthday_event(self, create_test_employee):
"""Удаление существующего события."""
# Arrange
employee = create_test_employee(
username='todelete',
email='todelete@example.com',
first_name='To',
last_name='Delete',
birth_date=date(1988, 4, 25)
)
# Создаем событие
UpsertBirthdayEventService.execute({'employee': employee})
assert Event.objects.filter(creator_id=employee.pk).exists()
# Act
result = DeleteBirthdayEventService.execute({'employee': employee})
# Assert
assert result['success'] is True
assert result['deleted_count'] == 1
assert not Event.objects.filter(creator_id=employee.pk).exists()
def test_delete_nonexistent_event_returns_false(self, create_test_employee):
"""Попытка удалить несуществующее событие."""
# Arrange
employee = create_test_employee(
username='noevents',
email='noevents@example.com',
first_name='No',
last_name='Events',
birth_date=date(1995, 8, 30)
)
# Act
result = DeleteBirthdayEventService.execute({'employee': employee})
# Assert
assert result['success'] is False
assert result['deleted_count'] == 0
@pytest.mark.django_db
class TestBulkSyncBirthdaysService:
"""Тесты для массовой синхронизации дней рождений."""
def test_bulk_sync_creates_events_for_all_employees(self, create_test_employee):
"""Массовая синхронизация создает события для всех сотрудников."""
# Arrange
employees_data = [
('user1', 'Alice', 'Johnson', date(1990, 1, 10)),
('user2', 'Bob', 'Williams', date(1985, 5, 20)),
('user3', 'Charlie', 'Brown', date(1992, 9, 15)),
]
for username, first, last, birth in employees_data:
create_test_employee(
username=username,
email=f'{username}@example.com',
first_name=first,
last_name=last,
birth_date=birth
)
# Act
result = BulkSyncBirthdaysService.execute({})
# Assert
assert result['total'] == 3
assert result['created'] == 3
assert result['updated'] == 0
assert result['skipped'] == 0
assert len(result['errors']) == 0
# Проверяем, что события созданы
assert Event.objects.filter(title__startswith='🎂 День рождения:').count() == 3
def test_bulk_sync_updates_existing_events(self, create_test_employee):
"""Массовая синхронизация обновляет существующие события."""
# Arrange
employee = create_test_employee(
username='existing',
email='existing@example.com',
first_name='Existing',
last_name='User',
birth_date=date(1987, 6, 12)
)
# Создаем событие заранее
UpsertBirthdayEventService.execute({'employee': employee})
# Act
result = BulkSyncBirthdaysService.execute({})
# Assert
assert result['total'] == 1
assert result['created'] == 0
assert result['updated'] == 1
assert result['skipped'] == 0
def test_bulk_sync_skips_employees_without_birth_date(self, create_test_employee):
"""Пропускает сотрудников без даты рождения."""
# Arrange
create_test_employee(
username='withbd',
email='withbd@example.com',
first_name='With',
last_name='BirthDate',
birth_date=date(1990, 1, 1)
)
create_test_employee(
username='nobd',
email='nobd@example.com',
first_name='No',
last_name='BirthDate',
birth_date=None
)
# Act
result = BulkSyncBirthdaysService.execute({})
# Assert
assert result['total'] == 2
assert result['created'] == 1
assert result['skipped'] == 1
def test_bulk_sync_handles_errors_gracefully(self, create_test_employee):
"""Обрабатывает ошибки без падения всего процесса."""
# Arrange
create_test_employee(
username='valid',
email='valid@example.com',
first_name='Valid',
last_name='Employee',
birth_date=date(1993, 3, 5)
)
# Act
result = BulkSyncBirthdaysService.execute({})
# Assert
assert result['total'] >= 1
assert result['created'] + result['updated'] >= 1
# Часть может быть успешной даже если есть ошибки

85
backend/employees/urls_front.py
View File

@ -1,85 +0,0 @@
# backend/employees/urls_front.py
from django.urls import path
from . import views_front
app_name = "employees"
urlpatterns = [
# --- Профиль и сотрудники ---
path("list/", views_front.employee_list, name="employees_list"),
path("me/", views_front.employee_profile, name="profile"),
path("me/edit/", views_front.employee_edit_me, name="employee_edit_me"),
path("<int:pk>/", views_front.employee_profile, name="employee_detail"),
path("<int:pk>/edit/", views_front.employee_edit, name="employee_edit"),
path("create/", views_front.employee_create, name="employee_create"),
path(
"create/modal/",
views_front.employee_create_modal,
name="employee_create_modal"
),
# --- Группы ---
path("employees/<int:pk>/groups/bulk/", views_front.employee_groups_bulk, name="employee_groups_bulk"),
path("groups/create/", views_front.group_create, name="group_create"),
path("groups/delete/", views_front.group_delete, name="group_delete"),
# --- Отделы ---
path("departments/create/", views_front.department_create, name="department_create"),
path(
"departments/<int:pk>/roles/edit/",
views_front.edit_department_role,
name="edit_department_role",
),
path("departments/", views_front.department_list, name="department_list"),
path(
"departments/<int:pk>/", views_front.department_detail, name="department_detail"
),
path(
"departments/<int:pk>/edit/",
views_front.department_edit,
name="department_edit",
),
path(
"departments/<int:pk>/set-head/",
views_front.department_set_head,
name="department_set_head",
),
path(
"<int:emp_id>/departments/<int:dept_id>/set-role/",
views_front.set_member_role,
name="set_member_role",
),
path(
"departments/<int:pk>/add-member/",
views_front.department_add_member,
name="department_add_member",
),
path(
"departments/<int:pk>/remove-member/",
views_front.department_remove_member,
name="department_remove_member",
),
# --- Навыки ---
path("<int:pk>/skill/add/", views_front.skill_add, name="skill_add"),
path("<int:pk>/skill/remove/", views_front.skill_remove, name="skill_remove"),
# --- Должности ---
path(
"positions/create/", views_front.position_create_front, name="position_create"
),
path(
"positions/<int:pos_id>/update/",
views_front.position_update_front,
name="position_update",
),
path(
"<int:emp_id>/set-position/",
views_front.employee_set_position_front,
name="employee_set_position",
),
path(
"me/set-position/",
views_front.employee_set_position_me_front,
name="employee_set_position_me",
),
path("positions/<int:pos_id>/", views_front.position_detail_front, name="position_detail"),
]

18
backend/employees/urls_front_auth.py
View File

@ -1,18 +0,0 @@
# backend\eusrr_backend\urls_front_auth.py
from django.urls import path
from . import views_auth
app_name = "auth_front"
urlpatterns = [
path("login/", views_auth.EmailOrPhoneLoginView.as_view(), name="login"),
path("logout/", views_auth.ConfirmLogoutView.as_view(), name="logout"),
path("register/", views_auth.register_view, name="register"),
path("verify-email/", views_auth.verify_email_view, name="verify_email"),
path("resend-email/", views_auth.resend_email_view, name="resend_email"),
path("password-reset/", views_auth.PasswordResetView.as_view(), name="password_reset"),
path("password-reset/done/", views_auth.PasswordResetDoneView.as_view(), name="password_reset_done"),
path("reset/<uidb64>/<token>/", views_auth.PasswordResetConfirmView.as_view(), name="password_reset_confirm"),
path("reset/done/", views_auth.PasswordResetCompleteView.as_view(), name="password_reset_complete"),
]

256
backend/employees/views_auth.py
View File

@ -1,256 +0,0 @@
from __future__ import annotations
import logging
from api.client import ApiClient, get_api_client
from django.contrib import messages
from django.contrib.auth import logout
from django.contrib.auth.mixins import LoginRequiredMixin
from django.contrib.auth.views import LoginView as DjangoLoginView
from django.contrib.auth.views import \
PasswordResetCompleteView as DjangoPasswordResetCompleteView
from django.contrib.auth.views import \
PasswordResetConfirmView as DjangoPasswordResetConfirmView
from django.contrib.auth.views import \
PasswordResetDoneView as DjangoPasswordResetDoneView
from django.contrib.auth.views import \
PasswordResetView as DjangoPasswordResetView
from django.shortcuts import redirect, render
from django.urls import reverse
from django.utils.http import urlencode
from django.views import View
from django.views.decorators.http import require_http_methods
from .forms_auth import (EmailOrPhoneAuthenticationForm, RegistrationForm,
VerifyEmailForm, CustomPasswordResetForm)
logger = logging.getLogger(__name__)
def _apply_api_errors_to_form(form, data: dict):
"""Развешиваем ошибки из API по полям/неполям."""
if not isinstance(data, dict):
form.add_error(None, "Неизвестная ошибка сервера.")
return
if "ok" in data and not data.get("ok"):
form.add_error(
None, data.get("error") or data.get("detail") or "Ошибка запроса."
)
detail = data.get("detail")
if isinstance(detail, str):
form.add_error(None, detail)
elif isinstance(detail, (list, tuple)):
for msg in detail:
form.add_error(None, msg)
for field, errs in data.items():
if field in {"ok", "error", "detail"}:
continue
target = "phone_number" if field == "phone" else field
if target in form.fields:
if isinstance(errs, (list, tuple)):
for e in errs:
form.add_error(target, e)
else:
form.add_error(target, str(errs))
else:
if isinstance(errs, (list, tuple)):
for e in errs:
form.add_error(None, f"{field}: {e}")
else:
form.add_error(None, f"{field}: {errs}")
@require_http_methods(["GET", "POST"])
def register_view(request):
"""
Страница регистрации: показываем Django-форму и шлём валидные данные в /api/v1/auth/register/
(аватар пока локально, в API не отправляем — загрузим после верификации).
"""
api = get_api_client(request)
if request.method == "GET":
form = RegistrationForm()
return render(request, "auth/register.html", {"form": form})
# ВАЖНО: учитывать файлы (чтобы form.avatar был BoundField)
form = RegistrationForm(request.POST, request.FILES)
# Логируем полученные данные
logger.warning(f"[REGISTER_VIEW] POST data: {request.POST.dict()}")
logger.warning(f"[REGISTER_VIEW] FILES: {list(request.FILES.keys())}")
if not form.is_valid():
logger.warning(f"[REGISTER_VIEW] Form errors: {form.errors}")
return render(request, "auth/register.html", {"form": form}, status=400)
payload = form.to_api_payload() # формирует json для API, включая password
logger.warning(f"[REGISTER_VIEW] API payload keys: {list(payload.keys())}")
payload_debug = {k: v for k, v in payload.items() if k != 'avatar'}
logger.warning(f"[REGISTER_VIEW] API payload (без avatar): {payload_debug}")
logger.warning(f"[REGISTER_VIEW] API base URL: {api.cfg.base_url}")
resp = api.post("v1/auth/register/", json=payload)
logger.warning(f"[REGISTER_VIEW] API response status: {resp.status}")
resp_data = resp.json if resp.json else resp.text
logger.warning(f"[REGISTER_VIEW] API response: {resp_data}")
if resp.status == 201:
messages.success(
request,
"Мы отправили код подтверждения на вашу почту. Введите его на следующем шаге.",
)
q = urlencode({"email": payload["email"]})
return redirect(f'{reverse("auth_front:verify_email")}?{q}')
data = resp.json or {}
if resp.status == 200 and data.get("ok") and data.get("pending_verification"):
messages.info(request, "Аккаунт уже создан, подтвердите email кодом из письма.")
q = urlencode({"email": payload["email"]})
return redirect(f'{reverse("auth_front:verify_email")}?{q}')
# Остальные ошибки — показать в форме
_apply_api_errors_to_form(form, data)
return render(request, "auth/register.html", {"form": form}, status=400)
@require_http_methods(["GET", "POST"])
def verify_email_view(request):
"""Ввод кода подтверждения. Использует /api/v1/auth/verify-email/."""
api = get_api_client(request)
if request.method == "GET":
form = VerifyEmailForm(initial={"email": request.GET.get("email", "")})
return render(request, "auth/verify_email.html", {"form": form})
form = VerifyEmailForm(request.POST)
if not form.is_valid():
return render(request, "auth/verify_email.html", {"form": form}, status=400)
payload = {"email": form.cleaned_data["email"], "code": form.cleaned_data["code"]}
r = api.post("v1/auth/verify-email/", json=payload)
if r.ok and (r.json or {}).get("ok"):
messages.success(request, "Email подтверждён! Теперь вы можете войти.")
return redirect("auth_front:login")
data = r.json or {}
err = data.get("error") or data.get("detail") or "Не удалось подтвердить email."
if err == "invalid_code":
err = "Неверный код подтверждения."
elif err == "user_not_found":
err = "Пользователь с таким email не найден."
elif err == "ldap_record_missing":
err = "Ошибка синхронизации с LDAP. Обратитесь к администратору."
form.add_error(None, err)
return render(request, "auth/verify_email.html", {"form": form}, status=400)
@require_http_methods(["POST"])
def resend_email_view(request):
"""Кнопка «Отправить код ещё раз» — /api/v1/auth/resend-email/."""
api = get_api_client(request)
email = (request.POST.get("email") or "").strip()
r = api.post("v1/auth/resend-email/", json={"email": email})
if r.ok and (r.json or {}).get("ok"):
messages.success(request, "Код подтверждения повторно отправлен на вашу почту.")
else:
data = r.json or {}
err = (
data.get("error")
or data.get("detail")
or "Не удалось отправить код повторно."
)
if err == "already_verified":
err = "Email уже подтверждён."
elif err == "user_not_found":
err = "Пользователь с таким email не найден."
messages.error(request, err)
q = urlencode({"email": email}) if email else ""
return redirect(f'{reverse("auth_front:verify_email")}{("?" + q) if q else ""}')
class EmailOrPhoneLoginView(DjangoLoginView):
"""
Страница входа: одно поле 'Email или телефон' + пароль.
После успешной Django-аутентификации — берём JWT у DRF и кладём в сессию.
"""
template_name = "auth/login.html"
authentication_form = EmailOrPhoneAuthenticationForm
def form_valid(self, form):
# 1) Django-логин
response = super().form_valid(form)
# 2) Получаем JWT от API и кладём его в сессию
user = self.request.user
password = form.cleaned_data.get("password")
client = ApiClient(self.request)
ok = client.login(user.email, password) # USERNAME_FIELD=email
if not ok:
messages.error(
self.request, "Не удалось получить токен API. Повторите вход позже."
)
return response
class ConfirmLogoutView(LoginRequiredMixin, View):
template_name = "auth/logout.html"
def get(self, request):
next_url = request.GET.get("next") or request.META.get("HTTP_REFERER") or ""
return render(request, self.template_name, {"next": next_url})
def post(self, request):
next_url = request.POST.get("next") or reverse("auth_front:login")
logout(request)
messages.success(request, "Вы вышли из аккаунта.")
return redirect(next_url)
# ===== Сброс пароля =====
class PasswordResetView(DjangoPasswordResetView):
template_name = "auth/password_reset.html"
email_template_name = "auth/password_reset_email.txt"
html_email_template_name = "emails/password_reset_email.html"
subject_template_name = "auth/password_reset_subject.txt"
success_url = "/auth/password-reset/done/"
form_class = CustomPasswordResetForm
def post(self, request, *args, **kwargs):
"""Переопределяем post для логирования"""
logger.info(
f"\n{'='*80}\n"
f"[PasswordResetView.post] 🔑 ЗАПРОС НА ВОССТАНОВЛЕНИЕ ПАРОЛЯ\n"
f" Email/телефон: {request.POST.get('email', '?')}\n"
f" IP: {request.META.get('REMOTE_ADDR')}\n"
f"{'='*80}"
)
return super().post(request, *args, **kwargs)
def form_valid(self, form):
"""Логирование успешной валидации формы"""
logger.info(
f"[PasswordResetView.form_valid] ✅ Форма валидна, email={form.cleaned_data.get('email')}"
)
return super().form_valid(form)
class PasswordResetDoneView(DjangoPasswordResetDoneView):
template_name = "auth/password_reset_done.html"
class PasswordResetConfirmView(DjangoPasswordResetConfirmView):
template_name = "auth/password_reset_confirm.html"
success_url = "/auth/reset/done/"
class PasswordResetCompleteView(DjangoPasswordResetCompleteView):
template_name = "auth/password_reset_complete.html"

1741
backend/employees/views_front.py
View File

File diff suppressed because it is too large Load Diff

6
backend/eusrr_backend/urls.py
View File

@ -27,19 +27,13 @@ def serve_service_worker(request):
urlpatterns = [
path("admin/", admin.site.urls),
path("auth/", include("employees.urls_front_auth", namespace="auth_front")),
path("documents/", include("documents.urls", namespace="documents")),
path("employees/", include("employees.urls_front", namespace="employees")),
path("communications/", include("communications.urls", namespace="communications")),
path("notifications/", include("notifications.urls", namespace="notifications")),
path("search/", include("search.urls", namespace="search")),
path("finance/", include("finance.urls", namespace="finance")),
path("api/", include(("api.urls", "api"), namespace="api")),
# Service Worker должен быть в корне для правильного scope
path("sw.js", serve_service_worker, name="sw"),
# django-filer URLs для обработки приватных файлов (с проверкой прав доступа)
path("", include("filer.server.urls")),
path("", include("feed.urls_front", namespace="feed")),
]
# Публичные медиа файлы (без проверки прав) - только в режиме разработки

3
backend/finance/tests.py
View File

@ -1,3 +0,0 @@
from django.test import TestCase
# Create your tests here.

11
backend/finance/urls.py
View File

@ -1,11 +0,0 @@
"""URL configuration for finance app."""
from django.urls import path
from . import views
app_name = "finance"
urlpatterns = [
path("", views.finance_dashboard, name="dashboard"),
]

15
backend/finance/views.py
View File

@ -1,15 +0,0 @@
"""Views for finance app."""
from django.contrib.auth.decorators import login_required
from django.shortcuts import render
@login_required
def finance_dashboard(request):
"""
Отображает финансовую панель с встроенным Яндекс.Диском.
"""
context = {
"yandex_disk_url": "https://disk.yandex.ru/i/0JWL-v5zyWJi0Q"
}
return render(request, "finance/dashboard.html", context)

3
backend/search/models.py
View File

@ -1,3 +0,0 @@
from django.db import models
# Create your models here.

3
backend/search/tests.py
View File

@ -1,3 +0,0 @@
from django.test import TestCase
# Create your tests here.

9
backend/search/urls.py
View File

@ -1,9 +0,0 @@
from django.urls import path
from .views import search_view
app_name = "search"
urlpatterns = [
path("", search_view, name="search"),
]

158
backend/search/views.py
View File

@ -1,158 +0,0 @@
# backend/search/views.py
"""
Полнотекстовый поиск через django-watson.
Заменяет предыдущую реализацию с ручными Q-объектами на
автоматическую индексацию watson.
"""
from __future__ import annotations
from typing import Any, Dict, List, Literal, TypedDict
from collections import Counter
from django.contrib.auth.decorators import login_required
from django.http import HttpRequest, HttpResponse
from django.shortcuts import render
from django.utils.html import escape
from watson.search import search as watson_search
from feed.models import Post
from employees.models import Employee, Department
from requests_app.models import Request
class SearchItem(TypedDict):
"""Элемент единой выдачи поиска.
Attributes:
model_name (Literal["post","employee","department","request","chat","message","event"]):
Тип найденной сущности.
object (Any): Экземпляр модели, который будет отрисован в шаблоне.
"""
model_name: Literal[
"post", "employee", "department", "request", "chat", "message", "event"
]
object: Any
# ------------------------ ВСПОМОГАТЕЛЬНЫЕ УТИЛИТЫ ------------------------
def _is_hr(user: Employee) -> bool:
"""Проверяет расширенные права на просмотр заявлений.
Args:
user (Employee): Текущий пользователь.
Returns:
bool: True, если пользователь может видеть все заявления.
Notes:
Используются пермишены из requests_app.
"""
return user.has_perm("requests_app.can_view_all_requests") or user.has_perm(
"requests_app.can_process_requests"
)
def _get_model_name(obj: Any) -> str:
"""Определяет тип модели для SearchItem.
Args:
obj: Объект модели из результатов поиска.
Returns:
str: Название типа модели (post, employee, department, etc.)
"""
model = type(obj)
model_name = model.__name__.lower()
# Маппинг моделей на типы для SearchItem
mapping = {
"post": "post",
"employee": "employee",
"department": "department",
"request": "request",
"chat": "chat",
"message": "message",
"calendarevent": "event",
}
return mapping.get(model_name, model_name)
# -------------------------------- ВЬЮХА --------------------------------
@login_required
def search_view(request: HttpRequest) -> HttpResponse:
"""Единый поиск по системе через django-watson.
Args:
request (HttpRequest): Запрос (`GET["q"]`).
Returns:
HttpResponse: Рендер `templates/search/results.html` с плоским списком.
Контекст шаблона:
- query: str — исходная строка запроса
- results: List[SearchItem] — плоский список элементов
- counts: Dict[str, int] — счётчики по типам:
keys: post, employee, department, request, chat, message, event
- total: int — сумма всех счётчиков
"""
raw_q = request.GET.get("q", "") or ""
query = escape(raw_q.strip())
items: List[SearchItem] = []
counts: Dict[str, int] = {
"post": 0,
"employee": 0,
"department": 0,
"request": 0,
"chat": 0,
"message": 0,
"event": 0,
}
if query:
# Выполняем поиск через watson
search_results = watson_search(query)
# Группируем по типам моделей
by_model: Dict[str, List[Any]] = {}
for result in search_results:
obj = result.object
model_name = _get_model_name(obj)
# Фильтрация заявлений по правам доступа
if model_name == "request":
is_hr = _is_hr(request.user) # type: ignore[arg-type]
# Проверяем, что объект - это действительно Request с полем employee
if not is_hr and hasattr(obj, 'employee') and obj.employee != request.user: # type: ignore[attr-defined]
continue # Пропускаем чужие заявления
if model_name not in by_model:
by_model[model_name] = []
by_model[model_name].append(obj)
# Формируем итоговый список с ограничением 10 элементов на тип
for model_name in ["post", "employee", "department", "request",
"chat", "message", "event"]:
objects = by_model.get(model_name, [])
counts[model_name] = len(objects)
# Берем первые 10 для отображения
for obj in objects[:10]:
items.append({
"model_name": model_name, # type: ignore[typeddict-item]
"object": obj
})
ctx = {
"query": query,
"results": items,
"counts": counts,
"total": sum(counts.values()),
}
return render(request, "search/results.html", ctx)