Merge branch 'develop'
This commit is contained in:
commit
a0c65b073e
@ -1,11 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from io import BytesIO
|
||||
from datetime import date, timedelta
|
||||
from calendar import monthrange
|
||||
from typing import Any
|
||||
|
||||
from django.conf import settings
|
||||
from django.conf import settings as django_settings
|
||||
from django.db import transaction
|
||||
from django.utils import timezone
|
||||
from django.utils.dateparse import parse_date
|
||||
@ -85,6 +86,8 @@ MATRIX_STATUS_SHORT_LABELS = {
|
||||
"normal": "OK",
|
||||
}
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PERSONNEL_STATUS_SHORT_LABELS = {
|
||||
ACTION_ON_LEAVE: "ОТП",
|
||||
ACTION_ON_SICK_LEAVE: "БЛ",
|
||||
@ -258,6 +261,34 @@ def get_attendance_auto_sync_employees():
|
||||
return Employee.objects.prefetch_related("actions").order_by("-id")
|
||||
|
||||
|
||||
def get_attendance_auto_sync_stale_after() -> timedelta:
|
||||
configured_seconds = getattr(
|
||||
django_settings,
|
||||
"ATTENDANCE_AUTO_SYNC_STALE_AFTER_SECONDS",
|
||||
None,
|
||||
)
|
||||
if configured_seconds is None:
|
||||
configured_seconds = (
|
||||
getattr(django_settings, "CELERY_TASK_TIME_LIMIT", 30 * 60) + 300
|
||||
)
|
||||
try:
|
||||
seconds = int(configured_seconds)
|
||||
except (TypeError, ValueError):
|
||||
seconds = 30 * 60 + 300
|
||||
return timedelta(seconds=max(seconds, 60))
|
||||
|
||||
|
||||
def is_attendance_auto_sync_stale(
|
||||
settings_obj: AttendanceAutoSyncSettings,
|
||||
now,
|
||||
) -> bool:
|
||||
if settings_obj.last_status != AttendanceAutoSyncSettings.STATUS_RUNNING:
|
||||
return False
|
||||
if not settings_obj.last_started_at:
|
||||
return True
|
||||
return now - settings_obj.last_started_at > get_attendance_auto_sync_stale_after()
|
||||
|
||||
|
||||
def run_attendance_auto_sync(*, force: bool = False) -> AttendanceAutoSyncSettings:
|
||||
from common.logstorm_attendance import (
|
||||
analyze_employee_attendance,
|
||||
@ -265,6 +296,7 @@ def run_attendance_auto_sync(*, force: bool = False) -> AttendanceAutoSyncSettin
|
||||
)
|
||||
now = timezone.now()
|
||||
with transaction.atomic():
|
||||
recovered_stale_running = False
|
||||
settings = (
|
||||
AttendanceAutoSyncSettings.objects.select_for_update()
|
||||
.filter(singleton=True)
|
||||
@ -274,11 +306,42 @@ def run_attendance_auto_sync(*, force: bool = False) -> AttendanceAutoSyncSettin
|
||||
settings = AttendanceAutoSyncSettings.objects.create(singleton=True)
|
||||
|
||||
if settings.last_status == AttendanceAutoSyncSettings.STATUS_RUNNING:
|
||||
return settings
|
||||
if not is_attendance_auto_sync_stale(settings, now):
|
||||
return settings
|
||||
|
||||
recovered_stale_running = True
|
||||
stale_started_at = settings.last_started_at
|
||||
stale_message = (
|
||||
"Recovered stale running attendance auto-sync"
|
||||
f" started at {stale_started_at.isoformat()}"
|
||||
if stale_started_at
|
||||
else "Recovered stale running attendance auto-sync without start time"
|
||||
)
|
||||
logger.warning(stale_message)
|
||||
settings.last_status = AttendanceAutoSyncSettings.STATUS_FAILED
|
||||
settings.last_finished_at = now
|
||||
settings.last_error = stale_message
|
||||
settings.last_error_count = 1
|
||||
settings.next_run_at = now if settings.enabled else None
|
||||
settings.save(
|
||||
update_fields=[
|
||||
"last_status",
|
||||
"last_finished_at",
|
||||
"last_error",
|
||||
"last_error_count",
|
||||
"next_run_at",
|
||||
"updated_at",
|
||||
]
|
||||
)
|
||||
|
||||
if not force:
|
||||
if not settings.enabled:
|
||||
return settings
|
||||
if settings.next_run_at and settings.next_run_at > now:
|
||||
if (
|
||||
not recovered_stale_running
|
||||
and settings.next_run_at
|
||||
and settings.next_run_at > now
|
||||
):
|
||||
return settings
|
||||
|
||||
settings.last_status = AttendanceAutoSyncSettings.STATUS_RUNNING
|
||||
@ -632,7 +695,11 @@ def _append_attendance_matrix_legend_sheet(workbook: Workbook) -> None:
|
||||
|
||||
|
||||
def _attendance_record_day_events_url(record_id: int) -> str:
|
||||
site_url = getattr(settings, "SITE_URL", "https://corp.robotail.pro").rstrip("/")
|
||||
site_url = getattr(
|
||||
django_settings,
|
||||
"SITE_URL",
|
||||
"https://corp.robotail.pro",
|
||||
).rstrip("/")
|
||||
return f"{site_url}/attendance?record={record_id}&events=1"
|
||||
|
||||
|
||||
|
||||
@ -5,6 +5,7 @@
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
from celery import shared_task
|
||||
from django.conf import settings
|
||||
@ -432,6 +433,25 @@ def process_ldap_queue():
|
||||
return
|
||||
|
||||
now = timezone.now()
|
||||
task_time_limit = getattr(settings, "CELERY_TASK_TIME_LIMIT", 30 * 60)
|
||||
stale_cutoff = now - timedelta(seconds=task_time_limit + 300)
|
||||
stale_count = LdapSyncQueue.objects.filter(
|
||||
status=LdapSyncQueue.Status.IN_PROGRESS,
|
||||
updated_at__lt=stale_cutoff,
|
||||
).update(
|
||||
status=LdapSyncQueue.Status.PENDING,
|
||||
next_retry_at=now,
|
||||
last_error=(
|
||||
"Recovered stale in-progress LDAP sync item after worker interruption"
|
||||
),
|
||||
updated_at=now,
|
||||
)
|
||||
if stale_count:
|
||||
logger.warning(
|
||||
"Recovered %d stale LDAP queue items from in_progress to pending",
|
||||
stale_count,
|
||||
)
|
||||
|
||||
pending = LdapSyncQueue.objects.filter(
|
||||
status=LdapSyncQueue.Status.PENDING,
|
||||
next_retry_at__lte=now,
|
||||
|
||||
@ -3,7 +3,6 @@
|
||||
"""
|
||||
import os
|
||||
from celery import Celery
|
||||
from celery.schedules import crontab
|
||||
|
||||
# Устанавливаем настройки Django по умолчанию
|
||||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'eusrr_backend.settings')
|
||||
@ -18,22 +17,6 @@ app.config_from_object('django.conf:settings', namespace='CELERY')
|
||||
# Автоматически находит tasks.py в каждом приложении
|
||||
app.autodiscover_tasks()
|
||||
|
||||
# Periodic tasks (Celery Beat)
|
||||
app.conf.beat_schedule = {
|
||||
'cleanup-missed-returns': {
|
||||
'task': 'requests_app.tasks.cleanup_missed_returns',
|
||||
'schedule': crontab(hour=0, minute=5), # Каждый день в 00:05
|
||||
},
|
||||
'process-due-personnel-actions': {
|
||||
'task': 'requests_app.tasks.process_due_personnel_actions',
|
||||
'schedule': crontab(hour=0, minute=10), # Каждый день в 00:10
|
||||
},
|
||||
'attendance-auto-sync-dispatcher': {
|
||||
'task': 'attendance.tasks.dispatch_attendance_auto_sync',
|
||||
'schedule': 60.0,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@app.task(bind=True, ignore_result=True)
|
||||
def debug_task(self):
|
||||
|
||||
@ -3,6 +3,7 @@ import tempfile
|
||||
from datetime import timedelta
|
||||
from pathlib import Path
|
||||
|
||||
from celery.schedules import crontab
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from dotenv import load_dotenv
|
||||
@ -708,10 +709,24 @@ CELERY_BEAT_SCHEDULE = {
|
||||
"task": "communications.tasks.cleanup_orphaned_attachments",
|
||||
"schedule": 3600.0, # Каждый час
|
||||
},
|
||||
"notifications-daily-digests": {
|
||||
"task": "notifications.send_digest_emails",
|
||||
"schedule": crontab(hour=9, minute=0),
|
||||
"args": ("daily",),
|
||||
},
|
||||
"notifications-weekly-digests": {
|
||||
"task": "notifications.send_digest_emails",
|
||||
"schedule": crontab(hour=9, minute=0, day_of_week="monday"),
|
||||
"args": ("weekly",),
|
||||
},
|
||||
"process-ldap-sync-queue": {
|
||||
"task": "employees.tasks.process_ldap_queue",
|
||||
"schedule": 60.0, # Каждую минуту проверяем очередь LDAP retry
|
||||
},
|
||||
"process-due-personnel-actions": {
|
||||
"task": "requests_app.tasks.process_due_personnel_actions",
|
||||
"schedule": crontab(hour=0, minute=10), # Каждый день в 00:10
|
||||
},
|
||||
"attendance-auto-sync-dispatcher": {
|
||||
"task": "attendance.tasks.dispatch_attendance_auto_sync",
|
||||
"schedule": 60.0,
|
||||
|
||||
@ -21,6 +21,20 @@ import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _enqueue_notification_task(task, *args, **kwargs) -> bool:
|
||||
try:
|
||||
task.delay(*args, **kwargs)
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Failed to enqueue notification task %s: %s",
|
||||
getattr(task, "name", repr(task)),
|
||||
exc,
|
||||
exc_info=True,
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
@receiver(post_save, sender="notifications.Notification")
|
||||
def route_notification_to_channels(sender, instance, created, **kwargs):
|
||||
"""
|
||||
@ -54,7 +68,9 @@ def route_notification_to_channels(sender, instance, created, **kwargs):
|
||||
# Если настроек нет - создаем с дефолтными значениями
|
||||
from .models import UserChannelPreferences
|
||||
|
||||
prefs = UserChannelPreferences.objects.create(user=user)
|
||||
prefs, _created = UserChannelPreferences.objects.get_or_create(
|
||||
user=user
|
||||
)
|
||||
|
||||
# Проверяем, не отключен ли этот тип уведомлений
|
||||
if not prefs.is_verb_enabled(notification.verb):
|
||||
@ -88,18 +104,26 @@ def route_notification_to_channels(sender, instance, created, **kwargs):
|
||||
} is in DND period, only web notifications (silent)"
|
||||
)
|
||||
if prefs.web_enabled:
|
||||
send_websocket_notification.delay(notification.id, silent=True)
|
||||
_enqueue_notification_task(
|
||||
send_websocket_notification,
|
||||
notification.id,
|
||||
silent=True,
|
||||
)
|
||||
return
|
||||
|
||||
# Отправляем по каналам асинхронно через Celery
|
||||
if prefs.web_enabled:
|
||||
send_websocket_notification.delay(notification.id, silent=False)
|
||||
_enqueue_notification_task(
|
||||
send_websocket_notification,
|
||||
notification.id,
|
||||
silent=False,
|
||||
)
|
||||
|
||||
if prefs.email_enabled and prefs.email_frequency == "instant":
|
||||
send_email_notification.delay(notification.id)
|
||||
_enqueue_notification_task(send_email_notification, notification.id)
|
||||
|
||||
if prefs.push_enabled:
|
||||
send_push_notification.delay(notification.id)
|
||||
_enqueue_notification_task(send_push_notification, notification.id)
|
||||
|
||||
transaction.on_commit(send_to_channels)
|
||||
|
||||
|
||||
@ -14,6 +14,7 @@ from django.core.mail import send_mail
|
||||
|
||||
from .base import BaseNotificationSender
|
||||
from notifications import config
|
||||
from notifications.models import Notification
|
||||
|
||||
|
||||
class EmailNotificationSender(BaseNotificationSender):
|
||||
@ -127,9 +128,7 @@ class EmailNotificationSender(BaseNotificationSender):
|
||||
self.log_error(notification, e, recipient_email)
|
||||
return False
|
||||
|
||||
def send_digest(
|
||||
self, user, notifications: list, frequency: str = "daily"
|
||||
) -> bool:
|
||||
def send_digest(self, user, notifications, frequency: str = "daily") -> bool:
|
||||
"""
|
||||
Отправляет email дайджест уведомлений.
|
||||
|
||||
@ -142,6 +141,7 @@ class EmailNotificationSender(BaseNotificationSender):
|
||||
True если отправлено успешно, False иначе
|
||||
"""
|
||||
try:
|
||||
notifications = list(notifications)
|
||||
if not notifications:
|
||||
return False
|
||||
|
||||
@ -198,8 +198,13 @@ class EmailNotificationSender(BaseNotificationSender):
|
||||
fail_silently=False,
|
||||
)
|
||||
|
||||
# Отмечаем как отправленные
|
||||
notifications.update(emailed=True)
|
||||
# Отмечаем как отправленные. На вход может прийти как QuerySet,
|
||||
# так и список из Celery-задачи, поэтому обновляем по id.
|
||||
notification_ids = [item.id for item in notifications if item.id]
|
||||
if notification_ids:
|
||||
Notification.objects.filter(id__in=notification_ids).update(
|
||||
emailed=True
|
||||
)
|
||||
|
||||
self.logger.info(
|
||||
f"✅ Email дайджест отправлен: "
|
||||
|
||||
@ -87,6 +87,32 @@ class PushNotificationSender(BaseNotificationSender):
|
||||
|
||||
return self._get_data_title(notification) or self.DEFAULT_TITLE
|
||||
|
||||
def _is_permanent_device_error(self, error: Exception) -> bool:
|
||||
error_text = str(error).lower()
|
||||
permanent_markers = (
|
||||
"expired",
|
||||
"unregistered",
|
||||
"invalid subscription",
|
||||
"invalidregistration",
|
||||
"not found",
|
||||
"gone",
|
||||
"404",
|
||||
"410",
|
||||
"400 bad request",
|
||||
)
|
||||
return any(marker in error_text for marker in permanent_markers)
|
||||
|
||||
def _deactivate_device(self, device, reason: str) -> None:
|
||||
if not device.active:
|
||||
return
|
||||
device.active = False
|
||||
device.save(update_fields=["active"])
|
||||
self.logger.info(
|
||||
"Push device %s deactivated after permanent delivery error: %s",
|
||||
device.id,
|
||||
reason,
|
||||
)
|
||||
|
||||
def send(self, notification, **kwargs) -> bool:
|
||||
"""
|
||||
Отправляет Web Push уведомление.
|
||||
@ -172,18 +198,13 @@ class PushNotificationSender(BaseNotificationSender):
|
||||
failed_count += 1
|
||||
# Вычисляем размер payload для диагностики
|
||||
message_size = len(message.encode("utf-8"))
|
||||
error_text = str(e)
|
||||
self.logger.error(
|
||||
f"Ошибка отправки push на device {device.id}: {e} "
|
||||
f"(payload size: {message_size} bytes)"
|
||||
)
|
||||
# Деактивируем устройство если ошибка
|
||||
if (
|
||||
"expired" in str(e).lower()
|
||||
or "unregistered" in str(e).lower()
|
||||
):
|
||||
device.active = False
|
||||
device.save()
|
||||
self.logger.info(f"Device {device.id} деактивирован")
|
||||
if self._is_permanent_device_error(e):
|
||||
self._deactivate_device(device, error_text)
|
||||
|
||||
if sent_count > 0:
|
||||
self.log_success(notification, f"{sent_count} devices")
|
||||
|
||||
@ -9,7 +9,11 @@ Celery задачи для асинхронной отправки уведом
|
||||
- Единообразное логирование, retry, rate limiting
|
||||
"""
|
||||
|
||||
from .email import send_email_notification, send_digest_email
|
||||
from .email import (
|
||||
send_digest_email,
|
||||
send_digest_emails,
|
||||
send_email_notification,
|
||||
)
|
||||
from .push import send_push_notification
|
||||
from .websocket import send_websocket_notification
|
||||
|
||||
@ -18,4 +22,5 @@ __all__ = [
|
||||
"send_push_notification",
|
||||
"send_websocket_notification",
|
||||
"send_digest_email",
|
||||
"send_digest_emails",
|
||||
]
|
||||
|
||||
@ -2,6 +2,8 @@
|
||||
Celery задачи для отправки email уведомлений
|
||||
"""
|
||||
|
||||
from celery import shared_task
|
||||
|
||||
from .base import BaseNotificationTask
|
||||
from notifications import config
|
||||
|
||||
@ -122,3 +124,36 @@ class DigestEmailTask(BaseNotificationTask):
|
||||
# Регистрируем задачи в Celery
|
||||
send_email_notification = EmailNotificationTask.register_task()
|
||||
send_digest_email = DigestEmailTask.register_task()
|
||||
|
||||
|
||||
@shared_task(name="notifications.send_digest_emails")
|
||||
def send_digest_emails(frequency: str = "daily") -> int:
|
||||
"""
|
||||
Dispatches digest email tasks for users who selected daily/weekly delivery.
|
||||
"""
|
||||
if frequency not in {"daily", "weekly"}:
|
||||
raise ValueError(f"Unsupported digest frequency: {frequency}")
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from django.utils import timezone
|
||||
|
||||
from notifications.models import UserChannelPreferences
|
||||
|
||||
cutoff = timezone.now() - timedelta(days=7 if frequency == "weekly" else 1)
|
||||
preferences = UserChannelPreferences.objects.select_related("user").filter(
|
||||
email_enabled=True,
|
||||
email_frequency=frequency,
|
||||
user__email__isnull=False,
|
||||
user__notifications__unread=True,
|
||||
user__notifications__emailed=False,
|
||||
user__notifications__deleted=False,
|
||||
user__notifications__timestamp__gte=cutoff,
|
||||
).exclude(user__email="").distinct()
|
||||
|
||||
dispatched = 0
|
||||
for prefs in preferences.iterator():
|
||||
send_digest_email.delay(prefs.user_id, frequency)
|
||||
dispatched += 1
|
||||
|
||||
return dispatched
|
||||
|
||||
@ -4,6 +4,7 @@ from datetime import date, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.test import override_settings
|
||||
from django.urls import reverse
|
||||
from django.utils import timezone
|
||||
from openpyxl import load_workbook
|
||||
@ -731,6 +732,7 @@ def test_auto_sync_skips_parallel_running_state(user_factory):
|
||||
AttendanceAutoSyncSettings.objects.create(
|
||||
enabled=True,
|
||||
last_status=AttendanceAutoSyncSettings.STATUS_RUNNING,
|
||||
last_started_at=timezone.now(),
|
||||
next_run_at=timezone.now() - timedelta(minutes=1),
|
||||
)
|
||||
|
||||
@ -741,6 +743,59 @@ def test_auto_sync_skips_parallel_running_state(user_factory):
|
||||
assert settings.last_status == "running"
|
||||
|
||||
|
||||
@override_settings(ATTENDANCE_AUTO_SYNC_STALE_AFTER_SECONDS=1800)
|
||||
def test_auto_sync_recovers_stale_running_state_and_runs(user_factory):
|
||||
from attendance.services import run_attendance_auto_sync
|
||||
|
||||
employee = user_factory(first_name="Recovered", last_name="Employee")
|
||||
today = timezone.localdate()
|
||||
AttendanceAutoSyncSettings.objects.create(
|
||||
enabled=True,
|
||||
frequency_minutes=60,
|
||||
lookback_days=1,
|
||||
last_status=AttendanceAutoSyncSettings.STATUS_RUNNING,
|
||||
last_started_at=timezone.now() - timedelta(minutes=40),
|
||||
next_run_at=timezone.now() + timedelta(hours=6),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"common.logstorm_attendance.analyze_employee_attendance",
|
||||
return_value=_auto_sync_result(employee, today),
|
||||
) as analyze:
|
||||
settings = run_attendance_auto_sync(force=False)
|
||||
|
||||
analyze.assert_called_once()
|
||||
assert settings.last_status == "success"
|
||||
assert settings.last_success_count == 1
|
||||
assert settings.last_error_count == 0
|
||||
assert settings.next_run_at > settings.last_finished_at
|
||||
assert AttendanceRecord.objects.filter(employee=employee, date=today).exists()
|
||||
|
||||
|
||||
@override_settings(ATTENDANCE_AUTO_SYNC_STALE_AFTER_SECONDS=1800)
|
||||
def test_auto_sync_recovers_stale_running_state_when_disabled(user_factory):
|
||||
from attendance.services import run_attendance_auto_sync
|
||||
|
||||
user_factory()
|
||||
started_at = timezone.now() - timedelta(minutes=40)
|
||||
AttendanceAutoSyncSettings.objects.create(
|
||||
enabled=False,
|
||||
last_status=AttendanceAutoSyncSettings.STATUS_RUNNING,
|
||||
last_started_at=started_at,
|
||||
next_run_at=timezone.now() + timedelta(hours=6),
|
||||
)
|
||||
|
||||
with patch("common.logstorm_attendance.analyze_employee_attendance") as analyze:
|
||||
settings = run_attendance_auto_sync(force=False)
|
||||
|
||||
analyze.assert_not_called()
|
||||
assert settings.last_status == "failed"
|
||||
assert settings.last_finished_at is not None
|
||||
assert settings.last_error_count == 1
|
||||
assert "Recovered stale running attendance auto-sync" in settings.last_error
|
||||
assert settings.next_run_at is None
|
||||
|
||||
|
||||
def test_staff_can_run_auto_sync_now(auth_client_factory, user_factory):
|
||||
staff = user_factory(staff=True)
|
||||
client = auth_client_factory(staff)
|
||||
|
||||
70
backend/tests/api/v1/notifications/test_channels.py
Normal file
70
backend/tests/api/v1/notifications/test_channels.py
Normal file
@ -0,0 +1,70 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from notifications.models import Notification, UserChannelPreferences
|
||||
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
class TestNotificationChannelRouting:
|
||||
def test_enqueue_error_does_not_abort_notification_creation(
|
||||
self, user_factory, django_capture_on_commit_callbacks
|
||||
):
|
||||
user = user_factory(email="route-recipient@example.com")
|
||||
UserChannelPreferences.objects.create(
|
||||
user=user,
|
||||
web_enabled=True,
|
||||
email_enabled=True,
|
||||
email_frequency="instant",
|
||||
push_enabled=True,
|
||||
)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"notifications.tasks.send_websocket_notification.delay",
|
||||
side_effect=RuntimeError("broker unavailable"),
|
||||
) as websocket_delay,
|
||||
patch(
|
||||
"notifications.tasks.send_email_notification.delay"
|
||||
) as email_delay,
|
||||
patch(
|
||||
"notifications.tasks.send_push_notification.delay"
|
||||
) as push_delay,
|
||||
):
|
||||
with django_capture_on_commit_callbacks(execute=True):
|
||||
notification = Notification.objects.create(
|
||||
recipient=user,
|
||||
verb="request_new",
|
||||
description="Новая заявка",
|
||||
)
|
||||
|
||||
assert Notification.objects.filter(pk=notification.pk).exists()
|
||||
websocket_delay.assert_called_once_with(
|
||||
notification.id,
|
||||
silent=False,
|
||||
)
|
||||
email_delay.assert_called_once_with(notification.id)
|
||||
push_delay.assert_called_once_with(notification.id)
|
||||
|
||||
def test_missing_preferences_are_created_during_routing(
|
||||
self, user_factory, django_capture_on_commit_callbacks
|
||||
):
|
||||
user = user_factory(email="new-prefs-recipient@example.com")
|
||||
|
||||
with patch(
|
||||
"notifications.tasks.send_websocket_notification.delay"
|
||||
) as websocket_delay:
|
||||
with django_capture_on_commit_callbacks(execute=True):
|
||||
notification = Notification.objects.create(
|
||||
recipient=user,
|
||||
verb="request_new",
|
||||
description="Новая заявка",
|
||||
)
|
||||
|
||||
assert UserChannelPreferences.objects.filter(user=user).exists()
|
||||
websocket_delay.assert_called_once_with(
|
||||
notification.id,
|
||||
silent=False,
|
||||
)
|
||||
92
backend/tests/api/v1/notifications/test_email_sender.py
Normal file
92
backend/tests/api/v1/notifications/test_email_sender.py
Normal file
@ -0,0 +1,92 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from notifications.models import Notification, UserChannelPreferences
|
||||
from notifications.senders.email import EmailNotificationSender
|
||||
from notifications.tasks.email import send_digest_emails
|
||||
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def _disable_auto_delivery(user):
|
||||
UserChannelPreferences.objects.create(
|
||||
user=user,
|
||||
web_enabled=False,
|
||||
email_enabled=False,
|
||||
push_enabled=False,
|
||||
)
|
||||
|
||||
|
||||
class TestEmailNotificationSender:
|
||||
def test_digest_marks_list_notifications_as_emailed(self, user_factory):
|
||||
user = user_factory(email="digest-recipient@example.com")
|
||||
_disable_auto_delivery(user)
|
||||
first = Notification.objects.create(
|
||||
recipient=user,
|
||||
verb="request_new",
|
||||
description="Новая заявка",
|
||||
)
|
||||
second = Notification.objects.create(
|
||||
recipient=user,
|
||||
verb="document_ready",
|
||||
description="Документ готов",
|
||||
)
|
||||
|
||||
sender = EmailNotificationSender()
|
||||
with patch("notifications.senders.email.send_mail") as send_mail:
|
||||
result = sender.send_digest(user, [first, second], "daily")
|
||||
|
||||
assert result is True
|
||||
send_mail.assert_called_once()
|
||||
first.refresh_from_db()
|
||||
second.refresh_from_db()
|
||||
assert first.emailed is True
|
||||
assert second.emailed is True
|
||||
|
||||
|
||||
class TestDigestEmailDispatch:
|
||||
def test_dispatches_only_matching_users_with_pending_notifications(
|
||||
self, user_factory
|
||||
):
|
||||
daily_user = user_factory(email="daily-digest@example.com")
|
||||
weekly_user = user_factory(email="weekly-digest@example.com")
|
||||
empty_user = user_factory(email="empty-digest@example.com")
|
||||
instant_user = user_factory(email="instant@example.com")
|
||||
for user, frequency in (
|
||||
(daily_user, "daily"),
|
||||
(weekly_user, "weekly"),
|
||||
(empty_user, "daily"),
|
||||
(instant_user, "instant"),
|
||||
):
|
||||
UserChannelPreferences.objects.create(
|
||||
user=user,
|
||||
web_enabled=False,
|
||||
email_enabled=True,
|
||||
email_frequency=frequency,
|
||||
push_enabled=False,
|
||||
)
|
||||
Notification.objects.create(
|
||||
recipient=daily_user,
|
||||
verb="request_new",
|
||||
description="Новая заявка",
|
||||
)
|
||||
Notification.objects.create(
|
||||
recipient=weekly_user,
|
||||
verb="document_ready",
|
||||
description="Документ готов",
|
||||
)
|
||||
Notification.objects.create(
|
||||
recipient=instant_user,
|
||||
verb="request_new",
|
||||
description="Мгновенная заявка",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"notifications.tasks.email.send_digest_email.delay"
|
||||
) as digest_delay:
|
||||
dispatched = send_digest_emails("daily")
|
||||
|
||||
assert dispatched == 1
|
||||
digest_delay.assert_called_once_with(daily_user.id, "daily")
|
||||
@ -196,3 +196,51 @@ class TestPushNotificationSenderPayload:
|
||||
|
||||
assert len(payload["body"]) == 300
|
||||
assert payload["body"] == ("а" * 297) + "..."
|
||||
|
||||
def test_permanent_push_error_deactivates_device(self, user_factory):
|
||||
recipient = user_factory(email="stale-device-recipient@example.com")
|
||||
device = _make_push_device(recipient)
|
||||
notification = _make_notification(
|
||||
recipient,
|
||||
verb="document_ready",
|
||||
description="Документ готов",
|
||||
action_url="/documents",
|
||||
data={"title": "Документ"},
|
||||
)
|
||||
|
||||
sender = PushNotificationSender()
|
||||
with patch.object(
|
||||
WebPushDevice,
|
||||
"send_message",
|
||||
autospec=True,
|
||||
side_effect=Exception("Push failed: 400 Bad Request"),
|
||||
):
|
||||
result = sender.send(notification)
|
||||
|
||||
assert result is False
|
||||
device.refresh_from_db()
|
||||
assert device.active is False
|
||||
|
||||
def test_transient_push_error_keeps_device_active(self, user_factory):
|
||||
recipient = user_factory(email="transient-device-recipient@example.com")
|
||||
device = _make_push_device(recipient)
|
||||
notification = _make_notification(
|
||||
recipient,
|
||||
verb="document_ready",
|
||||
description="Документ готов",
|
||||
action_url="/documents",
|
||||
data={"title": "Документ"},
|
||||
)
|
||||
|
||||
sender = PushNotificationSender()
|
||||
with patch.object(
|
||||
WebPushDevice,
|
||||
"send_message",
|
||||
autospec=True,
|
||||
side_effect=TimeoutError("temporary timeout"),
|
||||
):
|
||||
result = sender.send(notification)
|
||||
|
||||
assert result is False
|
||||
device.refresh_from_db()
|
||||
assert device.active is True
|
||||
|
||||
@ -239,6 +239,31 @@ class TestProcessQueue:
|
||||
# Не подхватывает — retry ещё не наступил
|
||||
mock_task.delay.assert_not_called()
|
||||
|
||||
@override_settings(LDAP_ENABLED=True, CELERY_TASK_TIME_LIMIT=1800)
|
||||
def test_recovers_stale_in_progress_items(self, db):
|
||||
from employees.models import LdapSyncQueue
|
||||
from employees.tasks import process_ldap_queue
|
||||
|
||||
item = LdapSyncQueue.objects.create(
|
||||
operation="employee_save",
|
||||
model_name="employee",
|
||||
object_pk="1",
|
||||
payload={"object_pk": "1", "changes": {}},
|
||||
status=LdapSyncQueue.Status.IN_PROGRESS,
|
||||
)
|
||||
LdapSyncQueue.objects.filter(pk=item.pk).update(
|
||||
updated_at=timezone.now() - timedelta(minutes=40),
|
||||
)
|
||||
|
||||
with patch("employees.tasks.process_ldap_queue_item") as mock_task:
|
||||
mock_task.delay = MagicMock()
|
||||
process_ldap_queue()
|
||||
|
||||
item.refresh_from_db()
|
||||
assert item.status == LdapSyncQueue.Status.PENDING
|
||||
assert "Recovered stale" in item.last_error
|
||||
mock_task.delay.assert_called_once_with(item.pk)
|
||||
|
||||
|
||||
class TestDepartmentQueueExecutors:
|
||||
|
||||
|
||||
22
backend/tests/test_celery_schedule.py
Normal file
22
backend/tests/test_celery_schedule.py
Normal file
@ -0,0 +1,22 @@
|
||||
from django.conf import settings
|
||||
|
||||
from eusrr_backend.celery import app
|
||||
|
||||
|
||||
def test_due_personnel_actions_are_scheduled():
|
||||
entry = settings.CELERY_BEAT_SCHEDULE["process-due-personnel-actions"]
|
||||
|
||||
assert entry["task"] == "requests_app.tasks.process_due_personnel_actions"
|
||||
|
||||
|
||||
def test_scheduled_tasks_are_registered():
|
||||
app.loader.import_default_modules()
|
||||
app.autodiscover_tasks(force=True)
|
||||
|
||||
missing = [
|
||||
entry["task"]
|
||||
for entry in settings.CELERY_BEAT_SCHEDULE.values()
|
||||
if entry["task"] not in app.tasks
|
||||
]
|
||||
|
||||
assert missing == []
|
||||
@ -239,6 +239,7 @@ export function NotificationsProvider({ children }: { children: React.ReactNode
|
||||
if (['unread_count', 'notification_count_update'].includes(data.type) && typeof data.count === 'number') {
|
||||
setUnreadCount(data.count);
|
||||
void syncUnreadSummary();
|
||||
void syncUnreadNotifications(false);
|
||||
}
|
||||
});
|
||||
|
||||
@ -250,7 +251,7 @@ export function NotificationsProvider({ children }: { children: React.ReactNode
|
||||
unsubscribeMessages();
|
||||
unsubscribeStatus();
|
||||
};
|
||||
}, [syncUnreadSummary]);
|
||||
}, [syncUnreadNotifications, syncUnreadSummary]);
|
||||
|
||||
// Регулярная сверка с API нужна даже при живом WebSocket: часть событий может не дойти по сокету.
|
||||
useEffect(() => {
|
||||
@ -260,6 +261,7 @@ export function NotificationsProvider({ children }: { children: React.ReactNode
|
||||
|
||||
const sync = () => {
|
||||
void syncUnreadSummary();
|
||||
void syncUnreadNotifications(false);
|
||||
};
|
||||
|
||||
sync();
|
||||
@ -280,7 +282,7 @@ export function NotificationsProvider({ children }: { children: React.ReactNode
|
||||
window.removeEventListener('focus', handleFocus);
|
||||
document.removeEventListener('visibilitychange', handleVisibilityChange);
|
||||
};
|
||||
}, [syncUnreadSummary]);
|
||||
}, [syncUnreadNotifications, syncUnreadSummary]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!isWsConnected) {
|
||||
@ -288,7 +290,8 @@ export function NotificationsProvider({ children }: { children: React.ReactNode
|
||||
}
|
||||
|
||||
void syncUnreadSummary();
|
||||
}, [isWsConnected, syncUnreadSummary]);
|
||||
void syncUnreadNotifications(false);
|
||||
}, [isWsConnected, syncUnreadNotifications, syncUnreadSummary]);
|
||||
|
||||
const markAsRead = useCallback(async (id: number) => {
|
||||
await apiClient.markNotificationAsRead(id);
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user