How do you unit test a Celery task?

celery integration tests
celery_task_always_eager
celery_always_eager
django-celery test database
celery task apply
pytest mark celery
django test celery delay
celery task_always_eager

The Celery documentation mentions testing Celery within Django but doesn't explain how to test a Celery task if you are not using Django. How do you do this?


It is possible to test tasks synchronously using any unittest lib out there. I normaly do 2 different test sessions when working with celery tasks. The first one (as I'm suggesting bellow) is completely synchronous and should be the one that makes sure the algorithm does what it should do. The second session uses the whole system (including the broker) and makes sure I'm not having serialization issues or any other distribution, comunication problem.

So:

from celery import Celery

celery = Celery()

@celery.task
def add(x, y):
    return x + y

And your test:

from nose.tools import eq_

def test_add_task():
    rst = add.apply(args=(4, 4)).get()
    eq_(rst, 8)

Hope that helps!

Testing with Celery, distributedpython.com. Unit Testing Celery Tasks. How to unit test Celery tasks. Published on May 01, 2018. Estimated reading time: 2 minutes The full source  When testing with eager mode you are only testing an emulation of what happens in a worker, and there are many discrepancies between the emulation and what happens in reality. A Celery task is much like a web view, in that it should only define how to perform the action in the context of being called as a task.


I use this:

with mock.patch('celeryconfig.CELERY_ALWAYS_EAGER', True, create=True):
    ...

Docs: http://docs.celeryproject.org/en/3.1/configuration.html#celery-always-eager

CELERY_ALWAYS_EAGER lets you run your task synchronous, and you don't need a celery server.

Unit Testing Celery Tasks, Writing unit tests for celery tasks can be painful since they are asynchronous and long. I will be explaining how to write unit tests for Celery  Writing unit tests for celery tasks can be painful since they are asynchronous and long. I will be explaining how to write unit tests for Celery tasks and will try to address different cases for this.


Testing Celery Tasks - Arpit Solanki, Because Celery requires task functions to be synchronous, and the function we're looking to test is asynchronous, we need a way to run an  CELERY_ALWAYS_EAGER lets you run your task synchronous, and you don't need a celery server. It is possible to test tasks synchronously using any unittest lib out there. I normaly do 2 different test sessions when working with celery tasks.


unittest
import unittest

from myproject.myapp import celeryapp

class TestMyCeleryWorker(unittest.TestCase):

  def setUp(self):
      celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)
py.test fixtures
# conftest.py
from myproject.myapp import celeryapp

@pytest.fixture(scope='module')
def celery_app(request):
    celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)
    return celeryapp

# test_tasks.py
def test_some_task(celery_app):
    ...
Addendum: make send_task respect eager
from celery import current_app

def send_task(name, args=(), kwargs={}, **opts):
    # https://github.com/celery/celery/issues/581
    task = current_app.tasks[name]
    return task.apply(args, kwargs, **opts)

current_app.send_task = send_task

Writing Unit Tests for Celery Tasks with async Functions · JF's Dev , The first problem you'll run in to when trying to write a test that runs a task is that Django's test runner doesn't use the same database as your celery daemon is  Call tasks in tests directly and wrap all inner celery interactions with if self.request.called_directly and run task directly if True or with apply_async if False. Wrap task.ready() and other statuses check with functions where I check for ALWAYS_EAGER and task readiness.


For those on Celery 4 it's:

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)

Because the settings names have been changed and need updating if you choose to upgrade, see

http://docs.celeryproject.org/en/latest/whatsnew-4.0.html#lowercase-setting-names

Unit Testing, It is possible to test tasks synchronously using any unittest lib out there. I normaly do 2 different test sessions when working with celery tasks. The first one (as I'm  If you're writing unit tests that depend on a celery worker, though, you're doing it wrong. For unit tests, you'll want to mock your Celery methods and test them separately. You could use CeleryTestCaseMixin to write integration tests with Celery tasks, though.


python - How do you unit test a Celery task?, For unit tests, we run Celery in eager mode (configured in devilry.project.develop.​test). Testing with non-eager Celery¶. Install RabbitMQ¶. See http://www  How can we bypass task.delay() during the test (I tried setting CELERY_ALWAYS_EAGER = True but it made no difference)? How do we use the test settings that are recommended (if that is the best way) without actually changing our settings.py? Can we still use manage.py test or do we have to use a custom runner?


Developing and testing Celery background tasks, Run a monitored Celery worker for integration tests that depend on Celery tasks. from unittest import TestCase from celerytest.testcase import  Besides background tasks execution, Celery also supports so called delayed tasks (apply_async method). In order to use them all you have to do is to provide datetime when the task should be invoked via eta or countdown (in seconds).


celerytest - Integration testing with Celery, to write integration tests using pyTest not the god ol' python UnitTest framework​. class for your integration tests using a real celery worker in the background. The @task decorator replaces the function with a Task object (see documentation). If you mock the task itself you'll replace the (somewhat magic) Task object with a MagicMock and it won't schedule the task at all. Instead mock the Task object's run() method, like so: