Celery Testing Sketch

gistfile1.txt

from celery.contrib.testing import app as _app
import mock
import unittest
import celery



class TestCase(unittest.TestCase):
  def setUp(self):
    app = _app.TestApp(
        'my_app',
        config={
            'CELERY_TASK_EAGER_PROPAGATES': True,
            'CELERY_TASK_ALWAYS_EAGER': True,
        },
        set_as_current=True,
        enable_logging=True,
    )
    app.autodiscover_tasks(
        [
            'app_with_tasks',
        ]
    )
    
    def send_task_apply(*args, **kwargs):
        task_name = args[0].split('.')[-1]
        module = importlib.import_module(
            '.'.join(args[0].split('.')[:-1])
        )

        task = getattr(module, task_name).s(
            **kwargs['kwargs']
        ).apply()

        if task.status == 'FAILURE':
            raise Exception(
                "{}.{} {}".format(
                    module,
                    task_name,
                    task.result
                )
            )

    mock.patch.objects(
        celery.current_app,
        'send_task',
        side_effect=send_Task_applym
        return_value=True
    ).start()