Mock Celery task in Pytest

Boost Your Testing Mojo: Unleash the Power of Mock Celery Tasks in PyTest.

2023-03-05 Python, Tips

Python logo and Celery logo

Mock Celery task in Pytest

Let’s get straight to the point. Although there are several ways to do it, and depending on the characteristics of each project, one may be more interesting than another, let’s look at one that generally can be useful for us.

Pytest Celery Fixture

This could be an example of celery config:

import pytest

@pytest.fixture(scope="session")
def celery_config(celery_envvars: None):
    return {
        "task_serializer": "pickle", # default json
        "result_serializer": "pickle", # default json
        "broker_url": "memory://", # default "amqp://"
        "result_backend": "rpc", # no dafault
        "imports": ["<your_module>"],
    }

It shouldn’t be mandatory/recommendable to set task_always_eager or CELERY_ALWAYS_EAGER to True, per Celery docs advice. This executed locally instead of being sent to the queue.

Creating a Celery task test

import celery

@celery.shared_task(bind=True)
def add_task(self, a, b):
    return a+b;

def test_one_pending_iterations(
    celery_app: celery.Celery,
    celery_worker: celery.Celery.WorkController,
) -> None:
    assert add.delay(2, 2) == 4

Stay updated on Python tips

Happy testing and may your bugs tremble in fear at the sight of your PyTest prowess!

If you want to stay updated…

Python Pytest

Mock Celery task in Pytest

Mock Celery task in Pytest

Pytest Celery Fixture

Creating a Celery task test

Stay updated on Python tips

Carlos Vecina

Senior Data Scientist at Jobandtalent

Related