[Django]-In-Memory broker for celery unit tests

You can specify the broker_backend in your settings :

if 'test' in sys.argv[1:]:
    BROKER_BACKEND = 'memory'
    CELERY_TASK_ALWAYS_EAGER = True
    CELERY_TASK_EAGER_PROPAGATES = True

or you can override the settings with a decorator directly in your test

import unittest
from django.test.utils import override_settings


class MyTestCase(unittest.TestCase):

    @override_settings(CELERY_TASK_EAGER_PROPAGATES=True,
                       CELERY_TASK_ALWAYS_EAGER=True,
                       BROKER_BACKEND='memory')
    def test_mytask(self):
        ...

You can use the Kombu in-memory broker to run unit tests, however to do so you need to spin-up a Celery worker using the same Celery app object as the Django server.

To use the in-memory broker, set BROKER_URL to memory://localhost/

Then, to spin up a small celery worker you can do the following:

app = <Django Celery App>

# Set the worker up to run in-place instead of using a pool
app.conf.CELERYD_CONCURRENCY = 1
app.conf.CELERYD_POOL = 'solo'

# Code to start the worker
def run_worker():
    app.worker_main()

# Create a thread and run the worker in it
import threading
t = threading.Thread(target=run_worker)
t.setDaemon(True)
t.start()

You need to make sure you use the same app as the Django celery app instance.

Note that starting the worker will print many things and modify logging settings.

Here’s a more fully featured example of a Django TransactionTestCase that works with Celery 4.x.

import threading

from django.test import TransactionTestCase
from django.db import connections

from myproj.celery import app  # your Celery app


class CeleryTestCase(TransactionTestCase):
    """Test case with Celery support."""

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        app.control.purge()
        cls._worker = app.Worker(app=app, pool='solo', concurrency=1)
        connections.close_all()
        cls._thread = threading.Thread(target=cls._worker.start)
        cls._thread.daemon = True
        cls._thread.start()

    @classmethod
    def tearDownClass(cls):
        cls._worker.stop()
        super().tearDownClass()

Be aware this doesn’t change your queue names to a testing queues, so if you’re also running the app you’ll want to do that too.

To use a memory broker for all celery tests in django, this pytest fixture can be used.

test.py

import pytest
from django.conf import settings
from django.test import TestCase


class TestStartFeatureDetectionView(TestCase):
    
    @pytest.fixture(autouse=True)
    def set_celery_broker_to_memory(self):
        settings.BROKER_BACKEND = 'memory://'
        settings.CELERY_BROKER_URL = 'memory://'
        settings.CELERY_RESULT_BACKEND = 'db+sqlite:///results.sqlite'

A SQLite database from SQLAlchemy is used to store the results.