Mock Celery task in Pytest

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

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

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

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…

Carlos Vecina
Carlos Vecina
Senior Data Scientist at Jobandtalent

Senior Data Scientist at Jobandtalent | AI & Data Science for Business