from __future__ import annotations import contextlib import random from typing import Optional from typing import TYPE_CHECKING from sqlalchemy import event from sqlalchemy import exc as sa_exc from sqlalchemy import func from sqlalchemy import inspect from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import select from sqlalchemy import String from sqlalchemy import Table from sqlalchemy import testing from sqlalchemy import text from sqlalchemy.orm import attributes from sqlalchemy.orm import clear_mappers from sqlalchemy.orm import exc as orm_exc from sqlalchemy.orm import relationship from sqlalchemy.orm import Session from sqlalchemy.orm import session as _session from sqlalchemy.orm import sessionmaker from sqlalchemy.orm.util import identity_key from sqlalchemy.sql import elements from sqlalchemy.testing import assert_raises from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import assert_warnings from sqlalchemy.testing import engines from sqlalchemy.testing import eq_ from sqlalchemy.testing import expect_raises_message from sqlalchemy.testing import expect_warnings from sqlalchemy.testing import fixtures from sqlalchemy.testing import is_ from sqlalchemy.testing import is_not from sqlalchemy.testing import mock from sqlalchemy.testing.config import Variation from sqlalchemy.testing.fixtures import fixture_session from sqlalchemy.testing.schema import Column from sqlalchemy.testing.util import gc_collect from test.orm._fixtures import FixtureTest if TYPE_CHECKING: from sqlalchemy import NestedTransaction from sqlalchemy import Transaction class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest): run_inserts = None __backend__ = True def test_no_close_transaction_on_flush(self, connection): User, users = self.classes.User, self.tables.users c = connection self.mapper_registry.map_imperatively(User, users) s = Session(bind=c) s.begin() tran = s.get_transaction() s.add(User(name="first")) s.flush() c.exec_driver_sql("select * from users") u = User(name="two") s.add(u) s.flush() u = User(name="third") s.add(u) s.flush() assert s.get_transaction() is tran tran.close() def test_subtransaction_on_external_no_begin(self, connection_no_trans): users, User = self.tables.users, self.classes.User connection = connection_no_trans self.mapper_registry.map_imperatively(User, users) trans = connection.begin() sess = Session(bind=connection, autoflush=True) u = User(name="ed") sess.add(u) sess.flush() sess.commit() # commit does nothing trans.rollback() # rolls back assert len(sess.query(User).all()) == 0 sess.close() @testing.requires.savepoints def test_external_nested_transaction(self, connection_no_trans): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) connection = connection_no_trans trans = connection.begin() sess = Session(bind=connection, autoflush=True) u1 = User(name="u1") sess.add(u1) sess.flush() savepoint = sess.begin_nested() u2 = User(name="u2") sess.add(u2) sess.flush() savepoint.rollback() trans.commit() assert len(sess.query(User).all()) == 1 join_transaction_mode = testing.variation( "join_transaction_mode", [ "none", "conditional_savepoint", "create_savepoint", "control_fully", "rollback_only", ], ) @join_transaction_mode @testing.variation("operation", ["commit", "close", "rollback", "nothing"]) @testing.variation("external_state", ["none", "transaction", "savepoint"]) def test_join_transaction_modes( self, connection_no_trans, join_transaction_mode, operation, external_state: testing.Variation, ): """test new join_transaction modes added in #9015""" connection = connection_no_trans t1: Optional[Transaction] s1: Optional[NestedTransaction] if external_state.none: t1 = s1 = None elif external_state.transaction: t1 = connection.begin() s1 = None elif external_state.savepoint: t1 = connection.begin() s1 = connection.begin_nested() else: external_state.fail() if join_transaction_mode.none: sess = Session(connection) else: sess = Session( connection, join_transaction_mode=join_transaction_mode.name ) sess.connection() if operation.close: sess.close() elif operation.commit: sess.commit() elif operation.rollback: sess.rollback() elif operation.nothing: pass else: operation.fail() if external_state.none: if operation.nothing: assert connection.in_transaction() else: assert not connection.in_transaction() elif external_state.transaction: assert t1 is not None if ( join_transaction_mode.none or join_transaction_mode.conditional_savepoint or join_transaction_mode.rollback_only ): if operation.rollback: assert t1._deactivated_from_connection assert not t1.is_active else: assert not t1._deactivated_from_connection assert t1.is_active elif join_transaction_mode.create_savepoint: assert not t1._deactivated_from_connection assert t1.is_active elif join_transaction_mode.control_fully: if operation.nothing: assert not t1._deactivated_from_connection assert t1.is_active else: assert t1._deactivated_from_connection assert not t1.is_active else: join_transaction_mode.fail() if t1.is_active: t1.rollback() elif external_state.savepoint: assert s1 is not None assert t1 is not None assert not t1._deactivated_from_connection assert t1.is_active if join_transaction_mode.rollback_only: if operation.rollback: assert s1._deactivated_from_connection assert not s1.is_active else: assert not s1._deactivated_from_connection assert s1.is_active elif join_transaction_mode.control_fully: if operation.nothing: assert not s1._deactivated_from_connection assert s1.is_active else: assert s1._deactivated_from_connection assert not s1.is_active else: if operation.nothing: # session is still open in the sub-savepoint, # so we are not activated on connection assert s1._deactivated_from_connection # but we are still an active savepoint assert s1.is_active # close session, then we're good sess.close() assert not s1._deactivated_from_connection assert s1.is_active if s1.is_active: s1.rollback() if t1.is_active: t1.rollback() else: external_state.fail() @join_transaction_mode def test_join_transaction_mode_with_event(self, join_transaction_mode): eng = engines.testing_engine() @event.listens_for(eng, "engine_connect") def make_transaction(conn): conn.begin() if join_transaction_mode.none: s = Session(eng) else: s = Session(eng, join_transaction_mode=join_transaction_mode.name) if ( join_transaction_mode.none or join_transaction_mode.conditional_savepoint ): with expect_warnings( "The engine provided as bind produced a " "connection that is already in a transaction. " "This is usually caused by a core event, " "such as 'engine_connect', that has left a " "transaction open. The effective join " "transaction mode used by this session is " "'rollback_only'. To silence this " "warning, do not leave transactions open" ): s.connection() else: s.connection() def test_subtransaction_on_external_commit(self, connection_no_trans): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) connection = connection_no_trans connection.begin() sess = Session(bind=connection, autoflush=True) u = User(name="ed") sess.add(u) sess.flush() sess.commit() # commit does nothing connection.rollback() # rolls back assert len(sess.query(User).all()) == 0 sess.close() def test_subtransaction_on_external_rollback(self, connection_no_trans): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) connection = connection_no_trans connection.begin() sess = Session(bind=connection, autoflush=True) u = User(name="ed") sess.add(u) sess.flush() sess.rollback() # rolls back connection.commit() # nothing to commit assert len(sess.query(User).all()) == 0 sess.close() @testing.requires.savepoints def test_savepoint_on_external(self, connection_no_trans): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) connection = connection_no_trans connection.begin() sess = Session(bind=connection, autoflush=True) u1 = User(name="u1") sess.add(u1) sess.flush() n1 = sess.begin_nested() u2 = User(name="u2") sess.add(u2) sess.flush() n1.rollback() connection.commit() assert len(sess.query(User).all()) == 1 @testing.requires.savepoints def test_nested_accounting_new_items_removed(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) session = fixture_session() session.begin() n1 = session.begin_nested() u1 = User(name="u1") session.add(u1) n1.commit() assert u1 in session session.rollback() assert u1 not in session @testing.requires.savepoints def test_nested_accounting_deleted_items_restored(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) session = fixture_session() session.begin() u1 = User(name="u1") session.add(u1) session.commit() session.begin() u1 = session.query(User).first() n1 = session.begin_nested() session.delete(u1) n1.commit() assert u1 not in session session.rollback() assert u1 in session @testing.requires.savepoints def test_dirty_state_transferred_deep_nesting(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) with fixture_session() as s: u1 = User(name="u1") s.add(u1) s.commit() nt1 = s.begin_nested() nt2 = s.begin_nested() u1.name = "u2" assert attributes.instance_state(u1) not in nt2._dirty assert attributes.instance_state(u1) not in nt1._dirty s.flush() assert attributes.instance_state(u1) in nt2._dirty assert attributes.instance_state(u1) not in nt1._dirty nt2.commit() assert attributes.instance_state(u1) in nt2._dirty assert attributes.instance_state(u1) in nt1._dirty nt1.rollback() assert attributes.instance_state(u1).expired eq_(u1.name, "u1") @testing.requires.independent_connections def test_transactions_isolated(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) s1 = fixture_session() s2 = fixture_session() u1 = User(name="u1") s1.add(u1) s1.flush() assert s2.query(User).all() == [] @testing.requires.two_phase_transactions def test_twophase(self): users, Address, addresses, User = ( self.tables.users, self.classes.Address, self.tables.addresses, self.classes.User, ) # TODO: mock up a failure condition here # to ensure a rollback succeeds self.mapper_registry.map_imperatively(User, users) self.mapper_registry.map_imperatively(Address, addresses) engine2 = engines.testing_engine() sess = fixture_session(autoflush=False, twophase=True) sess.bind_mapper(User, testing.db) sess.bind_mapper(Address, engine2) sess.begin() u1 = User(name="u1") a1 = Address(email_address="u1@e") sess.add_all((u1, a1)) sess.commit() sess.close() engine2.dispose() with testing.db.connect() as conn: eq_(conn.scalar(select(func.count("*")).select_from(users)), 1) eq_(conn.scalar(select(func.count("*")).select_from(addresses)), 1) @testing.requires.independent_connections def test_invalidate(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) sess = fixture_session() u = User(name="u1") sess.add(u) sess.flush() c1 = sess.connection(bind_arguments={"mapper": User}) dbapi_conn = c1.connection assert dbapi_conn.is_valid sess.invalidate() # Connection object is closed assert c1.closed # "invalidated" is not part of "closed" state assert not c1.invalidated # but the DBAPI conn (really ConnectionFairy) # is invalidated assert not dbapi_conn.is_valid eq_(sess.query(User).all(), []) c2 = sess.connection(bind_arguments={"mapper": User}) assert not c2.invalidated assert c2.connection.is_valid @testing.requires.savepoints def test_nested_transaction(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) sess = fixture_session() sess.begin() u = User(name="u1") sess.add(u) sess.flush() n1 = sess.begin_nested() # nested transaction u2 = User(name="u2") sess.add(u2) sess.flush() n1.rollback() sess.commit() assert len(sess.query(User).all()) == 1 sess.close() @testing.requires.savepoints def test_nested_autotrans(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) sess = fixture_session() u = User(name="u1") sess.add(u) sess.flush() sess.begin_nested() # nested transaction u2 = User(name="u2") sess.add(u2) sess.flush() sess.rollback() # rolls back the whole trans sess.commit() assert len(sess.query(User).all()) == 0 sess.close() @testing.requires.savepoints def test_nested_transaction_connection_add(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() sess.begin() sess.begin_nested() u1 = User(name="u1") sess.add(u1) sess.flush() sess.rollback() u2 = User(name="u2") sess.add(u2) sess.commit() eq_(set(sess.query(User).all()), {u2}) sess.rollback() sess.begin() n1 = sess.begin_nested() u3 = User(name="u3") sess.add(u3) n1.commit() # commit the nested transaction sess.rollback() eq_(set(sess.query(User).all()), {u2}) sess.close() @testing.requires.savepoints def test_mixed_transaction_close(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() sess.begin_nested() sess.add(User(name="u1")) sess.flush() sess.close() sess.add(User(name="u2")) sess.commit() sess.close() eq_(len(sess.query(User).all()), 1) def test_begin_fails_connection_is_closed(self): eng = engines.testing_engine() state = [] @event.listens_for(eng, "begin") def do_begin(conn): state.append((conn, conn.connection)) raise Exception("failure") s1 = Session(eng) assert_raises_message( Exception, "failure", s1.execute, text("select 1") ) conn, fairy = state[0] assert not fairy.is_valid assert conn.closed assert not conn.invalidated s1.close() # close does not occur because references were not saved, however # the underlying DBAPI connection was closed assert not fairy.is_valid assert conn.closed assert not conn.invalidated def test_begin_savepoint_fails_connection_is_not_closed(self): eng = engines.testing_engine() state = [] @event.listens_for(eng, "savepoint") def do_begin(conn, name): state.append((conn, conn.connection)) raise Exception("failure") s1 = Session(eng) s1.begin_nested() assert_raises_message( Exception, "failure", s1.execute, text("select 1") ) conn, fairy = state[0] assert fairy.is_valid assert not conn.closed assert not conn.invalidated s1.close() assert conn.closed assert not fairy.is_valid @testing.requires.independent_connections def test_no_rollback_in_committed_state(self): """test #7388 Prior to the fix, using the session.begin() context manager would produce the error "This session is in 'committed' state; no further SQL can be emitted ", when it attempted to call .rollback() if the connection.close() operation failed inside of session.commit(). While the real exception was chained inside, this still proved to be misleading so we now skip the rollback() in this specific case and allow the original error to be raised. """ sess = fixture_session() def fail(*arg, **kw): raise BaseException("some base exception") with mock.patch.object( testing.db.dialect, "do_rollback", side_effect=fail ) as fail_mock, mock.patch.object( testing.db.dialect, "do_commit", side_effect=testing.db.dialect.do_commit, ) as succeed_mock: # sess.begin() -> commit(). why would do_rollback() be called? # because of connection pool finalize_fairy *after* the commit. # this will cause the conn.close() in session.commit() to fail, # but after the DB commit succeeded. with expect_raises_message(BaseException, "some base exception"): with sess.begin(): conn = sess.connection() fairy_conn = conn.connection eq_(succeed_mock.mock_calls, [mock.call(fairy_conn)]) eq_(fail_mock.mock_calls, [mock.call(fairy_conn)]) def test_continue_flushing_on_commit(self): """test that post-flush actions get flushed also if we're in commit()""" users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() to_flush = [User(name="ed"), User(name="jack"), User(name="wendy")] @event.listens_for(sess, "after_flush_postexec") def add_another_user(session, ctx): if to_flush: session.add(to_flush.pop(0)) x = [1] @event.listens_for(sess, "after_commit") # noqa def add_another_user(session): # noqa x[0] += 1 sess.add(to_flush.pop()) sess.commit() eq_(x, [2]) eq_(sess.scalar(select(func.count(users.c.id))), 3) def test_continue_flushing_guard(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() @event.listens_for(sess, "after_flush_postexec") def add_another_user(session, ctx): session.add(User(name="x")) sess.add(User(name="x")) assert_raises_message( orm_exc.FlushError, "Over 100 subsequent flushes have occurred", sess.commit, ) def test_no_sql_during_commit(self): sess = fixture_session() @event.listens_for(sess, "after_commit") def go(session): session.execute(text("select 1")) assert_raises_message( sa_exc.InvalidRequestError, "This session is in 'committed' state; no further " "SQL can be emitted within this transaction.", sess.commit, ) def test_no_sql_during_prepare(self): sess = fixture_session(twophase=True) sess.prepare() assert_raises_message( sa_exc.InvalidRequestError, "This session is in 'prepared' state; no further " "SQL can be emitted within this transaction.", sess.execute, text("select 1"), ) def test_no_sql_during_rollback(self): sess = fixture_session() sess.connection() @event.listens_for(sess, "after_rollback") def go(session): session.execute(text("select 1")) assert_raises_message( sa_exc.InvalidRequestError, "This session is in 'inactive' state, due to the SQL transaction " "being rolled back; no further SQL can be emitted within this " "transaction.", sess.rollback, ) @testing.requires.independent_connections @testing.emits_warning(".*previous exception") def test_failed_rollback_deactivates_transaction(self): # test #4050 users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) session = Session(bind=testing.db) rollback_error = testing.db.dialect.dbapi.InterfaceError( "Can't roll back to savepoint" ) def prevent_savepoint_rollback( cursor, statement, parameters, context=None ): if ( context is not None and context.compiled and isinstance( context.compiled.statement, elements.RollbackToSavepointClause, ) ): raise rollback_error self.event_listen( testing.db.dialect, "do_execute", prevent_savepoint_rollback ) with session.begin(): session.add(User(id=1, name="x")) session.begin_nested() # raises IntegrityError on flush session.add(User(id=1, name="x")) assert_raises_message( sa_exc.InterfaceError, "Can't roll back to savepoint", session.commit, ) # rollback succeeds, because the Session is deactivated eq_(session._transaction._state, _session.DEACTIVE) eq_(session.is_active, False) session.rollback() is_(session._transaction, None) session.connection() # back to normal eq_(session._transaction._state, _session.ACTIVE) eq_(session.is_active, True) trans = session._transaction # leave the outermost trans session.rollback() # trans is now closed eq_(trans._state, _session.CLOSED) # outermost transaction is new is_not(session._transaction, trans) is_(session._transaction, None) eq_(session.is_active, True) def test_no_prepare_wo_twophase(self): sess = fixture_session() assert_raises_message( sa_exc.InvalidRequestError, "'twophase' mode not enabled, or not root " "transaction; can't prepare.", sess.prepare, ) def test_closed_status_check(self): sess = fixture_session() trans = sess.begin() trans.rollback() assert_raises_message( sa_exc.ResourceClosedError, "This transaction is closed", trans.rollback, ) assert_raises_message( sa_exc.ResourceClosedError, "This transaction is closed", trans.commit, ) def _inactive_flushed_session_fixture(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() u1 = User(id=1, name="u1") sess.add(u1) sess.commit() sess.add(User(id=1, name="u2")) with expect_warnings("New instance"): assert_raises(sa_exc.IntegrityError, sess.flush) return sess, u1 def test_execution_options_begin_transaction(self): bind = mock.Mock( connect=mock.Mock( return_value=mock.Mock( _is_future=False, execution_options=mock.Mock( return_value=mock.Mock( _is_future=False, in_transaction=mock.Mock(return_value=False), ) ), ) ) ) sess = Session(bind=bind) c1 = sess.connection(execution_options={"isolation_level": "FOO"}) eq_(bind.mock_calls, [mock.call.connect()]) eq_( bind.connect().mock_calls, [mock.call.execution_options(isolation_level="FOO")], ) eq_(c1, bind.connect().execution_options()) def test_execution_options_ignored_mid_transaction(self): bind = mock.Mock() conn = mock.Mock( engine=bind, in_transaction=mock.Mock(return_value=False) ) bind.connect = mock.Mock(return_value=conn) sess = Session(bind=bind) sess.execute(text("select 1")) with expect_warnings( "Connection is already established for the " "given bind; execution_options ignored" ): sess.connection(execution_options={"isolation_level": "FOO"}) def test_warning_on_using_inactive_session_new(self): User = self.classes.User sess, u1 = self._inactive_flushed_session_fixture() u2 = User(name="u2") sess.add(u2) def go(): sess.rollback() assert_warnings( go, [ "Session's state has been changed on a " "non-active transaction - this state " "will be discarded." ], ) assert u2 not in sess assert u1 in sess def test_warning_on_using_inactive_session_dirty(self): sess, u1 = self._inactive_flushed_session_fixture() u1.name = "newname" def go(): sess.rollback() assert_warnings( go, [ "Session's state has been changed on a " "non-active transaction - this state " "will be discarded." ], ) assert u1 in sess assert u1 not in sess.dirty def test_warning_on_using_inactive_session_delete(self): sess, u1 = self._inactive_flushed_session_fixture() sess.delete(u1) def go(): sess.rollback() assert_warnings( go, [ "Session's state has been changed on a " "non-active transaction - this state " "will be discarded." ], ) assert u1 in sess assert u1 not in sess.deleted def test_warning_on_using_inactive_session_rollback_evt(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() u1 = User(id=1, name="u1") sess.add(u1) sess.commit() u3 = User(name="u3") @event.listens_for(sess, "after_rollback") def evt(s): sess.add(u3) sess.add(User(id=1, name="u2")) def go(): assert_raises(orm_exc.FlushError, sess.flush) assert u3 not in sess def test_preserve_flush_error(self): User = self.classes.User sess, u1 = self._inactive_flushed_session_fixture() for i in range(5): assert_raises_message( sa_exc.PendingRollbackError, "^This Session's transaction has been " r"rolled back due to a previous exception " "during flush. To " "begin a new transaction with this " "Session, first issue " r"Session.rollback\(\). Original exception " "was:", sess.commit, ) sess.rollback() sess.add(User(id=5, name="some name")) sess.commit() def test_no_autobegin_after_explicit_commit(self): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) session = fixture_session() session.add(User(name="ed")) session.get_transaction().commit() is_(session.get_transaction(), None) session.connection() is_not(session.get_transaction(), None) class _LocalFixture(FixtureTest): run_setup_mappers = "once" run_inserts = None @classmethod def setup_mappers(cls): User, Address = cls.classes.User, cls.classes.Address users, addresses = cls.tables.users, cls.tables.addresses cls.mapper_registry.map_imperatively( User, users, properties={ "addresses": relationship( Address, backref="user", cascade="all, delete-orphan", order_by=addresses.c.id, ) }, ) cls.mapper_registry.map_imperatively(Address, addresses) def subtransaction_recipe_one(self): @contextlib.contextmanager def transaction(session): if session.in_transaction(): outermost = False else: outermost = True session.begin() try: yield except: if session.in_transaction(): session.rollback() raise else: if outermost and session.in_transaction(): session.commit() return transaction def subtransaction_recipe_two(self): # shorter recipe @contextlib.contextmanager def transaction(session): if not session.in_transaction(): with session.begin(): yield else: yield return transaction def subtransaction_recipe_three(self): @contextlib.contextmanager def transaction(session): if not session.in_transaction(): session.begin() try: yield except: if session.in_transaction(): session.rollback() else: session.commit() else: try: yield except: if session.in_transaction(): session.rollback() raise return transaction @testing.combinations( (subtransaction_recipe_one, True), (subtransaction_recipe_two, False), (subtransaction_recipe_three, True), argnames="target_recipe,recipe_rollsback_early", id_="ns", ) class SubtransactionRecipeTest(FixtureTest): run_inserts = None __backend__ = True @testing.fixture def subtransaction_recipe(self): return self.target_recipe() @testing.requires.savepoints def test_recipe_heavy_nesting(self, subtransaction_recipe): users = self.tables.users with fixture_session() as session: with subtransaction_recipe(session): session.connection().execute( users.insert().values(name="user1") ) with subtransaction_recipe(session): savepoint = session.begin_nested() session.connection().execute( users.insert().values(name="user2") ) assert ( session.connection() .exec_driver_sql("select count(1) from users") .scalar() == 2 ) savepoint.rollback() with subtransaction_recipe(session): assert ( session.connection() .exec_driver_sql("select count(1) from users") .scalar() == 1 ) session.connection().execute( users.insert().values(name="user3") ) assert ( session.connection() .exec_driver_sql("select count(1) from users") .scalar() == 2 ) @engines.close_open_connections def test_recipe_subtransaction_on_external_subtrans( self, subtransaction_recipe ): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) with testing.db.connect() as conn: trans = conn.begin() sess = Session(conn) with subtransaction_recipe(sess): u = User(name="ed") sess.add(u) sess.flush() # commit does nothing trans.rollback() # rolls back assert len(sess.query(User).all()) == 0 sess.close() def test_recipe_commit_one(self, subtransaction_recipe): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) with fixture_session() as sess: with subtransaction_recipe(sess): u = User(name="u1") sess.add(u) sess.close() assert len(sess.query(User).all()) == 1 def test_recipe_subtransaction_on_noautocommit( self, subtransaction_recipe ): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) with fixture_session() as sess: sess.begin() with subtransaction_recipe(sess): u = User(name="u1") sess.add(u) sess.flush() sess.rollback() # rolls back assert len(sess.query(User).all()) == 0 sess.close() @testing.requires.savepoints def test_recipe_mixed_transaction_control(self, subtransaction_recipe): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) with fixture_session() as sess: sess.begin() sess.begin_nested() with subtransaction_recipe(sess): sess.add(User(name="u1")) sess.commit() sess.commit() eq_(len(sess.query(User).all()), 1) sess.close() t1 = sess.begin() t2 = sess.begin_nested() sess.add(User(name="u2")) t2.commit() assert sess.get_transaction() is t1 def test_recipe_error_on_using_inactive_session_commands( self, subtransaction_recipe ): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) with fixture_session() as sess: sess.begin() try: with subtransaction_recipe(sess): sess.add(User(name="u1")) sess.flush() raise Exception("force rollback") except: pass if self.recipe_rollsback_early: # that was a real rollback, so no transaction assert not sess.in_transaction() is_(sess.get_transaction(), None) else: assert sess.in_transaction() sess.close() assert not sess.in_transaction() def test_recipe_multi_nesting(self, subtransaction_recipe): with fixture_session() as sess: with subtransaction_recipe(sess): assert sess.in_transaction() try: with subtransaction_recipe(sess): assert sess.get_transaction() raise Exception("force rollback") except: pass if self.recipe_rollsback_early: assert not sess.in_transaction() else: assert sess.in_transaction() assert not sess.in_transaction() def test_recipe_deactive_status_check(self, subtransaction_recipe): with fixture_session() as sess: sess.begin() with subtransaction_recipe(sess): sess.rollback() assert not sess.in_transaction() sess.commit() # no error class FixtureDataTest(_LocalFixture): run_inserts = "each" __backend__ = True def test_attrs_on_rollback(self): User = self.classes.User sess = fixture_session() u1 = sess.get(User, 7) u1.name = "ed" sess.rollback() eq_(u1.name, "jack") def test_commit_persistent(self): User = self.classes.User sess = fixture_session() u1 = sess.get(User, 7) u1.name = "ed" sess.flush() sess.commit() eq_(u1.name, "ed") def test_concurrent_commit_persistent(self): User = self.classes.User s1 = fixture_session() u1 = s1.get(User, 7) u1.name = "ed" s1.commit() s2 = fixture_session() u2 = s2.get(User, 7) assert u2.name == "ed" u2.name = "will" s2.commit() assert u1.name == "will" class CleanSavepointTest(FixtureTest): """test the behavior for [ticket:2452] - rollback on begin_nested() only expires objects tracked as being modified in that transaction. """ run_inserts = None __backend__ = True def _run_test(self, update_fn): User, users = self.classes.User, self.tables.users self.mapper_registry.map_imperatively(User, users) with fixture_session() as s: u1 = User(name="u1") u2 = User(name="u2") s.add_all([u1, u2]) s.commit() u1.name u2.name trans = s._transaction assert trans is not None s.begin_nested() update_fn(s, u2) eq_(u2.name, "u2modified") s.rollback() assert s._transaction is None assert "name" not in u1.__dict__ assert "name" not in u2.__dict__ eq_(u2.name, "u2") @testing.requires.savepoints def test_rollback_ignores_clean_on_savepoint(self): def update_fn(s, u2): u2.name = "u2modified" self._run_test(update_fn) @testing.requires.savepoints def test_rollback_ignores_clean_on_savepoint_agg_upd_eval(self): User = self.classes.User def update_fn(s, u2): s.query(User).filter_by(name="u2").update( dict(name="u2modified"), synchronize_session="evaluate" ) self._run_test(update_fn) @testing.requires.savepoints def test_rollback_ignores_clean_on_savepoint_agg_upd_fetch(self): User = self.classes.User def update_fn(s, u2): s.query(User).filter_by(name="u2").update( dict(name="u2modified"), synchronize_session="fetch" ) self._run_test(update_fn) class AutoExpireTest(_LocalFixture): __backend__ = True def test_expunge_pending_on_rollback(self): User = self.classes.User sess = fixture_session() u2 = User(name="newuser") sess.add(u2) assert u2 in sess sess.rollback() assert u2 not in sess def test_trans_pending_cleared_on_commit(self): User = self.classes.User sess = fixture_session() u2 = User(name="newuser") sess.add(u2) assert u2 in sess sess.commit() assert u2 in sess u3 = User(name="anotheruser") sess.add(u3) sess.rollback() assert u3 not in sess assert u2 in sess def test_update_deleted_on_rollback(self): User = self.classes.User s = fixture_session() u1 = User(name="ed") s.add(u1) s.commit() # this actually tests that the delete() operation, # when cascaded to the "addresses" collection, does not # trigger a flush (via lazyload) before the cascade is complete. s.delete(u1) assert u1 in s.deleted s.rollback() assert u1 in s assert u1 not in s.deleted def test_trans_deleted_cleared_on_rollback(self): User = self.classes.User s = fixture_session() u1 = User(name="ed") s.add(u1) s.commit() s.delete(u1) s.commit() assert u1 not in s s.rollback() assert u1 not in s def test_update_deleted_on_rollback_cascade(self): User, Address = self.classes.User, self.classes.Address s = fixture_session() u1 = User(name="ed", addresses=[Address(email_address="foo")]) s.add(u1) s.commit() s.delete(u1) assert u1 in s.deleted assert u1.addresses[0] in s.deleted s.rollback() assert u1 in s assert u1 not in s.deleted assert u1.addresses[0] not in s.deleted def test_update_deleted_on_rollback_orphan(self): User, Address = self.classes.User, self.classes.Address s = fixture_session() u1 = User(name="ed", addresses=[Address(email_address="foo")]) s.add(u1) s.commit() a1 = u1.addresses[0] u1.addresses.remove(a1) s.flush() eq_(s.query(Address).filter(Address.email_address == "foo").all(), []) s.rollback() assert a1 not in s.deleted assert u1.addresses == [a1] def test_commit_pending(self): User = self.classes.User sess = fixture_session() u1 = User(name="newuser") sess.add(u1) sess.flush() sess.commit() eq_(u1.name, "newuser") def test_concurrent_commit_pending(self): User = self.classes.User s1 = fixture_session() u1 = User(name="edward") s1.add(u1) s1.commit() s2 = fixture_session() u2 = s2.query(User).filter(User.name == "edward").one() u2.name = "will" s2.commit() assert u1.name == "will" class TwoPhaseTest(_LocalFixture): __backend__ = True @testing.requires.two_phase_transactions def test_rollback_on_prepare(self): User = self.classes.User s = fixture_session(twophase=True) u = User(name="ed") s.add(u) s.prepare() s.rollback() assert u not in s class RollbackRecoverTest(_LocalFixture): __backend__ = True def test_pk_violation(self): User, Address = self.classes.User, self.classes.Address s = fixture_session() a1 = Address(email_address="foo") u1 = User(id=1, name="ed", addresses=[a1]) s.add(u1) s.commit() a2 = Address(email_address="bar") u2 = User(id=1, name="jack", addresses=[a2]) u1.name = "edward" a1.email_address = "foober" s.add(u2) with expect_warnings("New instance"): assert_raises(sa_exc.IntegrityError, s.commit) assert_raises(sa_exc.InvalidRequestError, s.commit) s.rollback() assert u2 not in s assert a2 not in s assert u1 in s assert a1 in s assert u1.name == "ed" assert a1.email_address == "foo" u1.name = "edward" a1.email_address = "foober" s.commit() eq_( s.query(User).all(), [ User( id=1, name="edward", addresses=[Address(email_address="foober")], ) ], ) @testing.requires.savepoints def test_pk_violation_with_savepoint(self): User, Address = self.classes.User, self.classes.Address s = fixture_session() a1 = Address(email_address="foo") u1 = User(id=1, name="ed", addresses=[a1]) s.add(u1) s.commit() a2 = Address(email_address="bar") u2 = User(id=1, name="jack", addresses=[a2]) u1.name = "edward" a1.email_address = "foober" nt1 = s.begin_nested() s.add(u2) with expect_warnings("New instance"): assert_raises(sa_exc.IntegrityError, s.commit) assert_raises(sa_exc.InvalidRequestError, s.commit) nt1.rollback() assert u2 not in s assert a2 not in s assert u1 in s assert a1 in s s.commit() eq_( s.query(User).all(), [ User( id=1, name="edward", addresses=[Address(email_address="foober")], ) ], ) class SavepointTest(_LocalFixture): __backend__ = True @testing.requires.savepoints def test_savepoint_rollback(self): User = self.classes.User s = fixture_session() u1 = User(name="ed") u2 = User(name="jack") s.add_all([u1, u2]) nt1 = s.begin_nested() u3 = User(name="wendy") u4 = User(name="foo") u1.name = "edward" u2.name = "jackward" s.add_all([u3, u4]) eq_( s.query(User.name).order_by(User.id).all(), [("edward",), ("jackward",), ("wendy",), ("foo",)], ) nt1.rollback() assert u1.name == "ed" assert u2.name == "jack" eq_(s.query(User.name).order_by(User.id).all(), [("ed",), ("jack",)]) s.commit() assert u1.name == "ed" assert u2.name == "jack" eq_(s.query(User.name).order_by(User.id).all(), [("ed",), ("jack",)]) @testing.requires.savepoints def test_savepoint_delete(self): User = self.classes.User s = fixture_session() u1 = User(name="ed") s.add(u1) s.commit() eq_(s.query(User).filter_by(name="ed").count(), 1) s.begin_nested() s.delete(u1) s.commit() eq_(s.query(User).filter_by(name="ed").count(), 0) s.commit() @testing.requires.savepoints def test_savepoint_commit(self): User = self.classes.User s = fixture_session() u1 = User(name="ed") u2 = User(name="jack") s.add_all([u1, u2]) nt1 = s.begin_nested() u3 = User(name="wendy") u4 = User(name="foo") u1.name = "edward" u2.name = "jackward" s.add_all([u3, u4]) eq_( s.query(User.name).order_by(User.id).all(), [("edward",), ("jackward",), ("wendy",), ("foo",)], ) nt1.commit() def go(): assert u1.name == "edward" assert u2.name == "jackward" eq_( s.query(User.name).order_by(User.id).all(), [("edward",), ("jackward",), ("wendy",), ("foo",)], ) self.assert_sql_count(testing.db, go, 1) s.commit() eq_( s.query(User.name).order_by(User.id).all(), [("edward",), ("jackward",), ("wendy",), ("foo",)], ) @testing.requires.savepoints def test_savepoint_rollback_collections(self): User, Address = self.classes.User, self.classes.Address s = fixture_session() u1 = User(name="ed", addresses=[Address(email_address="foo")]) s.add(u1) s.commit() u1.name = "edward" u1.addresses.append(Address(email_address="bar")) nt1 = s.begin_nested() u2 = User(name="jack", addresses=[Address(email_address="bat")]) s.add(u2) eq_( s.query(User).order_by(User.id).all(), [ User( name="edward", addresses=[ Address(email_address="foo"), Address(email_address="bar"), ], ), User(name="jack", addresses=[Address(email_address="bat")]), ], ) nt1.rollback() eq_( s.query(User).order_by(User.id).all(), [ User( name="edward", addresses=[ Address(email_address="foo"), Address(email_address="bar"), ], ) ], ) s.commit() eq_( s.query(User).order_by(User.id).all(), [ User( name="edward", addresses=[ Address(email_address="foo"), Address(email_address="bar"), ], ) ], ) @testing.requires.savepoints def test_savepoint_commit_collections(self): User, Address = self.classes.User, self.classes.Address s = fixture_session() u1 = User(name="ed", addresses=[Address(email_address="foo")]) s.add(u1) s.commit() u1.name = "edward" u1.addresses.append(Address(email_address="bar")) s.begin_nested() u2 = User(name="jack", addresses=[Address(email_address="bat")]) s.add(u2) eq_( s.query(User).order_by(User.id).all(), [ User( name="edward", addresses=[ Address(email_address="foo"), Address(email_address="bar"), ], ), User(name="jack", addresses=[Address(email_address="bat")]), ], ) s.commit() eq_( s.query(User).order_by(User.id).all(), [ User( name="edward", addresses=[ Address(email_address="foo"), Address(email_address="bar"), ], ), User(name="jack", addresses=[Address(email_address="bat")]), ], ) s.commit() eq_( s.query(User).order_by(User.id).all(), [ User( name="edward", addresses=[ Address(email_address="foo"), Address(email_address="bar"), ], ), User(name="jack", addresses=[Address(email_address="bat")]), ], ) @testing.requires.savepoints def test_expunge_pending_on_rollback(self): User = self.classes.User sess = fixture_session() sess.begin_nested() u2 = User(name="newuser") sess.add(u2) assert u2 in sess sess.rollback() assert u2 not in sess @testing.requires.savepoints def test_update_deleted_on_rollback(self): User = self.classes.User s = fixture_session() u1 = User(name="ed") s.add(u1) s.commit() s.begin_nested() s.delete(u1) assert u1 in s.deleted s.rollback() assert u1 in s assert u1 not in s.deleted @testing.requires.savepoints_w_release def test_savepoint_lost_still_runs(self): User = self.classes.User s = fixture_session() trans = s.begin_nested() s.connection() u1 = User(name="ed") s.add(u1) # kill off the transaction nested_trans = trans._connections[self.bind][1] nested_trans._do_commit() is_(s.get_nested_transaction(), trans) with expect_warnings("nested transaction already deassociated"): # this previously would raise # "savepoint "sa_savepoint_1" does not exist", however as of # #5327 the savepoint already knows it's inactive s.rollback() assert u1 not in s.new is_(trans._state, _session.CLOSED) is_not(s.get_transaction(), trans) s.connection() is_(s.get_transaction()._state, _session.ACTIVE) is_(s.get_transaction().nested, False) is_(s.get_transaction()._parent, None) class AccountingFlagsTest(_LocalFixture): __backend__ = True def test_no_expire_on_commit(self): User, users = self.classes.User, self.tables.users sess = fixture_session(expire_on_commit=False) u1 = User(name="ed") sess.add(u1) sess.commit() sess.execute( users.update().where(users.c.name == "ed").values(name="edward") ) assert u1.name == "ed" sess.expire_all() assert u1.name == "edward" class ContextManagerPlusFutureTest(FixtureTest): run_inserts = None __backend__ = True @testing.requires.savepoints @engines.close_open_connections def test_contextmanager_nested_rollback(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() def go(): with sess.begin_nested(): sess.add(User()) # name can't be null sess.flush() # and not InvalidRequestError assert_raises(sa_exc.DBAPIError, go) with sess.begin_nested(): sess.add(User(name="u1")) eq_(sess.query(User).count(), 1) def test_contextmanager_commit(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() with sess.begin(): sess.add(User(name="u1")) sess.rollback() eq_(sess.query(User).count(), 1) def test_contextmanager_rollback(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) sess = fixture_session() def go(): with sess.begin(): sess.add(User()) # name can't be null assert_raises(sa_exc.DBAPIError, go) eq_(sess.query(User).count(), 0) sess.close() with sess.begin(): sess.add(User(name="u1")) eq_(sess.query(User).count(), 1) def test_explicit_begin(self): with fixture_session() as s1: with s1.begin() as trans: is_(trans, s1.get_transaction()) s1.connection() is_(s1._transaction, None) def test_no_double_begin_explicit(self): with fixture_session() as s1: s1.begin() assert_raises_message( sa_exc.InvalidRequestError, "A transaction is already begun on this Session.", s1.begin, ) @testing.requires.savepoints def test_rollback_is_global(self): users = self.tables.users with fixture_session() as s1: s1.begin() s1.connection().execute(users.insert(), [{"id": 1, "name": "n1"}]) s1.begin_nested() s1.connection().execute( users.insert(), [{"id": 2, "name": "n2"}, {"id": 3, "name": "n3"}], ) eq_( s1.connection().scalar( select(func.count()).select_from(users) ), 3, ) # rolls back the whole transaction s1.rollback() is_(s1.get_transaction(), None) eq_( s1.connection().scalar( select(func.count()).select_from(users) ), 0, ) s1.commit() is_(s1.get_transaction(), None) def test_session_as_ctx_manager_one(self): users = self.tables.users with fixture_session() as sess: is_(sess.get_transaction(), None) sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")] ) is_not(sess.get_transaction(), None) is_(sess.get_transaction(), None) # did not commit eq_(sess.connection().execute(users.select()).all(), []) def test_session_as_ctx_manager_two(self): users = self.tables.users try: with fixture_session() as sess: is_(sess.get_transaction(), None) sess.connection().execute( users.insert().values(id=1, name="user1") ) raise Exception("force rollback") except: pass is_(sess.get_transaction(), None) def test_begin_context_manager(self): users = self.tables.users with fixture_session() as sess: with sess.begin(): sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")], ) # committed eq_(sess.connection().execute(users.select()).all(), [(1, "user1")]) def test_sessionmaker_begin_context_manager(self): users = self.tables.users session = sessionmaker(testing.db) with session.begin() as sess: sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")], ) # committed eq_(sess.connection().execute(users.select()).all(), [(1, "user1")]) sess.close() def test_begin_context_manager_rollback_trans(self): users = self.tables.users try: with fixture_session() as sess: with sess.begin(): sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")], ) raise Exception("force rollback") except: pass # rolled back eq_(sess.connection().execute(users.select()).all(), []) sess.close() def test_begin_context_manager_rollback_outer(self): users = self.tables.users try: with fixture_session() as sess: with sess.begin(): sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")], ) raise Exception("force rollback") except: pass # committed eq_(sess.connection().execute(users.select()).all(), [(1, "user1")]) sess.close() def test_sessionmaker_begin_context_manager_rollback_trans(self): users = self.tables.users session = sessionmaker(testing.db) try: with session.begin() as sess: sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")], ) raise Exception("force rollback") except: pass # rolled back eq_(sess.connection().execute(users.select()).all(), []) sess.close() def test_sessionmaker_begin_context_manager_rollback_outer(self): users = self.tables.users session = sessionmaker(testing.db) try: with session.begin() as sess: sess.connection().execute( users.insert().values(id=1, name="user1") ) eq_( sess.connection().execute(users.select()).all(), [(1, "user1")], ) raise Exception("force rollback") except: pass # committed eq_(sess.connection().execute(users.select()).all(), [(1, "user1")]) sess.close() def test_interrupt_ctxmanager(self, trans_ctx_manager_fixture): fn = trans_ctx_manager_fixture session = fixture_session() fn(session, trans_on_subject=True, execute_on_subject=True) @testing.combinations((True,), (False,), argnames="rollback") @testing.combinations((True,), (False,), argnames="expire_on_commit") @testing.combinations( ("add",), ("modify",), ("delete",), ("begin",), argnames="check_operation", ) def test_interrupt_ctxmanager_ops( self, rollback, expire_on_commit, check_operation ): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) session = fixture_session(expire_on_commit=expire_on_commit) with session.begin(): u1 = User(id=7, name="u1") session.add(u1) with session.begin(): u1.name # unexpire u2 = User(id=8, name="u1") session.add(u2) session.flush() if rollback: session.rollback() else: session.commit() with expect_raises_message( sa_exc.InvalidRequestError, "Can't operate on closed transaction " "inside context manager. Please complete the context " "manager before emitting further commands.", ): if check_operation == "add": u3 = User(id=9, name="u2") session.add(u3) elif check_operation == "begin": session.begin() elif check_operation == "modify": u1.name = "newname" elif check_operation == "delete": session.delete(u1) class TransactionFlagsTest(fixtures.TestBase): def test_in_transaction(self): with fixture_session() as s1: eq_(s1.in_transaction(), False) trans = s1.begin() eq_(s1.in_transaction(), True) is_(s1.get_transaction(), trans) n1 = s1.begin_nested() eq_(s1.in_transaction(), True) is_(s1.get_transaction(), trans) is_(s1.get_nested_transaction(), n1) n1.rollback() is_(s1.get_nested_transaction(), None) is_(s1.get_transaction(), trans) eq_(s1.in_transaction(), True) s1.commit() eq_(s1.in_transaction(), False) is_(s1.get_transaction(), None) def test_in_transaction_subtransactions(self): """we'd like to do away with subtransactions for future sessions entirely. at the moment we are still using them internally. it might be difficult to keep the internals working in exactly the same way if remove this concept, so for now just test that the external API works. """ with fixture_session() as s1: eq_(s1.in_transaction(), False) trans = s1.begin() eq_(s1.in_transaction(), True) is_(s1.get_transaction(), trans) subtrans = s1._autobegin_t()._begin() is_(s1.get_transaction(), trans) eq_(s1.in_transaction(), True) is_(s1._transaction, subtrans) s1.rollback() eq_(s1.in_transaction(), False) is_(s1._transaction, None) s1.rollback() eq_(s1.in_transaction(), False) is_(s1._transaction, None) def test_in_transaction_nesting(self): with fixture_session() as s1: eq_(s1.in_transaction(), False) trans = s1.begin() eq_(s1.in_transaction(), True) is_(s1.get_transaction(), trans) sp1 = s1.begin_nested() eq_(s1.in_transaction(), True) is_(s1.get_transaction(), trans) is_(s1.get_nested_transaction(), sp1) sp2 = s1.begin_nested() eq_(s1.in_transaction(), True) eq_(s1.in_nested_transaction(), True) is_(s1.get_transaction(), trans) is_(s1.get_nested_transaction(), sp2) sp2.rollback() eq_(s1.in_transaction(), True) eq_(s1.in_nested_transaction(), True) is_(s1.get_transaction(), trans) is_(s1.get_nested_transaction(), sp1) sp1.rollback() is_(s1.get_nested_transaction(), None) eq_(s1.in_transaction(), True) eq_(s1.in_nested_transaction(), False) is_(s1.get_transaction(), trans) s1.rollback() eq_(s1.in_transaction(), False) is_(s1.get_transaction(), None) class NaturalPKRollbackTest(fixtures.MappedTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table("users", metadata, Column("name", String(50), primary_key=True)) @classmethod def setup_classes(cls): class User(cls.Comparable): pass def test_rollback_recover(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) session = fixture_session() u1, u2, u3 = User(name="u1"), User(name="u2"), User(name="u3") session.add_all([u1, u2, u3]) session.commit() session.delete(u2) u4 = User(name="u2") session.add(u4) session.flush() u5 = User(name="u3") session.add(u5) with expect_warnings("New instance"): assert_raises(sa_exc.IntegrityError, session.flush) assert u5 not in session assert u2 not in session.deleted session.rollback() def test_reloaded_deleted_checked_for_expiry(self): """test issue #3677""" users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) u1 = User(name="u1") s = fixture_session() s.add(u1) s.flush() del u1 gc_collect() u1 = s.query(User).first() # noqa s.rollback() u2 = User(name="u1") s.add(u2) s.commit() assert inspect(u2).persistent def test_key_replaced_by_update(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) u1 = User(name="u1") u2 = User(name="u2") s = fixture_session() s.add_all([u1, u2]) s.commit() s.delete(u1) s.flush() u2.name = "u1" s.flush() assert u1 not in s s.rollback() assert u1 in s assert u2 in s assert s.identity_map[identity_key(User, ("u1",))] is u1 assert s.identity_map[identity_key(User, ("u2",))] is u2 @testing.requires.savepoints def test_key_replaced_by_update_nested(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) u1 = User(name="u1") s = fixture_session() s.add(u1) s.commit() with s.begin_nested(): u2 = User(name="u2") s.add(u2) s.flush() u2.name = "u3" s.rollback() assert u1 in s assert u2 not in s u1.name = "u5" s.commit() def test_multiple_key_replaced_by_update(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) u1 = User(name="u1") u2 = User(name="u2") u3 = User(name="u3") s = fixture_session() s.add_all([u1, u2, u3]) s.commit() s.delete(u1) s.delete(u2) s.flush() u3.name = "u1" s.flush() u3.name = "u2" s.flush() s.rollback() assert u1 in s assert u2 in s assert u3 in s assert s.identity_map[identity_key(User, ("u1",))] is u1 assert s.identity_map[identity_key(User, ("u2",))] is u2 assert s.identity_map[identity_key(User, ("u3",))] is u3 def test_key_replaced_by_oob_insert(self): users, User = self.tables.users, self.classes.User self.mapper_registry.map_imperatively(User, users) u1 = User(name="u1") s = fixture_session() s.add(u1) s.commit() s.delete(u1) s.flush() s.execute(users.insert().values(name="u1")) u2 = s.get(User, "u1") assert u1 not in s s.rollback() assert u1 in s assert u2 not in s assert s.identity_map[identity_key(User, ("u1",))] is u1 class JoinIntoAnExternalTransactionFixture: """Test the "join into an external transaction" examples""" def setup_test(self): self.engine = engines.testing_engine( options={"use_reaper": False, "sqlite_savepoint": True} ) self.connection = self.engine.connect() self.metadata = MetaData() self.table = Table( "t1", self.metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), ) with self.connection.begin(): self.table.create(self.connection, checkfirst=True) self.setup_session() def teardown_test(self): self.teardown_session() with self.connection.begin(): self._assert_count(0) with self.connection.begin(): self.table.drop(self.connection) self.connection.close() def test_something(self, connection): A = self.A a1 = A() self.session.add(a1) self.session.commit() self._assert_count(1) def _assert_count(self, count): result = self.connection.scalar( select(func.count()).select_from(self.table) ) eq_(result, count) class CtxManagerJoinIntoAnExternalTransactionFixture( JoinIntoAnExternalTransactionFixture ): @testing.requires.compat_savepoints def test_something_with_context_managers(self): A = self.A a1 = A() with self.session.begin(): self.session.add(a1) self.session.flush() self._assert_count(1) self.session.rollback() self._assert_count(0) a1 = A() with self.session.begin(): self.session.add(a1) self._assert_count(1) a2 = A() with self.session.begin(): self.session.add(a2) self.session.flush() self._assert_count(2) self.session.rollback() self._assert_count(1) @testing.requires.compat_savepoints def test_super_abusive_nesting(self): session = self.session for i in range(random.randint(5, 30)): choice = random.randint(1, 3) if choice == 1: if session.in_transaction(): session.begin_nested() else: session.begin() elif choice == 2: session.rollback() elif choice == 3: session.commit() session.connection() # remaining nested / etc. are cleanly cleared out session.close() class NewStyleJoinIntoAnExternalTransactionTest( CtxManagerJoinIntoAnExternalTransactionFixture, fixtures.MappedTest ): """test the 1.4 join to an external transaction fixture. In 1.4, this works for both legacy and future engines/sessions """ def setup_session(self): # begin a non-ORM transaction self.trans = self.connection.begin() class A: pass clear_mappers() self.mapper_registry.map_imperatively(A, self.table) self.A = A # bind an individual Session to the connection self.session = Session(bind=self.connection) if testing.requires.compat_savepoints.enabled: self.nested = self.connection.begin_nested() @event.listens_for(self.session, "after_transaction_end") def end_savepoint(session, transaction): if not self.nested.is_active: self.nested = self.connection.begin_nested() def teardown_session(self): self.session.close() # rollback - everything that happened with the # Session above (including calls to commit()) # is rolled back. if self.trans.is_active: self.trans.rollback() @testing.combinations( *Variation.generate_cases( "join_mode", [ "create_savepoint", "conditional_w_savepoint", "create_savepoint_w_savepoint", ], ), argnames="join_mode", id_="s", ) class ReallyNewJoinIntoAnExternalTransactionTest( CtxManagerJoinIntoAnExternalTransactionFixture, fixtures.MappedTest ): """2.0 only recipe for "join into an external transaction" that works without event handlers """ def setup_session(self): self.trans = self.connection.begin() if ( self.join_mode.conditional_w_savepoint or self.join_mode.create_savepoint_w_savepoint ): self.nested = self.connection.begin_nested() class A: pass clear_mappers() self.mapper_registry.map_imperatively(A, self.table) self.A = A self.session = Session( self.connection, join_transaction_mode=( "create_savepoint" if ( self.join_mode.create_savepoint or self.join_mode.create_savepoint_w_savepoint ) else "conditional_savepoint" ), ) def teardown_session(self): self.session.close() if ( self.join_mode.conditional_w_savepoint or self.join_mode.create_savepoint_w_savepoint ): assert not self.nested._deactivated_from_connection assert self.nested.is_active self.nested.rollback() assert not self.trans._deactivated_from_connection assert self.trans.is_active self.trans.rollback() class LegacyJoinIntoAnExternalTransactionTest( JoinIntoAnExternalTransactionFixture, fixtures.MappedTest, ): """test the 1.3 join to an external transaction fixture""" def setup_session(self): # begin a non-ORM transaction self.trans = self.connection.begin() class A: pass # TODO: py2 is not hitting this correctly for some reason, # some mro issue self.mapper_registry.map_imperatively(A, self.table) self.A = A # bind an individual Session to the connection self.session = Session(bind=self.connection) if testing.requires.compat_savepoints.enabled: # start the session in a SAVEPOINT... self.session.begin_nested() # then each time that SAVEPOINT ends, reopen it @event.listens_for(self.session, "after_transaction_end") def restart_savepoint(session, transaction): if transaction.nested and not transaction._parent.nested: # ensure that state is expired the way # session.commit() at the top level normally does # (optional step) session.expire_all() session.begin_nested() def teardown_session(self): self.session.close() # rollback - everything that happened with the # Session above (including calls to commit()) # is rolled back. self.trans.rollback()