Add RRO, portal section perms, dislocation presets, and hiring intern flow.

Ship cash-register registry, section-based portal access, configurable dislocation columns, and trial-hire intern accounts with credentials PDF.
This commit is contained in:
2026-08-09 14:58:17 +03:00
parent 237275abd0
commit ae5c1f1848
63 changed files with 5987 additions and 2217 deletions
+21 -4
View File
@@ -1,14 +1,31 @@
from django.contrib import admin
from unfold.admin import ModelAdmin
from apps.hiring.models import HiringRequest
from apps.hiring.models import HiringRequest, InternAccount
@admin.register(InternAccount)
class InternAccountAdmin(ModelAdmin):
list_display = ['label', 'username', 'is_active', 'assigned_to', 'updated_at']
list_filter = ['is_active']
search_fields = ['label', 'username']
list_select_related = ['assigned_to']
ordering = ['label']
readonly_fields = ['assigned_to', 'created_at', 'updated_at', 'uuid']
@admin.register(HiringRequest)
class HiringRequestAdmin(ModelAdmin):
list_display = ['last_name', 'first_name', 'status', 'hire_date', 'trial_days', 'requested_by', 'created_at']
list_filter = ['status', 'user_location_type']
list_display = [
'last_name', 'first_name', 'status', 'hire_date', 'trial_days',
'requested_by', 'granted_ticket', 'created_at',
]
list_filter = ['status', 'user_location_type']
search_fields = ['last_name', 'first_name', 'middle_name', 'candidate_email']
list_select_related = ['requested_by', 'decision_by', 'employee', 'post', 'department']
list_select_related = [
'requested_by', 'decision_by', 'employee', 'post', 'department',
'user_pharmacy', 'user_location', 'granted_ticket',
]
filter_horizontal = ['requested_access']
ordering = ['-created_at']
readonly_fields = ['issued_password', 'created_at', 'updated_at', 'uuid']
@@ -0,0 +1,46 @@
# Generated by Django 6.0.3 on 2026-07-28 20:16
import core.models.fields
import django.db.models.deletion
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('hiring', '0003_hiringrequest_access_task'),
('locations', '0009_pharmacy_carbonio_password'),
]
operations = [
migrations.AddField(
model_name='hiringrequest',
name='issued_password',
field=core.models.fields.EncryptedCharField(blank=True, max_length=512, verbose_name='Виданий пароль (PDF)'),
),
migrations.AddField(
model_name='hiringrequest',
name='user_location',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='hiring_requests_location', to='locations.location', verbose_name='Локація (офіс/склад)'),
),
migrations.CreateModel(
name='InternAccount',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('uuid', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('label', models.CharField(max_length=50, unique=True, verbose_name='Мітка')),
('username', core.models.fields.EncryptedCharField(max_length=512, verbose_name='Логін')),
('password', core.models.fields.EncryptedCharField(max_length=512, verbose_name='Пароль')),
('is_active', models.BooleanField(default=True, verbose_name='Активний')),
('assigned_to', models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='intern_account', to='hiring.hiringrequest', verbose_name='Призначено заявці')),
],
options={
'verbose_name': 'Інтерн-обліковий запис',
'verbose_name_plural': 'Пул інтерн-облікових записів',
'ordering': ['label'],
},
),
]
@@ -0,0 +1,36 @@
"""Seed pool of 5 shared intern accounts for hiring trial period."""
from django.db import migrations
def seed_intern_accounts(apps, schema_editor):
# Concrete model — EncryptedCharField must encrypt on save.
from apps.hiring.models import InternAccount
for i in range(1, 6):
InternAccount.objects.get_or_create(
label=f'Intern-{i}',
defaults={
'username': f'intern{i}',
'password': f'InternPass{i}!',
'is_active': True,
},
)
def unseed_intern_accounts(apps, schema_editor):
InternAccount = apps.get_model('hiring', 'InternAccount')
InternAccount.objects.filter(
label__in=[f'Intern-{i}' for i in range(1, 6)],
assigned_to__isnull=True,
).delete()
class Migration(migrations.Migration):
dependencies = [
('hiring', '0004_intern_account_and_user_location'),
]
operations = [
migrations.RunPython(seed_intern_accounts, unseed_intern_accounts),
]
@@ -0,0 +1,140 @@
# Generated by Django 6.0.3 on 2026-07-29 08:35
import core.models.fields
import django.core.validators
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('core', '0008_access_location_types_help'),
('hiring', '0005_seed_intern_accounts'),
('locations', '0011_seed_dislocation_column_presets'),
('tasks', '0005_alter_task_priority_alter_task_status'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AlterModelOptions(
name='internaccount',
options={'ordering': ['label'], 'verbose_name': 'Интерн-аккаунт', 'verbose_name_plural': 'Пул интерн-аккаунтов'},
),
migrations.AlterField(
model_name='hiringrequest',
name='access_note',
field=models.TextField(blank=True, verbose_name='Примечание по доступам'),
),
migrations.AlterField(
model_name='hiringrequest',
name='access_task',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='+', to='tasks.task', verbose_name='Задача на выдачу доступов'),
),
migrations.AlterField(
model_name='hiringrequest',
name='candidate_email',
field=models.EmailField(blank=True, max_length=254, verbose_name='Личный email кандидата'),
),
migrations.AlterField(
model_name='hiringrequest',
name='decision_at',
field=models.DateTimeField(blank=True, null=True, verbose_name='Дата решения'),
),
migrations.AlterField(
model_name='hiringrequest',
name='decision_by',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='hiring_requests_decided', to=settings.AUTH_USER_MODEL, verbose_name='Решение принял'),
),
migrations.AlterField(
model_name='hiringrequest',
name='decision_note',
field=models.TextField(blank=True, verbose_name='Комментарий к решению'),
),
migrations.AlterField(
model_name='hiringrequest',
name='department',
field=models.ForeignKey(blank=True, limit_choices_to={'directory_type__code': 'department', 'is_active': True}, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='hiring_requests_department', to='core.directory', verbose_name='Отдел'),
),
migrations.AlterField(
model_name='hiringrequest',
name='employee',
field=models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='hiring_request', to=settings.AUTH_USER_MODEL, verbose_name='Учётная запись'),
),
migrations.AlterField(
model_name='hiringrequest',
name='first_name',
field=models.CharField(max_length=150, verbose_name='Имя'),
),
migrations.AlterField(
model_name='hiringrequest',
name='issued_password',
field=core.models.fields.EncryptedCharField(blank=True, max_length=512, verbose_name='Выданный пароль (PDF)'),
),
migrations.AlterField(
model_name='hiringrequest',
name='last_name',
field=models.CharField(max_length=150, verbose_name='Фамилия'),
),
migrations.AlterField(
model_name='hiringrequest',
name='middle_name',
field=models.CharField(blank=True, max_length=150, verbose_name='Отчество'),
),
migrations.AlterField(
model_name='hiringrequest',
name='post',
field=models.ForeignKey(blank=True, limit_choices_to={'directory_type__code': 'post', 'is_active': True}, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='hiring_requests_post', to='core.directory', verbose_name='Должность'),
),
migrations.AlterField(
model_name='hiringrequest',
name='requested_access',
field=models.ManyToManyField(blank=True, limit_choices_to={'directory_type__code': 'access_system', 'is_active': True}, related_name='hiring_requests_access', to='core.directory', verbose_name='Требуемые доступы'),
),
migrations.AlterField(
model_name='hiringrequest',
name='requested_by',
field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='hiring_requests_submitted', to=settings.AUTH_USER_MODEL, verbose_name='Подал HR'),
),
migrations.AlterField(
model_name='hiringrequest',
name='status',
field=models.CharField(choices=[('trial', 'На испытании'), ('approved', 'Утверждено'), ('rejected', 'Отклонено')], default='trial', max_length=10, verbose_name='Статус'),
),
migrations.AlterField(
model_name='hiringrequest',
name='trial_days',
field=models.PositiveSmallIntegerField(choices=[(1, '1 дн.'), (2, '2 дн.'), (3, '3 дн.'), (4, '4 дн.'), (5, '5 дн.')], default=3, validators=[django.core.validators.MinValueValidator(1), django.core.validators.MaxValueValidator(5)], verbose_name='Срок испытания, дней'),
),
migrations.AlterField(
model_name='hiringrequest',
name='user_location',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='hiring_requests_location', to='locations.location', verbose_name='Локация (офис/склад)'),
),
migrations.AlterField(
model_name='hiringrequest',
name='user_location_type',
field=models.CharField(choices=[('pharmacy', 'Аптека'), ('warehouse', 'Склад'), ('office', 'Офис')], max_length=20, verbose_name='Тип локации'),
),
migrations.AlterField(
model_name='internaccount',
name='assigned_to',
field=models.OneToOneField(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='intern_account', to='hiring.hiringrequest', verbose_name='Назначено заявке'),
),
migrations.AlterField(
model_name='internaccount',
name='is_active',
field=models.BooleanField(default=True, verbose_name='Активный'),
),
migrations.AlterField(
model_name='internaccount',
name='label',
field=models.CharField(max_length=50, unique=True, verbose_name='Метка'),
),
migrations.AlterField(
model_name='internaccount',
name='username',
field=core.models.fields.EncryptedCharField(max_length=512, verbose_name='Логин'),
),
]
+91 -22
View File
@@ -1,30 +1,82 @@
from datetime import timedelta
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from core.models import BaseModel
from core.models.fields import EncryptedCharField
from core.utils.validators import validate_ukrainian_phone
class InternAccount(BaseModel):
"""Спільний інтерн-обліковий запис з пулу (5 слотів). Призначається заявці
на час випробування; після approve/reject звільняється."""
label = models.CharField(_('Метка'), max_length=50, unique=True)
username = EncryptedCharField(_('Логин'), max_length=512)
password = EncryptedCharField(_('Пароль'), max_length=512)
is_active = models.BooleanField(_('Активный'), default=True)
assigned_to = models.OneToOneField(
'hiring.HiringRequest', on_delete=models.SET_NULL,
null=True, blank=True,
related_name='intern_account', verbose_name=_('Назначено заявке'),
)
class Meta:
verbose_name = _('Интерн-аккаунт')
verbose_name_plural = _('Пул интерн-аккаунтов')
ordering = ['label']
def __str__(self):
status = _('занят') if self.assigned_to_id else _('свободен')
return f'{self.label} ({status})'
@classmethod
def allocate(cls, hiring_request):
"""Виділяє свободен слот під select_for_update (викликати всередині atomic)."""
account = (
cls.objects.select_for_update()
.filter(is_active=True, assigned_to__isnull=True)
.order_by('label')
.first()
)
if account is None:
raise ValidationError({
'__all__': _(
'Нет свободных интерн-аккаунтов. '
'Дождитесь завершения стажировки или добавьте слот в админке.'
),
})
account.assigned_to = hiring_request
account.save(update_fields=['assigned_to', 'updated_at'])
return account
def release(self):
if self.assigned_to_id is None:
return
self.assigned_to = None
self.save(update_fields=['assigned_to', 'updated_at'])
class HiringRequest(BaseModel):
"""Заявка HR на найм кандидата: створення тимчасового (стажерського)
облікового запису та подальше затвердження/відхилення."""
"""Заявка HR на найм кандидата: інтерн-слот + IT Ticket на випробуванні,
персональний Employee — лише після затвердження."""
STATUS_CHOICES = [
('trial', _('На випробуванні')),
('approved', _('Затверджено')),
('rejected', _('Відхилено')),
('trial', _('На испытании')),
('approved', _('Утверждено')),
('rejected', _('Отклонено')),
]
TRIAL_DAYS_CHOICES = [(i, _('%(n)d дн.') % {'n': i}) for i in range(1, 6)]
last_name = models.CharField(_('Прізвище'), max_length=150)
first_name = models.CharField(_('Ім\'я'), max_length=150)
middle_name = models.CharField(_('По батькові'), max_length=150, blank=True)
last_name = models.CharField(_('Фамилия'), max_length=150)
first_name = models.CharField(_('Имя'), max_length=150)
middle_name = models.CharField(_('Отчество'), max_length=150, blank=True)
candidate_email = models.EmailField(_('Особистий email кандидата'), blank=True)
candidate_email = models.EmailField(_('Личный email кандидата'), blank=True)
candidate_phone = models.CharField(
_('Телефон кандидата'), max_length=20, blank=True,
validators=[validate_ukrainian_phone],
@@ -33,60 +85,65 @@ class HiringRequest(BaseModel):
LOCATION_TYPES = [
('pharmacy', _('Аптека')),
('warehouse', _('Склад')),
('office', _('Офіс')),
('office', _('Офис')),
]
user_location_type = models.CharField(
_('Тип локації'), max_length=20, choices=LOCATION_TYPES,
_('Тип локации'), max_length=20, choices=LOCATION_TYPES,
)
department = models.ForeignKey(
'core.Directory', on_delete=models.SET_NULL,
null=True, blank=True,
limit_choices_to={'directory_type__code': 'department', 'is_active': True},
related_name='hiring_requests_department', verbose_name=_('Відділ'),
related_name='hiring_requests_department', verbose_name=_('Отдел'),
)
user_pharmacy = models.ForeignKey(
'locations.Pharmacy', on_delete=models.SET_NULL,
null=True, blank=True,
related_name='hiring_requests', verbose_name=_('Аптека'),
)
user_location = models.ForeignKey(
'locations.Location', on_delete=models.SET_NULL,
null=True, blank=True,
related_name='hiring_requests_location', verbose_name=_('Локация (офис/склад)'),
)
post = models.ForeignKey(
'core.Directory', on_delete=models.SET_NULL,
null=True, blank=True,
limit_choices_to={'directory_type__code': 'post', 'is_active': True},
related_name='hiring_requests_post', verbose_name=_('Посада'),
related_name='hiring_requests_post', verbose_name=_('Должность'),
)
hire_date = models.DateField(_('Дата виходу'), default=timezone.localdate)
trial_days = models.PositiveSmallIntegerField(
_('Термін випробування, днів'), choices=TRIAL_DAYS_CHOICES, default=3,
_('Срок испытания, дней'), choices=TRIAL_DAYS_CHOICES, default=3,
validators=[MinValueValidator(1), MaxValueValidator(5)],
)
requested_access = models.ManyToManyField(
'core.Directory', blank=True,
limit_choices_to={'directory_type__code': 'access_system', 'is_active': True},
related_name='hiring_requests_access', verbose_name=_('Потрібні доступи'),
related_name='hiring_requests_access', verbose_name=_('Требуемые доступы'),
)
access_note = models.TextField(_('Примітка щодо доступів'), blank=True)
access_note = models.TextField(_('Примечание по доступам'), blank=True)
status = models.CharField(_('Статус'), max_length=10, choices=STATUS_CHOICES, default='trial')
requested_by = models.ForeignKey(
'employees.Employee', on_delete=models.PROTECT,
related_name='hiring_requests_submitted', verbose_name=_('Подав HR'),
related_name='hiring_requests_submitted', verbose_name=_('Подал HR'),
)
employee = models.OneToOneField(
'employees.Employee', on_delete=models.SET_NULL,
null=True, blank=True,
related_name='hiring_request', verbose_name=_('Обліковий запис'),
related_name='hiring_request', verbose_name=_('Учётная запись'),
)
decision_by = models.ForeignKey(
'employees.Employee', on_delete=models.SET_NULL,
null=True, blank=True,
related_name='hiring_requests_decided', verbose_name=_('Рішення прийняв'),
related_name='hiring_requests_decided', verbose_name=_('Решение принял'),
)
decision_at = models.DateTimeField(_('Дата рішення'), null=True, blank=True)
decision_note = models.TextField(_('Коментар до рішення'), blank=True)
decision_at = models.DateTimeField(_('Дата решения'), null=True, blank=True)
decision_note = models.TextField(_('Комментарий к решению'), blank=True)
granted_ticket = models.ForeignKey(
'it_manager.Ticket', on_delete=models.SET_NULL,
@@ -96,7 +153,12 @@ class HiringRequest(BaseModel):
access_task = models.ForeignKey(
'tasks.Task', on_delete=models.SET_NULL,
null=True, blank=True,
related_name='+', verbose_name=_('Задача на видачу доступів'),
related_name='+', verbose_name=_('Задача на выдачу доступов'),
)
# Пароль персонального акаунта після approve — для одноразового PDF.
issued_password = EncryptedCharField(
_('Выданный пароль (PDF)'), max_length=512, blank=True,
)
class Meta:
@@ -117,3 +179,10 @@ class HiringRequest(BaseModel):
@property
def is_overdue(self):
return self.status == 'trial' and self.trial_end_date < timezone.localdate()
def release_intern(self):
try:
account = self.intern_account
except Exception:
return
account.release()
+83
View File
@@ -0,0 +1,83 @@
"""PDF з обліковими даними після затвердження кандидата."""
from pathlib import Path
from django.http import HttpResponse
from django.utils.translation import gettext as _
_FONT_CANDIDATES = (
Path('/usr/share/fonts/google-noto/NotoSans-Regular.ttf'),
Path('/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf'),
Path('/usr/share/fonts/dejavu/DejaVuSans.ttf'),
)
def _resolve_font():
for path in _FONT_CANDIDATES:
if path.is_file():
return str(path)
return None
def build_credentials_pdf(hiring_request) -> bytes:
from fpdf import FPDF
from fpdf.enums import XPos, YPos
employee = hiring_request.employee
if employee is None:
raise ValueError('HiringRequest has no employee')
pdf = FPDF()
pdf.set_auto_page_break(auto=True, margin=15)
pdf.add_page()
pdf.set_margins(20, 20, 20)
font_path = _resolve_font()
font_name = 'Portal' if font_path else 'Helvetica'
if font_path:
pdf.add_font('Portal', fname=font_path)
pdf.set_font(font_name, size=14)
pdf.cell(
0, 10, _('Учётные данные сотрудника'),
new_x=XPos.LMARGIN, new_y=YPos.NEXT,
)
pdf.ln(4)
pdf.set_font(font_name, size=11)
dash = '-'
lines = [
f"{_('ФИО')}: {employee.last_name} {employee.first_name} {employee.middle_name}".strip(),
f"{_('Должность')}: {employee.post.name if employee.post_id else dash}",
f"{_('Тип локации')}: {hiring_request.get_user_location_type_display()}",
f"{_('Логин')}: {employee.username}",
f"{_('Пароль')}: {hiring_request.issued_password or dash}",
f"{_('Рабочий email')}: {employee.email or dash}",
'',
f"{_('Запрошенные доступы')}:",
]
accesses = list(hiring_request.requested_access.all())
if accesses:
for access in accesses:
lines.append(f' - {access.name}')
else:
lines.append(f" {_('(не указано)')}")
for line in lines:
if not line:
pdf.ln(4)
continue
pdf.set_x(pdf.l_margin)
pdf.multi_cell(
pdf.epw, 7, line,
new_x=XPos.LMARGIN, new_y=YPos.NEXT,
)
return bytes(pdf.output())
def credentials_pdf_response(hiring_request) -> HttpResponse:
content = build_credentials_pdf(hiring_request)
filename = f'credentials-{hiring_request.employee.username}.pdf'
response = HttpResponse(content, content_type='application/pdf')
response['Content-Disposition'] = f'attachment; filename="{filename}"'
return response
+189 -87
View File
@@ -1,13 +1,14 @@
"""Бізнес-логіка заявок на найм: створення тимчасового акаунта стажера та
подальше затвердження/відхилення. Використовує вже робочу AD/Mattermost
синхронізацію через сигнали Employee.save() — жодного нового коду для
зовнішніх систем тут немає."""
"""Бізнес-логіка найму: інтерн-слот + IT Ticket на випробуванні;
персональний Employee і PDF — лише після затвердження."""
import secrets
from django.db import transaction
from django.utils import timezone
from django.utils.translation import gettext as _
from pytils.translit import slugify
from apps.employees.models.base import Employee
from apps.hiring.models import HiringRequest
from apps.hiring.models import HiringRequest, InternAccount
def _generate_username(last_name, first_name):
@@ -20,32 +21,155 @@ def _generate_username(last_name, first_name):
return username
@transaction.atomic
def create_hiring_request(*, requester, last_name, first_name, middle_name='',
candidate_email='', candidate_phone='',
user_location_type, department_id=None, user_pharmacy_id=None, post_id=None,
hire_date, trial_days, requested_access_ids=None, access_note=''):
"""Створює одразу особистий обмежений обліковий запис (Employee) і
заявку (HiringRequest) на випробувальний термін."""
username = _generate_username(last_name, first_name)
def _generate_password():
return secrets.token_urlsafe(12)
employee = Employee.objects.create(
username=username,
last_name=last_name,
first_name=first_name,
middle_name=middle_name,
personal_phone=candidate_phone,
def _coerce_location_fields(user_location_type, department_id, user_pharmacy_id, user_location_id):
"""Приводить поля локації до правил Employee.clean(): аптека XOR офіс/склад."""
if user_location_type == 'pharmacy':
return None, user_pharmacy_id, None
if user_location_type in ('office', 'warehouse'):
return department_id, None, user_location_id
return department_id, user_pharmacy_id, user_location_id
def _filter_access_ids_for_location(user_location_type, requested_access_ids):
"""Залишає лише системи доступу, дозволені для типу локації.
Порожній location_types у системи = доступна скрізь (як у посад).
"""
if not requested_access_ids:
return []
from core.models import Directory
allowed = []
for access in Directory.objects.filter(
pk__in=requested_access_ids,
directory_type__code='access_system',
is_active=True,
).prefetch_related('location_types'):
types = [lt.code for lt in access.location_types.all()]
if not types or user_location_type in types:
allowed.append(access.pk)
return allowed
def _validate_hire_location(*, user_location_type, department_id, user_pharmacy_id,
user_location_id, post_id, hire_date):
"""Перевіряє правила локації Employee.clean() без збереження в БД."""
emp = Employee(
username='__hire_validate__',
last_name='_',
first_name='_',
hire_date=hire_date,
probation_end_date=hire_date + timezone.timedelta(days=trial_days),
user_location_type=user_location_type,
department_id=department_id,
user_pharmacy_id=user_pharmacy_id,
user_location_id=user_location_id,
post_id=post_id,
# AD-синхронізація стосується лише офісних співробітників — це вже
# чинне правило в apps/ad_connector/signals.py (_is_office), тут лише
# вмикаємо прапорець, без якого сигнал взагалі не спрацює.
ad_sync_enabled=(user_location_type == 'office'),
)
emp.clean()
def _directory(type_code, code):
from core.models import Directory
return Directory.objects.get(directory_type__code=type_code, code=code)
def _ticket_location(hiring_request):
if hiring_request.user_location_type == 'pharmacy' and hiring_request.user_pharmacy_id:
return hiring_request.user_pharmacy
return hiring_request.user_location
def _build_ticket_description(hiring_request, intern):
loc_label = hiring_request.get_user_location_type_display()
if hiring_request.user_pharmacy_id:
loc_detail = str(hiring_request.user_pharmacy)
elif hiring_request.user_location_id:
loc_detail = str(hiring_request.user_location)
elif hiring_request.department_id:
loc_detail = hiring_request.department.name
else:
loc_detail = '—'
lines = [
f"{_('Кандидат')}: {hiring_request.last_name} {hiring_request.first_name} {hiring_request.middle_name}".strip(),
f"{_('Должность')}: {hiring_request.post.name if hiring_request.post_id else '—'}",
f"{_('Тип локации')}: {loc_label}",
f"{_('Локация / отдел')}: {loc_detail}",
f"{_('Дата выхода')}: {hiring_request.hire_date:%d.%m.%Y}",
f"{_('Испытательный срок')}: {hiring_request.trial_days} {_('дн.')}",
f"{_('Интерн-аккаунт')}: {intern.label} / {_('логин')} «{intern.username}»",
'',
f"{_('Требуемые доступы')}:",
]
accesses = list(hiring_request.requested_access.all())
if accesses:
for access in accesses:
lines.append(f' • {access.name}')
else:
lines.append(f" {_('(не указано)')}")
if hiring_request.access_note:
lines.append('')
lines.append(f"{_('Примечание')}: {hiring_request.access_note}")
lines.append('')
lines.append(_(
'Подготовьте рабочее место и выдайте доступы к интерн-аккаунту на время '
'испытания. После утверждения HR будет создан персональный аккаунт.'
))
return '\n'.join(lines)
def _create_access_ticket(hiring_request, requester, intern):
from apps.it_manager.models import Ticket
fio = f'{hiring_request.last_name} {hiring_request.first_name}'.strip()
ticket = Ticket.objects.create(
title=f"{_('Подготовка рабочего места')}: {fio}",
ticket_type=_directory('ticket_type', 'access'),
priority=_directory('ticket_priority', 'high'),
requester=requester,
location=_ticket_location(hiring_request),
description=_build_ticket_description(hiring_request, intern),
)
hiring_request.granted_ticket = ticket
hiring_request.save(update_fields=['granted_ticket', 'updated_at'])
return ticket
def _close_ticket(ticket):
if ticket is None:
return
closed = _directory('ticket_status', 'closed')
if ticket.status_id == closed.pk:
return
ticket.status = closed
ticket.save(update_fields=['status', 'closed_at', 'updated_at'])
@transaction.atomic
def create_hiring_request(*, requester, last_name, first_name, middle_name='',
candidate_email='', candidate_phone='',
user_location_type, department_id=None, user_pharmacy_id=None,
user_location_id=None, post_id=None,
hire_date, trial_days, requested_access_ids=None, access_note=''):
"""Створює заявку на випробування: виділяє інтерн-слот і IT Ticket.
Персональний Employee ще не створюється.
"""
department_id, user_pharmacy_id, user_location_id = _coerce_location_fields(
user_location_type, department_id, user_pharmacy_id, user_location_id,
)
_validate_hire_location(
user_location_type=user_location_type,
department_id=department_id,
user_pharmacy_id=user_pharmacy_id,
user_location_id=user_location_id,
post_id=post_id,
hire_date=hire_date,
)
requested_access_ids = _filter_access_ids_for_location(
user_location_type, requested_access_ids or [],
)
hiring_request = HiringRequest.objects.create(
@@ -57,72 +181,30 @@ def create_hiring_request(*, requester, last_name, first_name, middle_name='',
user_location_type=user_location_type,
department_id=department_id,
user_pharmacy_id=user_pharmacy_id,
user_location_id=user_location_id,
post_id=post_id,
hire_date=hire_date,
trial_days=trial_days,
access_note=access_note,
status='trial',
requested_by=requester,
employee=employee,
)
if requested_access_ids:
hiring_request.requested_access.set(requested_access_ids)
_create_access_task(hiring_request, requester)
intern = InternAccount.allocate(hiring_request)
_create_access_ticket(hiring_request, requester, intern)
return hiring_request
def _create_access_task(hiring_request, requester):
"""Ставить задачу IT-відділу на видачу доступів новому співробітнику
(чек-лист по обраних базах), призначає її IT-групі, рахує строк за
робочим часом і повідомляє постановника."""
from apps.tasks.models import Task, TaskChecklistItem
from apps.visit_manager.services import add_working_hours
accesses = list(hiring_request.requested_access.all())
if not accesses:
return None
fio = f'{hiring_request.last_name} {hiring_request.first_name}'.strip()
due_dt = add_working_hours(timezone.now(), 24)
parts = ['Надати новому співробітнику доступ до обраних систем/баз (відмітьте виконане у чек-листі).']
if hiring_request.post_id:
parts.append(f'Посада: {hiring_request.post.name}.')
if hiring_request.access_note:
parts.append(f'Примітка: {hiring_request.access_note}')
task = Task.objects.create(
title=f'Доступи для {fio}',
description=' '.join(parts),
creator=requester,
priority='high',
due_date=due_dt.date(),
)
for i, access in enumerate(accesses):
TaskChecklistItem.objects.create(task=task, label=f'Додати в «{access.name}»', order=i)
# Призначаємо IT-відділ (co_executors) — це тригерить сповіщення виконавцям (сигнали tasks).
it_staff = list(Employee.objects.filter(groups__name='IT', is_active=True))
if it_staff:
task.co_executors.add(*it_staff)
_notify_requester_taken(requester, fio, due_dt)
hiring_request.access_task = task
hiring_request.save(update_fields=['access_task', 'updated_at'])
return task
def _should_assign_office_email(user_location_type) -> bool:
return user_location_type == 'office'
def _notify_requester_taken(requester, fio, due_dt):
"""Повідомити постановника (HR), що заявку взято в роботу і строк — 24 год робочого часу."""
from apps.mattermost_connector.services.bot_service import send_direct_message
local = timezone.localtime(due_dt)
message = (
f'Заявку на доступи для {fio} прийнято в роботу IT-відділом. '
f'Виконаємо протягом 24 годин робочого часу — орієнтовно до {local:%d.%m %H:%M}.'
)
send_direct_message(requester, message)
def _office_email_for(username: str) -> str:
from apps.employees.models import EmployeeDefaults
domain = EmployeeDefaults.load().normalized_office_domain
return f'{username}@{domain}'
@transaction.atomic
@@ -130,20 +212,43 @@ def approve_hiring_request(hiring_request, decided_by, note=''):
if hiring_request.status != 'trial':
return hiring_request
# Доступи видаються задачею IT-відділу ще на етапі створення заявки
# (_create_access_task), тому затвердження лише завершує випробувальний термін.
employee = hiring_request.employee
if employee is not None:
employee.probation_end_date = timezone.localdate()
employee.save(update_fields=['probation_end_date'])
username = _generate_username(hiring_request.last_name, hiring_request.first_name)
password = _generate_password()
employee = Employee(
username=username,
last_name=hiring_request.last_name,
first_name=hiring_request.first_name,
middle_name=hiring_request.middle_name,
personal_phone=hiring_request.candidate_phone or '',
personal_email=hiring_request.candidate_email or '',
is_active=True,
hire_date=hiring_request.hire_date,
probation_end_date=timezone.localdate(),
user_location_type=hiring_request.user_location_type,
department_id=hiring_request.department_id,
user_pharmacy_id=hiring_request.user_pharmacy_id,
user_location_id=hiring_request.user_location_id,
post_id=hiring_request.post_id,
ad_sync_enabled=(hiring_request.user_location_type == 'office'),
)
if _should_assign_office_email(hiring_request.user_location_type):
employee.email = _office_email_for(username)
employee.set_password(password)
employee.full_clean()
employee.save()
hiring_request.employee = employee
hiring_request.issued_password = password
hiring_request.status = 'approved'
hiring_request.decision_by = decided_by
hiring_request.decision_at = timezone.now()
hiring_request.decision_note = note
hiring_request.save(update_fields=[
'status', 'decision_by', 'decision_at', 'decision_note', 'updated_at',
'employee', 'issued_password', 'status',
'decision_by', 'decision_at', 'decision_note', 'updated_at',
])
hiring_request.release_intern()
return hiring_request
@@ -152,11 +257,8 @@ def reject_hiring_request(hiring_request, decided_by, note=''):
if hiring_request.status != 'trial':
return hiring_request
employee = hiring_request.employee
if employee is not None:
employee.is_active = False
employee.termination_date = timezone.localdate()
employee.save(update_fields=['is_active', 'termination_date'])
hiring_request.release_intern()
_close_ticket(hiring_request.granted_ticket)
hiring_request.status = 'rejected'
hiring_request.decision_by = decided_by
+258 -89
View File
@@ -1,21 +1,32 @@
"""Тести бізнес-логіки заявок на найм: створення тимчасового акаунта,
задача IT на доступи (чек-лист), затвердження і відхилення."""
"""Тести найму: пул інтернів, IT Ticket, Employee лише після approve, PDF."""
from datetime import date, timedelta
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.test import TestCase
from django.core.exceptions import ValidationError
from django.test import Client, TestCase
from django.urls import reverse
from core.models import Directory
from apps.hiring.models import HiringRequest
from core.models import Directory, DirectoryType
from apps.hiring.models import HiringRequest, InternAccount
from apps.hiring.services import approve_hiring_request, create_hiring_request, reject_hiring_request
from apps.tasks.models import Task
from apps.it_manager.models import Ticket
from apps.locations.models import Location, Pharmacy
Employee = get_user_model()
_employee_seq = 0
def login_2fa(client, user):
client.force_login(user)
if user.is_staff and not user.totp_enabled:
user.totp_enabled = True
user.save(update_fields=['totp_enabled'])
session = client.session
session['2fa_verified'] = True
session.save()
def make_employee(**overrides):
global _employee_seq
_employee_seq += 1
@@ -34,133 +45,291 @@ def make_employee(**overrides):
return Employee.objects.create(**defaults)
def make_intern(label='Intern-1', username=None, password='intern-pass'):
return InternAccount.objects.create(
label=label,
username=username or f'intern.{label.lower().replace("-", "")}',
password=password,
is_active=True,
)
class HiringServiceTests(TestCase):
@classmethod
def setUpTestData(cls):
org_type = DirectoryType.objects.create(code='organization', name='Организация')
region_type = DirectoryType.objects.create(code='region', name='Регион')
dept_type = DirectoryType.objects.create(code='department', name='Відділ')
DirectoryType.objects.get_or_create(code='access_system', defaults={'name': 'Системи доступу'})
cls.region = Directory.objects.create(directory_type=region_type, code='od', name='Одеса')
cls.org = Directory.objects.create(directory_type=org_type, code='akc', name='АКС')
cls.department = Directory.objects.create(directory_type=dept_type, code='hr', name='HR')
cls.pharmacy = Pharmacy.objects.create(
organization=cls.org, location_region=cls.region,
pharmacy_number=1, location_type='pharmacy',
)
cls.office = Location.objects.create(
location_type='office', location_region=cls.region,
street='Дерибасівська', house_number='1',
)
def setUp(self):
self.hr_user = make_employee()
# Сид-миграция 0005 наполняет пул Intern-1..5 — очищаем, чтобы тесты сами
# управляли составом стажёров (иначе конфликт лейблов и «свободные» из сида).
InternAccount.objects.all().delete()
self.intern = make_intern('Intern-1')
def test_create_hiring_request_creates_trial_employee(self):
hire_date = date.today() + timedelta(days=1)
hr = create_hiring_request(
def _pharmacy_kwargs(self, **extra):
defaults = dict(
requester=self.hr_user,
last_name='Кандидат',
first_name='Іван',
user_location_type='pharmacy',
hire_date=hire_date,
user_pharmacy_id=self.pharmacy.pk,
hire_date=date.today() + timedelta(days=1),
trial_days=3,
)
defaults.update(extra)
return defaults
def _office_kwargs(self, **extra):
defaults = dict(
requester=self.hr_user,
last_name='Офісний',
first_name='Кандидат',
user_location_type='office',
department_id=self.department.pk,
user_location_id=self.office.pk,
hire_date=date.today() + timedelta(days=1),
trial_days=2,
)
defaults.update(extra)
return defaults
def test_create_assigns_intern_and_ticket_without_employee(self):
hire_date = date.today() + timedelta(days=1)
hr = create_hiring_request(**self._pharmacy_kwargs(hire_date=hire_date))
self.assertEqual(hr.status, 'trial')
self.assertIsNotNone(hr.employee)
self.assertTrue(hr.employee.is_active)
self.assertEqual(hr.employee.probation_end_date, hire_date + timedelta(days=3))
self.assertIsNone(hr.employee)
self.assertIsNotNone(hr.granted_ticket)
self.assertEqual(hr.granted_ticket.ticket_type.code, 'access')
self.assertEqual(hr.granted_ticket.priority.code, 'high')
self.assertEqual(hr.intern_account.pk, self.intern.pk)
self.assertEqual(hr.trial_end_date, hire_date + timedelta(days=3))
self.assertFalse(hr.is_overdue)
self.assertIsNone(hr.access_task)
def test_ad_sync_enabled_only_for_office_employees(self):
office_hr = create_hiring_request(
requester=self.hr_user, last_name='Офісний', first_name='Кандидат',
user_location_type='office', hire_date=date.today() + timedelta(days=1), trial_days=2,
)
pharmacy_hr = create_hiring_request(
requester=self.hr_user, last_name='Аптечний', first_name='Кандидат',
user_location_type='pharmacy', hire_date=date.today() + timedelta(days=1), trial_days=2,
)
def test_no_free_intern_raises(self):
create_hiring_request(**self._pharmacy_kwargs())
with self.assertRaises(ValidationError) as cm:
create_hiring_request(**self._pharmacy_kwargs(
last_name='Другий', first_name='Кандидат',
))
self.assertIn('__all__', cm.exception.message_dict)
self.assertTrue(office_hr.employee.ad_sync_enabled)
self.assertFalse(pharmacy_hr.employee.ad_sync_enabled)
def test_pharmacy_without_pharmacy_raises(self):
with self.assertRaises(ValidationError) as cm:
create_hiring_request(**self._pharmacy_kwargs(user_pharmacy_id=None))
self.assertIn('user_pharmacy', cm.exception.message_dict)
def test_office_without_department_raises(self):
with self.assertRaises(ValidationError) as cm:
create_hiring_request(**self._office_kwargs(department_id=None))
self.assertIn('department', cm.exception.message_dict)
def test_office_without_location_raises(self):
with self.assertRaises(ValidationError) as cm:
create_hiring_request(**self._office_kwargs(user_location_id=None))
self.assertIn('user_location', cm.exception.message_dict)
def test_is_overdue_when_trial_period_passed(self):
hr = create_hiring_request(
requester=self.hr_user, last_name='Прострочений', first_name='Тест',
user_location_type='pharmacy', hire_date=date(2026, 1, 1), trial_days=1,
)
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Прострочений', first_name='Тест',
hire_date=date(2026, 1, 1), trial_days=1,
))
self.assertTrue(hr.is_overdue)
def test_create_hiring_request_generates_unique_username(self):
hr1 = create_hiring_request(
requester=self.hr_user, last_name='Дубль', first_name='Петро',
user_location_type='pharmacy', hire_date=date(2026, 6, 1), trial_days=1,
)
hr2 = create_hiring_request(
requester=self.hr_user, last_name='Дубль', first_name='Петро',
user_location_type='pharmacy', hire_date=date(2026, 6, 1), trial_days=1,
)
self.assertNotEqual(hr1.employee.username, hr2.employee.username)
# --- задача IT на доступи ---
def _access_ids(self, n=2):
ids = list(Directory.objects.filter(directory_type__code='access_system', is_active=True)
.values_list('id', flat=True)[:n])
self.assertTrue(ids, 'у тестовій БД немає довідника access_system')
return ids
def test_create_with_access_creates_it_task_with_checklist(self):
it_group, _ = Group.objects.get_or_create(name='IT')
it_user = make_employee(username='it.tech', email='it.tech@company.local', last_name='Айтішник')
it_user.groups.add(it_group)
def test_create_with_access_creates_ticket_not_task(self):
access_ids = self._access_ids(2)
hr = create_hiring_request(
requester=self.hr_user, last_name='Доступний', first_name='Кандидат',
user_location_type='pharmacy', hire_date=date.today() + timedelta(days=1), trial_days=3,
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Доступний', first_name='Кандидат',
requested_access_ids=access_ids,
)
))
self.assertIsNotNone(hr.access_task)
task = hr.access_task
self.assertEqual(task.creator, self.hr_user)
self.assertEqual(task.priority, 'high')
self.assertIsNotNone(task.due_date)
self.assertEqual(task.checklist.count(), len(access_ids))
self.assertIn(it_user, task.co_executors.all())
# виконавець може відмічати пункти
item = task.checklist.first()
item.set_done(True, user=it_user)
self.assertTrue(task.checklist.get(pk=item.pk).is_done)
self.assertEqual(task.checklist_done, 1)
def test_create_without_access_creates_no_task(self):
hr = create_hiring_request(
requester=self.hr_user, last_name='Без', first_name='Доступів',
user_location_type='pharmacy', hire_date=date.today() + timedelta(days=1), trial_days=3,
)
self.assertIsNone(hr.access_task)
self.assertIsNotNone(hr.granted_ticket)
self.assertEqual(hr.requested_access.count(), len(access_ids))
self.assertIn('Доступний', hr.granted_ticket.description)
self.assertEqual(Ticket.objects.filter(pk=hr.granted_ticket_id).count(), 1)
def test_approve_ends_probation_without_ticket(self):
hr = create_hiring_request(
requester=self.hr_user, last_name='Кандидат', first_name='Марія',
user_location_type='pharmacy', hire_date=date(2026, 6, 1), trial_days=5,
)
def test_create_without_access_still_creates_ticket(self):
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Без', first_name='Доступів',
))
self.assertIsNone(hr.access_task)
self.assertIsNotNone(hr.granted_ticket)
def test_access_for_office_only_skipped_for_pharmacy(self):
access = Directory.objects.filter(
directory_type__code='access_system', is_active=True,
).first()
self.assertIsNotNone(access)
loc_type = DirectoryType.objects.get(code='location_type')
office_lt = Directory.objects.get_or_create(
directory_type=loc_type, code='office',
defaults={'name': 'Офіс', 'is_active': True},
)[0]
access.location_types.set([office_lt])
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Фільтр', first_name='Доступ',
requested_access_ids=[access.pk],
))
self.assertEqual(hr.requested_access.count(), 0)
make_intern('Intern-2')
hr_office = create_hiring_request(**self._office_kwargs(
last_name='Фільтр', first_name='Офіс',
requested_access_ids=[access.pk],
))
self.assertEqual(list(hr_office.requested_access.values_list('pk', flat=True)), [access.pk])
def test_approve_creates_employee_and_frees_intern(self):
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Кандидат', first_name='Марія',
hire_date=date(2026, 6, 1), trial_days=5,
))
approved = approve_hiring_request(hr, decided_by=self.hr_user, note='ok')
self.assertEqual(approved.status, 'approved')
self.assertEqual(approved.decision_by, self.hr_user)
self.assertIsNone(approved.granted_ticket) # доступи — задачею при створенні, не тікетом
approved.employee.refresh_from_db()
self.assertIsNotNone(approved.employee)
self.assertTrue(approved.issued_password)
self.assertEqual(approved.employee.probation_end_date, date.today())
self.assertTrue(approved.employee.check_password(approved.issued_password))
def test_reject_deactivates_employee(self):
hr = create_hiring_request(
requester=self.hr_user, last_name='Кандидат', first_name='Олег',
user_location_type='pharmacy', hire_date=date(2026, 6, 1), trial_days=2,
self.intern.refresh_from_db()
self.assertIsNone(self.intern.assigned_to_id)
def test_approve_office_assigns_corporate_email(self):
from apps.employees.models import EmployeeDefaults
EmployeeDefaults.load()
hr = create_hiring_request(**self._office_kwargs(
last_name='Петренко', first_name='Іван',
))
approve_hiring_request(hr, decided_by=self.hr_user)
hr.refresh_from_db()
self.assertEqual(
hr.employee.email,
f'{hr.employee.username}@aks-apteka.com.ua',
)
def test_approve_office_uses_admin_domain_setting(self):
from apps.employees.models import EmployeeDefaults
defaults = EmployeeDefaults.load()
defaults.office_email_domain = 'office.example.com'
defaults.save()
hr = create_hiring_request(**self._office_kwargs(
last_name='Сидоренко', first_name='Оля',
))
approve_hiring_request(hr, decided_by=self.hr_user)
hr.refresh_from_db()
self.assertEqual(
hr.employee.email,
f'{hr.employee.username}@office.example.com',
)
def test_approve_pharmacy_uses_placeholder_email(self):
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Аптечна', first_name='Марія',
))
approve_hiring_request(hr, decided_by=self.hr_user)
hr.refresh_from_db()
self.assertTrue(hr.employee.email.endswith('@local.domain'))
def test_approve_ad_sync_only_for_office(self):
office_hr = create_hiring_request(**self._office_kwargs())
make_intern('Intern-2')
pharmacy_hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Аптечний', first_name='Кандидат',
))
approve_hiring_request(office_hr, decided_by=self.hr_user)
approve_hiring_request(pharmacy_hr, decided_by=self.hr_user)
office_hr.refresh_from_db()
pharmacy_hr.refresh_from_db()
self.assertTrue(office_hr.employee.ad_sync_enabled)
self.assertFalse(pharmacy_hr.employee.ad_sync_enabled)
def test_approve_generates_unique_username(self):
hr1 = create_hiring_request(**self._pharmacy_kwargs(
last_name='Дубль', first_name='Петро', hire_date=date(2026, 6, 1), trial_days=1,
))
make_intern('Intern-2')
hr2 = create_hiring_request(**self._pharmacy_kwargs(
last_name='Дубль', first_name='Петро', hire_date=date(2026, 6, 1), trial_days=1,
))
approve_hiring_request(hr1, decided_by=self.hr_user)
approve_hiring_request(hr2, decided_by=self.hr_user)
hr1.refresh_from_db()
hr2.refresh_from_db()
self.assertNotEqual(hr1.employee.username, hr2.employee.username)
def test_reject_frees_intern_closes_ticket_no_employee(self):
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Кандидат', first_name='Олег',
hire_date=date(2026, 6, 1), trial_days=2,
))
ticket = hr.granted_ticket
rejected = reject_hiring_request(hr, decided_by=self.hr_user, note='не пройшов')
self.assertEqual(rejected.status, 'rejected')
rejected.employee.refresh_from_db()
self.assertFalse(rejected.employee.is_active)
self.assertEqual(rejected.employee.termination_date, date.today())
self.assertIsNone(rejected.employee)
self.intern.refresh_from_db()
self.assertIsNone(self.intern.assigned_to_id)
ticket.refresh_from_db()
self.assertEqual(ticket.status.code, 'closed')
def test_decision_is_a_noop_once_already_decided(self):
hr = create_hiring_request(
requester=self.hr_user, last_name='Кандидат', first_name='Ліда',
user_location_type='pharmacy', hire_date=date(2026, 6, 1), trial_days=2,
)
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='Кандидат', first_name='Ліда',
hire_date=date(2026, 6, 1), trial_days=2,
))
approve_hiring_request(hr, decided_by=self.hr_user)
reject_hiring_request(hr, decided_by=self.hr_user)
hr.refresh_from_db()
self.assertEqual(hr.status, 'approved')
def test_credentials_pdf_after_approve(self):
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='PDF', first_name='Тест',
))
approve_hiring_request(hr, decided_by=self.hr_user)
client = Client()
login_2fa(client, self.hr_user)
url = reverse('hiring_credentials_pdf', kwargs={'pk': hr.pk})
response = client.get(url)
self.assertEqual(response.status_code, 200)
self.assertEqual(response['Content-Type'], 'application/pdf')
self.assertTrue(response.content.startswith(b'%PDF'))
def test_credentials_pdf_forbidden_while_trial(self):
hr = create_hiring_request(**self._pharmacy_kwargs(
last_name='PDF', first_name='Trial',
))
client = Client()
login_2fa(client, self.hr_user)
response = client.get(reverse('hiring_credentials_pdf', kwargs={'pk': hr.pk}))
self.assertEqual(response.status_code, 403)