test_task.py 文件源码

def test_transaction_manager_abort(celery_worker, task_app_request, dbsession, demo_user):
    """Test that transaction are not executed to schedule if transaction manager aborts."""

    try:
        # Tasks do not execute if transaction is never commit because of exception
        with transaction.manager:
            assert dbsession.query(User).count() == 1
            assert dbsession.query(User).get(1).username == "test"
            demotasks.modify_username.apply_async(args=[1], tm=transaction.manager)
            raise RuntimeError("aargh")
    except RuntimeError:
        pass

    # Let the transaction commit (which should not happen)
    time.sleep(0.5)

    # Because of exception, Celery never fires
    with transaction.manager:
        u = dbsession.query(User).get(1)
        assert u.username == "test"