test_celery_from_threads.py

# -*- coding: utf-8 -*-
import threading
from nose.tools import eq_
import celery
from celery._state import _get_current_app


eq_('3.0.22', celery.__version__)
eq_('default', _get_current_app().main)


app = celery.Celery()
app.config_from_object(dict(CELERY_ALWAYS_EAGER=True))


@app.task
def hello():
    return 'hello'


def test_main():
    eq_(app, hello._app)
    eq_(app, celery.current_app)
    # celery.group.type depends on current_app so essentially the same assertion
    eq_(app, celery.group(hello.subtask()).type._app)
    print 'ok'


def test_with_set_current():
    app.set_current()
    test_main()


def test_from_thread(target=test_main):
    t = threading.Thread(target=target)
    t.start()
    t.join()


if __name__ == '__main__':
    # passes
    print 'test_main()'
    test_main()

    # AssertionError: <Celery __main__:0x10605f990> != <Celery default:0x106261e90>
    print 'test_from_thread()'
    test_from_thread(test_main)

    # passes
    print 'test_with_set_current()'
    test_from_thread(test_with_set_current)