93 lines
4.2 KiB
Python
93 lines
4.2 KiB
Python
"""Тесты постановки задач backend-продюсерами `Celery.send_task`.
|
||
|
||
`pipeline_producer`/`invitations_producer` создают собственный (голый)
|
||
Celery-клиент без `task_routes` из `workers/celery_app.py` (backend не
|
||
импортирует пакет `workers`, см. докстрины модулей) — очередь этим клиентам
|
||
нужно передавать явным параметром `queue=` при каждой отправке, иначе задача
|
||
уйдёт в дефолтную очередь `celery`, а не в изолированную очередь семьи задач.
|
||
"""
|
||
|
||
import uuid
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
from kombu.exceptions import OperationalError
|
||
|
||
from services import invitations_producer, pipeline_producer
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _fake_celery_client(monkeypatch: pytest.MonkeyPatch) -> MagicMock:
|
||
"""Подменить `celery.Celery` фейком, чтобы не создавать реальное broker-подключение."""
|
||
instance = MagicMock()
|
||
factory = MagicMock(return_value=instance)
|
||
monkeypatch.setattr(pipeline_producer, "Celery", factory)
|
||
monkeypatch.setattr(invitations_producer, "Celery", factory)
|
||
return instance
|
||
|
||
|
||
def test_enqueue_pipeline_uses_transcription_queue(_fake_celery_client: MagicMock) -> None:
|
||
session_id = uuid.uuid4()
|
||
pipeline_producer.enqueue_pipeline(session_id)
|
||
_fake_celery_client.send_task.assert_called_once_with(
|
||
pipeline_producer.RUN_PIPELINE_TASK_NAME,
|
||
args=[str(session_id)],
|
||
queue="transcription",
|
||
)
|
||
|
||
|
||
def test_enqueue_invitations_uses_notify_queue(_fake_celery_client: MagicMock) -> None:
|
||
conference_id = uuid.uuid4()
|
||
invitations_producer.enqueue_invitations(conference_id, emails=["a@example.com"])
|
||
_fake_celery_client.send_task.assert_called_once_with(
|
||
invitations_producer.SEND_INVITATIONS_TASK_NAME,
|
||
args=[str(conference_id), ["a@example.com"]],
|
||
queue="notify",
|
||
)
|
||
|
||
|
||
def test_transcription_queue_served_true_when_worker_reports_queue(
|
||
_fake_celery_client: MagicMock,
|
||
) -> None:
|
||
"""`active_queues()` вернул хотя бы
|
||
одного воркера с очередью `transcription` среди прочих его очередей."""
|
||
_fake_celery_client.control.inspect.return_value.active_queues.return_value = {
|
||
"celery@worker1": [{"name": "notify"}, {"name": "transcription"}],
|
||
}
|
||
|
||
assert pipeline_producer.transcription_queue_served() is True
|
||
_fake_celery_client.control.inspect.assert_called_once_with(timeout=1.0)
|
||
|
||
|
||
def test_transcription_queue_served_false_when_no_worker_replies(
|
||
_fake_celery_client: MagicMock,
|
||
) -> None:
|
||
"""`active_queues()` вернул `None` (нет воркеров либо брокер недоступен) → `False`."""
|
||
_fake_celery_client.control.inspect.return_value.active_queues.return_value = None
|
||
|
||
assert pipeline_producer.transcription_queue_served() is False
|
||
|
||
|
||
def test_transcription_queue_served_false_when_worker_active_but_other_queue(
|
||
_fake_celery_client: MagicMock,
|
||
) -> None:
|
||
"""Воркер(ы) ответили, но никто не слушает именно `transcription` → `False`."""
|
||
_fake_celery_client.control.inspect.return_value.active_queues.return_value = {
|
||
"celery@worker1": [{"name": "notify"}],
|
||
}
|
||
|
||
assert pipeline_producer.transcription_queue_served() is False
|
||
|
||
|
||
def test_transcription_queue_served_false_when_broker_unavailable(
|
||
_fake_celery_client: MagicMock,
|
||
) -> None:
|
||
"""Недоступный брокер (Redis лежит) кидает `OperationalError`, а не
|
||
возвращает `None`, — по контракту («нет воркеров ИЛИ брокер недоступен → `False`»)
|
||
исключение не должно пробрасываться наружу (500 в `GET`/`PUT /admin/settings`)."""
|
||
_fake_celery_client.control.inspect.return_value.active_queues.side_effect = OperationalError(
|
||
"Error 61 connecting to localhost:6379. Connection refused."
|
||
)
|
||
|
||
assert pipeline_producer.transcription_queue_served() is False
|