533 lines
18 KiB
Python
533 lines
18 KiB
Python
from datetime import date, datetime, time, timedelta
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.utils import timezone
|
|
|
|
from attendance.models import AttendanceAnalysisRun, AttendanceRecord
|
|
from employees.constants import (
|
|
ACTION_DISMISSED,
|
|
ACTION_HIRED,
|
|
ACTION_ON_DAY_OFF,
|
|
ACTION_ON_SICK_LEAVE,
|
|
ACTION_REMOTE,
|
|
)
|
|
from employees.models import (
|
|
EmployeeAction,
|
|
EmployeeSkillProgress,
|
|
SecretSkillSettings,
|
|
Skill,
|
|
)
|
|
from employees.services.secret_skills import recalculate_attendance_secret_skills
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _use_full_test_history():
|
|
settings = SecretSkillSettings.get_solo()
|
|
settings.calculation_start_date = date(2000, 1, 1)
|
|
settings.save(update_fields=["calculation_start_date", "updated_at"])
|
|
|
|
|
|
def _run(employee, start: date, end: date):
|
|
return AttendanceAnalysisRun.objects.create(
|
|
employee=employee,
|
|
period_start=start,
|
|
period_end=end,
|
|
status=AttendanceAnalysisRun.STATUS_SUCCESS,
|
|
)
|
|
|
|
|
|
def _record(run, employee, record_date, **overrides):
|
|
values = {
|
|
"arrival_time": "08:00",
|
|
"departure_time": "17:00",
|
|
"work_hours": 9,
|
|
"expected_hours": 9,
|
|
"is_workday": True,
|
|
"effective_is_workday": True,
|
|
"personnel_status": "normal",
|
|
}
|
|
values.update(overrides)
|
|
return AttendanceRecord.objects.create(
|
|
analysis_run=run,
|
|
employee=employee,
|
|
date=record_date,
|
|
**values,
|
|
)
|
|
|
|
|
|
def _action(employee, action, action_date, **overrides):
|
|
values = {
|
|
"date": timezone.make_aware(datetime.combine(action_date, time.min)),
|
|
}
|
|
values.update(overrides)
|
|
return EmployeeAction.objects.create(
|
|
employee=employee,
|
|
action=action,
|
|
**values,
|
|
)
|
|
|
|
|
|
def test_punctuality_uses_best_streak_and_pauses_unknown_days(user_factory):
|
|
employee = user_factory()
|
|
start = date(2026, 1, 1)
|
|
run = _run(employee, start, start + timedelta(days=21))
|
|
|
|
for offset in range(10):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=10),
|
|
arrival_time=None,
|
|
departure_time=None,
|
|
work_hours=0,
|
|
technical_issues=["Нет данных контроллера"],
|
|
)
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=11),
|
|
arrival_time=None,
|
|
departure_time=None,
|
|
work_hours=0,
|
|
personnel_status=ACTION_REMOTE,
|
|
)
|
|
for offset in range(12, 22):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
|
|
assert recalculate_attendance_secret_skills(employee.id) == ["Пунктуальность"]
|
|
skill = Skill.objects.get(system_code="attendance_punctuality")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["best_streak"] == 20
|
|
assert progress.progress_data["current_streak"] == 20
|
|
assert employee.skills.filter(pk=skill.pk).exists()
|
|
|
|
late_record = _record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=22),
|
|
is_late=True,
|
|
late_minutes=15,
|
|
)
|
|
assert recalculate_attendance_secret_skills(employee.id) == []
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["current_streak"] == 0
|
|
assert progress.progress_data["anti_streak"] == 1
|
|
|
|
late_record.delete()
|
|
assert recalculate_attendance_secret_skills(employee.id) == []
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["anti_streak"] == 0
|
|
|
|
|
|
def test_attendance_recalculation_respects_since_boundary(user_factory):
|
|
employee = user_factory()
|
|
start = date(2026, 1, 1)
|
|
run = _run(employee, start, start + timedelta(days=24))
|
|
for offset in range(25):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
skill = Skill.objects.get(system_code="attendance_punctuality")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
|
|
recalculate_attendance_secret_skills(
|
|
employee.id,
|
|
since=start + timedelta(days=10),
|
|
)
|
|
|
|
progress.refresh_from_db()
|
|
assert progress.progress_data["best_streak"] == 15
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
|
|
|
|
def test_self_organization_requires_every_bronze_condition(
|
|
user_factory,
|
|
):
|
|
employee = user_factory()
|
|
start = date(2024, 1, 1)
|
|
run = _run(employee, start, start + timedelta(days=64))
|
|
|
|
for offset in range(60):
|
|
overtime = offset < 5
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_overtime=overtime,
|
|
overtime_hours=1.5 if overtime else None,
|
|
)
|
|
off_schedule_records = []
|
|
for offset in range(60, 65):
|
|
off_schedule_records.append(
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_workday=False,
|
|
effective_is_workday=False,
|
|
is_overtime=True,
|
|
overtime_hours=2,
|
|
)
|
|
)
|
|
|
|
changes = recalculate_attendance_secret_skills(employee.id)
|
|
assert "Самоорганизация" in changes
|
|
skill = Skill.objects.get(system_code="attendance_self_organization")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["clean_days"] == 60
|
|
assert progress.progress_data["best_clean_streak"] == 60
|
|
assert progress.progress_data["off_schedule_days"] == 5
|
|
assert progress.progress_data["overtime_days"] == 10
|
|
|
|
off_schedule_records[0].is_overtime = False
|
|
off_schedule_records[0].overtime_hours = None
|
|
off_schedule_records[0].save(update_fields=["is_overtime", "overtime_hours"])
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.progress_data["off_schedule_days"] == 4
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
assert not employee.skills.filter(pk=skill.pk).exists()
|
|
|
|
|
|
def test_punctuality_anti_streak_demotes_revokes_and_recovers(user_factory):
|
|
employee = user_factory()
|
|
start = date(2025, 1, 1)
|
|
run = _run(employee, start, start + timedelta(days=90))
|
|
|
|
for offset in range(45):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
skill = Skill.objects.get(system_code="attendance_punctuality")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "silver"
|
|
|
|
for offset in range(45, 48):
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_late=True,
|
|
late_minutes=10,
|
|
)
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["penalty_levels"] == 1
|
|
assert progress.progress_data["anti_streak"] == 0
|
|
|
|
for offset in range(48, 51):
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_late=True,
|
|
late_minutes=10,
|
|
)
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
assert progress.progress_data["penalty_levels"] == 2
|
|
assert not employee.skills.filter(pk=skill.pk).exists()
|
|
|
|
for offset in range(51, 71):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.is_active is True
|
|
assert progress.progress_data["penalty_levels"] == 1
|
|
assert employee.skills.filter(pk=skill.pk).exists()
|
|
|
|
for offset in range(71, 91):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "silver"
|
|
assert progress.progress_data["penalty_levels"] == 0
|
|
|
|
|
|
def test_self_organization_anti_streak_revokes_and_recovers(user_factory):
|
|
employee = user_factory()
|
|
start = date(2024, 1, 1)
|
|
run = _run(employee, start, start + timedelta(days=99))
|
|
|
|
for offset in range(60):
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_overtime=offset < 5,
|
|
overtime_hours=1.5 if offset < 5 else None,
|
|
)
|
|
for offset in range(60, 65):
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_workday=False,
|
|
effective_is_workday=False,
|
|
is_overtime=True,
|
|
overtime_hours=2,
|
|
)
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
skill = Skill.objects.get(system_code="attendance_self_organization")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
|
|
for offset in range(65, 70):
|
|
_record(
|
|
run,
|
|
employee,
|
|
start + timedelta(days=offset),
|
|
is_late=True,
|
|
late_minutes=10,
|
|
)
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
assert progress.progress_data["penalty_levels"] == 1
|
|
|
|
for offset in range(70, 100):
|
|
_record(run, employee, start + timedelta(days=offset))
|
|
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.is_active is True
|
|
assert progress.progress_data["penalty_levels"] == 0
|
|
|
|
|
|
def test_reliability_uses_only_personnel_events_and_recovers_rank(user_factory):
|
|
employee = user_factory()
|
|
start = date(2025, 1, 1)
|
|
_action(employee, ACTION_HIRED, start)
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=59),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
skill = Skill.objects.get(system_code="attendance_reliability")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["worked_days"] == 60
|
|
|
|
for offset in (60, 65, 70):
|
|
_action(
|
|
employee,
|
|
ACTION_ON_DAY_OFF,
|
|
start + timedelta(days=offset),
|
|
date_to=start + timedelta(days=offset),
|
|
)
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=70),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
assert progress.progress_data["penalty_levels"] == 1
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=130),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.is_active is True
|
|
assert progress.progress_data["worked_days"] == 128
|
|
assert progress.progress_data["leave_or_day_off_episodes"] == 3
|
|
assert progress.progress_data["penalty_levels"] == 0
|
|
|
|
|
|
def test_personnel_recalculation_counts_days_only_since_boundary(user_factory):
|
|
employee = user_factory()
|
|
start = date(2025, 1, 1)
|
|
_action(employee, ACTION_HIRED, start)
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=89),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
skill = Skill.objects.get(system_code="attendance_reliability")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
|
|
recalculate_attendance_secret_skills(
|
|
employee.id,
|
|
since=start + timedelta(days=40),
|
|
)
|
|
|
|
progress.refresh_from_db()
|
|
assert progress.progress_data["worked_days"] == 50
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
|
|
|
|
def test_personnel_cutoff_after_dismissal_saves_json_state(user_factory):
|
|
employee = user_factory()
|
|
start = date(2025, 1, 1)
|
|
_action(employee, ACTION_HIRED, start)
|
|
_action(employee, ACTION_DISMISSED, start + timedelta(days=10))
|
|
skill = Skill.objects.get(system_code="attendance_reliability")
|
|
progress = EmployeeSkillProgress.objects.create(
|
|
employee=employee,
|
|
skill=skill,
|
|
special_rank="bronze",
|
|
is_active=True,
|
|
)
|
|
employee.skills.add(skill)
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=100),
|
|
):
|
|
recalculate_attendance_secret_skills(
|
|
employee.id,
|
|
since=start + timedelta(days=50),
|
|
)
|
|
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
assert progress.progress_data["recent_violation_dates"] == []
|
|
assert "_rolling_dates" not in progress.progress_data
|
|
|
|
|
|
def test_remote_reactivates_employment_without_resetting_active_period(user_factory):
|
|
employee = user_factory()
|
|
start = date(2025, 1, 1)
|
|
_action(employee, ACTION_HIRED, start)
|
|
_action(employee, ACTION_REMOTE, start + timedelta(days=30))
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=59),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
|
|
skill = Skill.objects.get(system_code="attendance_reliability")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["worked_days"] == 60
|
|
|
|
_action(employee, ACTION_DISMISSED, start + timedelta(days=60))
|
|
_action(employee, ACTION_REMOTE, start + timedelta(days=70))
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=129),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["worked_days"] == 60
|
|
assert (
|
|
progress.progress_data["last_evaluated_date"]
|
|
== (start + timedelta(days=129)).isoformat()
|
|
)
|
|
|
|
|
|
def test_resilience_counts_sick_episodes_and_recovers_rank(user_factory):
|
|
employee = user_factory()
|
|
start = date(2024, 1, 1)
|
|
_action(employee, ACTION_HIRED, start)
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=179),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
skill = Skill.objects.get(system_code="personnel_resilience")
|
|
progress = EmployeeSkillProgress.objects.get(employee=employee, skill=skill)
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["best_resilience_streak"] == 180
|
|
|
|
_action(
|
|
employee,
|
|
ACTION_ON_SICK_LEAVE,
|
|
start + timedelta(days=180),
|
|
date_to=start + timedelta(days=189),
|
|
)
|
|
_action(employee, ACTION_ON_SICK_LEAVE, start + timedelta(days=195))
|
|
_action(employee, ACTION_ON_SICK_LEAVE, start + timedelta(days=200))
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=200),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "none"
|
|
assert progress.is_active is False
|
|
assert progress.progress_data["sick_leave_episodes"] == 3
|
|
assert progress.progress_data["penalty_levels"] == 1
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=290),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
progress.refresh_from_db()
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.is_active is True
|
|
assert progress.progress_data["current_resilience_streak"] == 90
|
|
assert progress.progress_data["penalty_levels"] == 0
|
|
|
|
|
|
def test_resilience_counts_calendar_days_without_attendance(user_factory):
|
|
employee = user_factory()
|
|
start = date(2024, 1, 1)
|
|
_action(employee, ACTION_HIRED, start)
|
|
|
|
with patch(
|
|
"employees.services.secret_skills.timezone.localdate",
|
|
return_value=start + timedelta(days=179),
|
|
):
|
|
recalculate_attendance_secret_skills(employee.id)
|
|
|
|
skill = Skill.objects.get(system_code="personnel_resilience")
|
|
progress = EmployeeSkillProgress.objects.get(
|
|
employee=employee,
|
|
skill=skill,
|
|
)
|
|
assert progress.special_rank == "bronze"
|
|
assert progress.progress_data["best_resilience_streak"] == 180
|
|
|
|
|
|
def test_personnel_action_schedules_secret_skill_recalculation(
|
|
user_factory,
|
|
django_capture_on_commit_callbacks,
|
|
):
|
|
employee = user_factory()
|
|
with patch(
|
|
"employees.services.secret_skills.recalculate_attendance_secret_skills"
|
|
) as recalculate:
|
|
with django_capture_on_commit_callbacks(execute=True):
|
|
_action(employee, ACTION_ON_SICK_LEAVE, date(2026, 8, 1))
|
|
|
|
recalculate.assert_called_once_with(employee.id)
|