New About Yours API Help
1.2 KB, Python 3
# Failed
def test_conflict2(test_request, dbsession_factory, test_instance):
    TestModel = get_test_model()

    s1 = dbsession_factory()
    s2 = dbsession_factory()
    s2.transaction_manager.begin()

    def target():
        s1.transaction_manager.begin()
        q = s1.query(TestModel).get(1)
        q.balance += 1
        time.sleep(1)
        s1.transaction_manager.commit()
        
    thread = threading.Thread(target=target)
    thread.start()


    q = s2.query(TestModel).get(1)
    q.balance += 1
    thread.join()
    s2.transaction_manager.commit()
    q = s1.query(TestModel).get(1)
    assert q.balance == 12

#-----------
# PASSED
# with @tm.run decorator

def test_conflict2(test_request, dbsession_factory, test_instance):
    TestModel = get_test_model()

    s1 = dbsession_factory()
    s2 = dbsession_factory()
    s2.transaction_manager.begin()

    @s1.transaction_manager.run(2)
    def target():
        q = s1.query(TestModel).get(1)
        q.balance += 1
        time.sleep(1)

    thread = threading.Thread(target=target)
    thread.start()


    q = s2.query(TestModel).get(1)
    q.balance += 1
    thread.join()
    s2.transaction_manager.commit()
    q = s1.query(TestModel).get(1)
    assert q.balance == 12
Pasted 5 months, 1 week ago — Expires in 202 days
URL: http://dpaste.com/2H0DWSY