from sqlalchemy.test.testing import eq_, assert_raises, \ assert_raises_message import sys import time import threading from sqlalchemy import create_engine, MetaData, INT, VARCHAR, Sequence, \ select, Integer, String, func, text, exc from sqlalchemy.test.schema import Table from sqlalchemy.test.schema import Column from sqlalchemy.test import TestBase, testing users, metadata = None, None class TransactionTest(TestBase): @classmethod def setup_class(cls): global users, metadata metadata = MetaData() users = Table('query_users', metadata, Column('user_id', INT, primary_key = True), Column('user_name', VARCHAR(20)), test_needs_acid=True, ) users.create(testing.db) def teardown(self): testing.db.execute(users.delete()).close() @classmethod def teardown_class(cls): users.drop(testing.db) def test_commits(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') transaction.commit() transaction = connection.begin() connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') transaction.commit() transaction = connection.begin() result = connection.execute("select * from query_users") assert len(result.fetchall()) == 3 transaction.commit() connection.close() def test_rollback(self): """test a basic rollback""" connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') transaction.rollback() result = connection.execute("select * from query_users") assert len(result.fetchall()) == 0 connection.close() def test_raise(self): connection = testing.db.connect() transaction = connection.begin() try: connection.execute(users.insert(), user_id=1, user_name='user1') connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=1, user_name='user3') transaction.commit() assert False except Exception , e: print "Exception: ", e transaction.rollback() result = connection.execute("select * from query_users") assert len(result.fetchall()) == 0 connection.close() def test_transaction_container(self): def go(conn, table, data): for d in data: conn.execute(table.insert(), d) testing.db.transaction(go, users, [dict(user_id=1, user_name='user1')]) eq_(testing.db.execute(users.select()).fetchall(), [(1, 'user1' )]) assert_raises(exc.DBAPIError, testing.db.transaction, go, users, [{'user_id': 2, 'user_name': 'user2'}, {'user_id': 1, 'user_name': 'user3'}]) eq_(testing.db.execute(users.select()).fetchall(), [(1, 'user1' )]) def test_nested_rollback(self): connection = testing.db.connect() try: transaction = connection.begin() try: connection.execute(users.insert(), user_id=1, user_name='user1') connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') trans2 = connection.begin() try: connection.execute(users.insert(), user_id=4, user_name='user4') connection.execute(users.insert(), user_id=5, user_name='user5') raise Exception('uh oh') trans2.commit() except: trans2.rollback() raise transaction.rollback() except Exception, e: transaction.rollback() raise except Exception, e: try: assert str(e) == 'uh oh' # and not "This transaction is # inactive" finally: connection.close() def test_retains_through_options(self): connection = testing.db.connect() try: transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') conn2 = connection.execution_options(dummy=True) conn2.execute(users.insert(), user_id=2, user_name='user2') transaction.rollback() eq_(connection.scalar("select count(1) from query_users"), 0) finally: connection.close() def test_nesting(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') trans2 = connection.begin() connection.execute(users.insert(), user_id=4, user_name='user4') connection.execute(users.insert(), user_id=5, user_name='user5') trans2.commit() transaction.rollback() self.assert_(connection.scalar('select count(1) from ' 'query_users') == 0) result = connection.execute('select * from query_users') assert len(result.fetchall()) == 0 connection.close() def test_close(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') trans2 = connection.begin() connection.execute(users.insert(), user_id=4, user_name='user4') connection.execute(users.insert(), user_id=5, user_name='user5') assert connection.in_transaction() trans2.close() assert connection.in_transaction() transaction.commit() assert not connection.in_transaction() self.assert_(connection.scalar('select count(1) from ' 'query_users') == 5) result = connection.execute('select * from query_users') assert len(result.fetchall()) == 5 connection.close() def test_close2(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') trans2 = connection.begin() connection.execute(users.insert(), user_id=4, user_name='user4') connection.execute(users.insert(), user_id=5, user_name='user5') assert connection.in_transaction() trans2.close() assert connection.in_transaction() transaction.close() assert not connection.in_transaction() self.assert_(connection.scalar('select count(1) from ' 'query_users') == 0) result = connection.execute('select * from query_users') assert len(result.fetchall()) == 0 connection.close() @testing.requires.savepoints def test_nested_subtransaction_rollback(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') trans2 = connection.begin_nested() connection.execute(users.insert(), user_id=2, user_name='user2') trans2.rollback() connection.execute(users.insert(), user_id=3, user_name='user3') transaction.commit() eq_(connection.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (3, )]) connection.close() @testing.requires.savepoints @testing.crashes('oracle+zxjdbc', 'Errors out and causes subsequent tests to ' 'deadlock') def test_nested_subtransaction_commit(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') trans2 = connection.begin_nested() connection.execute(users.insert(), user_id=2, user_name='user2') trans2.commit() connection.execute(users.insert(), user_id=3, user_name='user3') transaction.commit() eq_(connection.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (2, ), (3, )]) connection.close() @testing.requires.savepoints def test_rollback_to_subtransaction(self): connection = testing.db.connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') trans2 = connection.begin_nested() connection.execute(users.insert(), user_id=2, user_name='user2') trans3 = connection.begin() connection.execute(users.insert(), user_id=3, user_name='user3') trans3.rollback() connection.execute(users.insert(), user_id=4, user_name='user4') transaction.commit() eq_(connection.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (4, )]) connection.close() @testing.requires.two_phase_transactions def test_two_phase_transaction(self): connection = testing.db.connect() transaction = connection.begin_twophase() connection.execute(users.insert(), user_id=1, user_name='user1') transaction.prepare() transaction.commit() transaction = connection.begin_twophase() connection.execute(users.insert(), user_id=2, user_name='user2') transaction.commit() transaction.close() transaction = connection.begin_twophase() connection.execute(users.insert(), user_id=3, user_name='user3') transaction.rollback() transaction = connection.begin_twophase() connection.execute(users.insert(), user_id=4, user_name='user4') transaction.prepare() transaction.rollback() transaction.close() eq_(connection.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (2, )]) connection.close() @testing.requires.two_phase_transactions @testing.requires.savepoints def test_mixed_two_phase_transaction(self): connection = testing.db.connect() transaction = connection.begin_twophase() connection.execute(users.insert(), user_id=1, user_name='user1') transaction2 = connection.begin() connection.execute(users.insert(), user_id=2, user_name='user2') transaction3 = connection.begin_nested() connection.execute(users.insert(), user_id=3, user_name='user3') transaction4 = connection.begin() connection.execute(users.insert(), user_id=4, user_name='user4') transaction4.commit() transaction3.rollback() connection.execute(users.insert(), user_id=5, user_name='user5') transaction2.commit() transaction.prepare() transaction.commit() eq_(connection.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (2, ), (5, )]) connection.close() @testing.requires.two_phase_transactions @testing.crashes('mysql+oursql', 'Times out in full test runs only, causing ' 'subsequent tests to fail') @testing.crashes('mysql+zxjdbc', 'Deadlocks, causing subsequent tests to fail') @testing.fails_on('mysql', 'FIXME: unknown') def test_two_phase_recover(self): # MySQL recovery doesn't currently seem to work correctly # Prepared transactions disappear when connections are closed # and even when they aren't it doesn't seem possible to use the # recovery id. connection = testing.db.connect() transaction = connection.begin_twophase() connection.execute(users.insert(), user_id=1, user_name='user1') transaction.prepare() connection.close() connection2 = testing.db.connect() eq_(connection2.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), []) recoverables = connection2.recover_twophase() assert transaction.xid in recoverables connection2.commit_prepared(transaction.xid, recover=True) eq_(connection2.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, )]) connection2.close() @testing.requires.two_phase_transactions def test_multiple_two_phase(self): conn = testing.db.connect() xa = conn.begin_twophase() conn.execute(users.insert(), user_id=1, user_name='user1') xa.prepare() xa.commit() xa = conn.begin_twophase() conn.execute(users.insert(), user_id=2, user_name='user2') xa.prepare() xa.rollback() xa = conn.begin_twophase() conn.execute(users.insert(), user_id=3, user_name='user3') xa.rollback() xa = conn.begin_twophase() conn.execute(users.insert(), user_id=4, user_name='user4') xa.prepare() xa.commit() result = \ conn.execute(select([users.c.user_name]). order_by(users.c.user_id)) eq_(result.fetchall(), [('user1', ), ('user4', )]) conn.close() class AutoRollbackTest(TestBase): @classmethod def setup_class(cls): global metadata metadata = MetaData() @classmethod def teardown_class(cls): metadata.drop_all(testing.db) def test_rollback_deadlock(self): """test that returning connections to the pool clears any object locks.""" conn1 = testing.db.connect() conn2 = testing.db.connect() users = Table('deadlock_users', metadata, Column('user_id', INT, primary_key=True), Column('user_name', VARCHAR(20)), test_needs_acid=True) users.create(conn1) conn1.execute('select * from deadlock_users') conn1.close() # without auto-rollback in the connection pool's return() logic, # this deadlocks in PostgreSQL, because conn1 is returned to the # pool but still has a lock on "deadlock_users". comment out the # rollback in pool/ConnectionFairy._close() to see ! users.drop(conn2) conn2.close() class ExplicitAutoCommitTest(TestBase): """test the 'autocommit' flag on select() and text() objects. Requires PostgreSQL so that we may define a custom function which modifies the database. """ __only_on__ = 'postgresql' @classmethod def setup_class(cls): global metadata, foo metadata = MetaData(testing.db) foo = Table('foo', metadata, Column('id', Integer, primary_key=True), Column('data', String(100))) metadata.create_all() testing.db.execute("create function insert_foo(varchar) " "returns integer as 'insert into foo(data) " "values ($1);select 1;' language sql") def teardown(self): foo.delete().execute().close() @classmethod def teardown_class(cls): testing.db.execute('drop function insert_foo(varchar)') metadata.drop_all() def test_control(self): # test that not using autocommit does not commit conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execute(select([func.insert_foo('data1')])) assert conn2.execute(select([foo.c.data])).fetchall() == [] conn1.execute(text("select insert_foo('moredata')")) assert conn2.execute(select([foo.c.data])).fetchall() == [] trans = conn1.begin() trans.commit() assert conn2.execute(select([foo.c.data])).fetchall() \ == [('data1', ), ('moredata', )] conn1.close() conn2.close() def test_explicit_compiled(self): conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execute(select([func.insert_foo('data1' )]).execution_options(autocommit=True)) assert conn2.execute(select([foo.c.data])).fetchall() \ == [('data1', )] conn1.close() conn2.close() def test_explicit_connection(self): conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execution_options(autocommit=True).\ execute(select([func.insert_foo('data1' )])) eq_(conn2.execute(select([foo.c.data])).fetchall(), [('data1', )]) # connection supercedes statement conn1.execution_options(autocommit=False).\ execute(select([func.insert_foo('data2' )]).execution_options(autocommit=True)) eq_(conn2.execute(select([foo.c.data])).fetchall(), [('data1', )]) # ditto conn1.execution_options(autocommit=True).\ execute(select([func.insert_foo('data3' )]).execution_options(autocommit=False)) eq_(conn2.execute(select([foo.c.data])).fetchall(), [('data1', ), ('data2', ), ('data3', )]) conn1.close() conn2.close() def test_explicit_text(self): conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execute(text("select insert_foo('moredata')" ).execution_options(autocommit=True)) assert conn2.execute(select([foo.c.data])).fetchall() \ == [('moredata', )] conn1.close() conn2.close() @testing.uses_deprecated(r'autocommit on select\(\) is deprecated', r'autocommit\(\) is deprecated') def test_explicit_compiled_deprecated(self): conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execute(select([func.insert_foo('data1')], autocommit=True)) assert conn2.execute(select([foo.c.data])).fetchall() \ == [('data1', )] conn1.execute(select([func.insert_foo('data2')]).autocommit()) assert conn2.execute(select([foo.c.data])).fetchall() \ == [('data1', ), ('data2', )] conn1.close() conn2.close() @testing.uses_deprecated(r'autocommit on text\(\) is deprecated') def test_explicit_text_deprecated(self): conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execute(text("select insert_foo('moredata')", autocommit=True)) assert conn2.execute(select([foo.c.data])).fetchall() \ == [('moredata', )] conn1.close() conn2.close() def test_implicit_text(self): conn1 = testing.db.connect() conn2 = testing.db.connect() conn1.execute(text("insert into foo (data) values " "('implicitdata')")) assert conn2.execute(select([foo.c.data])).fetchall() \ == [('implicitdata', )] conn1.close() conn2.close() tlengine = None class TLTransactionTest(TestBase): @classmethod def setup_class(cls): global users, metadata, tlengine tlengine = create_engine(testing.db.url, strategy='threadlocal') metadata = MetaData() users = Table('query_users', metadata, Column('user_id', INT, Sequence('query_users_id_seq', optional=True), primary_key=True), Column('user_name', VARCHAR(20)), test_needs_acid=True) metadata.create_all(tlengine) def teardown(self): tlengine.execute(users.delete()).close() @classmethod def teardown_class(cls): metadata.drop_all(tlengine) tlengine.dispose() def setup(self): # ensure tests start with engine closed tlengine.close() def test_connection_close(self): """test that when connections are closed for real, transactions are rolled back and disposed.""" c = tlengine.contextual_connect() c.begin() assert c.in_transaction() c.close() assert not c.in_transaction() def test_transaction_close(self): c = tlengine.contextual_connect() t = c.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.execute(users.insert(), user_id=2, user_name='user2') t2 = c.begin() tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.execute(users.insert(), user_id=4, user_name='user4') t2.close() result = c.execute('select * from query_users') assert len(result.fetchall()) == 4 t.close() external_connection = tlengine.connect() result = external_connection.execute('select * from query_users' ) try: assert len(result.fetchall()) == 0 finally: c.close() external_connection.close() def test_rollback(self): """test a basic rollback""" tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.rollback() external_connection = tlengine.connect() result = external_connection.execute('select * from query_users' ) try: assert len(result.fetchall()) == 0 finally: external_connection.close() def test_commit(self): """test a basic commit""" tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.commit() external_connection = tlengine.connect() result = external_connection.execute('select * from query_users' ) try: assert len(result.fetchall()) == 3 finally: external_connection.close() def test_commits(self): connection = tlengine.connect() assert connection.execute('select count(1) from query_users' ).scalar() == 0 connection.close() connection = tlengine.contextual_connect() transaction = connection.begin() connection.execute(users.insert(), user_id=1, user_name='user1') transaction.commit() transaction = connection.begin() connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') transaction.commit() transaction = connection.begin() result = connection.execute('select * from query_users') l = result.fetchall() assert len(l) == 3, 'expected 3 got %d' % len(l) transaction.commit() connection.close() def test_rollback_off_conn(self): # test that a TLTransaction opened off a TLConnection allows # that TLConnection to be aware of the transactional context conn = tlengine.contextual_connect() trans = conn.begin() conn.execute(users.insert(), user_id=1, user_name='user1') conn.execute(users.insert(), user_id=2, user_name='user2') conn.execute(users.insert(), user_id=3, user_name='user3') trans.rollback() external_connection = tlengine.connect() result = external_connection.execute('select * from query_users' ) try: assert len(result.fetchall()) == 0 finally: conn.close() external_connection.close() def test_morerollback_off_conn(self): # test that an existing TLConnection automatically takes place # in a TLTransaction opened on a second TLConnection conn = tlengine.contextual_connect() conn2 = tlengine.contextual_connect() trans = conn2.begin() conn.execute(users.insert(), user_id=1, user_name='user1') conn.execute(users.insert(), user_id=2, user_name='user2') conn.execute(users.insert(), user_id=3, user_name='user3') trans.rollback() external_connection = tlengine.connect() result = external_connection.execute('select * from query_users' ) try: assert len(result.fetchall()) == 0 finally: conn.close() conn2.close() external_connection.close() def test_commit_off_connection(self): conn = tlengine.contextual_connect() trans = conn.begin() conn.execute(users.insert(), user_id=1, user_name='user1') conn.execute(users.insert(), user_id=2, user_name='user2') conn.execute(users.insert(), user_id=3, user_name='user3') trans.commit() external_connection = tlengine.connect() result = external_connection.execute('select * from query_users' ) try: assert len(result.fetchall()) == 3 finally: conn.close() external_connection.close() def test_nesting_rollback(self): """tests nesting of transactions, rollback at the end""" external_connection = tlengine.connect() self.assert_(external_connection.connection is not tlengine.contextual_connect().connection) tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.begin() tlengine.execute(users.insert(), user_id=4, user_name='user4') tlengine.execute(users.insert(), user_id=5, user_name='user5') tlengine.commit() tlengine.rollback() try: self.assert_(external_connection.scalar( 'select count(1) from query_users' ) == 0) finally: external_connection.close() def test_nesting_commit(self): """tests nesting of transactions, commit at the end.""" external_connection = tlengine.connect() self.assert_(external_connection.connection is not tlengine.contextual_connect().connection) tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.begin() tlengine.execute(users.insert(), user_id=4, user_name='user4') tlengine.execute(users.insert(), user_id=5, user_name='user5') tlengine.commit() tlengine.commit() try: self.assert_(external_connection.scalar( 'select count(1) from query_users' ) == 5) finally: external_connection.close() def test_mixed_nesting(self): """tests nesting of transactions off the TLEngine directly inside of tranasctions off the connection from the TLEngine""" external_connection = tlengine.connect() self.assert_(external_connection.connection is not tlengine.contextual_connect().connection) conn = tlengine.contextual_connect() trans = conn.begin() trans2 = conn.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.begin() tlengine.execute(users.insert(), user_id=4, user_name='user4') tlengine.begin() tlengine.execute(users.insert(), user_id=5, user_name='user5') tlengine.execute(users.insert(), user_id=6, user_name='user6') tlengine.execute(users.insert(), user_id=7, user_name='user7') tlengine.commit() tlengine.execute(users.insert(), user_id=8, user_name='user8') tlengine.commit() trans2.commit() trans.rollback() conn.close() try: self.assert_(external_connection.scalar( 'select count(1) from query_users' ) == 0) finally: external_connection.close() def test_more_mixed_nesting(self): """tests nesting of transactions off the connection from the TLEngine inside of tranasctions off thbe TLEngine directly.""" external_connection = tlengine.connect() self.assert_(external_connection.connection is not tlengine.contextual_connect().connection) tlengine.begin() connection = tlengine.contextual_connect() connection.execute(users.insert(), user_id=1, user_name='user1') tlengine.begin() connection.execute(users.insert(), user_id=2, user_name='user2') connection.execute(users.insert(), user_id=3, user_name='user3') trans = connection.begin() connection.execute(users.insert(), user_id=4, user_name='user4') connection.execute(users.insert(), user_id=5, user_name='user5') trans.commit() tlengine.commit() tlengine.rollback() connection.close() try: self.assert_(external_connection.scalar( 'select count(1) from query_users' ) == 0) finally: external_connection.close() @testing.requires.savepoints def test_nested_subtransaction_rollback(self): tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.begin_nested() tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.rollback() tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.commit() tlengine.close() eq_(tlengine.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (3, )]) tlengine.close() @testing.requires.savepoints @testing.crashes('oracle+zxjdbc', 'Errors out and causes subsequent tests to ' 'deadlock') def test_nested_subtransaction_commit(self): tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.begin_nested() tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.commit() tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.commit() tlengine.close() eq_(tlengine.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (2, ), (3, )]) tlengine.close() @testing.requires.savepoints def test_rollback_to_subtransaction(self): tlengine.begin() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.begin_nested() tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.begin() tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.rollback() tlengine.rollback() tlengine.execute(users.insert(), user_id=4, user_name='user4') tlengine.commit() tlengine.close() eq_(tlengine.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (4, )]) tlengine.close() def test_connections(self): """tests that contextual_connect is threadlocal""" c1 = tlengine.contextual_connect() c2 = tlengine.contextual_connect() assert c1.connection is c2.connection c2.close() assert not c1.closed assert not tlengine.closed @testing.requires.independent_cursors def test_result_closing(self): """tests that contextual_connect is threadlocal""" r1 = tlengine.execute(select([1])) r2 = tlengine.execute(select([1])) row1 = r1.fetchone() row2 = r2.fetchone() r1.close() assert r2.connection is r1.connection assert not r2.connection.closed assert not tlengine.closed # close again, nothing happens since resultproxy calls close() # only once r1.close() assert r2.connection is r1.connection assert not r2.connection.closed assert not tlengine.closed r2.close() assert r2.connection.closed assert tlengine.closed def test_dispose(self): eng = create_engine(testing.db.url, strategy='threadlocal') result = eng.execute(select([1])) eng.dispose() eng.execute(select([1])) @testing.requires.two_phase_transactions def test_two_phase_transaction(self): tlengine.begin_twophase() tlengine.execute(users.insert(), user_id=1, user_name='user1') tlengine.prepare() tlengine.commit() tlengine.begin_twophase() tlengine.execute(users.insert(), user_id=2, user_name='user2') tlengine.commit() tlengine.begin_twophase() tlengine.execute(users.insert(), user_id=3, user_name='user3') tlengine.rollback() tlengine.begin_twophase() tlengine.execute(users.insert(), user_id=4, user_name='user4') tlengine.prepare() tlengine.rollback() eq_(tlengine.execute(select([users.c.user_id]). order_by(users.c.user_id)).fetchall(), [(1, ), (2, )]) counters = None class ForUpdateTest(TestBase): @classmethod def setup_class(cls): global counters, metadata metadata = MetaData() counters = Table('forupdate_counters', metadata, Column('counter_id', INT, primary_key=True), Column('counter_value', INT), test_needs_acid=True) counters.create(testing.db) def teardown(self): testing.db.execute(counters.delete()).close() @classmethod def teardown_class(cls): counters.drop(testing.db) def increment( self, count, errors, update_style=True, delay=0.005, ): con = testing.db.connect() sel = counters.select(for_update=update_style, whereclause=counters.c.counter_id == 1) for i in xrange(count): trans = con.begin() try: existing = con.execute(sel).first() incr = existing['counter_value'] + 1 time.sleep(delay) con.execute(counters.update(counters.c.counter_id == 1, values={'counter_value': incr})) time.sleep(delay) readback = con.execute(sel).first() if readback['counter_value'] != incr: raise AssertionError('Got %s post-update, expected ' '%s' % (readback['counter_value'], incr)) trans.commit() except Exception, e: trans.rollback() errors.append(e) break con.close() @testing.crashes('mssql', 'FIXME: unknown') @testing.crashes('firebird', 'FIXME: unknown') @testing.crashes('sybase', 'FIXME: unknown') @testing.crashes('access', 'FIXME: unknown') @testing.requires.independent_connections def test_queued_update(self): """Test SELECT FOR UPDATE with concurrent modifications. Runs concurrent modifications on a single row in the users table, with each mutator trying to increment a value stored in user_name. """ db = testing.db db.execute(counters.insert(), counter_id=1, counter_value=0) iterations, thread_count = 10, 5 threads, errors = [], [] for i in xrange(thread_count): thrd = threading.Thread(target=self.increment, args=(iterations, ), kwargs={'errors': errors, 'update_style': True}) thrd.start() threads.append(thrd) for thrd in threads: thrd.join() for e in errors: sys.stdout.write('Failure: %s\n' % e) self.assert_(len(errors) == 0) sel = counters.select(whereclause=counters.c.counter_id == 1) final = db.execute(sel).first() self.assert_(final['counter_value'] == iterations * thread_count) def overlap( self, ids, errors, update_style, ): sel = counters.select(for_update=update_style, whereclause=counters.c.counter_id.in_(ids)) con = testing.db.connect() trans = con.begin() try: rows = con.execute(sel).fetchall() time.sleep(0.25) trans.commit() except Exception, e: trans.rollback() errors.append(e) con.close() def _threaded_overlap( self, thread_count, groups, update_style=True, pool=5, ): db = testing.db for cid in range(pool - 1): db.execute(counters.insert(), counter_id=cid + 1, counter_value=0) errors, threads = [], [] for i in xrange(thread_count): thrd = threading.Thread(target=self.overlap, args=(groups.pop(0), errors, update_style)) thrd.start() threads.append(thrd) for thrd in threads: thrd.join() return errors @testing.crashes('mssql', 'FIXME: unknown') @testing.crashes('firebird', 'FIXME: unknown') @testing.crashes('sybase', 'FIXME: unknown') @testing.crashes('access', 'FIXME: unknown') @testing.requires.independent_connections def test_queued_select(self): """Simple SELECT FOR UPDATE conflict test""" errors = self._threaded_overlap(2, [(1, 2, 3), (3, 4, 5)]) for e in errors: sys.stderr.write('Failure: %s\n' % e) self.assert_(len(errors) == 0) @testing.crashes('mssql', 'FIXME: unknown') @testing.fails_on('mysql', 'No support for NOWAIT') @testing.crashes('firebird', 'FIXME: unknown') @testing.crashes('sybase', 'FIXME: unknown') @testing.crashes('access', 'FIXME: unknown') @testing.requires.independent_connections def test_nowait_select(self): """Simple SELECT FOR UPDATE NOWAIT conflict test""" errors = self._threaded_overlap(2, [(1, 2, 3), (3, 4, 5)], update_style='nowait') self.assert_(len(errors) != 0)