from sqlalchemy import event from sqlalchemy import exc from sqlalchemy import func from sqlalchemy import INT from sqlalchemy import MetaData from sqlalchemy import pool as _pool from sqlalchemy import select from sqlalchemy import testing from sqlalchemy import util from sqlalchemy import VARCHAR from sqlalchemy.engine import base from sqlalchemy.engine import characteristics from sqlalchemy.engine import default from sqlalchemy.engine import url from sqlalchemy.pool import AsyncAdaptedQueuePool from sqlalchemy.pool import QueuePool from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import eq_ from sqlalchemy.testing import expect_warnings from sqlalchemy.testing import fixtures from sqlalchemy.testing import mock from sqlalchemy.testing import ne_ from sqlalchemy.testing.assertions import expect_raises_message from sqlalchemy.testing.engines import testing_engine from sqlalchemy.testing.schema import Column from sqlalchemy.testing.schema import Table class TransactionTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "users", metadata, Column("user_id", INT, primary_key=True), Column("user_name", VARCHAR(20)), test_needs_acid=True, ) @testing.fixture def local_connection(self): with testing.db.connect() as conn: yield conn def test_commits(self, local_connection): users = self.tables.users connection = local_connection transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) transaction.commit() transaction = connection.begin() connection.execute(users.insert(), dict(user_id=2, user_name="user2")) connection.execute(users.insert(), dict(user_id=3, user_name="user3")) transaction.commit() transaction = connection.begin() result = connection.exec_driver_sql("select * from users") assert len(result.fetchall()) == 3 transaction.commit() connection.close() def test_rollback(self, local_connection): """test a basic rollback""" users = self.tables.users connection = local_connection transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) connection.execute(users.insert(), dict(user_id=2, user_name="user2")) connection.execute(users.insert(), dict(user_id=3, user_name="user3")) transaction.rollback() result = connection.exec_driver_sql("select * from users") assert len(result.fetchall()) == 0 def test_raise(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin() try: connection.execute( users.insert(), dict(user_id=1, user_name="user1") ) connection.execute( users.insert(), dict(user_id=2, user_name="user2") ) connection.execute( users.insert(), dict(user_id=1, user_name="user3") ) transaction.commit() assert False except Exception as e: print("Exception: ", e) transaction.rollback() result = connection.exec_driver_sql("select * from users") assert len(result.fetchall()) == 0 def test_rollback_end_ctx_manager_autocommit(self, local_connection): m1 = mock.Mock() event.listen(local_connection, "rollback", m1.rollback) event.listen(local_connection, "commit", m1.commit) with local_connection.begin() as trans: assert local_connection.in_transaction() trans.rollback() assert not local_connection.in_transaction() # previously, would be subject to autocommit. # now it raises with expect_raises_message( exc.InvalidRequestError, "Can't operate on closed transaction inside context manager. " "Please complete the context manager before emitting " "further commands.", ): local_connection.execute(select(1)) assert not local_connection.in_transaction() @testing.combinations((True,), (False,), argnames="roll_back_in_block") @testing.requires.savepoints def test_ctxmanager_rolls_back_savepoint( self, local_connection, roll_back_in_block ): m1 = mock.Mock() event.listen( local_connection, "rollback_savepoint", m1.rollback_savepoint ) event.listen(local_connection, "rollback", m1.rollback) event.listen(local_connection, "commit", m1.commit) with local_connection.begin() as trans: with expect_raises_message(Exception, "test"): with local_connection.begin_nested() as nested_trans: if roll_back_in_block: nested_trans.rollback() if 1 == 1: raise Exception("test") assert not nested_trans.is_active assert nested_trans._deactivated_from_connection assert trans.is_active assert local_connection.in_transaction() assert not trans._deactivated_from_connection eq_( m1.mock_calls, [ mock.call.rollback_savepoint( local_connection, mock.ANY, mock.ANY ), mock.call.commit(local_connection), ], ) def test_deactivated_warning_straight(self, local_connection): with expect_warnings( "transaction already deassociated from connection" ): trans = local_connection.begin() trans.rollback() trans.rollback() @testing.requires.savepoints def test_deactivated_savepoint_warning_straight(self, local_connection): with expect_warnings( "nested transaction already deassociated from connection" ): with local_connection.begin(): savepoint = local_connection.begin_nested() savepoint.rollback() savepoint.rollback() def test_commit_fails_flat(self, local_connection): connection = local_connection t1 = connection.begin() with mock.patch.object( connection, "_commit_impl", mock.Mock(side_effect=exc.DBAPIError("failure", None, None, None)), ): assert_raises_message(exc.DBAPIError, r"failure", t1.commit) assert not t1.is_active t1.rollback() # no error def test_commit_fails_ctxmanager(self, local_connection): connection = local_connection transaction = [None] def go(): with mock.patch.object( connection, "_commit_impl", mock.Mock( side_effect=exc.DBAPIError("failure", None, None, None) ), ): with connection.begin() as t1: transaction[0] = t1 assert_raises_message(exc.DBAPIError, r"failure", go) t1 = transaction[0] assert not t1.is_active with expect_warnings( "transaction already deassociated from connection" ): t1.rollback() # no error @testing.requires.savepoints_w_release def test_savepoint_rollback_fails_flat(self, local_connection): connection = local_connection t1 = connection.begin() s1 = connection.begin_nested() # force the "commit" of the savepoint that occurs # when the "with" block fails, e.g. # the RELEASE, to fail, because the savepoint is already # released. connection.dialect.do_release_savepoint(connection, s1._savepoint) assert_raises_message( exc.DBAPIError, r".*SQL\:.*ROLLBACK TO SAVEPOINT", s1.rollback ) assert not s1.is_active with testing.expect_warnings("nested transaction already"): s1.rollback() # no error (though it warns) # this test was previously calling "commit", but note relies on # buggy behavior in PostgreSQL as the transaction block is in fact # aborted. pg8000 enforces this on the client as of 1.29 t1.rollback() # no error @testing.requires.savepoints_w_release def test_savepoint_release_fails_flat(self): with testing.db.connect() as connection: t1 = connection.begin() s1 = connection.begin_nested() # force the "commit" of the savepoint that occurs # when the "with" block fails, e.g. # the RELEASE, to fail, because the savepoint is already # released. connection.dialect.do_release_savepoint(connection, s1._savepoint) assert_raises_message( exc.DBAPIError, r".*SQL\:.*RELEASE SAVEPOINT", s1.commit ) assert not s1.is_active s1.rollback() # no error. prior to 1.4 this would try to rollback # this test was previously calling "commit", but note relies on # buggy behavior in PostgreSQL as the transaction block is in fact # aborted. pg8000 enforces this on the client as of 1.29 t1.rollback() # no error @testing.requires.savepoints_w_release def test_savepoint_release_fails_ctxmanager(self, local_connection): connection = local_connection connection.begin() savepoint = [None] def go(): with connection.begin_nested() as sp: savepoint[0] = sp # force the "commit" of the savepoint that occurs # when the "with" block fails, e.g. # the RELEASE, to fail, because the savepoint is already # released. connection.dialect.do_release_savepoint( connection, sp._savepoint ) # prior to SQLAlchemy 1.4, the above release would fail # and then the savepoint would try to rollback, and that failed # also, causing a long exception chain that under Python 2 # was particularly hard to diagnose, leading to issue # #2696 which eventually impacted Openstack, and we # had to add warnings that show what the "context" for an # exception was. The SQL for the exception was # ROLLBACK TO SAVEPOINT, and up the exception chain would be # the RELEASE failing. # # now, when the savepoint "commit" fails, it sets itself as # inactive. so it does not try to rollback and it cleans # itself out appropriately. # exc_ = assert_raises_message( exc.DBAPIError, r".*SQL\:.*RELEASE SAVEPOINT", go ) savepoint = savepoint[0] assert not savepoint.is_active # ensure cause comes from the DBAPI assert isinstance(exc_.__cause__, testing.db.dialect.dbapi.Error) def test_retains_through_options(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) conn2 = connection.execution_options(dummy=True) conn2.execute(users.insert(), dict(user_id=2, user_name="user2")) transaction.rollback() eq_( connection.exec_driver_sql("select count(*) from users").scalar(), 0, ) def test_ctxmanager_interface(self, local_connection): # a legacy test, adapted for 2.x style, was called # "test_with_interface". this is likely an early test for when # the "with" construct was first added. connection = local_connection users = self.tables.users trans = connection.begin() with trans: connection.execute( users.insert(), dict(user_id=1, user_name="user1") ) connection.execute( users.insert(), dict(user_id=2, user_name="user2") ) assert trans.is_active assert not trans.is_active eq_( connection.exec_driver_sql("select count(*) from users").scalar(), 2, ) connection.rollback() def test_close(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) connection.execute(users.insert(), dict(user_id=2, user_name="user2")) connection.execute(users.insert(), dict(user_id=3, user_name="user3")) assert connection.in_transaction() transaction.commit() assert not connection.in_transaction() result = connection.exec_driver_sql("select * from users") eq_(len(result.fetchall()), 3) def test_close2(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) connection.execute(users.insert(), dict(user_id=2, user_name="user2")) connection.execute(users.insert(), dict(user_id=3, user_name="user3")) assert connection.in_transaction() transaction.close() assert not connection.in_transaction() result = connection.exec_driver_sql("select * from users") assert len(result.fetchall()) == 0 @testing.requires.independent_connections def test_no_rollback_in_deactive(self, local_connection): """test #7388""" def fail(*arg, **kw): raise BaseException("some base exception") with mock.patch.object(testing.db.dialect, "do_commit", fail): with expect_raises_message(BaseException, "some base exception"): with local_connection.begin(): pass @testing.requires.independent_connections @testing.requires.savepoints def test_no_rollback_in_deactive_savepoint(self, local_connection): """test #7388""" def fail(*arg, **kw): raise BaseException("some base exception") with mock.patch.object( testing.db.dialect, "do_release_savepoint", fail ): with local_connection.begin(): with expect_raises_message( BaseException, "some base exception" ): with local_connection.begin_nested(): pass @testing.requires.savepoints def test_nested_subtransaction_rollback(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) trans2 = connection.begin_nested() connection.execute(users.insert(), dict(user_id=2, user_name="user2")) trans2.rollback() connection.execute(users.insert(), dict(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,)], ) @testing.requires.savepoints def test_nested_subtransaction_commit(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) trans2 = connection.begin_nested() connection.execute(users.insert(), dict(user_id=2, user_name="user2")) trans2.commit() connection.execute(users.insert(), dict(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,)], ) @testing.requires.two_phase_transactions def test_two_phase_transaction(self, local_connection): connection = local_connection users = self.tables.users transaction = connection.begin_twophase() connection.execute(users.insert(), dict(user_id=1, user_name="user1")) transaction.prepare() transaction.commit() transaction = connection.begin_twophase() connection.execute(users.insert(), dict(user_id=2, user_name="user2")) transaction.commit() transaction.close() transaction = connection.begin_twophase() connection.execute(users.insert(), dict(user_id=3, user_name="user3")) transaction.rollback() transaction = connection.begin_twophase() connection.execute(users.insert(), dict(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,)], ) @testing.requires.two_phase_transactions @testing.requires.two_phase_recovery @testing.variation("commit", [True, False]) def test_two_phase_recover(self, commit): users = self.tables.users # 2020, still can't get this to work w/ modern MySQL or MariaDB. # the XA RECOVER comes back as bytes, OK, convert to string, # XA COMMIT then says Unknown XID. Also, the drivers seem to be # killing off the XID if I use the connection.invalidate() before # trying to access in another connection. Not really worth it # unless someone wants to step through how mysqlclient / pymysql # support this correctly. connection = testing.db.connect() transaction = connection.begin_twophase() connection.execute( users.insert(), dict(dict(user_id=1, user_name="user1")) ) transaction.prepare() connection.invalidate() with testing.db.connect() as connection2: eq_( connection2.execute( select(users.c.user_id).order_by(users.c.user_id) ).fetchall(), [], ) # recover_twophase needs to be run in a new transaction with testing.db.connect() as connection3: # oracle transactions can't be recovered for commit after... # about 1 second? OK with testing.skip_if_timeout( 0.50, cleanup=( lambda: connection3.rollback_prepared( transaction.xid, recover=True ) ), ): recoverables = connection3.recover_twophase() assert transaction.xid in recoverables if commit: connection3.commit_prepared(transaction.xid, recover=True) res = [(1,)] else: connection3.rollback_prepared(transaction.xid, recover=True) res = [] stmt = select(users.c.user_id).order_by(users.c.user_id) eq_(connection3.execute(stmt).fetchall(), res) @testing.requires.two_phase_transactions def test_multiple_two_phase(self, local_connection): conn = local_connection users = self.tables.users xa = conn.begin_twophase() conn.execute(users.insert(), dict(user_id=1, user_name="user1")) xa.prepare() xa.commit() xa = conn.begin_twophase() conn.execute(users.insert(), dict(user_id=2, user_name="user2")) xa.prepare() xa.rollback() xa = conn.begin_twophase() conn.execute(users.insert(), dict(user_id=3, user_name="user3")) xa.rollback() xa = conn.begin_twophase() conn.execute(users.insert(), dict(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",)]) @testing.requires.two_phase_transactions def test_reset_rollback_two_phase_no_rollback(self): # test [ticket:2907], essentially that the # TwoPhaseTransaction is given the job of "reset on return" # so that picky backends like MySQL correctly clear out # their state when a connection is closed without handling # the transaction explicitly. users = self.tables.users eng = testing_engine() # MySQL raises if you call straight rollback() on # a connection with an XID present @event.listens_for(eng, "invalidate") def conn_invalidated(dbapi_con, con_record, exception): dbapi_con.close() raise exception with eng.connect() as conn: rec = conn.connection._connection_record raw_dbapi_con = rec.dbapi_connection conn.begin_twophase() conn.execute(users.insert(), dict(user_id=1, user_name="user1")) assert rec.dbapi_connection is raw_dbapi_con with eng.connect() as conn: result = conn.execute( select(users.c.user_name).order_by(users.c.user_id) ) eq_(result.fetchall(), []) def test_interrupt_ctxmanager_engine(self, trans_ctx_manager_fixture): fn = trans_ctx_manager_fixture fn(testing.db, trans_on_subject=False, execute_on_subject=False) @testing.combinations((True,), (False,), argnames="trans_on_subject") def test_interrupt_ctxmanager_connection( self, trans_ctx_manager_fixture, trans_on_subject ): fn = trans_ctx_manager_fixture with testing.db.connect() as conn: fn( conn, trans_on_subject=trans_on_subject, execute_on_subject=True, ) def test_autobegin_rollback(self): users = self.tables.users with testing.db.connect() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) conn.rollback() eq_(conn.scalar(select(func.count(1)).select_from(users)), 0) @testing.requires.autocommit def test_autocommit_isolation_level(self): users = self.tables.users with testing.db.connect().execution_options( isolation_level="AUTOCOMMIT" ) as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) conn.rollback() with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) @testing.requires.autocommit def test_no_autocommit_w_begin(self): with testing.db.begin() as conn: assert_raises_message( exc.InvalidRequestError, r"This connection has already initialized a SQLAlchemy " r"Transaction\(\) object via begin\(\) or autobegin; " r"isolation_level may not be altered unless rollback\(\) or " r"commit\(\) is called first.", conn.execution_options, isolation_level="AUTOCOMMIT", ) @testing.requires.autocommit def test_no_autocommit_w_autobegin(self): with testing.db.connect() as conn: conn.execute(select(1)) assert_raises_message( exc.InvalidRequestError, r"This connection has already initialized a SQLAlchemy " r"Transaction\(\) object via begin\(\) or autobegin; " r"isolation_level may not be altered unless rollback\(\) or " r"commit\(\) is called first.", conn.execution_options, isolation_level="AUTOCOMMIT", ) conn.rollback() conn.execution_options(isolation_level="AUTOCOMMIT") def test_autobegin_commit(self): users = self.tables.users with testing.db.connect() as conn: assert not conn.in_transaction() conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) assert conn.in_transaction() conn.commit() assert not conn.in_transaction() eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) conn.execute(users.insert(), {"user_id": 2, "user_name": "name 2"}) eq_( conn.scalar(select(func.count(1)).select_from(users)), 2, ) assert conn.in_transaction() conn.rollback() assert not conn.in_transaction() eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) def test_rollback_on_close(self): canary = mock.Mock() with testing.db.connect() as conn: event.listen(conn, "rollback", canary) conn.execute(select(1)) assert conn.in_transaction() eq_(canary.mock_calls, [mock.call(conn)]) def test_no_on_close_no_transaction(self): canary = mock.Mock() with testing.db.connect() as conn: event.listen(conn, "rollback", canary) conn.execute(select(1)) conn.rollback() assert not conn.in_transaction() eq_(canary.mock_calls, [mock.call(conn)]) def test_rollback_on_exception(self): canary = mock.Mock() try: with testing.db.connect() as conn: event.listen(conn, "rollback", canary) conn.execute(select(1)) assert conn.in_transaction() raise Exception("some error") assert False except: pass eq_(canary.mock_calls, [mock.call(conn)]) def test_rollback_on_exception_if_no_trans(self): canary = mock.Mock() try: with testing.db.connect() as conn: event.listen(conn, "rollback", canary) assert not conn.in_transaction() raise Exception("some error") assert False except: pass eq_(canary.mock_calls, []) def test_commit_no_begin(self): with testing.db.connect() as conn: assert not conn.in_transaction() conn.commit() @testing.requires.independent_connections def test_commit_inactive(self): with testing.db.connect() as conn: conn.begin() conn.invalidate() assert_raises_message( exc.InvalidRequestError, "Can't reconnect until", conn.commit ) @testing.requires.independent_connections def test_rollback_inactive(self): users = self.tables.users with testing.db.connect() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) conn.commit() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) conn.invalidate() assert_raises_message( exc.PendingRollbackError, "Can't reconnect", conn.execute, select(1), ) conn.rollback() eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) def test_rollback_no_begin(self): with testing.db.connect() as conn: assert not conn.in_transaction() conn.rollback() def test_rollback_end_ctx_manager(self): with testing.db.begin() as conn: assert conn.in_transaction() conn.rollback() assert not conn.in_transaction() def test_rollback_end_ctx_manager_autobegin(self, local_connection): m1 = mock.Mock() event.listen(local_connection, "rollback", m1.rollback) event.listen(local_connection, "commit", m1.commit) with local_connection.begin() as trans: assert local_connection.in_transaction() trans.rollback() assert not local_connection.in_transaction() # previously, would be subject to autocommit. # now it raises with expect_raises_message( exc.InvalidRequestError, "Can't operate on closed transaction inside context manager. " "Please complete the context manager before emitting " "further commands.", ): local_connection.execute(select(1)) assert not local_connection.in_transaction() @testing.combinations((True,), (False,), argnames="roll_back_in_block") def test_ctxmanager_rolls_back(self, local_connection, roll_back_in_block): m1 = mock.Mock() event.listen(local_connection, "rollback", m1.rollback) event.listen(local_connection, "commit", m1.commit) with expect_raises_message(Exception, "test"): with local_connection.begin() as trans: if roll_back_in_block: trans.rollback() if 1 == 1: raise Exception("test") assert not trans.is_active assert not local_connection.in_transaction() assert trans._deactivated_from_connection eq_(m1.mock_calls, [mock.call.rollback(local_connection)]) @testing.requires.savepoints def test_ctxmanager_autobegins_real_trans_from_nested( self, local_connection ): # the legacy version of this test in 1.4 # was test_ctxmanager_commits_real_trans_from_nested m1 = mock.Mock() event.listen( local_connection, "rollback_savepoint", m1.rollback_savepoint ) event.listen( local_connection, "release_savepoint", m1.release_savepoint ) event.listen(local_connection, "rollback", m1.rollback) event.listen(local_connection, "commit", m1.commit) event.listen(local_connection, "begin", m1.begin) event.listen(local_connection, "savepoint", m1.savepoint) with local_connection.begin_nested() as nested_trans: pass assert not nested_trans.is_active assert nested_trans._deactivated_from_connection eq_( m1.mock_calls, [ mock.call.begin(local_connection), mock.call.savepoint(local_connection, mock.ANY), mock.call.release_savepoint( local_connection, mock.ANY, mock.ANY ), ], ) def test_explicit_begin(self): users = self.tables.users with testing.db.connect() as conn: assert not conn.in_transaction() conn.begin() assert conn.in_transaction() conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) conn.commit() eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) def test_no_double_begin(self): with testing.db.connect() as conn: conn.begin() assert_raises_message( exc.InvalidRequestError, r"This connection has already initialized a SQLAlchemy " r"Transaction\(\) object via begin\(\) or autobegin; can't " r"call begin\(\) here unless rollback\(\) or commit\(\) is " r"called first.", conn.begin, ) def test_no_autocommit(self): users = self.tables.users with testing.db.connect() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 0, ) def test_begin_block(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) @testing.requires.savepoints def test_savepoint_one(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) savepoint = conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) eq_( conn.scalar(select(func.count(1)).select_from(users)), 2, ) savepoint.rollback() eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) @testing.requires.savepoints def test_savepoint_two(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) savepoint = conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) eq_( conn.scalar(select(func.count(1)).select_from(users)), 2, ) savepoint.commit() eq_( conn.scalar(select(func.count(1)).select_from(users)), 2, ) with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 2, ) @testing.requires.savepoints def test_savepoint_three(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) conn.rollback() assert not conn.in_transaction() with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 0, ) @testing.requires.savepoints def test_savepoint_four(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) sp1 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) sp2 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 3, "user_name": "name3"}) sp2.rollback() assert not sp2.is_active assert sp1.is_active assert conn.in_transaction() assert not sp1.is_active with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 2, ) @testing.requires.savepoints def test_savepoint_five(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) sp2 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 3, "user_name": "name3"}) sp2.commit() assert conn.in_transaction() with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 3, ) @testing.requires.savepoints def test_savepoint_six(self): users = self.tables.users with testing.db.begin() as conn: conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) sp1 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) assert conn._nested_transaction is sp1 sp2 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 3, "user_name": "name3"}) assert conn._nested_transaction is sp2 sp2.commit() assert conn._nested_transaction is sp1 sp1.rollback() assert conn._nested_transaction is None assert conn.in_transaction() with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 1, ) @testing.requires.savepoints def test_savepoint_seven(self): users = self.tables.users with testing.db.connect() as conn: trans = conn.begin() conn.execute(users.insert(), {"user_id": 1, "user_name": "name"}) sp1 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 2, "user_name": "name2"}) sp2 = conn.begin_nested() conn.execute(users.insert(), {"user_id": 3, "user_name": "name3"}) assert conn.in_transaction() trans.close() assert not sp1.is_active assert not sp2.is_active assert not trans.is_active assert conn._transaction is None assert conn._nested_transaction is None with testing.db.connect() as conn: eq_( conn.scalar(select(func.count(1)).select_from(users)), 0, ) class AutoRollbackTest(fixtures.TestBase): __backend__ = True @classmethod def setup_test_class(cls): global metadata metadata = MetaData() @classmethod def teardown_test_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, ) with conn1.begin(): users.create(conn1) conn1.exec_driver_sql("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 ! with conn2.begin(): users.drop(conn2) conn2.close() class IsolationLevelTest(fixtures.TestBase): """see also sqlalchemy/testing/suite/test_dialect.py::IsolationLevelTest""" __requires__ = ( "isolation_level", "ad_hoc_engines", ) __backend__ = True def _default_isolation_level(self): return testing.requires.get_isolation_levels(testing.config)["default"] def _non_default_isolation_level(self): levels = testing.requires.get_isolation_levels(testing.config) default = levels["default"] supported = levels["supported"] s = set(supported).difference(["AUTOCOMMIT", default]) if s: return s.pop() else: assert False, "no non-default isolation level available" def test_engine_param_stays(self): eng = testing_engine() with eng.connect() as conn: isolation_level = eng.dialect.get_isolation_level( conn.connection.dbapi_connection ) level = self._non_default_isolation_level() ne_(isolation_level, level) eng = testing_engine(options=dict(isolation_level=level)) with eng.connect() as conn: eq_( eng.dialect.get_isolation_level( conn.connection.dbapi_connection ), level, ) # check that it stays with eng.connect() as conn: eq_( eng.dialect.get_isolation_level( conn.connection.dbapi_connection ), level, ) with eng.connect() as conn: eq_( eng.dialect.get_isolation_level( conn.connection.dbapi_connection ), level, ) def test_default_level(self): eng = testing_engine(options=dict()) with eng.connect() as conn: isolation_level = eng.dialect.get_isolation_level( conn.connection.dbapi_connection ) eq_(isolation_level, self._default_isolation_level()) def test_reset_level(self): eng = testing_engine(options=dict()) conn = eng.connect() eq_( eng.dialect.get_isolation_level(conn.connection.dbapi_connection), self._default_isolation_level(), ) eng.dialect.set_isolation_level( conn.connection.dbapi_connection, self._non_default_isolation_level(), ) eq_( eng.dialect.get_isolation_level(conn.connection.dbapi_connection), self._non_default_isolation_level(), ) eng.dialect.reset_isolation_level(conn.connection.dbapi_connection) eq_( eng.dialect.get_isolation_level(conn.connection.dbapi_connection), self._default_isolation_level(), ) conn.close() def test_reset_level_with_setting(self): eng = testing_engine( options=dict(isolation_level=self._non_default_isolation_level()) ) conn = eng.connect() eq_( eng.dialect.get_isolation_level(conn.connection.dbapi_connection), self._non_default_isolation_level(), ) eng.dialect.set_isolation_level( conn.connection.dbapi_connection, self._default_isolation_level() ) eq_( eng.dialect.get_isolation_level(conn.connection.dbapi_connection), self._default_isolation_level(), ) eng.dialect.reset_isolation_level(conn.connection.dbapi_connection) eq_( eng.dialect.get_isolation_level(conn.connection.dbapi_connection), self._non_default_isolation_level(), ) conn.close() def test_underscore_replacement(self, connection_no_trans): conn = connection_no_trans with mock.patch.object( conn.dialect, "set_isolation_level" ) as mock_sil, mock.patch.object( conn.dialect, "_gen_allowed_isolation_levels", mock.Mock(return_value=["READ COMMITTED", "REPEATABLE READ"]), ): conn.execution_options(isolation_level="REPEATABLE_READ") dbapi_conn = conn.connection.dbapi_connection eq_(mock_sil.mock_calls, [mock.call(dbapi_conn, "REPEATABLE READ")]) def test_casing_replacement(self, connection_no_trans): conn = connection_no_trans with mock.patch.object( conn.dialect, "set_isolation_level" ) as mock_sil, mock.patch.object( conn.dialect, "_gen_allowed_isolation_levels", mock.Mock(return_value=["READ COMMITTED", "REPEATABLE READ"]), ): conn.execution_options(isolation_level="repeatable_read") dbapi_conn = conn.connection.dbapi_connection eq_(mock_sil.mock_calls, [mock.call(dbapi_conn, "REPEATABLE READ")]) def test_dialect_doesnt_follow_naming_guidelines( self, connection_no_trans ): conn = connection_no_trans conn.dialect.__dict__.pop("_gen_allowed_isolation_levels", None) with mock.patch.object( conn.dialect, "get_isolation_level_values", mock.Mock( return_value=[ "READ COMMITTED", "REPEATABLE_READ", "serializable", ] ), ): with expect_raises_message( ValueError, f"Dialect {conn.dialect.name!r} " r"get_isolation_level_values\(\) method " r"should " r"return names as UPPERCASE using spaces, not underscores; " r"got \['REPEATABLE_READ', 'serializable'\]", ): conn.execution_options(isolation_level="READ COMMITTED") def test_invalid_level_engine_param(self): eng = testing_engine(options=dict(isolation_level="FOO")) assert_raises_message( exc.ArgumentError, f"Invalid value 'FOO' for isolation_level. " f"Valid isolation levels for {eng.dialect.name!r} are " f"""{', '.join( testing.requires.get_isolation_levels( testing.config )['supported'] )}""", eng.connect, ) def test_invalid_level_execution_option(self): eng = testing_engine( options=dict(execution_options=dict(isolation_level="FOO")) ) assert_raises_message( exc.ArgumentError, f"Invalid value 'FOO' for isolation_level. " f"Valid isolation levels for {eng.dialect.name!r} are " f"""{', '.join( testing.requires.get_isolation_levels( testing.config )['supported'] )}""", eng.connect, ) def test_connection_invalidated(self): eng = testing_engine() with eng.connect() as conn: c2 = conn.execution_options( isolation_level=self._non_default_isolation_level() ) c2.invalidate() c2.connection # TODO: do we want to rebuild the previous isolation? # for now, this is current behavior so we will leave it. eq_(c2.get_isolation_level(), self._default_isolation_level()) def test_per_connection(self): eng = testing_engine( options=dict( poolclass=( QueuePool if not testing.db.dialect.is_async else AsyncAdaptedQueuePool ), pool_size=2, max_overflow=0, ) ) c1 = eng.connect() c1 = c1.execution_options( isolation_level=self._non_default_isolation_level() ) c2 = eng.connect() eq_( eng.dialect.get_isolation_level(c1.connection.dbapi_connection), self._non_default_isolation_level(), ) eq_( eng.dialect.get_isolation_level(c2.connection.dbapi_connection), self._default_isolation_level(), ) c1.close() c2.close() c3 = eng.connect() eq_( eng.dialect.get_isolation_level(c3.connection.dbapi_connection), self._default_isolation_level(), ) c4 = eng.connect() eq_( eng.dialect.get_isolation_level(c4.connection.dbapi_connection), self._default_isolation_level(), ) c3.close() c4.close() def test_exception_in_transaction(self): eng = testing_engine() with eng.connect() as c1: with expect_raises_message( exc.InvalidRequestError, r"This connection has already initialized a SQLAlchemy " r"Transaction\(\) object via begin\(\) or autobegin; " r"isolation_level may not be altered unless rollback\(\) or " r"commit\(\) is called first.", ): with c1.begin(): c1 = c1.execution_options( isolation_level=self._non_default_isolation_level() ) # was never set, so we are on original value eq_( eng.dialect.get_isolation_level( c1.connection.dbapi_connection ), self._default_isolation_level(), ) def test_per_statement_bzzt(self): assert_raises_message( exc.ArgumentError, r"'isolation_level' execution option may only be specified " r"on Connection.execution_options\(\), or " r"per-engine using the isolation_level " r"argument to create_engine\(\).", select(1).execution_options, isolation_level=self._non_default_isolation_level(), ) def test_per_engine(self): # new in 0.9 eng = testing_engine( testing.db.url, options=dict( execution_options={ "isolation_level": self._non_default_isolation_level() } ), ) with eng.connect() as conn: eq_( eng.dialect.get_isolation_level( conn.connection.dbapi_connection ), self._non_default_isolation_level(), ) def test_per_option_engine(self): eng = testing_engine(testing.db.url).execution_options( isolation_level=self._non_default_isolation_level() ) with eng.connect() as conn: eq_( eng.dialect.get_isolation_level( conn.connection.dbapi_connection ), self._non_default_isolation_level(), ) def test_isolation_level_accessors_connection_default(self): eng = testing_engine(testing.db.url) with eng.connect() as conn: eq_(conn.default_isolation_level, self._default_isolation_level()) with eng.connect() as conn: eq_(conn.get_isolation_level(), self._default_isolation_level()) def test_isolation_level_accessors_connection_option_modified(self): eng = testing_engine(testing.db.url) with eng.connect() as conn: c2 = conn.execution_options( isolation_level=self._non_default_isolation_level() ) eq_(conn.default_isolation_level, self._default_isolation_level()) eq_( conn.get_isolation_level(), self._non_default_isolation_level() ) eq_(c2.get_isolation_level(), self._non_default_isolation_level()) class ConnectionCharacteristicTest(fixtures.TestBase): @testing.fixture def characteristic_fixture(self): class FooCharacteristic(characteristics.ConnectionCharacteristic): transactional = True def reset_characteristic(self, dialect, dbapi_conn): dialect.reset_foo(dbapi_conn) def set_characteristic(self, dialect, dbapi_conn, value): dialect.set_foo(dbapi_conn, value) def get_characteristic(self, dialect, dbapi_conn): return dialect.get_foo(dbapi_conn) class FooDialect(default.DefaultDialect): connection_characteristics = util.immutabledict( {"foo": FooCharacteristic()} ) def reset_foo(self, dbapi_conn): dbapi_conn.foo = "original_value" def set_foo(self, dbapi_conn, value): dbapi_conn.foo = value def get_foo(self, dbapi_conn): return dbapi_conn.foo connection = mock.Mock() def creator(): connection.foo = "original_value" return connection pool = _pool.SingletonThreadPool(creator=creator) u = url.make_url("foo://") return base.Engine(pool, FooDialect(), u), connection def test_engine_param_stays(self, characteristic_fixture): engine, connection = characteristic_fixture foo_level = engine.dialect.get_foo(engine.connect().connection) new_level = "new_level" ne_(foo_level, new_level) eng = engine.execution_options(foo=new_level) eq_(eng.dialect.get_foo(eng.connect().connection), new_level) # check that it stays conn = eng.connect() eq_(eng.dialect.get_foo(conn.connection), new_level) conn.close() conn = eng.connect() eq_(eng.dialect.get_foo(conn.connection), new_level) conn.close() def test_default_level(self, characteristic_fixture): engine, connection = characteristic_fixture eq_( engine.dialect.get_foo(engine.connect().connection), "original_value", ) def test_connection_invalidated(self, characteristic_fixture): engine, connection = characteristic_fixture conn = engine.connect() c2 = conn.execution_options(foo="new_value") eq_(connection.foo, "new_value") c2.invalidate() c2.connection eq_(connection.foo, "original_value") def test_exception_in_transaction(self, characteristic_fixture): # this was a warning in 1.x. it appears we did not test the # 2.0 error case in 1.4 engine, connection = characteristic_fixture c1 = engine.connect() with expect_raises_message( exc.InvalidRequestError, r"This connection has already initialized a SQLAlchemy " r"Transaction\(\) object via begin\(\) or autobegin; " r"foo may not be altered unless rollback\(\) or " r"commit\(\) is called first.", ): with c1.begin(): c1 = c1.execution_options(foo="new_foo") # was never set, so we are on original value eq_(engine.dialect.get_foo(c1.connection), "original_value") @testing.fails("no error is raised yet here.") def test_per_statement_bzzt(self, characteristic_fixture): engine, connection = characteristic_fixture # this would need some on-execute mechanism to look inside of # the characteristics list. unfortunately this would # add some latency. assert_raises_message( exc.ArgumentError, r"'foo' execution option may only be specified " r"on Connection.execution_options\(\), or " r"per-engine using the isolation_level " r"argument to create_engine\(\).", connection.execute, select([1]).execution_options(foo="bar"), ) def test_per_engine(self, characteristic_fixture): engine, connection = characteristic_fixture pool, dialect, url = engine.pool, engine.dialect, engine.url eng = base.Engine( pool, dialect, url, execution_options={"foo": "new_value"} ) conn = eng.connect() eq_(eng.dialect.get_foo(conn.connection), "new_value") def test_per_option_engine(self, characteristic_fixture): engine, connection = characteristic_fixture eng = engine.execution_options(foo="new_value") conn = eng.connect() eq_( eng.dialect.get_foo(conn.connection), "new_value", ) class ResetFixture: @testing.fixture() def reset_agent(self, testing_engine): engine = testing_engine() engine.connect().close() harness = mock.Mock( do_rollback=mock.Mock(side_effect=testing.db.dialect.do_rollback), do_commit=mock.Mock(side_effect=testing.db.dialect.do_commit), engine=engine, ) event.listen(engine, "rollback", harness.rollback) event.listen(engine, "commit", harness.commit) event.listen(engine, "rollback_savepoint", harness.rollback_savepoint) event.listen(engine, "rollback_twophase", harness.rollback_twophase) event.listen(engine, "commit_twophase", harness.commit_twophase) with mock.patch.object( engine.dialect, "do_rollback", harness.do_rollback ), mock.patch.object(engine.dialect, "do_commit", harness.do_commit): yield harness event.remove(engine, "rollback", harness.rollback) event.remove(engine, "commit", harness.commit) event.remove(engine, "rollback_savepoint", harness.rollback_savepoint) event.remove(engine, "rollback_twophase", harness.rollback_twophase) event.remove(engine, "commit_twophase", harness.commit_twophase) class ResetAgentTest(ResetFixture, fixtures.TestBase): # many of these tests illustate rollback-on-return being redundant # vs. what the transaction just did, however this is to ensure # even if statements were invoked on the DBAPI connection directly, # the state is cleared. options to optimize this with clear # docs etc. should be added. __backend__ = True def test_begin_close(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() assert not trans.is_active eq_( reset_agent.mock_calls, [mock.call.rollback(connection), mock.call.do_rollback(mock.ANY)], ) def test_begin_rollback(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() trans.rollback() assert not trans.is_active eq_( reset_agent.mock_calls, [ mock.call.rollback(connection), mock.call.do_rollback(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) def test_begin_commit(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() trans.commit() assert not trans.is_active eq_( reset_agent.mock_calls, [ mock.call.commit(connection), mock.call.do_commit(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) def test_trans_close(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() trans.close() eq_( reset_agent.mock_calls, [ mock.call.rollback(connection), mock.call.do_rollback(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) @testing.requires.savepoints def test_begin_nested_trans_close_one(self, reset_agent): with reset_agent.engine.connect() as connection: t1 = connection.begin() t2 = connection.begin_nested() assert connection._nested_transaction is t2 assert connection._transaction is t1 t2.close() assert connection._nested_transaction is None assert connection._transaction is t1 t1.close() assert not t1.is_active eq_( reset_agent.mock_calls, [ mock.call.rollback_savepoint(connection, mock.ANY, mock.ANY), mock.call.rollback(connection), mock.call.do_rollback(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) @testing.requires.savepoints def test_begin_nested_trans_close_two(self, reset_agent): with reset_agent.engine.connect() as connection: t1 = connection.begin() t2 = connection.begin_nested() assert connection._nested_transaction is t2 assert connection._transaction is t1 t1.close() assert connection._nested_transaction is None assert connection._transaction is None assert not t1.is_active eq_( reset_agent.mock_calls, [ mock.call.rollback(connection), mock.call.do_rollback(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) @testing.requires.savepoints def test_begin_nested_trans_rollback(self, reset_agent): with reset_agent.engine.connect() as connection: t1 = connection.begin() t2 = connection.begin_nested() assert connection._nested_transaction is t2 assert connection._transaction is t1 t2.close() assert connection._nested_transaction is None assert connection._transaction is t1 t1.rollback() assert connection._transaction is None assert not t2.is_active assert not t1.is_active eq_( reset_agent.mock_calls, [ mock.call.rollback_savepoint(connection, mock.ANY, mock.ANY), mock.call.rollback(connection), mock.call.do_rollback(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) @testing.requires.savepoints def test_begin_nested_close(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin_nested() # it's a savepoint, but root made sure it closed assert not trans.is_active eq_( reset_agent.mock_calls, [mock.call.rollback(connection), mock.call.do_rollback(mock.ANY)], ) @testing.requires.savepoints def test_begin_begin_nested_close(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() trans2 = connection.begin_nested() assert not trans2.is_active assert not trans.is_active eq_( reset_agent.mock_calls, [mock.call.rollback(connection), mock.call.do_rollback(mock.ANY)], ) @testing.requires.savepoints def test_begin_begin_nested_rollback_commit(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() trans2 = connection.begin_nested() trans2.rollback() # this is not a connection level event trans.commit() eq_( reset_agent.mock_calls, [ mock.call.rollback_savepoint(connection, mock.ANY, None), mock.call.commit(connection), mock.call.do_commit(mock.ANY), mock.call.do_rollback(mock.ANY), ], ) @testing.requires.savepoints def test_begin_begin_nested_rollback_rollback(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin() trans2 = connection.begin_nested() trans2.rollback() trans.rollback() eq_( reset_agent.mock_calls, [ mock.call.rollback_savepoint(connection, mock.ANY, mock.ANY), mock.call.rollback(connection), mock.call.do_rollback(mock.ANY), mock.call.do_rollback(mock.ANY), # this is the reset on return ], ) @testing.requires.two_phase_transactions def test_reset_via_agent_begin_twophase(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin_twophase() assert not trans.is_active # pg8000 uses the rollback_twophase as the full rollback. eq_( reset_agent.mock_calls[0], mock.call.rollback_twophase(connection, mock.ANY, False), ) @testing.requires.two_phase_transactions def test_reset_via_agent_begin_twophase_commit(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin_twophase() trans.commit() # again pg8000 vs. other PG drivers have different API eq_( reset_agent.mock_calls[0], mock.call.commit_twophase(connection, mock.ANY, False), ) eq_(reset_agent.mock_calls[-1], mock.call.do_rollback(mock.ANY)) @testing.requires.two_phase_transactions def test_reset_via_agent_begin_twophase_rollback(self, reset_agent): with reset_agent.engine.connect() as connection: trans = connection.begin_twophase() trans.rollback() # pg8000 vs. the other postgresql drivers have different # twophase implementations. the base postgresql driver emits # "ROLLBACK PREPARED" explicitly then calls do_rollback(). # pg8000 has a dedicated API method. so we get either one or # two do_rollback() at the end, just need at least one. eq_( reset_agent.mock_calls[0:2], [ mock.call.rollback_twophase(connection, mock.ANY, False), mock.call.do_rollback(mock.ANY), # mock.call.do_rollback(mock.ANY), ], ) eq_(reset_agent.mock_calls[-1], mock.call.do_rollback(mock.ANY)) def test_reset_agent_no_conn_transaction(self, reset_agent): with reset_agent.engine.connect(): pass eq_(reset_agent.mock_calls, [mock.call.do_rollback(mock.ANY)])