52 lines
2.9 KiB
Python
52 lines
2.9 KiB
Python
"""Тесты маршрутизации задач Celery по очередям.
|
||
|
||
Проверяем не сами паттерны `task_routes` (это конфигурация), а результат их
|
||
резолвинга Celery-роутером `app.amqp.router` для КОНКРЕТНЫХ имён задач,
|
||
которые реально ставятся в очередь (`workers/tasks/dispatch.py`,
|
||
`workers/tasks/maintenance.py`, `backend/services/pipeline_producer.py`,
|
||
`backend/services/invitations_producer.py`) — регрессия на опечатку в
|
||
паттерне (например, `workers.tasks.notify.*` вместо `workers.tasks.notify`)
|
||
будет поймана здесь, а не только при ручном прогоне в проде. Не требует
|
||
подключения к брокеру: `Router.route` — чистая функция над `task_routes`.
|
||
"""
|
||
|
||
from workers.celery_app import app
|
||
|
||
|
||
def _resolved_queue(task_name: str) -> str:
|
||
"""Имя очереди, в которую Celery-роутер направит задачу `task_name`."""
|
||
route = app.amqp.router.route({}, task_name)
|
||
queue = route["queue"]
|
||
return str(queue.name) if hasattr(queue, "name") else str(queue)
|
||
|
||
|
||
def test_run_pipeline_routes_to_transcription_queue() -> None:
|
||
"""`run_pipeline` — выделенная очередь `transcription` (только `worker-transcriber`)."""
|
||
assert _resolved_queue("workers.tasks.pipeline.run_pipeline") == "transcription"
|
||
|
||
|
||
def test_summarize_session_routes_to_summarize_queue() -> None:
|
||
"""`summarize_session` — очередь `summarize`."""
|
||
assert _resolved_queue("workers.tasks.summarize.summarize_session") == "summarize"
|
||
|
||
|
||
def test_notify_session_routes_to_notify_queue() -> None:
|
||
"""`notify_session` — очередь `notify`."""
|
||
assert _resolved_queue("workers.tasks.notify.notify_session") == "notify"
|
||
|
||
|
||
def test_send_invitations_routes_to_notify_queue() -> None:
|
||
"""`send_invitations` (рассылка .ics-приглашений) — тоже очередь `notify`."""
|
||
assert _resolved_queue("workers.tasks.invitations.send_invitations") == "notify"
|
||
|
||
|
||
def test_maintenance_tasks_route_to_default_celery_queue() -> None:
|
||
"""Задачи обслуживания без явного маршрута остаются на дефолтной очереди `celery`.
|
||
|
||
Базовый `worker` слушает `celery,summarize,notify` (`docs/deploy/scaling.md`) —
|
||
маршрутизация обслуживающих задач намеренно не переопределяется.
|
||
"""
|
||
assert _resolved_queue("workers.tasks.maintenance.cleanup_conferences") == "celery"
|
||
assert _resolved_queue("workers.tasks.maintenance.recover_stuck_summaries") == "celery"
|
||
assert _resolved_queue("workers.tasks.maintenance.recover_stuck_notifications") == "celery"
|