Python Celery could start with a threading inprocess ?
Question: I want make a testcase with my celery codes. But usually celery need start
with a new process like `$ celery -A CELERY_MODULE worker`, It's means I can't
run my testcase code directly ?

I'm configurate the Celery with memory store to void the extra I/O in the
testcase. That's config can't sample share the task queue in different
process.


Answer: Here is my naive implements. The celery entry from
`celery.bin.celeryd.WorkCommand`, it's parse the args and execute works.  
Use the `solo` to void the MultiProcess use in the case. Of course you need
install that's lib first.

You could use this before your celery testcase start.

    
    
    #!/usr/bin/env python
    #vim: encoding=utf-8
    
    import time
    import unittest
    from threading import Thread
    
    from celery import Celery, states
    from celery.bin.celeryd import WorkerCommand
    
    class CELERY_CONFIG(object):
        BROKER_URL = "memory://"
        CELERY_CACHE_BACKEND = "memory"
        CELERY_RESULT_BACKEND = "cache"
        CELERYD_POOL = "solo"
    
    class CeleryTestCase(unittest.TestCase):
        def test_inprocess(self):
            app = Celery(__name__)
            app.config_from_object(CELERY_CONFIG)
    
            @app.task
            def dumpy_task(dct):
                return 321
    
            worker = WorkerCommand(app)
            #worker.execute_from_commandline(["-P solo"])
            t = Thread(target=worker.execute_from_commandline, args=(["-c 1"],))
            t.daemon = True
            t.start()
            ar = dumpy_task.apply_async(({"a": 123},))
            while ar.status != states.SUCCESS:
                time.sleep(.01)
            self.assertEqual(states.SUCCESS, ar.status)
            self.assertEqual(ar.result, 321)
            t.join(0)
    

