Test result serialization when CELERY_ALWAYS_EAGER = True

test_serialization.py

# -*- coding: utf-8 -*-
import cPickle as pickle
from nose.tools import *
from celery import Celery


celery = Celery()
celery.conf.CELERY_ALWAYS_EAGER = True


class UnpicklableObject(object):
    __slots__ = ('a', 'b')


@celery.task
def unpicklable_task():
    return UnpicklableObject()


def test_unpicklable_object():
    def pickle_the_unpicklable():
        pickle.dumps(UnpicklableObject())
    assert_raises(TypeError, pickle_the_unpicklable)


def test_unpicklable_task():
    def call_unpicklable_task():
        unpicklable_task.delay().get()
    # How do I make it raise TypeError?
    assert_raises(TypeError, call_unpicklable_task)


if __name__ == '__main__':
    import nose
    nose.main(argv=['nosetests', __file__])

workaround.py

# -*- coding: utf-8 -*-
from celery import Celery


celery = Celery()
celery.conf.CELERY_ALWAYS_EAGER = True


def init_celery():
    celery.finalize()
    if celery.conf.CELERY_ALWAYS_EAGER:
        from celery.signals import task_success
        task_success.connect(check_result_serialization)


def check_result_serialization(result, **kwargs):
    celery.backend.encode(dict(result=result))