Celery's pytest fixtures (celery_worker and celery_app) does not work

Question

I'm trying to write a Celery(v. 4.2.1) integration test for my Django(v. 2.2.3) application.

There is a bunch of outdated articles about this around, but non of them seem to use stuff from the latest celery testing documentation - https://docs.celeryproject.org/en/v4.2.1/userguide/testing.html#fixtures

Seems like Celery comes with two fixtures for testing: celery_app and celery_worker which should allow to actually run worker in the background thread of your tests.

As the doc's say I've added

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

into my conftest.py

I've wrapped my test function with

@pytest.mark.celery_app
@pytest.mark.celery_worker

usually I wrap my celery tasks with

from celery import shared_task
@shared_task
def blabla(...):
    pass

but I've even tried to replace it with

from myapp.celery import celery
@celery.task
def blabla():
    pass

What else... I run my celery task via apply_async providing eta argument.

Tried tons of ways but the celery fixtures do not affect how things work and the task call goes to actual redis instance and is picked by a worker in a separate process (if I run it) and hence my assert_called fails along with my efforts to access the object which are in the testing database.

This way it it does not load fixtures.

This way it does not use specified fixtures because they should appear in the method arguments and break it by exceeding the number of arguments.

Thought that the Celery pytest plugin might be missing at all, but that's not true - tried to register it explicitly:

Though the fixtures are available to pytest:

But I've got into theis source code, added some wild prints there and I don't see them logged.

Jara

Answer 1

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

Jara

The mock idea is simple and works, celery should have an easy way to test – Envelope

Do you have any simple project example of your decision? Can't get the result code form the article – Parenthood

Answer 2

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"

  . . .

Wylen

If one is using rabbitmq-server , how can kombu from celery make connection with rabbitmq-server without starting the live server(rabbit) on the local machine ? and how can this be replicated on github ci.yml since running the test there cause similar issue – Wagner

Your answer seems to work fine with tasks created with @shared_task. How do you test a task created with @app.task where app is a Celery app? – Radman

I have posted my question with more details @ https://mcmap.net/q/823799/-how-to-test-the-tasks-of-a-celery-instance-using-pytest/4635580 – Radman