testapp.py 文件源码

def test_mixed():
    from celery import chain
    from celery_deadline import job

    task = chain(
        job('Python', '1-2',
            plugin_info=dict(
                ScriptFile='/Users/chad/python/untitled.py',
                Version='2.7')),
        sum.s()  # <-- will go to a celery worker
    )
    result = task.apply_async()
    print("waiting for results:")
    print(result.get())