Source code for sqlobject.tests.test_transactions

import pytest
from sqlobject import SQLObject, SQLObjectNotFound, StringCol
from sqlobject.tests.dbtest import raises, setupClass, supports


########################################
# Transaction test
########################################


try:
    support_transactions = supports('transactions')
except NameError:
    # The module was imported during documentation building
    pass
else:
    if not support_transactions:
        pytestmark = pytest.mark.skip('')


[docs]class SOTestSOTrans(SQLObject): class sqlmeta: defaultOrder = 'name' name = StringCol(length=10, alternateID=True, dbName='name_col')
[docs]def test_transaction(): setupClass(SOTestSOTrans) SOTestSOTrans(name='bob') SOTestSOTrans(name='tim') trans = SOTestSOTrans._connection.transaction() try: SOTestSOTrans._connection.autoCommit = 'exception' SOTestSOTrans(name='joe', connection=trans) trans.rollback() trans.begin() assert ([n.name for n in SOTestSOTrans.select(connection=trans)] == ['bob', 'tim']) b = SOTestSOTrans.byName('bob', connection=trans) b.name = 'robert' trans.commit() assert b.name == 'robert' b.name = 'bob' trans.rollback() trans.begin() assert b.name == 'robert' finally: SOTestSOTrans._connection.autoCommit = True
[docs]def test_transaction_commit_sync(): setupClass(SOTestSOTrans) trans = SOTestSOTrans._connection.transaction() try: SOTestSOTrans(name='bob') bOut = SOTestSOTrans.byName('bob') bIn = SOTestSOTrans.byName('bob', connection=trans) bIn.name = 'robert' assert bOut.name == 'bob' trans.commit() assert bOut.name == 'robert' finally: SOTestSOTrans._connection.autoCommit = True
[docs]def test_transaction_delete(close=False): setupClass(SOTestSOTrans) connection = SOTestSOTrans._connection if (connection.dbName == 'sqlite') and connection._memory: pytest.skip("The following test requires a different connection") trans = connection.transaction() try: SOTestSOTrans(name='bob') bIn = SOTestSOTrans.byName('bob', connection=trans) bIn.destroySelf() bOut = SOTestSOTrans.select(SOTestSOTrans.q.name == 'bob') assert bOut.count() == 1 bOutInst = bOut[0] bOutID = bOutInst.id # noqa: bOutID is used in the string code below trans.commit(close=close) assert bOut.count() == 0 raises(SQLObjectNotFound, "SOTestSOTrans.get(bOutID)") raises(SQLObjectNotFound, "bOutInst.name") finally: trans.rollback() connection.autoCommit = True connection.close()
[docs]def test_transaction_delete_with_close(): test_transaction_delete(close=True)