Celery only gives instructions on how to write integration tests using pyTest not the god ol' python UnitTest framework. So I wrote a CeleryTestCase which can be used as a base class for your integration tests using a real celery worker in the background.

CeleryTestCase.py

from unittest import TestCase
from celery.contrib.testing import worker


class CeleryTestCase(TestCase):
    @property
    def celery_app(self):
        """
            Override this method to provide this testcase with the celery app under test
        """
        raise NotImplementedError()

    def run(self, result=None):
        from kombu.asynchronous.hub import set_event_loop

        set_event_loop(None)
        self.celery_app.loader.import_module('celery.contrib.testing.tasks')

        with worker.start_worker(self.celery_app, loglevel="debug"):
            super(CeleryTestCase, self).run(result)

CodeUnderTest.py

from celery import Celery

celery_app = Celery("CeleryUnderTest", broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')


@celery_app.task
def mul(x, y):
    return x * y

MyTestCase.py

from .CodeUnderTest import celery_app, mul
from .CeleryTestCase import CeleryTestCase


class MyTestCase(CeleryTestCase):
    @property
    def celery_app(self):
        return celery_app

    def test_mul(self):
        self.assertEqual(mul.delay(4, 4).get(timeout=10), 16)

The line set_event_loop(None) is quite important, otherwise, you'll run into the following error:

AttributeError: 'NoneType' object has no attribute 'poll'

As far as I understand it, the second half of the worker.start_worker context manager stops the worker and it's Hub and while doing that, sets the poller property of the Hub to None. Starting a new worker instance reuses the same event loop and doesn't initialize a new Hub, which leads to a partially initialized Hub with a poller property set to None. That's why I am explicitly removing the current event loop.

Importing self.celery_app.loader.import_module('celery.contrib.testing.tasks') is essential because otherwise you'll run into the follwing assert:

assert 'celery.ping' in app.tasks

This is because the test worker context waits, for the worker to be started, by pinging it using the imported task. If this taks is not available the assert above is triggered.

If you getting django.db.utils.InterfaceError: connection already closed change class CeleryTestCase(TestCase): to class CeleryTestCase(TransactionTestCase):

Worked for me