524 lines
19 KiB
Python
524 lines
19 KiB
Python
"""Тесты обслуживания конференций (`workers.tasks.maintenance.cleanup_conferences_async`).
|
||
|
||
Как и старый `test_auto_release.py`, не использует savepoint-фикстуру
|
||
`db_session`: `cleanup_conferences_async` открывает собственную сессию с
|
||
отдельным engine (`workers/db.py::open_session`), которая не видит
|
||
незакоммиченные изменения другой сессии. Тестовые данные заводятся и
|
||
коммитятся напрямую через `core.db.engine`, очистка — вручную по завершении
|
||
теста. Время передаётся параметром `now` — без freezegun.
|
||
"""
|
||
|
||
import uuid
|
||
from collections.abc import AsyncGenerator
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import cast
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import AsyncConnection
|
||
|
||
from core.db import engine
|
||
from core.plugins.config import ChatConfig, InstanceConfig, SummarizerConfig, TranscriberConfig
|
||
from services.conference_ids import generate_number, generate_slug
|
||
from workers.tasks import maintenance as maintenance_module
|
||
|
||
NOW = datetime.now(UTC)
|
||
|
||
|
||
class _Fixture:
|
||
"""Id тестовой конференции + (опционально) её сеанса/участника."""
|
||
|
||
def __init__(self) -> None:
|
||
self.conference_id = uuid.uuid4()
|
||
self.session_id = uuid.uuid4()
|
||
self.user_id = uuid.uuid4()
|
||
self.number = generate_number()
|
||
self.slug = generate_slug()
|
||
|
||
|
||
@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, 'Maintenance Tester', 'x')"
|
||
),
|
||
{"id": f.user_id, "email": f"maintenance-{f.user_id}@example.com"},
|
||
)
|
||
await conn.commit()
|
||
yield f
|
||
async with engine.connect() as conn:
|
||
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 conferences WHERE id = :id"), {"id": f.conference_id})
|
||
await conn.execute(text("DELETE FROM users WHERE id = :id"), {"id": f.user_id})
|
||
await conn.commit()
|
||
|
||
|
||
async def _insert_conference(
|
||
conn: AsyncConnection,
|
||
fx: _Fixture,
|
||
*,
|
||
is_pinned: bool,
|
||
status: str,
|
||
scheduled_at: datetime | None = None,
|
||
duration_minutes: int | None = None,
|
||
) -> None:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conferences "
|
||
"(id, number, slug, title, status, is_pinned, scheduled_at, duration_minutes) "
|
||
"VALUES (:id, :number, :slug, 'Maintenance test', :status, :is_pinned, "
|
||
":scheduled_at, :duration_minutes)"
|
||
),
|
||
{
|
||
"id": fx.conference_id,
|
||
"number": fx.number,
|
||
"slug": fx.slug,
|
||
"status": status,
|
||
"is_pinned": is_pinned,
|
||
"scheduled_at": scheduled_at,
|
||
"duration_minutes": duration_minutes,
|
||
},
|
||
)
|
||
|
||
|
||
async def _insert_session(
|
||
conn: AsyncConnection, fx: _Fixture, *, t_start: datetime, t_end: datetime | None = None
|
||
) -> None:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_sessions (id, conference_id, title, t_start, t_end) "
|
||
"VALUES (:id, :conference_id, 'Maintenance session', :t_start, :t_end)"
|
||
),
|
||
{
|
||
"id": fx.session_id,
|
||
"conference_id": fx.conference_id,
|
||
"t_start": t_start,
|
||
"t_end": t_end,
|
||
},
|
||
)
|
||
|
||
|
||
async def _insert_summarizing_session(
|
||
conn: AsyncConnection,
|
||
fx: _Fixture,
|
||
*,
|
||
t_end: datetime,
|
||
summary_data: str | None = None,
|
||
) -> None:
|
||
"""Сеанс на шаге суммаризации — для тестов `recover_stuck_summaries_async`."""
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_sessions "
|
||
"(id, conference_id, title, t_start, t_end, pipeline_status, summary_data) "
|
||
"VALUES (:id, :conference_id, 'Recovery session', :t_start, :t_end, "
|
||
"'summarizing', :summary_data)"
|
||
),
|
||
{
|
||
"id": fx.session_id,
|
||
"conference_id": fx.conference_id,
|
||
"t_start": t_end - timedelta(minutes=30),
|
||
"t_end": t_end,
|
||
"summary_data": summary_data,
|
||
},
|
||
)
|
||
|
||
|
||
async def _insert_active_participant(conn: AsyncConnection, fx: _Fixture) -> None:
|
||
await conn.execute(
|
||
text(
|
||
"INSERT INTO conference_participants (id, session_id, user_id, joined_at) "
|
||
"VALUES (:id, :session_id, :user_id, :joined_at)"
|
||
),
|
||
{"id": uuid.uuid4(), "session_id": fx.session_id, "user_id": fx.user_id, "joined_at": NOW},
|
||
)
|
||
|
||
|
||
async def _fetch_conference(fx: _Fixture) -> dict[str, object]:
|
||
async with engine.connect() as conn:
|
||
result = await conn.execute(
|
||
text("SELECT status, ended_at FROM conferences WHERE id = :id"),
|
||
{"id": fx.conference_id},
|
||
)
|
||
row = result.mappings().one()
|
||
return dict(row)
|
||
|
||
|
||
async def _fetch_session_t_end(fx: _Fixture) -> datetime | None:
|
||
async with engine.connect() as conn:
|
||
result = await conn.execute(
|
||
text("SELECT t_end FROM conference_sessions WHERE id = :id"), {"id": fx.session_id}
|
||
)
|
||
return cast("datetime | None", result.scalar_one())
|
||
|
||
|
||
def _sent_tasks_for_session(mock_send_task: MagicMock, fx: _Fixture) -> list[str]:
|
||
"""Имена задач, отправленных `send_task` именно для тестового сеанса `fx`.
|
||
|
||
Recovery-задачи сканируют ВСЮ таблицу `conference_sessions`, а общая
|
||
dev-БД может легитимно содержать чужие зависшие сеансы (реальные данные
|
||
разработчика) — глобальные `assert_called_once`/`assert_not_called`
|
||
от них флэкают. Проверяем только вызовы с id нашего сеанса.
|
||
"""
|
||
session_id = str(fx.session_id)
|
||
return [
|
||
c.args[0] for c in mock_send_task.call_args_list if c.kwargs.get("args") == [session_id]
|
||
]
|
||
|
||
|
||
def _summarizer_cfg(*, enabled: bool = True) -> InstanceConfig:
|
||
"""Конфиг для тестов `recover_stuck_summaries_async` — интересует
|
||
только `summarizer.enabled`."""
|
||
return InstanceConfig(
|
||
transcriber=TranscriberConfig(),
|
||
summarizer=SummarizerConfig(enabled=enabled, provider="fake"),
|
||
chat=ChatConfig(),
|
||
)
|
||
|
||
|
||
async def test_idle_session_of_unpinned_conference_closes_and_ends_conference(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="active")
|
||
await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
assert await _fetch_session_t_end(fx) is not None
|
||
conference = await _fetch_conference(fx)
|
||
assert conference["status"] == "ended"
|
||
assert conference["ended_at"] is not None
|
||
mock_delete.assert_awaited_once_with(fx.slug)
|
||
|
||
|
||
async def test_idle_session_of_pinned_conference_returns_to_scheduled(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=True, status="active")
|
||
await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
assert await _fetch_session_t_end(fx) is not None
|
||
conference = await _fetch_conference(fx)
|
||
assert conference["status"] == "scheduled"
|
||
assert conference["ended_at"] is None
|
||
|
||
|
||
async def test_session_with_active_participant_is_not_touched(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="active")
|
||
await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
|
||
await _insert_active_participant(conn, fx)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
assert await _fetch_session_t_end(fx) is None
|
||
mock_delete.assert_not_awaited()
|
||
|
||
|
||
async def test_fresh_session_is_not_touched(fx: _Fixture, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="active")
|
||
await _insert_session(conn, fx, t_start=NOW)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
assert await _fetch_session_t_end(fx) is None
|
||
mock_delete.assert_not_awaited()
|
||
|
||
|
||
async def test_repeated_run_is_a_noop(fx: _Fixture, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="active")
|
||
await _insert_session(conn, fx, t_start=NOW - timedelta(minutes=15))
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
first_t_end = await _fetch_session_t_end(fx)
|
||
assert first_t_end is not None
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW + timedelta(minutes=5))
|
||
second_t_end = await _fetch_session_t_end(fx)
|
||
assert second_t_end == first_t_end
|
||
mock_delete.assert_awaited_once()
|
||
|
||
|
||
async def test_expired_unpinned_scheduled_without_session_becomes_ended(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(
|
||
conn,
|
||
fx,
|
||
is_pinned=False,
|
||
status="scheduled",
|
||
scheduled_at=NOW - timedelta(hours=2),
|
||
duration_minutes=30,
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
conference = await _fetch_conference(fx)
|
||
assert conference["status"] == "ended"
|
||
assert conference["ended_at"] is not None
|
||
|
||
|
||
async def test_scheduled_conference_within_grace_is_not_touched(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(
|
||
conn,
|
||
fx,
|
||
is_pinned=False,
|
||
status="scheduled",
|
||
scheduled_at=NOW - timedelta(minutes=10),
|
||
duration_minutes=None,
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
conference = await _fetch_conference(fx)
|
||
assert conference["status"] == "scheduled"
|
||
|
||
|
||
async def test_pinned_expired_scheduled_conference_is_not_touched(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(
|
||
conn,
|
||
fx,
|
||
is_pinned=True,
|
||
status="scheduled",
|
||
scheduled_at=NOW - timedelta(hours=5),
|
||
duration_minutes=30,
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
conference = await _fetch_conference(fx)
|
||
assert conference["status"] == "scheduled"
|
||
|
||
|
||
async def test_expired_scheduled_conference_with_session_is_not_touched(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
mock_delete = AsyncMock()
|
||
monkeypatch.setattr(maintenance_module, "delete_livekit_room", mock_delete)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(
|
||
conn,
|
||
fx,
|
||
is_pinned=False,
|
||
status="scheduled",
|
||
scheduled_at=NOW - timedelta(hours=5),
|
||
duration_minutes=30,
|
||
)
|
||
# Сеанс уже был (и закрыт) — конференция не должна считаться "без сеансов".
|
||
await _insert_session(
|
||
conn, fx, t_start=NOW - timedelta(hours=5), t_end=NOW - timedelta(hours=4)
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.cleanup_conferences_async(NOW)
|
||
|
||
conference = await _fetch_conference(fx)
|
||
assert conference["status"] == "scheduled"
|
||
|
||
|
||
async def test_recover_stuck_summaries_resends_task_for_hung_session(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Сеанс завис в 'summarizing' без summary_data дольше порога — задача переставлена."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(
|
||
conn,
|
||
fx,
|
||
t_end=NOW - maintenance_module.STUCK_SUMMARIZING_THRESHOLD - timedelta(minutes=1),
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_summaries_async(NOW, plugins_config=_summarizer_cfg())
|
||
|
||
assert _sent_tasks_for_session(mock_send_task, fx) == [
|
||
"workers.tasks.summarize.summarize_session"
|
||
]
|
||
|
||
|
||
async def test_recover_stuck_summaries_ignores_fresh_session(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Сеанс моложе порога (недавно ушёл в summarizing) — трогать рано, no-op."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(conn, fx, t_end=NOW - timedelta(minutes=5))
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_summaries_async(NOW, plugins_config=_summarizer_cfg())
|
||
|
||
assert _sent_tasks_for_session(mock_send_task, fx) == []
|
||
|
||
|
||
async def test_recover_stuck_summaries_ignores_session_with_summary_already_present(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Сеанс уже готов к уведомлению (summary_data заполнен) — recovery его не трогает."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(
|
||
conn,
|
||
fx,
|
||
t_end=NOW - maintenance_module.STUCK_SUMMARIZING_THRESHOLD - timedelta(minutes=1),
|
||
summary_data="Уже готовое саммари",
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_summaries_async(NOW, plugins_config=_summarizer_cfg())
|
||
|
||
assert _sent_tasks_for_session(mock_send_task, fx) == []
|
||
|
||
|
||
async def test_recover_stuck_summaries_noop_when_summarizer_disabled(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""`summarizer.enabled=false` — recovery выходит сразу, не читая БД и не слишком часто
|
||
впустую рассылая задачу (иначе слала бы её каждые 5 минут без толку)."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(
|
||
conn,
|
||
fx,
|
||
t_end=NOW - maintenance_module.STUCK_SUMMARIZING_THRESHOLD - timedelta(minutes=1),
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_summaries_async(
|
||
NOW, plugins_config=_summarizer_cfg(enabled=False)
|
||
)
|
||
|
||
mock_send_task.assert_not_called()
|
||
|
||
|
||
async def test_recover_stuck_notifications_resends_task_for_hung_session(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Саммари готово, но сеанс завис в 'summarizing' дольше порога — notify_session
|
||
переставлена."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(
|
||
conn,
|
||
fx,
|
||
t_end=NOW - maintenance_module.STUCK_NOTIFYING_THRESHOLD - timedelta(minutes=1),
|
||
summary_data="Готовое саммари",
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_notifications_async(NOW)
|
||
|
||
assert _sent_tasks_for_session(mock_send_task, fx) == ["workers.tasks.notify.notify_session"]
|
||
|
||
|
||
async def test_recover_stuck_notifications_ignores_fresh_session(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Сеанс моложе порога (саммари только что готово) — трогать рано, no-op."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(
|
||
conn, fx, t_end=NOW - timedelta(minutes=5), summary_data="Готовое саммари"
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_notifications_async(NOW)
|
||
|
||
assert _sent_tasks_for_session(mock_send_task, fx) == []
|
||
|
||
|
||
async def test_recover_stuck_notifications_ignores_session_without_summary(
|
||
fx: _Fixture, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Саммари ещё не готово (summary_data NULL) — это забота recover_stuck_summaries, не notify."""
|
||
mock_send_task = MagicMock()
|
||
monkeypatch.setattr(maintenance_module.app, "send_task", mock_send_task)
|
||
|
||
async with engine.connect() as conn:
|
||
await _insert_conference(conn, fx, is_pinned=False, status="ended")
|
||
await _insert_summarizing_session(
|
||
conn,
|
||
fx,
|
||
t_end=NOW - maintenance_module.STUCK_NOTIFYING_THRESHOLD - timedelta(minutes=1),
|
||
)
|
||
await conn.commit()
|
||
|
||
await maintenance_module.recover_stuck_notifications_async(NOW)
|
||
|
||
assert _sent_tasks_for_session(mock_send_task, fx) == []
|