test_celery.py 文件源码

def test_celery__TransactionAwareTask____call____2(
        celery_session_worker, interaction):
    """It aborts the transaction and retries in case of an ConflictError."""
    result = conflict_task.delay()
    transaction.commit()
    with pytest.raises(Exception) as err:
        result.get()
    assert 'MaxRetriesExceededError' == err.value.__class__.__name__