[Fixed]-Celery's pytest fixtures (celery_worker and celery_app) does not work

18👍

OP here, I’ve figured it out and wrote an article – https://medium.com/@scythargon/how-to-use-celery-pytest-fixtures-for-celery-intergration-testing-6d61c91775d9

Main key:

@pytest.mark.usefixtures('celery_session_app')
@pytest.mark.usefixtures('celery_session_worker')
class MyTest():
    def test(self):
        assert mul.delay(4, 4).get(timeout=10) == 16

4👍

For how many developers use these tools, I’m surprised at how lacking the docs are on the topic. I struggled with this for about a half day and then found @scythargon‘s discussion. I solved it slightly differently, so I’m throwing my answer in the the mix for posterity (very close to the OP’s method):

tasks.py

from celery import shared_task

@shared_task
def add(x, y):
    return x + y

@shared_task()
def multiply(x, y):
    return x * y

conftest.py

import pytest

pytest_plugins = ('celery.contrib.pytest', )

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

tests.py

from api.app.tasks import add, multiply

def test_celery_worker_initializes(celery_app, celery_worker):
    assert True


def test_celery_tasks(celery_app, celery_worker):

    assert add.delay(4, 4).get(timeout=5) == 8
    assert multiply.delay(4, 4).get(timeout=5) == 16

As an added bonus, my redis broker and backend are running in Docker (as part of a swarm):

docker-compose.yml

version: "3.9"

services:

  . . .

  redis:
    image: redis:alpine
    networks:
      swarm-net:
        aliases:
          - redis
    ports:
      - "8001:6379"

  . . .

Leave a comment