test_celery.py 文件源码

def test_celery__TransactionAwareTask____call____4(
        celery_session_worker, interaction):
    """It propagates the task_id to the worker."""
    job = get_task_id.apply_async(task_id='my-nice-task-id')
    transaction.commit()
    assert 'my-nice-task-id' == job.get()