Тестирование с Celery

celery_app — приложение Celery для тестирования.

Эта фикстура возвращает приложение Celery, которое можно использовать для тестирования.

Пример:

def test_create_task(celery_app, celery_worker):
    @celery_app.task
    def mul(x, y):
        return x * y

    celery_worker.reload()
    assert mul.delay(4, 4).get(timeout=10) == 16

celery_worker — встроенный работающий worker.

Эта фикстура запускает экземпляр worker Celery, который можно использовать для интеграционных тестов. Worker запускается в отдельном потоке и останавливается сразу после завершения теста.

По умолчанию фикстура ждёт до 10 секунд, пока worker выполнит оставшиеся задачи, и вызывает исключение, если время ожидания превышено. Тайм-аут можно настроить, задав ключ shutdown_timeout в словаре, возвращаемом фикстурой celery_worker_parameters().

Пример:

# Put this in your conftest.py
@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'redis://'
    }

def test_add(celery_worker):
    mytask.delay()


# If you wish to override some setting in one test cases
# only - you can use the ``celery`` mark:
@pytest.mark.celery(result_backend='rpc')
def test_other(celery_worker):
    ...

По умолчанию heartbeat отключены, то есть тестовый worker не отправляет события для worker-online, worker-offline и worker-heartbeat. Чтобы включить heartbeat, измените фикстуру celery_worker_parameters():

# Put this in your conftest.py
@pytest.fixture(scope="session")
def celery_worker_parameters():
    return {"without_heartbeat": False}
    ...

celery_config — переопределение конфигурации тестового приложения Celery.

Эту фикстуру можно переопределить, чтобы настроить тестовое приложение Celery.

Конфигурация, возвращаемая фикстурой, будет использоваться для настройки фикстур celery_app() и celery_session_app().

Пример:

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'rpc',
    }

celery_parameters — переопределение параметров тестового приложения Celery.

Эту фикстуру можно переопределить, чтобы изменить параметры __init__ тестового приложения Celery. В отличие от celery_config(), они напрямую передаются при создании экземпляра Celery.

Конфигурация, возвращаемая фикстурой, будет использоваться для настройки фикстур celery_app() и celery_session_app().

Пример:

@pytest.fixture(scope='session')
def celery_parameters():
    return {
        'task_cls':  my.package.MyCustomTaskClass,
        'strict_typing': False,
    }

celery_worker_parameters — переопределение параметров worker Celery.

Эту фикстуру можно переопределить, чтобы изменить параметры __init__ тестовых worker Celery. Они напрямую передаются в WorkController при его создании.

Конфигурация, возвращаемая фикстурой, будет использоваться для настройки фикстур celery_worker() и celery_session_worker().

Пример:

@pytest.fixture(scope='session')
def celery_worker_parameters():
    return {
        'queues':  ('high-prio', 'low-prio'),
        'exclude_queues': ('celery'),
    }

celery_enable_logging — переопределение для включения журналирования во встроенных worker.

Эту фикстуру можно переопределить, чтобы включить журналирование во встроенных worker.

Пример:

@pytest.fixture(scope='session')
def celery_enable_logging():
    return True

celery_includes — добавление дополнительных импортов для встроенных worker.

Эту фикстуру можно переопределить, чтобы включать модули при запуске встроенного worker.

Можно вернуть список имён модулей для импорта: это могут быть модули задач, модули, регистрирующие сигналы, и так далее.

Пример:

@pytest.fixture(scope='session')
def celery_includes():
    return [
        'proj.tests.tasks',
        'proj.tests.celery_signal_handlers',
    ]

celery_worker_pool — переопределение пула для встроенных worker.

Эту фикстуру можно переопределить, чтобы настроить пул выполнения для встроенных worker.

Пример:

@pytest.fixture(scope='session')
def celery_worker_pool():
    return 'prefork'

celery_session_worker — встроенный worker, работающий на протяжении всего сеанса.

Эта фикстура запускает worker, который работает на протяжении всего сеанса тестирования (он не будет запускаться и останавливаться для каждого теста).

Пример:

# Add this to your conftest.py
@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'rpc',
    }

# Do this in your tests.
def test_add_task(celery_session_worker):
    assert add.delay(2, 2).get() == 4

celery_session_app — приложение Celery для тестирования (область действия сеанса).

Эту фикстуру можно использовать в других фикстурах с областью действия сеанса, которым требуется ссылка на экземпляр приложения Celery.

use_celery_app_trap — вызов исключения при переходе к приложению по умолчанию.

Эту фикстуру можно переопределить в conftest.py, чтобы включить «ловушку приложения»: если что-либо попытается обратиться к приложению по умолчанию или current_app, будет вызвано исключение.

Пример:

@pytest.fixture(scope='session')
def use_celery_app_trap():
    return True

Если тесту необходимо обратиться к приложению по умолчанию, его нужно пометить с помощью фикстуры depends_on_current_app:

@pytest.mark.usefixtures('depends_on_current_app')
def test_something():
    something()