<s> from __future__ import with_statement <EOL> from sqlalchemy import ( <EOL> testing , exc as sa_exc , event , String , Column , Table , select , func ) <EOL> from sqlalchemy . testing import ( <EOL> fixtures , engines , eq_ , assert_raises , assert_raises_message , <EOL> assert_warnings , mock , expect_warnings , is_ , is_not_ ) <EOL> from sqlalchemy . orm import ( <EOL> exc as orm_exc , Session , mapper , sessionmaker , create_session , <EOL> relationship , attributes , session as _session ) <EOL> from sqlalchemy . testing . util import gc_collect <EOL> from test . orm . _fixtures import FixtureTest <EOL> from sqlalchemy import inspect <EOL> class SessionTransactionTest ( FixtureTest ) : <EOL> run_inserts = None <EOL> __backend__ = True <EOL> def test_no_close_transaction_on_flush ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> c = testing . db . connect ( ) <EOL> try : <EOL> mapper ( User , users ) <EOL> s = create_session ( bind = c ) <EOL> s . begin ( ) <EOL> tran = s . transaction <EOL> s . add ( User ( name = '<STR_LIT>' ) ) <EOL> s . flush ( ) <EOL> c . execute ( "<STR_LIT>" ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> s . add ( u ) <EOL> s . flush ( ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> s . add ( u ) <EOL> s . flush ( ) <EOL> assert s . transaction is tran <EOL> tran . close ( ) <EOL> finally : <EOL> c . close ( ) <EOL> @ engines . close_open_connections <EOL> def test_subtransaction_on_external ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> conn = testing . db . connect ( ) <EOL> trans = conn . begin ( ) <EOL> sess = create_session ( bind = conn , autocommit = False , autoflush = True ) <EOL> sess . begin ( subtransactions = True ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> sess . commit ( ) <EOL> trans . rollback ( ) <EOL> assert len ( sess . query ( User ) . all ( ) ) == <NUM_LIT:0> <EOL> sess . close ( ) <EOL> @ testing . requires . savepoints <EOL> @ engines . close_open_connections <EOL> def test_external_nested_transaction ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> try : <EOL> conn = testing . db . connect ( ) <EOL> trans = conn . begin ( ) <EOL> sess = create_session ( bind = conn , autocommit = False , <EOL> autoflush = True ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . begin_nested ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> sess . flush ( ) <EOL> sess . rollback ( ) <EOL> trans . commit ( ) <EOL> assert len ( sess . query ( User ) . all ( ) ) == <NUM_LIT:1> <EOL> except : <EOL> conn . close ( ) <EOL> raise <EOL> @ testing . requires . savepoints <EOL> def test_nested_accounting_new_items_removed ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> session = create_session ( bind = testing . db ) <EOL> session . begin ( ) <EOL> session . begin_nested ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> session . add ( u1 ) <EOL> session . commit ( ) <EOL> assert u1 in session <EOL> session . rollback ( ) <EOL> assert u1 not in session <EOL> @ testing . requires . savepoints <EOL> def test_nested_accounting_deleted_items_restored ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> session = create_session ( bind = testing . db ) <EOL> session . begin ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> session . add ( u1 ) <EOL> session . commit ( ) <EOL> session . begin ( ) <EOL> u1 = session . query ( User ) . first ( ) <EOL> session . begin_nested ( ) <EOL> session . delete ( u1 ) <EOL> session . commit ( ) <EOL> assert u1 not in session <EOL> session . rollback ( ) <EOL> assert u1 in session <EOL> @ testing . requires . savepoints <EOL> def test_heavy_nesting ( self ) : <EOL> users = self . tables . users <EOL> session = create_session ( bind = testing . db ) <EOL> session . begin ( ) <EOL> session . connection ( ) . execute ( users . insert ( ) . values ( <EOL> name = '<STR_LIT>' ) ) <EOL> session . begin ( subtransactions = True ) <EOL> session . begin_nested ( ) <EOL> session . connection ( ) . execute ( users . insert ( ) . values ( <EOL> name = '<STR_LIT>' ) ) <EOL> assert session . connection ( ) . execute ( <EOL> '<STR_LIT>' ) . scalar ( ) == <NUM_LIT:2> <EOL> session . rollback ( ) <EOL> assert session . connection ( ) . execute ( <EOL> '<STR_LIT>' ) . scalar ( ) == <NUM_LIT:1> <EOL> session . connection ( ) . execute ( users . insert ( ) . values ( <EOL> name = '<STR_LIT>' ) ) <EOL> session . commit ( ) <EOL> assert session . connection ( ) . execute ( <EOL> '<STR_LIT>' ) . scalar ( ) == <NUM_LIT:2> <EOL> @ testing . requires . savepoints <EOL> def test_dirty_state_transferred_deep_nesting ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> s = Session ( testing . db ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> nt1 = s . begin_nested ( ) <EOL> nt2 = s . begin_nested ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> assert attributes . instance_state ( u1 ) not in nt2 . _dirty <EOL> assert attributes . instance_state ( u1 ) not in nt1 . _dirty <EOL> s . flush ( ) <EOL> assert attributes . instance_state ( u1 ) in nt2 . _dirty <EOL> assert attributes . instance_state ( u1 ) not in nt1 . _dirty <EOL> s . commit ( ) <EOL> assert attributes . instance_state ( u1 ) in nt2 . _dirty <EOL> assert attributes . instance_state ( u1 ) in nt1 . _dirty <EOL> s . rollback ( ) <EOL> assert attributes . instance_state ( u1 ) . expired <EOL> eq_ ( u1 . name , '<STR_LIT>' ) <EOL> @ testing . requires . independent_connections <EOL> def test_transactions_isolated ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> s1 = create_session ( bind = testing . db , autocommit = False ) <EOL> s2 = create_session ( bind = testing . db , autocommit = False ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s1 . add ( u1 ) <EOL> s1 . flush ( ) <EOL> assert s2 . query ( User ) . all ( ) == [ ] <EOL> @ testing . requires . two_phase_transactions <EOL> def test_twophase ( self ) : <EOL> users , Address , addresses , User = ( self . tables . users , <EOL> self . classes . Address , <EOL> self . tables . addresses , <EOL> self . classes . User ) <EOL> mapper ( User , users ) <EOL> mapper ( Address , addresses ) <EOL> engine2 = engines . testing_engine ( ) <EOL> sess = create_session ( autocommit = True , autoflush = False , <EOL> twophase = True ) <EOL> sess . bind_mapper ( User , testing . db ) <EOL> sess . bind_mapper ( Address , engine2 ) <EOL> sess . begin ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> sess . add_all ( ( u1 , a1 ) ) <EOL> sess . commit ( ) <EOL> sess . close ( ) <EOL> engine2 . dispose ( ) <EOL> assert users . count ( ) . scalar ( ) == <NUM_LIT:1> <EOL> assert addresses . count ( ) . scalar ( ) == <NUM_LIT:1> <EOL> @ testing . requires . independent_connections <EOL> def test_invalidate ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> sess = Session ( ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> c1 = sess . connection ( User ) <EOL> sess . invalidate ( ) <EOL> assert c1 . invalidated <EOL> eq_ ( sess . query ( User ) . all ( ) , [ ] ) <EOL> c2 = sess . connection ( User ) <EOL> assert not c2 . invalidated <EOL> def test_subtransaction_on_noautocommit ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> sess = create_session ( autocommit = False , autoflush = True ) <EOL> sess . begin ( subtransactions = True ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> sess . commit ( ) <EOL> sess . rollback ( ) <EOL> assert len ( sess . query ( User ) . all ( ) ) == <NUM_LIT:0> <EOL> sess . close ( ) <EOL> @ testing . requires . savepoints <EOL> def test_nested_transaction ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> sess = create_session ( ) <EOL> sess . begin ( ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> sess . begin_nested ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> sess . flush ( ) <EOL> sess . rollback ( ) <EOL> sess . commit ( ) <EOL> assert len ( sess . query ( User ) . all ( ) ) == <NUM_LIT:1> <EOL> sess . close ( ) <EOL> @ testing . requires . savepoints <EOL> def test_nested_autotrans ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> sess = create_session ( autocommit = False ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> sess . begin_nested ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> sess . flush ( ) <EOL> sess . rollback ( ) <EOL> sess . commit ( ) <EOL> assert len ( sess . query ( User ) . all ( ) ) == <NUM_LIT:1> <EOL> sess . close ( ) <EOL> @ testing . requires . savepoints <EOL> def test_nested_transaction_connection_add ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = create_session ( autocommit = True ) <EOL> sess . begin ( ) <EOL> sess . begin_nested ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . rollback ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> sess . commit ( ) <EOL> eq_ ( set ( sess . query ( User ) . all ( ) ) , set ( [ u2 ] ) ) <EOL> sess . begin ( ) <EOL> sess . begin_nested ( ) <EOL> u3 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u3 ) <EOL> sess . commit ( ) <EOL> sess . rollback ( ) <EOL> eq_ ( set ( sess . query ( User ) . all ( ) ) , set ( [ u2 ] ) ) <EOL> sess . close ( ) <EOL> @ testing . requires . savepoints <EOL> def test_mixed_transaction_control ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = create_session ( autocommit = True ) <EOL> sess . begin ( ) <EOL> sess . begin_nested ( ) <EOL> transaction = sess . begin ( subtransactions = True ) <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> transaction . commit ( ) <EOL> sess . commit ( ) <EOL> sess . commit ( ) <EOL> sess . close ( ) <EOL> eq_ ( len ( sess . query ( User ) . all ( ) ) , <NUM_LIT:1> ) <EOL> t1 = sess . begin ( ) <EOL> t2 = sess . begin_nested ( ) <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> t2 . commit ( ) <EOL> assert sess . transaction is t1 <EOL> sess . close ( ) <EOL> @ testing . requires . savepoints <EOL> def test_mixed_transaction_close ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = create_session ( autocommit = False ) <EOL> sess . begin_nested ( ) <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> sess . flush ( ) <EOL> sess . close ( ) <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> sess . commit ( ) <EOL> sess . close ( ) <EOL> eq_ ( len ( sess . query ( User ) . all ( ) ) , <NUM_LIT:1> ) <EOL> def test_continue_flushing_on_commit ( self ) : <EOL> """<STR_LIT>""" <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( ) <EOL> to_flush = [ User ( name = '<STR_LIT>' ) , User ( name = '<STR_LIT>' ) , User ( name = '<STR_LIT>' ) ] <EOL> @ event . listens_for ( sess , "<STR_LIT>" ) <EOL> def add_another_user ( session , ctx ) : <EOL> if to_flush : <EOL> session . add ( to_flush . pop ( <NUM_LIT:0> ) ) <EOL> x = [ <NUM_LIT:1> ] <EOL> @ event . listens_for ( sess , "<STR_LIT>" ) <EOL> def add_another_user ( session ) : <EOL> x [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> sess . add ( to_flush . pop ( ) ) <EOL> sess . commit ( ) <EOL> eq_ ( x , [ <NUM_LIT:2> ] ) <EOL> eq_ ( <EOL> sess . scalar ( select ( [ func . count ( users . c . id ) ] ) ) , <NUM_LIT:3> <EOL> ) <EOL> def test_continue_flushing_guard ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( ) <EOL> @ event . listens_for ( sess , "<STR_LIT>" ) <EOL> def add_another_user ( session , ctx ) : <EOL> session . add ( User ( name = '<STR_LIT:x>' ) ) <EOL> sess . add ( User ( name = '<STR_LIT:x>' ) ) <EOL> assert_raises_message ( <EOL> orm_exc . FlushError , <EOL> "<STR_LIT>" , <EOL> sess . commit <EOL> ) <EOL> def test_error_on_using_inactive_session_commands ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = create_session ( autocommit = True ) <EOL> sess . begin ( ) <EOL> sess . begin ( subtransactions = True ) <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> sess . flush ( ) <EOL> sess . rollback ( ) <EOL> assert_raises_message ( sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> "<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> sess . begin , subtransactions = True ) <EOL> sess . close ( ) <EOL> def test_no_sql_during_commit ( self ) : <EOL> sess = create_session ( bind = testing . db , autocommit = False ) <EOL> @ event . listens_for ( sess , "<STR_LIT>" ) <EOL> def go ( session ) : <EOL> session . execute ( "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . commit ) <EOL> def test_no_sql_during_prepare ( self ) : <EOL> sess = create_session ( bind = testing . db , autocommit = False , twophase = True ) <EOL> sess . prepare ( ) <EOL> assert_raises_message ( <EOL> sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . execute , "<STR_LIT>" ) <EOL> def test_no_prepare_wo_twophase ( self ) : <EOL> sess = create_session ( bind = testing . db , autocommit = False ) <EOL> assert_raises_message ( sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . prepare ) <EOL> def test_closed_status_check ( self ) : <EOL> sess = create_session ( ) <EOL> trans = sess . begin ( ) <EOL> trans . rollback ( ) <EOL> assert_raises_message ( <EOL> sa_exc . ResourceClosedError , "<STR_LIT>" , <EOL> trans . rollback ) <EOL> assert_raises_message ( <EOL> sa_exc . ResourceClosedError , "<STR_LIT>" , <EOL> trans . commit ) <EOL> def test_deactive_status_check ( self ) : <EOL> sess = create_session ( ) <EOL> trans = sess . begin ( ) <EOL> trans2 = sess . begin ( subtransactions = True ) <EOL> trans2 . rollback ( ) <EOL> assert_raises_message ( <EOL> sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> trans . commit <EOL> ) <EOL> def test_deactive_status_check_w_exception ( self ) : <EOL> sess = create_session ( ) <EOL> trans = sess . begin ( ) <EOL> trans2 = sess . begin ( subtransactions = True ) <EOL> try : <EOL> raise Exception ( "<STR_LIT:test>" ) <EOL> except : <EOL> trans2 . rollback ( _capture_exception = True ) <EOL> assert_raises_message ( <EOL> sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> trans . commit <EOL> ) <EOL> def _inactive_flushed_session_fixture ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> sess . add ( User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) ) <EOL> assert_raises ( <EOL> orm_exc . FlushError , sess . flush <EOL> ) <EOL> return sess , u1 <EOL> def test_execution_options_begin_transaction ( self ) : <EOL> bind = mock . Mock ( ) <EOL> sess = Session ( bind = bind ) <EOL> c1 = sess . connection ( execution_options = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> eq_ ( <EOL> bind . mock_calls , <EOL> [ <EOL> mock . call . contextual_connect ( ) , <EOL> mock . call . contextual_connect ( ) . <EOL> execution_options ( isolation_level = '<STR_LIT>' ) , <EOL> mock . call . contextual_connect ( ) . execution_options ( ) . begin ( ) <EOL> ] <EOL> ) <EOL> eq_ ( c1 , bind . contextual_connect ( ) . execution_options ( ) ) <EOL> def test_execution_options_ignored_mid_transaction ( self ) : <EOL> bind = mock . Mock ( ) <EOL> conn = mock . Mock ( engine = bind ) <EOL> bind . contextual_connect = mock . Mock ( return_value = conn ) <EOL> sess = Session ( bind = bind ) <EOL> sess . execute ( "<STR_LIT>" ) <EOL> with expect_warnings ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) : <EOL> sess . connection ( execution_options = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_warning_on_using_inactive_session_new ( self ) : <EOL> User = self . classes . User <EOL> sess , u1 = self . _inactive_flushed_session_fixture ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> def go ( ) : <EOL> sess . rollback ( ) <EOL> assert_warnings ( go , <EOL> [ "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> ) <EOL> assert u2 not in sess <EOL> assert u1 in sess <EOL> def test_warning_on_using_inactive_session_dirty ( self ) : <EOL> sess , u1 = self . _inactive_flushed_session_fixture ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> def go ( ) : <EOL> sess . rollback ( ) <EOL> assert_warnings ( go , <EOL> [ "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> ) <EOL> assert u1 in sess <EOL> assert u1 not in sess . dirty <EOL> def test_warning_on_using_inactive_session_delete ( self ) : <EOL> sess , u1 = self . _inactive_flushed_session_fixture ( ) <EOL> sess . delete ( u1 ) <EOL> def go ( ) : <EOL> sess . rollback ( ) <EOL> assert_warnings ( go , <EOL> [ "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> ) <EOL> assert u1 in sess <EOL> assert u1 not in sess . deleted <EOL> def test_warning_on_using_inactive_session_rollback_evt ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> u3 = User ( name = '<STR_LIT>' ) <EOL> @ event . listens_for ( sess , "<STR_LIT>" ) <EOL> def evt ( s ) : <EOL> sess . add ( u3 ) <EOL> sess . add ( User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) ) <EOL> def go ( ) : <EOL> assert_raises ( <EOL> orm_exc . FlushError , sess . flush <EOL> ) <EOL> assert_warnings ( go , <EOL> [ "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> ) <EOL> assert u3 not in sess <EOL> def test_preserve_flush_error ( self ) : <EOL> User = self . classes . User <EOL> sess , u1 = self . _inactive_flushed_session_fixture ( ) <EOL> for i in range ( <NUM_LIT:5> ) : <EOL> assert_raises_message ( sa_exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . commit ) <EOL> sess . rollback ( ) <EOL> sess . add ( User ( id = <NUM_LIT:5> , name = '<STR_LIT>' ) ) <EOL> sess . commit ( ) <EOL> def test_no_autocommit_with_explicit_commit ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> session = create_session ( autocommit = False ) <EOL> session . add ( User ( name = '<STR_LIT>' ) ) <EOL> session . transaction . commit ( ) <EOL> assert session . transaction is not None , '<STR_LIT>' <EOL> @ testing . requires . python2 <EOL> @ testing . requires . savepoints_w_release <EOL> def test_report_primary_error_when_rollback_fails ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> session = Session ( testing . db ) <EOL> with expect_warnings ( "<STR_LIT>" ) : <EOL> session . begin_nested ( ) <EOL> savepoint = session . connection ( ) . _Connection__transaction . _savepoint <EOL> session . connection ( ) . dialect . do_release_savepoint ( <EOL> session . connection ( ) , savepoint <EOL> ) <EOL> session . add_all ( [ User ( id = <NUM_LIT:1> ) , User ( id = <NUM_LIT:1> ) ] ) <EOL> assert_raises_message ( <EOL> sa_exc . DBAPIError , <EOL> "<STR_LIT>" , <EOL> session . flush <EOL> ) <EOL> class _LocalFixture ( FixtureTest ) : <EOL> run_setup_mappers = '<STR_LIT>' <EOL> run_inserts = None <EOL> session = sessionmaker ( ) <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> User , Address = cls . classes . User , cls . classes . Address <EOL> users , addresses = cls . tables . users , cls . tables . addresses <EOL> mapper ( <EOL> User , users , properties = { <EOL> '<STR_LIT>' : relationship ( <EOL> Address , backref = '<STR_LIT:user>' , cascade = "<STR_LIT>" , <EOL> order_by = addresses . c . id ) , <EOL> } ) <EOL> mapper ( Address , addresses ) <EOL> class FixtureDataTest ( _LocalFixture ) : <EOL> run_inserts = '<STR_LIT>' <EOL> __backend__ = True <EOL> def test_attrs_on_rollback ( self ) : <EOL> User = self . classes . User <EOL> sess = self . session ( ) <EOL> u1 = sess . query ( User ) . get ( <NUM_LIT:7> ) <EOL> u1 . name = '<STR_LIT>' <EOL> sess . rollback ( ) <EOL> eq_ ( u1 . name , '<STR_LIT>' ) <EOL> def test_commit_persistent ( self ) : <EOL> User = self . classes . User <EOL> sess = self . session ( ) <EOL> u1 = sess . query ( User ) . get ( <NUM_LIT:7> ) <EOL> u1 . name = '<STR_LIT>' <EOL> sess . flush ( ) <EOL> sess . commit ( ) <EOL> eq_ ( u1 . name , '<STR_LIT>' ) <EOL> def test_concurrent_commit_persistent ( self ) : <EOL> User = self . classes . User <EOL> s1 = self . session ( ) <EOL> u1 = s1 . query ( User ) . get ( <NUM_LIT:7> ) <EOL> u1 . name = '<STR_LIT>' <EOL> s1 . commit ( ) <EOL> s2 = self . session ( ) <EOL> u2 = s2 . query ( User ) . get ( <NUM_LIT:7> ) <EOL> assert u2 . name == '<STR_LIT>' <EOL> u2 . name = '<STR_LIT>' <EOL> s2 . commit ( ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> class CleanSavepointTest ( FixtureTest ) : <EOL> """<STR_LIT>""" <EOL> run_inserts = None <EOL> __backend__ = True <EOL> def _run_test ( self , update_fn ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> s = Session ( bind = testing . db ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> s . add_all ( [ u1 , u2 ] ) <EOL> s . commit ( ) <EOL> u1 . name <EOL> u2 . name <EOL> s . begin_nested ( ) <EOL> update_fn ( s , u2 ) <EOL> eq_ ( u2 . name , '<STR_LIT>' ) <EOL> s . rollback ( ) <EOL> eq_ ( u1 . __dict__ [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> assert '<STR_LIT:name>' not in u2 . __dict__ <EOL> eq_ ( u2 . name , '<STR_LIT>' ) <EOL> @ testing . requires . savepoints <EOL> def test_rollback_ignores_clean_on_savepoint ( self ) : <EOL> def update_fn ( s , u2 ) : <EOL> u2 . name = '<STR_LIT>' <EOL> self . _run_test ( update_fn ) <EOL> @ testing . requires . savepoints <EOL> def test_rollback_ignores_clean_on_savepoint_agg_upd_eval ( self ) : <EOL> User = self . classes . User <EOL> def update_fn ( s , u2 ) : <EOL> s . query ( User ) . filter_by ( name = '<STR_LIT>' ) . update ( <EOL> dict ( name = '<STR_LIT>' ) , synchronize_session = '<STR_LIT>' ) <EOL> self . _run_test ( update_fn ) <EOL> @ testing . requires . savepoints <EOL> def test_rollback_ignores_clean_on_savepoint_agg_upd_fetch ( self ) : <EOL> User = self . classes . User <EOL> def update_fn ( s , u2 ) : <EOL> s . query ( User ) . filter_by ( name = '<STR_LIT>' ) . update ( <EOL> dict ( name = '<STR_LIT>' ) , <EOL> synchronize_session = '<STR_LIT>' ) <EOL> self . _run_test ( update_fn ) <EOL> class ContextManagerTest ( FixtureTest ) : <EOL> run_inserts = None <EOL> __backend__ = True <EOL> @ testing . requires . savepoints <EOL> @ engines . close_open_connections <EOL> def test_contextmanager_nested_rollback ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( ) <EOL> def go ( ) : <EOL> with sess . begin_nested ( ) : <EOL> sess . add ( User ( ) ) <EOL> sess . flush ( ) <EOL> assert_raises ( <EOL> sa_exc . DBAPIError , <EOL> go <EOL> ) <EOL> with sess . begin_nested ( ) : <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> eq_ ( sess . query ( User ) . count ( ) , <NUM_LIT:1> ) <EOL> def test_contextmanager_commit ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( autocommit = True ) <EOL> with sess . begin ( ) : <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> sess . rollback ( ) <EOL> eq_ ( sess . query ( User ) . count ( ) , <NUM_LIT:1> ) <EOL> def test_contextmanager_rollback ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> sess = Session ( autocommit = True ) <EOL> def go ( ) : <EOL> with sess . begin ( ) : <EOL> sess . add ( User ( ) ) <EOL> assert_raises ( <EOL> sa_exc . DBAPIError , <EOL> go <EOL> ) <EOL> eq_ ( sess . query ( User ) . count ( ) , <NUM_LIT:0> ) <EOL> with sess . begin ( ) : <EOL> sess . add ( User ( name = '<STR_LIT>' ) ) <EOL> eq_ ( sess . query ( User ) . count ( ) , <NUM_LIT:1> ) <EOL> class AutoExpireTest ( _LocalFixture ) : <EOL> __backend__ = True <EOL> def test_expunge_pending_on_rollback ( self ) : <EOL> User = self . classes . User <EOL> sess = self . session ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> assert u2 in sess <EOL> sess . rollback ( ) <EOL> assert u2 not in sess <EOL> def test_trans_pending_cleared_on_commit ( self ) : <EOL> User = self . classes . User <EOL> sess = self . session ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> assert u2 in sess <EOL> sess . commit ( ) <EOL> assert u2 in sess <EOL> u3 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u3 ) <EOL> sess . rollback ( ) <EOL> assert u3 not in sess <EOL> assert u2 in sess <EOL> def test_update_deleted_on_rollback ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> assert u1 in s . deleted <EOL> s . rollback ( ) <EOL> assert u1 in s <EOL> assert u1 not in s . deleted <EOL> @ testing . requires . predictable_gc <EOL> def test_gced_delete_on_rollback ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> u1_state = attributes . instance_state ( u1 ) <EOL> assert u1_state in s . identity_map . all_states ( ) <EOL> assert u1_state in s . _deleted <EOL> s . flush ( ) <EOL> assert u1_state not in s . identity_map . all_states ( ) <EOL> assert u1_state not in s . _deleted <EOL> del u1 <EOL> gc_collect ( ) <EOL> assert u1_state . obj ( ) is None <EOL> s . rollback ( ) <EOL> assert u1_state not in s . identity_map . all_states ( ) <EOL> u1 = s . query ( User ) . filter_by ( name = '<STR_LIT>' ) . one ( ) <EOL> assert u1_state not in s . identity_map . all_states ( ) <EOL> assert s . scalar ( users . count ( ) ) == <NUM_LIT:1> <EOL> s . delete ( u1 ) <EOL> s . flush ( ) <EOL> assert s . scalar ( users . count ( ) ) == <NUM_LIT:0> <EOL> s . commit ( ) <EOL> def test_trans_deleted_cleared_on_rollback ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> s . commit ( ) <EOL> assert u1 not in s <EOL> s . rollback ( ) <EOL> assert u1 not in s <EOL> def test_update_deleted_on_rollback_cascade ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT:foo>' ) ] ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> assert u1 in s . deleted <EOL> assert u1 . addresses [ <NUM_LIT:0> ] in s . deleted <EOL> s . rollback ( ) <EOL> assert u1 in s <EOL> assert u1 not in s . deleted <EOL> assert u1 . addresses [ <NUM_LIT:0> ] not in s . deleted <EOL> def test_update_deleted_on_rollback_orphan ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT:foo>' ) ] ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> a1 = u1 . addresses [ <NUM_LIT:0> ] <EOL> u1 . addresses . remove ( a1 ) <EOL> s . flush ( ) <EOL> eq_ ( s . query ( Address ) . filter ( Address . email_address == '<STR_LIT:foo>' ) . all ( ) , [ ] ) <EOL> s . rollback ( ) <EOL> assert a1 not in s . deleted <EOL> assert u1 . addresses == [ a1 ] <EOL> def test_commit_pending ( self ) : <EOL> User = self . classes . User <EOL> sess = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . commit ( ) <EOL> eq_ ( u1 . name , '<STR_LIT>' ) <EOL> def test_concurrent_commit_pending ( self ) : <EOL> User = self . classes . User <EOL> s1 = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s1 . add ( u1 ) <EOL> s1 . commit ( ) <EOL> s2 = self . session ( ) <EOL> u2 = s2 . query ( User ) . filter ( User . name == '<STR_LIT>' ) . one ( ) <EOL> u2 . name = '<STR_LIT>' <EOL> s2 . commit ( ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> class TwoPhaseTest ( _LocalFixture ) : <EOL> __backend__ = True <EOL> @ testing . requires . two_phase_transactions <EOL> def test_rollback_on_prepare ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( twophase = True ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> s . add ( u ) <EOL> s . prepare ( ) <EOL> s . rollback ( ) <EOL> assert u not in s <EOL> class RollbackRecoverTest ( _LocalFixture ) : <EOL> __backend__ = True <EOL> def test_pk_violation ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> s = self . session ( ) <EOL> a1 = Address ( email_address = '<STR_LIT:foo>' ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' , addresses = [ a1 ] ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> a2 = Address ( email_address = '<STR_LIT:bar>' ) <EOL> u2 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' , addresses = [ a2 ] ) <EOL> u1 . name = '<STR_LIT>' <EOL> a1 . email_address = '<STR_LIT>' <EOL> s . add ( u2 ) <EOL> assert_raises ( orm_exc . FlushError , s . commit ) <EOL> assert_raises ( sa_exc . InvalidRequestError , s . commit ) <EOL> s . rollback ( ) <EOL> assert u2 not in s <EOL> assert a2 not in s <EOL> assert u1 in s <EOL> assert a1 in s <EOL> assert u1 . name == '<STR_LIT>' <EOL> assert a1 . email_address == '<STR_LIT:foo>' <EOL> u1 . name = '<STR_LIT>' <EOL> a1 . email_address = '<STR_LIT>' <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( User ) . all ( ) , <EOL> [ User ( id = <NUM_LIT:1> , name = '<STR_LIT>' , <EOL> addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) ] <EOL> ) <EOL> @ testing . requires . savepoints <EOL> def test_pk_violation_with_savepoint ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> s = self . session ( ) <EOL> a1 = Address ( email_address = '<STR_LIT:foo>' ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' , addresses = [ a1 ] ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> a2 = Address ( email_address = '<STR_LIT:bar>' ) <EOL> u2 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' , addresses = [ a2 ] ) <EOL> u1 . name = '<STR_LIT>' <EOL> a1 . email_address = '<STR_LIT>' <EOL> s . begin_nested ( ) <EOL> s . add ( u2 ) <EOL> assert_raises ( orm_exc . FlushError , s . commit ) <EOL> assert_raises ( sa_exc . InvalidRequestError , s . commit ) <EOL> s . rollback ( ) <EOL> assert u2 not in s <EOL> assert a2 not in s <EOL> assert u1 in s <EOL> assert a1 in s <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( User ) . all ( ) , <EOL> [ <EOL> User ( <EOL> id = <NUM_LIT:1> , name = '<STR_LIT>' , <EOL> addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) ] ) <EOL> class SavepointTest ( _LocalFixture ) : <EOL> __backend__ = True <EOL> @ testing . requires . savepoints <EOL> def test_savepoint_rollback ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> s . add_all ( [ u1 , u2 ] ) <EOL> s . begin_nested ( ) <EOL> u3 = User ( name = '<STR_LIT>' ) <EOL> u4 = User ( name = '<STR_LIT:foo>' ) <EOL> u1 . name = '<STR_LIT>' <EOL> u2 . name = '<STR_LIT>' <EOL> s . add_all ( [ u3 , u4 ] ) <EOL> eq_ ( <EOL> s . query ( User . name ) . order_by ( User . id ) . all ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT:foo>' , ) ] ) <EOL> s . rollback ( ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> assert u2 . name == '<STR_LIT>' <EOL> eq_ ( <EOL> s . query ( User . name ) . order_by ( User . id ) . all ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) ] ) <EOL> s . commit ( ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> assert u2 . name == '<STR_LIT>' <EOL> eq_ ( s . query ( User . name ) . order_by ( User . id ) . all ( ) , [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) ] ) <EOL> @ testing . requires . savepoints <EOL> def test_savepoint_delete ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> eq_ ( s . query ( User ) . filter_by ( name = '<STR_LIT>' ) . count ( ) , <NUM_LIT:1> ) <EOL> s . begin_nested ( ) <EOL> s . delete ( u1 ) <EOL> s . commit ( ) <EOL> eq_ ( s . query ( User ) . filter_by ( name = '<STR_LIT>' ) . count ( ) , <NUM_LIT:0> ) <EOL> s . commit ( ) <EOL> @ testing . requires . savepoints <EOL> def test_savepoint_commit ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> s . add_all ( [ u1 , u2 ] ) <EOL> s . begin_nested ( ) <EOL> u3 = User ( name = '<STR_LIT>' ) <EOL> u4 = User ( name = '<STR_LIT:foo>' ) <EOL> u1 . name = '<STR_LIT>' <EOL> u2 . name = '<STR_LIT>' <EOL> s . add_all ( [ u3 , u4 ] ) <EOL> eq_ ( <EOL> s . query ( User . name ) . order_by ( User . id ) . all ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT:foo>' , ) ] ) <EOL> s . commit ( ) <EOL> def go ( ) : <EOL> assert u1 . name == '<STR_LIT>' <EOL> assert u2 . name == '<STR_LIT>' <EOL> eq_ ( <EOL> s . query ( User . name ) . order_by ( User . id ) . all ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT:foo>' , ) ] ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:1> ) <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( User . name ) . order_by ( User . id ) . all ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT:foo>' , ) ] ) <EOL> @ testing . requires . savepoints <EOL> def test_savepoint_rollback_collections ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT:foo>' ) ] ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> u1 . addresses . append ( Address ( email_address = '<STR_LIT:bar>' ) ) <EOL> s . begin_nested ( ) <EOL> u2 = User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) <EOL> s . add ( u2 ) <EOL> eq_ ( <EOL> s . query ( User ) . order_by ( User . id ) . all ( ) , <EOL> [ <EOL> User ( <EOL> name = '<STR_LIT>' , <EOL> addresses = [ <EOL> Address ( email_address = '<STR_LIT:foo>' ) , <EOL> Address ( email_address = '<STR_LIT:bar>' ) ] ) , <EOL> User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) <EOL> ] ) <EOL> s . rollback ( ) <EOL> eq_ ( <EOL> s . query ( User ) . order_by ( User . id ) . all ( ) , <EOL> [ <EOL> User ( <EOL> name = '<STR_LIT>' , <EOL> addresses = [ <EOL> Address ( email_address = '<STR_LIT:foo>' ) , <EOL> Address ( email_address = '<STR_LIT:bar>' ) ] ) , <EOL> ] ) <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( User ) . order_by ( User . id ) . all ( ) , <EOL> [ <EOL> User ( <EOL> name = '<STR_LIT>' , <EOL> addresses = [ <EOL> Address ( email_address = '<STR_LIT:foo>' ) , <EOL> Address ( email_address = '<STR_LIT:bar>' ) ] ) , <EOL> ] <EOL> ) <EOL> @ testing . requires . savepoints <EOL> def test_savepoint_commit_collections ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT:foo>' ) ] ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> u1 . addresses . append ( Address ( email_address = '<STR_LIT:bar>' ) ) <EOL> s . begin_nested ( ) <EOL> u2 = User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) <EOL> s . add ( u2 ) <EOL> eq_ ( <EOL> s . query ( User ) . order_by ( User . id ) . all ( ) , <EOL> [ <EOL> User ( <EOL> name = '<STR_LIT>' , <EOL> addresses = [ <EOL> Address ( email_address = '<STR_LIT:foo>' ) , <EOL> Address ( email_address = '<STR_LIT:bar>' ) ] ) , <EOL> User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) <EOL> ] <EOL> ) <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( User ) . order_by ( User . id ) . all ( ) , <EOL> [ <EOL> User ( <EOL> name = '<STR_LIT>' , <EOL> addresses = [ <EOL> Address ( email_address = '<STR_LIT:foo>' ) , <EOL> Address ( email_address = '<STR_LIT:bar>' ) ] ) , <EOL> User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) <EOL> ] <EOL> ) <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( User ) . order_by ( User . id ) . all ( ) , <EOL> [ <EOL> User ( <EOL> name = '<STR_LIT>' , <EOL> addresses = [ <EOL> Address ( email_address = '<STR_LIT:foo>' ) , <EOL> Address ( email_address = '<STR_LIT:bar>' ) ] ) , <EOL> User ( name = '<STR_LIT>' , addresses = [ Address ( email_address = '<STR_LIT>' ) ] ) <EOL> ] <EOL> ) <EOL> @ testing . requires . savepoints <EOL> def test_expunge_pending_on_rollback ( self ) : <EOL> User = self . classes . User <EOL> sess = self . session ( ) <EOL> sess . begin_nested ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> assert u2 in sess <EOL> sess . rollback ( ) <EOL> assert u2 not in sess <EOL> @ testing . requires . savepoints <EOL> def test_update_deleted_on_rollback ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> s . begin_nested ( ) <EOL> s . delete ( u1 ) <EOL> assert u1 in s . deleted <EOL> s . rollback ( ) <EOL> assert u1 in s <EOL> assert u1 not in s . deleted <EOL> @ testing . requires . savepoints_w_release <EOL> def test_savepoint_lost_still_runs ( self ) : <EOL> User = self . classes . User <EOL> s = self . session ( bind = self . bind ) <EOL> trans = s . begin_nested ( ) <EOL> s . connection ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u1 ) <EOL> nested_trans = trans . _connections [ self . bind ] [ <NUM_LIT:1> ] <EOL> nested_trans . _do_commit ( ) <EOL> is_ ( s . transaction , trans ) <EOL> assert_raises ( <EOL> sa_exc . DBAPIError , <EOL> s . rollback <EOL> ) <EOL> assert u1 not in s . new <EOL> is_ ( trans . _state , _session . CLOSED ) <EOL> is_not_ ( s . transaction , trans ) <EOL> is_ ( s . transaction . _state , _session . ACTIVE ) <EOL> is_ ( s . transaction . nested , False ) <EOL> is_ ( s . transaction . _parent , None ) <EOL> class AccountingFlagsTest ( _LocalFixture ) : <EOL> __backend__ = True <EOL> def test_no_expire_on_commit ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> sess = sessionmaker ( expire_on_commit = False ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> testing . db . execute ( <EOL> users . update ( users . c . name == '<STR_LIT>' ) . values ( name = '<STR_LIT>' ) ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> sess . expire_all ( ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> def test_rollback_no_accounting ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> sess = sessionmaker ( _enable_transaction_accounting = False ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> sess . rollback ( ) <EOL> testing . db . execute ( <EOL> users . update ( users . c . name == '<STR_LIT>' ) . values ( name = '<STR_LIT>' ) ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> sess . expire_all ( ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> def test_commit_no_accounting ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> sess = sessionmaker ( _enable_transaction_accounting = False ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> sess . rollback ( ) <EOL> testing . db . execute ( <EOL> users . update ( users . c . name == '<STR_LIT>' ) . values ( name = '<STR_LIT>' ) ) <EOL> assert u1 . name == '<STR_LIT>' <EOL> sess . commit ( ) <EOL> assert testing . db . execute ( select ( [ users . c . name ] ) ) . fetchall ( ) == [ ( '<STR_LIT>' , ) ] <EOL> assert u1 . name == '<STR_LIT>' <EOL> sess . delete ( u1 ) <EOL> sess . commit ( ) <EOL> def test_preflush_no_accounting ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> sess = Session ( <EOL> _enable_transaction_accounting = False , autocommit = True , <EOL> autoflush = False ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . begin ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> u2 = User ( name = "<STR_LIT>" ) <EOL> sess . add ( u2 ) <EOL> sess . rollback ( ) <EOL> sess . begin ( ) <EOL> assert testing . db . execute ( select ( [ users . c . name ] ) ) . fetchall ( ) == [ ( '<STR_LIT>' , ) ] <EOL> class AutoCommitTest ( _LocalFixture ) : <EOL> __backend__ = True <EOL> def test_begin_nested_requires_trans ( self ) : <EOL> sess = create_session ( autocommit = True ) <EOL> assert_raises ( sa_exc . InvalidRequestError , sess . begin_nested ) <EOL> def test_begin_preflush ( self ) : <EOL> User = self . classes . User <EOL> sess = create_session ( autocommit = True ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . begin ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> sess . rollback ( ) <EOL> assert u2 not in sess <EOL> assert u1 in sess <EOL> assert sess . query ( User ) . filter_by ( name = '<STR_LIT>' ) . one ( ) is u1 <EOL> def test_accounting_commit_fails_add ( self ) : <EOL> User = self . classes . User <EOL> sess = create_session ( autocommit = True ) <EOL> fail = False <EOL> def fail_fn ( * arg , ** kw ) : <EOL> if fail : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> event . listen ( sess , "<STR_LIT>" , fail_fn ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> fail = True <EOL> assert_raises ( <EOL> Exception , <EOL> sess . flush <EOL> ) <EOL> fail = False <EOL> assert u1 not in sess <EOL> u1new = User ( id = <NUM_LIT:2> , name = '<STR_LIT>' ) <EOL> sess . add ( u1new ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> assert u1 in sess <EOL> eq_ ( <EOL> sess . query ( User . name ) . order_by ( User . name ) . all ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) ] <EOL> ) <EOL> def test_accounting_commit_fails_delete ( self ) : <EOL> User = self . classes . User <EOL> sess = create_session ( autocommit = True ) <EOL> fail = False <EOL> def fail_fn ( * arg , ** kw ) : <EOL> if fail : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> event . listen ( sess , "<STR_LIT>" , fail_fn ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( u1 ) <EOL> fail = True <EOL> assert_raises ( <EOL> Exception , <EOL> sess . flush <EOL> ) <EOL> fail = False <EOL> assert u1 in sess <EOL> assert u1 not in sess . deleted <EOL> sess . delete ( u1 ) <EOL> sess . flush ( ) <EOL> assert u1 not in sess <EOL> eq_ ( <EOL> sess . query ( User . name ) . order_by ( User . name ) . all ( ) , <EOL> [ ] <EOL> ) <EOL> @ testing . requires . updateable_autoincrement_pks <EOL> def test_accounting_no_select_needed ( self ) : <EOL> """<STR_LIT>""" <EOL> User = self . classes . User <EOL> sess = create_session ( autocommit = True , expire_on_commit = True ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> u1 . id = <NUM_LIT:3> <EOL> u1 . name = '<STR_LIT>' <EOL> self . assert_sql_count ( testing . db , sess . flush , <NUM_LIT:1> ) <EOL> assert '<STR_LIT:id>' not in u1 . __dict__ <EOL> eq_ ( u1 . id , <NUM_LIT:3> ) <EOL> class NaturalPKRollbackTest ( fixtures . MappedTest ) : <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT>' , metadata , Column ( '<STR_LIT:name>' , String ( <NUM_LIT:50> ) , primary_key = True ) ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class User ( cls . Comparable ) : <EOL> pass <EOL> def test_rollback_recover ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> session = sessionmaker ( ) ( ) <EOL> u1 , u2 , u3 = User ( name = '<STR_LIT>' ) , User ( name = '<STR_LIT>' ) , User ( name = '<STR_LIT>' ) <EOL> session . add_all ( [ u1 , u2 , u3 ] ) <EOL> session . commit ( ) <EOL> session . delete ( u2 ) <EOL> u4 = User ( name = '<STR_LIT>' ) <EOL> session . add ( u4 ) <EOL> session . flush ( ) <EOL> u5 = User ( name = '<STR_LIT>' ) <EOL> session . add ( u5 ) <EOL> assert_raises ( orm_exc . FlushError , session . flush ) <EOL> assert u5 not in session <EOL> assert u2 not in session . deleted <EOL> session . rollback ( ) <EOL> def test_reloaded_deleted_checked_for_expiry ( self ) : <EOL> """<STR_LIT>""" <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s = Session ( ) <EOL> s . add ( u1 ) <EOL> s . flush ( ) <EOL> del u1 <EOL> gc_collect ( ) <EOL> u1 = s . query ( User ) . first ( ) <EOL> s . rollback ( ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> s . add ( u2 ) <EOL> s . commit ( ) <EOL> assert inspect ( u2 ) . persistent <EOL> def test_key_replaced_by_update ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> s = Session ( ) <EOL> s . add_all ( [ u1 , u2 ] ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> s . flush ( ) <EOL> u2 . name = '<STR_LIT>' <EOL> s . flush ( ) <EOL> assert u1 not in s <EOL> s . rollback ( ) <EOL> assert u1 in s <EOL> assert u2 in s <EOL> assert s . identity_map [ ( User , ( '<STR_LIT>' , ) ) ] is u1 <EOL> assert s . identity_map [ ( User , ( '<STR_LIT>' , ) ) ] is u2 <EOL> def test_multiple_key_replaced_by_update ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> u3 = User ( name = '<STR_LIT>' ) <EOL> s = Session ( ) <EOL> s . add_all ( [ u1 , u2 , u3 ] ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> s . delete ( u2 ) <EOL> s . flush ( ) <EOL> u3 . name = '<STR_LIT>' <EOL> s . flush ( ) <EOL> u3 . name = '<STR_LIT>' <EOL> s . flush ( ) <EOL> s . rollback ( ) <EOL> assert u1 in s <EOL> assert u2 in s <EOL> assert u3 in s <EOL> assert s . identity_map [ ( User , ( '<STR_LIT>' , ) ) ] is u1 <EOL> assert s . identity_map [ ( User , ( '<STR_LIT>' , ) ) ] is u2 <EOL> assert s . identity_map [ ( User , ( '<STR_LIT>' , ) ) ] is u3 <EOL> def test_key_replaced_by_oob_insert ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> s = Session ( ) <EOL> s . add ( u1 ) <EOL> s . commit ( ) <EOL> s . delete ( u1 ) <EOL> s . flush ( ) <EOL> s . execute ( users . insert ( ) . values ( name = '<STR_LIT>' ) ) <EOL> u2 = s . query ( User ) . get ( '<STR_LIT>' ) <EOL> assert u1 not in s <EOL> s . rollback ( ) <EOL> assert u1 in s <EOL> assert u2 not in s <EOL> assert s . identity_map [ ( User , ( '<STR_LIT>' , ) ) ] is u1 </s>
