449 lines
18 KiB
Python
449 lines
18 KiB
Python
"""Тесты Celery-задачи уведомления (`workers.tasks.notify.notify_session_async`).
|
||
|
||
Как и `test_summarize_task.py`, не использует savepoint-фикстуру `db_session`:
|
||
`notify_session_async` открывает собственную сессию с отдельным engine
|
||
(`workers/db.py::open_session`), которая не видит незакоммиченные изменения
|
||
другой сессии. Тестовые данные заводятся и коммитятся напрямую через
|
||
`core.db.engine`; email-транспорт — фейковый `EmailBackend` за контрактом
|
||
`services.email.EmailBackend` (без реального SMTP/console).
|
||
"""
|
||
|
||
import uuid
|
||
from collections.abc import AsyncGenerator
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any, cast
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
from celery.exceptions import MaxRetriesExceededError
|
||
from sqlalchemy import text
|
||
|
||
from core.db import engine
|
||
from core.plugins.config import ChatConfig, InstanceConfig, SummarizerConfig, TranscriberConfig
|
||
from services.conference_ids import generate_number, generate_slug
|
||
from services.email import EmailSendError
|
||
from workers.tasks import notify as notify_module
|
||
from workers.tasks.notify import notify_session_async
|
||
|
||
NOW = datetime.now(UTC)
|
||
|
||
|
||
class _Request:
|
||
"""Заглушка `celery.Task.request`: нужен только счётчик уже сделанных попыток."""
|
||
|
||
def __init__(self, retries: int = 0) -> None:
|
||
self.retries = retries
|
||
|
||
|
||
class _FakeTask:
|
||
"""Заглушка bound-задачи Celery: фиксирует вызовы `retry`, не бросает исключение."""
|
||
|
||
def __init__(self, retries: int = 0) -> None:
|
||
self.request = _Request(retries)
|
||
self.retry = MagicMock()
|
||
|
||
|
||
class _ExhaustedRetryTask:
|
||
"""Заглушка bound-задачи: `retry` всегда бросает `MaxRetriesExceededError`."""
|
||
|
||
def __init__(self, retries: int = 0) -> None:
|
||
self.request = _Request(retries)
|
||
self.retry_calls = 0
|
||
|
||
def retry(self, countdown: int | None = None) -> None:
|
||
self.retry_calls += 1
|
||
raise MaxRetriesExceededError("исчерпаны попытки уведомления")
|
||
|
||
|
||
class _FakeEmailBackend:
|
||
"""Заглушка `EmailBackend`: запоминает успешные отправки, может «отказывать» части адресов."""
|
||
|
||
def __init__(self, *, fail_for: set[str] | None = None, retryable: bool = True) -> None:
|
||
self.sent: list[str] = []
|
||
self._fail_for = fail_for or set()
|
||
self._retryable = retryable
|
||
|
||
async def send(
|
||
self,
|
||
*,
|
||
to: str,
|
||
subject: str,
|
||
body: str,
|
||
html_body: str | None = None,
|
||
attachments: Any = (),
|
||
) -> None:
|
||
if to in self._fail_for:
|
||
raise EmailSendError(f"сбой отправки для {to}", retryable=self._retryable)
|
||
self.sent.append(to)
|
||
|
||
|
||
class _Fixture:
|
||
"""Id тестовой конференции/сеанса/участников с уже готовым summary_data."""
|
||
|
||
def __init__(self) -> None:
|
||
self.conference_id = uuid.uuid4()
|
||
self.session_id = uuid.uuid4()
|
||
self.owner_id = uuid.uuid4()
|
||
self.user_id = uuid.uuid4()
|
||
self.guest_with_email_id = uuid.uuid4()
|
||
self.guest_no_email_id = uuid.uuid4()
|
||
self.user_participant_id = uuid.uuid4()
|
||
self.guest_with_email_participant_id = uuid.uuid4()
|
||
self.guest_no_email_participant_id = uuid.uuid4()
|
||
self.number = generate_number()
|
||
self.slug = generate_slug()
|
||
self.t_start = NOW - timedelta(minutes=30)
|
||
self.t_end = NOW
|
||
self.owner_email = f"owner-{self.owner_id}@example.com"
|
||
self.user_email = f"user-{self.user_id}@example.com"
|
||
self.guest_email = f"guest-{self.guest_with_email_id}@example.com"
|
||
|
||
|
||
@pytest.fixture
|
||
async def fx() -> AsyncGenerator[_Fixture, None]:
|
||
f = _Fixture()
|
||
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO users (id, email, name_user, password_hash) "
|
||
"VALUES (:id, :email, 'Владелец Тестов', 'x')"
|
||
),
|
||
{"id": f.owner_id, "email": f.owner_email},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO users (id, email, name_user, password_hash) "
|
||
"VALUES (:id, :email, 'Участник Тестов', 'x')"
|
||
),
|
||
{"id": f.user_id, "email": f.user_email},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conferences (id, number, slug, title, status, is_pinned, owner_id) "
|
||
"VALUES (:id, :number, :slug, 'Notify test', 'ended', false, :owner_id)"
|
||
),
|
||
{"id": f.conference_id, "number": f.number, "slug": f.slug, "owner_id": f.owner_id},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO guest_access (id, conference_id, display_name, email) "
|
||
"VALUES (:id, :conference_id, 'Гость С Почтой', :email)"
|
||
),
|
||
{
|
||
"id": f.guest_with_email_id,
|
||
"conference_id": f.conference_id,
|
||
"email": f.guest_email,
|
||
},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO guest_access (id, conference_id, display_name, email) "
|
||
"VALUES (:id, :conference_id, 'Гость Без Почты', NULL)"
|
||
),
|
||
{"id": f.guest_no_email_id, "conference_id": f.conference_id},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_sessions "
|
||
"(id, conference_id, title, t_start, t_end, pipeline_status, summary_data) "
|
||
"VALUES (:id, :conference_id, 'Notify session', :t_start, :t_end, "
|
||
"'summarizing', 'Готовое саммари встречи')"
|
||
),
|
||
{
|
||
"id": f.session_id,
|
||
"conference_id": f.conference_id,
|
||
"t_start": f.t_start,
|
||
"t_end": f.t_end,
|
||
},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_participants (id, session_id, user_id, joined_at, left_at) "
|
||
"VALUES (:id, :session_id, :user_id, :joined_at, :left_at)"
|
||
),
|
||
{
|
||
"id": f.user_participant_id,
|
||
"session_id": f.session_id,
|
||
"user_id": f.user_id,
|
||
"joined_at": f.t_start,
|
||
"left_at": f.t_end,
|
||
},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_participants "
|
||
"(id, session_id, guest_id, joined_at, left_at) "
|
||
"VALUES (:id, :session_id, :guest_id, :joined_at, :left_at)"
|
||
),
|
||
{
|
||
"id": f.guest_with_email_participant_id,
|
||
"session_id": f.session_id,
|
||
"guest_id": f.guest_with_email_id,
|
||
"joined_at": f.t_start,
|
||
"left_at": f.t_end,
|
||
},
|
||
)
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_participants "
|
||
"(id, session_id, guest_id, joined_at, left_at) "
|
||
"VALUES (:id, :session_id, :guest_id, :joined_at, :left_at)"
|
||
),
|
||
{
|
||
"id": f.guest_no_email_participant_id,
|
||
"session_id": f.session_id,
|
||
"guest_id": f.guest_no_email_id,
|
||
"joined_at": f.t_start,
|
||
"left_at": f.t_end,
|
||
},
|
||
)
|
||
await conn.commit()
|
||
yield f
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text("DELETE FROM email_deliveries WHERE session_id = :id"), {"id": f.session_id}
|
||
)
|
||
await conn.execute(
|
||
text("DELETE FROM conference_participants WHERE session_id = :id"),
|
||
{"id": f.session_id},
|
||
)
|
||
await conn.execute(
|
||
text("DELETE FROM conference_sessions WHERE conference_id = :id"),
|
||
{"id": f.conference_id},
|
||
)
|
||
await conn.execute(
|
||
text("DELETE FROM guest_access WHERE conference_id = :id"), {"id": f.conference_id}
|
||
)
|
||
await conn.execute(text("DELETE FROM conferences WHERE id = :id"), {"id": f.conference_id})
|
||
await conn.execute(text("DELETE FROM users WHERE id = :id"), {"id": f.user_id})
|
||
await conn.execute(text("DELETE FROM users WHERE id = :id"), {"id": f.owner_id})
|
||
await conn.commit()
|
||
|
||
|
||
def _cfg(*, summary_recipients: str = "all") -> InstanceConfig:
|
||
return InstanceConfig(
|
||
transcriber=TranscriberConfig(),
|
||
summarizer=SummarizerConfig(),
|
||
chat=ChatConfig(),
|
||
summary_recipients=cast("Any", summary_recipients),
|
||
)
|
||
|
||
|
||
async def _fetch_session_status(session_id: uuid.UUID) -> str:
|
||
async with engine.connect() as conn:
|
||
result = await conn.execute(
|
||
text("SELECT pipeline_status FROM conference_sessions WHERE id = :id"),
|
||
{"id": session_id},
|
||
)
|
||
return cast("str", result.scalar_one())
|
||
|
||
|
||
async def _fetch_delivered_emails(session_id: uuid.UUID) -> set[str]:
|
||
async with engine.connect() as conn:
|
||
result = await conn.execute(
|
||
text(
|
||
"SELECT recipient_email FROM email_deliveries "
|
||
"WHERE session_id = :id AND kind = 'summary'"
|
||
),
|
||
{"id": session_id},
|
||
)
|
||
return {row[0] for row in result.all()}
|
||
|
||
|
||
async def _set_conference_summary_recipients(conference_id: uuid.UUID, value: str | None) -> None:
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text("UPDATE conferences SET summary_recipients = :value WHERE id = :id"),
|
||
{"id": conference_id, "value": value},
|
||
)
|
||
await conn.commit()
|
||
|
||
|
||
async def test_notify_session_mode_all_sends_to_participants_with_email(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Режим «всем»: пользователь и гость с email получают письмо, гость без email — нет."""
|
||
backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: backend)
|
||
|
||
await notify_session_async(
|
||
_FakeTask(), fx.session_id, plugins_config=_cfg(summary_recipients="all")
|
||
)
|
||
|
||
assert set(backend.sent) == {fx.user_email, fx.guest_email}
|
||
assert await _fetch_session_status(fx.session_id) == "notified"
|
||
assert await _fetch_delivered_emails(fx.session_id) == {fx.user_email, fx.guest_email}
|
||
|
||
|
||
async def test_notify_session_mode_owner_sends_only_to_owner(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Режим «только организатору» (дефолт инстанса) — письмо получает только владелец."""
|
||
backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: backend)
|
||
|
||
await notify_session_async(
|
||
_FakeTask(), fx.session_id, plugins_config=_cfg(summary_recipients="owner")
|
||
)
|
||
|
||
assert backend.sent == [fx.owner_email]
|
||
assert await _fetch_session_status(fx.session_id) == "notified"
|
||
|
||
|
||
async def test_notify_session_conference_override_wins_over_instance_default(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Переопределение `conference.summary_recipients` сильнее дефолта инстанса."""
|
||
backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: backend)
|
||
await _set_conference_summary_recipients(fx.conference_id, "owner")
|
||
|
||
await notify_session_async(
|
||
_FakeTask(), fx.session_id, plugins_config=_cfg(summary_recipients="all")
|
||
)
|
||
|
||
assert backend.sent == [fx.owner_email]
|
||
|
||
|
||
async def test_notify_session_repeated_run_does_not_duplicate(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Повторный запуск после успешной рассылки — email_deliveries не дублируются, писем не шлём."""
|
||
first_backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: first_backend)
|
||
await notify_session_async(_FakeTask(), fx.session_id, plugins_config=_cfg())
|
||
assert await _fetch_session_status(fx.session_id) == "notified"
|
||
|
||
# Второй прогон подаётся уже другим сеансом (pipeline_status уже notified) — guard
|
||
# должен остановить его до всякой попытки отправки; проверим и явный сценарий
|
||
# "уже отправлено part of pending", переставив статус обратно как это делает
|
||
# recover_stuck_notifications при гонке с уже отработавшей задачей.
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text("UPDATE conference_sessions SET pipeline_status = 'summarizing' WHERE id = :id"),
|
||
{"id": fx.session_id},
|
||
)
|
||
await conn.commit()
|
||
|
||
second_backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: second_backend)
|
||
await notify_session_async(_FakeTask(), fx.session_id, plugins_config=_cfg())
|
||
|
||
assert second_backend.sent == []
|
||
assert await _fetch_session_status(fx.session_id) == "notified"
|
||
assert await _fetch_delivered_emails(fx.session_id) == {fx.user_email, fx.guest_email}
|
||
|
||
|
||
async def test_notify_session_partial_failure_retry_delivers_only_undelivered(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Частичный сбой SMTP → retry; повторный запуск дошлёт только недоставленным."""
|
||
failing_backend = _FakeEmailBackend(fail_for={fx.guest_email}, retryable=True)
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: failing_backend)
|
||
|
||
task = _FakeTask(retries=0)
|
||
await notify_session_async(task, fx.session_id, plugins_config=_cfg())
|
||
|
||
task.retry.assert_called_once_with(countdown=60)
|
||
# Получатель, отправленный до сбоя, уже зафиксирован — статус пока не notified.
|
||
assert await _fetch_session_status(fx.session_id) == "summarizing"
|
||
delivered_before_retry = await _fetch_delivered_emails(fx.session_id)
|
||
assert fx.user_email in delivered_before_retry
|
||
assert fx.guest_email not in delivered_before_retry
|
||
|
||
# Повторная доставка задачи celery-ретраем (acks_late) — второй прогон уже
|
||
# без временного сбоя, дошлёт только недоставленного получателя.
|
||
recovering_backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: recovering_backend)
|
||
second_task = _FakeTask(retries=1)
|
||
await notify_session_async(second_task, fx.session_id, plugins_config=_cfg())
|
||
|
||
second_task.retry.assert_not_called()
|
||
assert recovering_backend.sent == [fx.guest_email]
|
||
assert await _fetch_session_status(fx.session_id) == "notified"
|
||
assert await _fetch_delivered_emails(fx.session_id) == {fx.user_email, fx.guest_email}
|
||
|
||
|
||
async def test_notify_session_marks_failed_when_retries_exhausted(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Исчерпание retry (SMTP постоянно недоступен) → pipeline_status='failed'."""
|
||
backend = _FakeEmailBackend(fail_for={fx.user_email, fx.guest_email}, retryable=True)
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: backend)
|
||
task = _ExhaustedRetryTask(retries=5)
|
||
|
||
await notify_session_async(task, fx.session_id, plugins_config=_cfg())
|
||
|
||
assert task.retry_calls == 1
|
||
assert await _fetch_session_status(fx.session_id) == "failed"
|
||
|
||
|
||
async def test_notify_session_no_recipients_marks_notified_immediately(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Режим «организатору», но у конференции нет владельца — получателей нет, сразу notified."""
|
||
backend = _FakeEmailBackend()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", lambda settings: backend)
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text("UPDATE conferences SET owner_id = NULL WHERE id = :id"),
|
||
{"id": fx.conference_id},
|
||
)
|
||
await conn.commit()
|
||
|
||
await notify_session_async(
|
||
_FakeTask(), fx.session_id, plugins_config=_cfg(summary_recipients="owner")
|
||
)
|
||
|
||
assert backend.sent == []
|
||
assert await _fetch_session_status(fx.session_id) == "notified"
|
||
|
||
|
||
async def test_notify_session_noop_when_session_not_found(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""Guard №1: сеанс не найден — no-op, send не вызывается."""
|
||
mock_factory = MagicMock()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", mock_factory)
|
||
|
||
await notify_session_async(_FakeTask(), uuid.uuid4(), plugins_config=_cfg())
|
||
|
||
mock_factory.assert_not_called()
|
||
|
||
|
||
async def test_notify_session_noop_when_not_on_notify_step(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Guard №2: pipeline_status не 'summarizing' — no-op."""
|
||
mock_factory = MagicMock()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", mock_factory)
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text("UPDATE conference_sessions SET pipeline_status = 'notified' WHERE id = :id"),
|
||
{"id": fx.session_id},
|
||
)
|
||
await conn.commit()
|
||
|
||
await notify_session_async(_FakeTask(), fx.session_id, plugins_config=_cfg())
|
||
|
||
mock_factory.assert_not_called()
|
||
|
||
|
||
async def test_notify_session_noop_when_summary_not_ready(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Guard №3: summary_data ещё NULL — no-op."""
|
||
mock_factory = MagicMock()
|
||
monkeypatch.setattr(notify_module, "create_email_backend", mock_factory)
|
||
async with engine.connect() as conn:
|
||
await conn.execute(
|
||
text("UPDATE conference_sessions SET summary_data = NULL WHERE id = :id"),
|
||
{"id": fx.session_id},
|
||
)
|
||
await conn.commit()
|
||
|
||
await notify_session_async(_FakeTask(), fx.session_id, plugins_config=_cfg())
|
||
|
||
mock_factory.assert_not_called()
|