Source code for sqlobject.tests.test_transactions

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

########################################
# 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("These tests require transactions")


[docs]class SOTestSOTrans(SQLObject): class sqlmeta: defaultOrder = 'name' name = StringCol(length=10, alternateID=True, dbName='name_col')
[docs]def make_watcher(): log = [] def watch(*args): log.append(args) watch.log = log return watch
[docs]def make_listen(signal): watcher = make_watcher() events.listen(watcher, sqlmeta, signal) return watcher
[docs]def test_transaction(): commit_watcher = make_listen(events.CommitSignal) rollback_watcher = make_listen(events.RollbackSignal) 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' assert len(commit_watcher.log) == 1 assert commit_watcher.log[0][0][0][0] == 'SOTestSOTrans' assert commit_watcher.log[0][0][0][1] == [1, 2] assert len(rollback_watcher.log) == 2 assert rollback_watcher.log[0][0][0][0] == 'SOTestSOTrans' assert rollback_watcher.log[0][0][0][1] == [3] assert rollback_watcher.log[1][0][0][0] == 'SOTestSOTrans' assert rollback_watcher.log[1][0][0][1] == [1, 2] finally: SOTestSOTrans._connection.autoCommit = True
[docs]def test_transaction_commit_sync(): setupClass(SOTestSOTrans) connection = SOTestSOTrans._connection trans = 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: trans.rollback() connection.autoCommit = True connection.close()
[docs]def test_transaction_delete(close=False): setupClass(SOTestSOTrans) connection = SOTestSOTrans._connection if (connection.dbName == 'sqlite') and connection._memory: pytest.skip("The test doesn't work with sqlite memory 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) with raises(SQLObjectNotFound): bOutInst.name finally: trans.rollback() connection.autoCommit = True connection.close()
[docs]def test_transaction_delete_with_close(): test_transaction_delete(close=True)