<s> from sqlalchemy . testing import eq_ , assert_raises_message <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import engines <EOL> from sqlalchemy . testing . schema import Table , Column <EOL> from test . orm import _fixtures <EOL> from sqlalchemy import exc , util <EOL> from sqlalchemy . testing import fixtures , config <EOL> from sqlalchemy import Integer , String , ForeignKey , func , literal , FetchedValue , text <EOL> from sqlalchemy . orm import mapper , relationship , backref , create_session , unitofwork , attributes , Session , exc as orm_exc <EOL> from sqlalchemy . testing . mock import Mock , patch <EOL> from sqlalchemy . testing . assertsql import AllOf , CompiledSQL <EOL> from sqlalchemy import event <EOL> class AssertsUOW ( object ) : <EOL> def _get_test_uow ( self , session ) : <EOL> uow = unitofwork . UOWTransaction ( session ) <EOL> deleted = set ( session . _deleted ) <EOL> new = set ( session . _new ) <EOL> dirty = set ( session . _dirty_states ) . difference ( deleted ) <EOL> for s in new . union ( dirty ) : <EOL> uow . register_object ( s ) <EOL> for d in deleted : <EOL> uow . register_object ( d , isdelete = True ) <EOL> return uow <EOL> def _assert_uow_size ( self , session , expected ) : <EOL> uow = self . _get_test_uow ( session ) <EOL> postsort_actions = uow . _generate_actions ( ) <EOL> print ( postsort_actions ) <EOL> eq_ ( len ( postsort_actions ) , expected , postsort_actions ) <EOL> class UOWTest ( <EOL> _fixtures . FixtureTest , <EOL> testing . AssertsExecutionResults , AssertsUOW ) : <EOL> run_inserts = None <EOL> class RudimentaryFlushTest ( UOWTest ) : <EOL> def test_one_to_many_save ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( Address ) , <EOL> } ) <EOL> mapper ( Address , addresses ) <EOL> sess = create_session ( ) <EOL> a1 , a2 = Address ( email_address = '<STR_LIT>' ) , Address ( email_address = '<STR_LIT>' ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 , a2 ] ) <EOL> sess . add ( u1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : u1 . id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : u1 . id } <EOL> ) , <EOL> ) <EOL> def test_one_to_many_delete_all ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( Address ) , <EOL> } ) <EOL> mapper ( Address , addresses ) <EOL> sess = create_session ( ) <EOL> a1 , a2 = Address ( email_address = '<STR_LIT>' ) , Address ( email_address = '<STR_LIT>' ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 , a2 ] ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( u1 ) <EOL> sess . delete ( a1 ) <EOL> sess . delete ( a2 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : a1 . id } , { '<STR_LIT:id>' : a2 . id } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:id>' : u1 . id } <EOL> ) , <EOL> ) <EOL> def test_one_to_many_delete_parent ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( Address ) , <EOL> } ) <EOL> mapper ( Address , addresses ) <EOL> sess = create_session ( ) <EOL> a1 , a2 = Address ( email_address = '<STR_LIT>' ) , Address ( email_address = '<STR_LIT>' ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 , a2 ] ) <EOL> sess . add ( u1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( u1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ <EOL> { '<STR_LIT>' : a1 . id , '<STR_LIT>' : None } , <EOL> { '<STR_LIT>' : a2 . id , '<STR_LIT>' : None } <EOL> ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:id>' : u1 . id } <EOL> ) , <EOL> ) <EOL> def test_many_to_one_save ( 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 , properties = { <EOL> '<STR_LIT:user>' : relationship ( User ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> a1 , a2 = Address ( email_address = '<STR_LIT>' , user = u1 ) , Address ( email_address = '<STR_LIT>' , user = u1 ) <EOL> sess . add_all ( [ a1 , a2 ] ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : u1 . id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : u1 . id } <EOL> ) , <EOL> ) <EOL> def test_many_to_one_delete_all ( 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 , properties = { <EOL> '<STR_LIT:user>' : relationship ( User ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> a1 , a2 = Address ( email_address = '<STR_LIT>' , user = u1 ) , Address ( email_address = '<STR_LIT>' , user = u1 ) <EOL> sess . add_all ( [ a1 , a2 ] ) <EOL> sess . flush ( ) <EOL> sess . delete ( u1 ) <EOL> sess . delete ( a1 ) <EOL> sess . delete ( a2 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : a1 . id } , { '<STR_LIT:id>' : a2 . id } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:id>' : u1 . id } <EOL> ) , <EOL> ) <EOL> def test_many_to_one_delete_target ( 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 , properties = { <EOL> '<STR_LIT:user>' : relationship ( User ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> a1 , a2 = Address ( email_address = '<STR_LIT>' , user = u1 ) , Address ( email_address = '<STR_LIT>' , user = u1 ) <EOL> sess . add_all ( [ a1 , a2 ] ) <EOL> sess . flush ( ) <EOL> sess . delete ( u1 ) <EOL> a1 . user = a2 . user = None <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ <EOL> { '<STR_LIT>' : a1 . id , '<STR_LIT>' : None } , <EOL> { '<STR_LIT>' : a2 . id , '<STR_LIT>' : None } <EOL> ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:id>' : u1 . id } <EOL> ) , <EOL> ) <EOL> def test_many_to_one_delete_unloaded ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( User ) <EOL> } ) <EOL> parent = User ( name = '<STR_LIT>' ) <EOL> c1 , c2 = Address ( email_address = '<STR_LIT>' , parent = parent ) , Address ( email_address = '<STR_LIT>' , parent = parent ) <EOL> session = Session ( ) <EOL> session . add_all ( [ c1 , c2 ] ) <EOL> session . add ( parent ) <EOL> session . flush ( ) <EOL> pid = parent . id <EOL> c1id = c1 . id <EOL> c2id = c2 . id <EOL> session . expire ( parent ) <EOL> session . expire ( c1 ) <EOL> session . expire ( c2 ) <EOL> session . delete ( c1 ) <EOL> session . delete ( c2 ) <EOL> session . delete ( parent ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> session . flush , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c1id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c2id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : pid } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : c1id } , { '<STR_LIT:id>' : c2id } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT:id>' : pid } <EOL> ) , <EOL> ) , <EOL> ) <EOL> def test_many_to_one_delete_childonly_unloaded ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( User ) <EOL> } ) <EOL> parent = User ( name = '<STR_LIT>' ) <EOL> c1 , c2 = Address ( email_address = '<STR_LIT>' , parent = parent ) , Address ( email_address = '<STR_LIT>' , parent = parent ) <EOL> session = Session ( ) <EOL> session . add_all ( [ c1 , c2 ] ) <EOL> session . add ( parent ) <EOL> session . flush ( ) <EOL> c1id = c1 . id <EOL> c2id = c2 . id <EOL> session . expire ( c1 ) <EOL> session . expire ( c2 ) <EOL> session . delete ( c1 ) <EOL> session . delete ( c2 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> session . flush , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c1id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c2id } <EOL> ) , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : c1id } , { '<STR_LIT:id>' : c2id } ] <EOL> ) , <EOL> ) <EOL> def test_many_to_one_delete_childonly_unloaded_expired ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( User ) <EOL> } ) <EOL> parent = User ( name = '<STR_LIT>' ) <EOL> c1 , c2 = Address ( email_address = '<STR_LIT>' , parent = parent ) , Address ( email_address = '<STR_LIT>' , parent = parent ) <EOL> session = Session ( ) <EOL> session . add_all ( [ c1 , c2 ] ) <EOL> session . add ( parent ) <EOL> session . flush ( ) <EOL> c1id = c1 . id <EOL> c2id = c2 . id <EOL> session . expire ( parent ) <EOL> session . expire ( c1 ) <EOL> session . expire ( c2 ) <EOL> session . delete ( c1 ) <EOL> session . delete ( c2 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> session . flush , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c1id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c2id } <EOL> ) , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : c1id } , { '<STR_LIT:id>' : c2id } ] <EOL> ) , <EOL> ) <EOL> def test_natural_ordering ( self ) : <EOL> """<STR_LIT>""" <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 , properties = { <EOL> '<STR_LIT>' : relationship ( User ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) <EOL> a1 = Address ( id = <NUM_LIT:1> , user_id = <NUM_LIT:1> , email_address = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , a1 ] ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:name>' : '<STR_LIT>' } ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:id>' : <NUM_LIT:1> } <EOL> ) <EOL> ) <EOL> sess . delete ( u1 ) <EOL> sess . delete ( a1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } ] <EOL> ) <EOL> ) <EOL> def test_natural_selfref ( self ) : <EOL> """<STR_LIT>""" <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( id = <NUM_LIT:1> ) <EOL> n2 = Node ( id = <NUM_LIT:2> , parent_id = <NUM_LIT:1> ) <EOL> n3 = Node ( id = <NUM_LIT:3> , parent_id = <NUM_LIT:2> ) <EOL> sess . add_all ( [ n1 , n2 , n3 ] ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : None , '<STR_LIT:data>' : None , '<STR_LIT:id>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:data>' : None , '<STR_LIT:id>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT:data>' : None , '<STR_LIT:id>' : <NUM_LIT:3> } ] <EOL> ) , <EOL> ) <EOL> def test_many_to_many ( self ) : <EOL> keywords , items , item_keywords , Keyword , Item = ( <EOL> self . tables . keywords , self . tables . items , self . tables . item_keywords , <EOL> self . classes . Keyword , self . classes . Item ) <EOL> mapper ( Item , items , properties = { <EOL> '<STR_LIT>' : relationship ( Keyword , secondary = item_keywords ) <EOL> } ) <EOL> mapper ( Keyword , keywords ) <EOL> sess = create_session ( ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keywords = [ k1 ] ) <EOL> sess . add ( i1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:description>' : '<STR_LIT>' } <EOL> ) , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : i1 . id , '<STR_LIT>' : k1 . id } <EOL> ) <EOL> ) <EOL> sess . expire ( i1 , [ '<STR_LIT>' ] ) <EOL> i1 . description = '<STR_LIT>' <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT:description>' : '<STR_LIT>' , '<STR_LIT>' : i1 . id } ) <EOL> ) <EOL> def test_m2o_flush_size ( 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 , properties = { <EOL> '<STR_LIT:user>' : relationship ( User , passive_updates = True ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> def test_o2m_flush_size ( 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 , properties = { <EOL> '<STR_LIT>' : relationship ( Address ) , <EOL> } ) <EOL> mapper ( Address , addresses ) <EOL> sess = create_session ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> sess . flush ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> sess . flush ( ) <EOL> a1 = Address ( email_address = '<STR_LIT:foo>' ) <EOL> sess . add ( a1 ) <EOL> sess . flush ( ) <EOL> u1 . addresses . append ( a1 ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:6> ) <EOL> sess . flush ( ) <EOL> sess = create_session ( ) <EOL> u1 = sess . query ( User ) . first ( ) <EOL> u1 . name = '<STR_LIT>' <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> u1 . addresses <EOL> self . _assert_uow_size ( sess , <NUM_LIT:6> ) <EOL> class SingleCycleTest ( UOWTest ) : <EOL> def teardown ( self ) : <EOL> engines . testing_reaper . rollback_all ( ) <EOL> testing . db . execute ( <EOL> self . tables . nodes . update ( ) . values ( parent_id = None ) <EOL> ) <EOL> super ( SingleCycleTest , self ) . teardown ( ) <EOL> def test_one_to_many_save ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' ) , Node ( data = '<STR_LIT>' ) <EOL> n1 = Node ( data = '<STR_LIT>' , children = [ n2 , n3 ] ) <EOL> sess . add ( n1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : None , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> ) <EOL> ) <EOL> def test_one_to_many_delete_all ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' , children = [ ] ) , Node ( data = '<STR_LIT>' , children = [ ] ) <EOL> n1 = Node ( data = '<STR_LIT>' , children = [ n2 , n3 ] ) <EOL> sess . add ( n1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( n1 ) <EOL> sess . delete ( n2 ) <EOL> sess . delete ( n3 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : n2 . id } , { '<STR_LIT:id>' : n3 . id } ] ) , <EOL> CompiledSQL ( "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT:id>' : n1 . id } ) <EOL> ) <EOL> def test_one_to_many_delete_parent ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' , children = [ ] ) , Node ( data = '<STR_LIT>' , children = [ ] ) <EOL> n1 = Node ( data = '<STR_LIT>' , children = [ n2 , n3 ] ) <EOL> sess . add ( n1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( n1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , sess . flush , AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , lambda ctx : [ <EOL> { '<STR_LIT>' : n3 . id , '<STR_LIT>' : None } , <EOL> { '<STR_LIT>' : n2 . id , '<STR_LIT>' : None } <EOL> ] <EOL> ) <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , lambda ctx : { <EOL> '<STR_LIT:id>' : n1 . id } ) ) <EOL> def test_many_to_one_save ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node , remote_side = nodes . c . id ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' , parent = n1 ) , Node ( data = '<STR_LIT>' , parent = n1 ) <EOL> sess . add_all ( [ n2 , n3 ] ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : None , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> ) <EOL> ) <EOL> def test_many_to_one_delete_all ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node , remote_side = nodes . c . id ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' , parent = n1 ) , Node ( data = '<STR_LIT>' , parent = n1 ) <EOL> sess . add_all ( [ n2 , n3 ] ) <EOL> sess . flush ( ) <EOL> sess . delete ( n1 ) <EOL> sess . delete ( n2 ) <EOL> sess . delete ( n3 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : n2 . id } , { '<STR_LIT:id>' : n3 . id } ] ) , <EOL> CompiledSQL ( "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT:id>' : n1 . id } ) <EOL> ) <EOL> def test_many_to_one_set_null_unloaded ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node , remote_side = nodes . c . id ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> n2 = Node ( data = '<STR_LIT>' , parent = n1 ) <EOL> sess . add_all ( [ n1 , n2 ] ) <EOL> sess . flush ( ) <EOL> sess . close ( ) <EOL> n2 = sess . query ( Node ) . filter_by ( data = '<STR_LIT>' ) . one ( ) <EOL> n2 . parent = None <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { "<STR_LIT>" : None , "<STR_LIT>" : n2 . id } <EOL> ) <EOL> ) <EOL> def test_cycle_rowswitch ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' , children = [ ] ) , Node ( data = '<STR_LIT>' , children = [ ] ) <EOL> n1 = Node ( data = '<STR_LIT>' , children = [ n2 ] ) <EOL> sess . add ( n1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( n2 ) <EOL> n3 . id = n2 . id <EOL> n1 . children . append ( n3 ) <EOL> sess . flush ( ) <EOL> def test_bidirectional_mutations_one ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node , <EOL> backref = backref ( '<STR_LIT>' , <EOL> remote_side = nodes . c . id ) ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n2 , n3 = Node ( data = '<STR_LIT>' , children = [ ] ) , Node ( data = '<STR_LIT>' , children = [ ] ) <EOL> n1 = Node ( data = '<STR_LIT>' , children = [ n2 ] ) <EOL> sess . add ( n1 ) <EOL> sess . flush ( ) <EOL> sess . delete ( n2 ) <EOL> n1 . children . append ( n3 ) <EOL> sess . flush ( ) <EOL> sess . delete ( n1 ) <EOL> sess . delete ( n3 ) <EOL> sess . flush ( ) <EOL> def test_bidirectional_multilevel_save ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( <EOL> Node , <EOL> nodes , <EOL> properties = { <EOL> '<STR_LIT>' : relationship ( <EOL> Node , <EOL> backref = backref ( <EOL> '<STR_LIT>' , <EOL> remote_side = nodes . c . id ) ) } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> n1 . children . append ( Node ( data = '<STR_LIT>' ) ) <EOL> n12 = Node ( data = '<STR_LIT>' ) <EOL> n1 . children . append ( n12 ) <EOL> n1 . children . append ( Node ( data = '<STR_LIT>' ) ) <EOL> n1 . children [ <NUM_LIT:1> ] . children . append ( Node ( data = '<STR_LIT>' ) ) <EOL> n1 . children [ <NUM_LIT:1> ] . children . append ( Node ( data = '<STR_LIT>' ) ) <EOL> n1 . children [ <NUM_LIT:1> ] . children . append ( Node ( data = '<STR_LIT>' ) ) <EOL> sess . add ( n1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : None , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n12 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n12 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n12 . id , '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> ) <EOL> def test_singlecycle_flush_size ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) <EOL> } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> sess . add ( n1 ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> sess . flush ( ) <EOL> n1 . data = '<STR_LIT>' <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> sess . flush ( ) <EOL> n2 = Node ( data = '<STR_LIT:foo>' ) <EOL> sess . add ( n2 ) <EOL> sess . flush ( ) <EOL> n1 . children . append ( n2 ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:3> ) <EOL> sess . flush ( ) <EOL> sess = create_session ( ) <EOL> n1 = sess . query ( Node ) . first ( ) <EOL> n1 . data = '<STR_LIT>' <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> n1 . children <EOL> self . _assert_uow_size ( sess , <NUM_LIT:2> ) <EOL> def test_delete_unloaded_m2o ( self ) : <EOL> Node , nodes = self . classes . Node , self . tables . nodes <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node , remote_side = nodes . c . id ) <EOL> } ) <EOL> parent = Node ( ) <EOL> c1 , c2 = Node ( parent = parent ) , Node ( parent = parent ) <EOL> session = Session ( ) <EOL> session . add_all ( [ c1 , c2 ] ) <EOL> session . add ( parent ) <EOL> session . flush ( ) <EOL> pid = parent . id <EOL> c1id = c1 . id <EOL> c2id = c2 . id <EOL> session . expire ( parent ) <EOL> session . expire ( c1 ) <EOL> session . expire ( c2 ) <EOL> session . delete ( c1 ) <EOL> session . delete ( c2 ) <EOL> session . delete ( parent ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> session . flush , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : pid } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c1id } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : c2id } <EOL> ) , <EOL> AllOf ( <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : c1id } , { '<STR_LIT:id>' : c2id } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT:id>' : pid } <EOL> ) , <EOL> ) , <EOL> ) , <EOL> ) <EOL> class SingleCyclePlusAttributeTest ( <EOL> fixtures . MappedTest , <EOL> testing . AssertsExecutionResults , <EOL> AssertsUOW ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:30> ) ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> ) <EOL> def test_flush_size ( self ) : <EOL> foobars , nodes = self . tables . foobars , self . tables . nodes <EOL> class Node ( fixtures . ComparableEntity ) : <EOL> pass <EOL> class FooBar ( fixtures . ComparableEntity ) : <EOL> pass <EOL> mapper ( Node , nodes , properties = { <EOL> '<STR_LIT>' : relationship ( Node ) , <EOL> '<STR_LIT>' : relationship ( FooBar ) <EOL> } ) <EOL> mapper ( FooBar , foobars ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> n2 = Node ( data = '<STR_LIT>' ) <EOL> n1 . children . append ( n2 ) <EOL> sess . add ( n1 ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:3> ) <EOL> n1 . foobars . append ( FooBar ( ) ) <EOL> self . _assert_uow_size ( sess , <NUM_LIT:6> ) <EOL> sess . flush ( ) <EOL> class SingleCycleM2MTest ( fixtures . MappedTest , <EOL> testing . AssertsExecutionResults , AssertsUOW ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( <EOL> '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( <EOL> '<STR_LIT:data>' , String ( <NUM_LIT:30> ) ) , Column ( <EOL> '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( <EOL> '<STR_LIT>' , Integer , <EOL> ForeignKey ( '<STR_LIT>' ) , primary_key = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , Integer , <EOL> ForeignKey ( '<STR_LIT>' ) , primary_key = True ) , <EOL> ) <EOL> def test_many_to_many_one ( self ) : <EOL> nodes , node_to_nodes = self . tables . nodes , self . tables . node_to_nodes <EOL> class Node ( fixtures . ComparableEntity ) : <EOL> pass <EOL> mapper ( <EOL> Node , <EOL> nodes , <EOL> properties = { <EOL> '<STR_LIT>' : relationship ( <EOL> Node , <EOL> secondary = node_to_nodes , <EOL> primaryjoin = nodes . c . id == node_to_nodes . c . left_node_id , <EOL> secondaryjoin = nodes . c . id == node_to_nodes . c . right_node_id , <EOL> backref = '<STR_LIT>' ) , <EOL> '<STR_LIT>' : relationship ( <EOL> Node , <EOL> remote_side = nodes . c . id ) } ) <EOL> sess = create_session ( ) <EOL> n1 = Node ( data = '<STR_LIT>' ) <EOL> n2 = Node ( data = '<STR_LIT>' ) <EOL> n3 = Node ( data = '<STR_LIT>' ) <EOL> n4 = Node ( data = '<STR_LIT>' ) <EOL> n5 = Node ( data = '<STR_LIT>' ) <EOL> n4 . favorite = n3 <EOL> n1 . favorite = n5 <EOL> n5 . favorite = n2 <EOL> n1 . children = [ n2 , n3 , n4 ] <EOL> n2 . children = [ n3 , n5 ] <EOL> n3 . children = [ n5 , n4 ] <EOL> sess . add_all ( [ n1 , n2 , n3 , n4 , n5 ] ) <EOL> sess . flush ( ) <EOL> eq_ ( <EOL> sess . query ( node_to_nodes . c . left_node_id , <EOL> node_to_nodes . c . right_node_id ) . <EOL> order_by ( node_to_nodes . c . left_node_id , <EOL> node_to_nodes . c . right_node_id ) . <EOL> all ( ) , <EOL> sorted ( [ <EOL> ( n1 . id , n2 . id ) , ( n1 . id , n3 . id ) , ( n1 . id , n4 . id ) , <EOL> ( n2 . id , n3 . id ) , ( n2 . id , n5 . id ) , <EOL> ( n3 . id , n5 . id ) , ( n3 . id , n4 . id ) <EOL> ] ) <EOL> ) <EOL> sess . delete ( n1 ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT>' : n1 . id } , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ <EOL> { '<STR_LIT>' : n2 . id , '<STR_LIT>' : n1 . id } , <EOL> { '<STR_LIT>' : n3 . id , '<STR_LIT>' : n1 . id } , <EOL> { '<STR_LIT>' : n4 . id , '<STR_LIT>' : n1 . id } <EOL> ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : { '<STR_LIT:id>' : n1 . id } <EOL> ) , <EOL> ) <EOL> for n in [ n2 , n3 , n4 , n5 ] : <EOL> sess . delete ( n ) <EOL> n4 . children <EOL> n5 . children <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ <EOL> { '<STR_LIT>' : n5 . id , '<STR_LIT>' : n3 . id } , <EOL> { '<STR_LIT>' : n4 . id , '<STR_LIT>' : n3 . id } , <EOL> { '<STR_LIT>' : n3 . id , '<STR_LIT>' : n2 . id } , <EOL> { '<STR_LIT>' : n5 . id , '<STR_LIT>' : n2 . id } <EOL> ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : n4 . id } , { '<STR_LIT:id>' : n5 . id } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> lambda ctx : [ { '<STR_LIT:id>' : n2 . id } , { '<STR_LIT:id>' : n3 . id } ] <EOL> ) , <EOL> ) <EOL> class RowswitchAccountingTest ( fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:data>' , Integer ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , ForeignKey ( '<STR_LIT>' ) , primary_key = True ) , <EOL> Column ( '<STR_LIT:data>' , Integer ) <EOL> ) <EOL> def _fixture ( self ) : <EOL> parent , child = self . tables . parent , self . tables . child <EOL> class Parent ( fixtures . BasicEntity ) : <EOL> pass <EOL> class Child ( fixtures . BasicEntity ) : <EOL> pass <EOL> mapper ( Parent , parent , properties = { <EOL> '<STR_LIT>' : relationship ( Child , uselist = False , <EOL> cascade = "<STR_LIT>" , <EOL> backref = "<STR_LIT>" ) <EOL> } ) <EOL> mapper ( Child , child ) <EOL> return Parent , Child <EOL> def test_switch_on_update ( self ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = create_session ( autocommit = False ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , child = Child ( ) ) <EOL> sess . add ( p1 ) <EOL> sess . commit ( ) <EOL> sess . close ( ) <EOL> p2 = Parent ( id = <NUM_LIT:1> , child = Child ( ) ) <EOL> p3 = sess . merge ( p2 ) <EOL> old = attributes . get_history ( p3 , '<STR_LIT>' ) [ <NUM_LIT:2> ] [ <NUM_LIT:0> ] <EOL> assert old in sess <EOL> sess . flush ( ) <EOL> assert p3 . child . _sa_instance_state . session_id == sess . hash_key <EOL> assert p3 . child in sess <EOL> p4 = Parent ( id = <NUM_LIT:1> , child = Child ( ) ) <EOL> p5 = sess . merge ( p4 ) <EOL> old = attributes . get_history ( p5 , '<STR_LIT>' ) [ <NUM_LIT:2> ] [ <NUM_LIT:0> ] <EOL> assert old in sess <EOL> sess . flush ( ) <EOL> def test_switch_on_delete ( self ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:2> , child = None ) <EOL> sess . add ( p1 ) <EOL> sess . flush ( ) <EOL> p1 . id = <NUM_LIT:5> <EOL> sess . delete ( p1 ) <EOL> eq_ ( p1 . id , <NUM_LIT:5> ) <EOL> sess . flush ( ) <EOL> eq_ ( sess . scalar ( self . tables . parent . count ( ) ) , <NUM_LIT:0> ) <EOL> class RowswitchM2OTest ( fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT:a>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> ) <EOL> Table ( <EOL> '<STR_LIT:b>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT>' , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> Table ( <EOL> '<STR_LIT:c>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> ) <EOL> def _fixture ( self ) : <EOL> a , b , c = self . tables . a , self . tables . b , self . tables . c <EOL> class A ( fixtures . BasicEntity ) : <EOL> pass <EOL> class B ( fixtures . BasicEntity ) : <EOL> pass <EOL> class C ( fixtures . BasicEntity ) : <EOL> pass <EOL> mapper ( A , a , properties = { <EOL> '<STR_LIT>' : relationship ( B , cascade = "<STR_LIT>" ) <EOL> } ) <EOL> mapper ( B , b , properties = { <EOL> '<STR_LIT:c>' : relationship ( C ) <EOL> } ) <EOL> mapper ( C , c ) <EOL> return A , B , C <EOL> def test_set_none_replaces_m2o ( self ) : <EOL> A , B , C = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> sess . add ( <EOL> A ( id = <NUM_LIT:1> , bs = [ B ( id = <NUM_LIT:1> , c = C ( id = <NUM_LIT:1> ) ) ] ) <EOL> ) <EOL> sess . commit ( ) <EOL> a1 = sess . query ( A ) . first ( ) <EOL> a1 . bs = [ B ( id = <NUM_LIT:1> , c = None ) ] <EOL> sess . commit ( ) <EOL> assert a1 . bs [ <NUM_LIT:0> ] . c is None <EOL> def test_set_none_w_get_replaces_m2o ( self ) : <EOL> A , B , C = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> sess . add ( <EOL> A ( id = <NUM_LIT:1> , bs = [ B ( id = <NUM_LIT:1> , c = C ( id = <NUM_LIT:1> ) ) ] ) <EOL> ) <EOL> sess . commit ( ) <EOL> a1 = sess . query ( A ) . first ( ) <EOL> b2 = B ( id = <NUM_LIT:1> ) <EOL> assert b2 . c is None <EOL> b2 . c = None <EOL> a1 . bs = [ b2 ] <EOL> sess . commit ( ) <EOL> assert a1 . bs [ <NUM_LIT:0> ] . c is None <EOL> def test_set_none_replaces_scalar ( self ) : <EOL> A , B , C = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> sess . add ( <EOL> A ( id = <NUM_LIT:1> , bs = [ B ( id = <NUM_LIT:1> , data = '<STR_LIT>' ) ] ) <EOL> ) <EOL> sess . commit ( ) <EOL> a1 = sess . query ( A ) . first ( ) <EOL> a1 . bs = [ B ( id = <NUM_LIT:1> , data = None ) ] <EOL> sess . commit ( ) <EOL> assert a1 . bs [ <NUM_LIT:0> ] . data is None <EOL> def test_set_none_w_get_replaces_scalar ( self ) : <EOL> A , B , C = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> sess . add ( <EOL> A ( id = <NUM_LIT:1> , bs = [ B ( id = <NUM_LIT:1> , data = '<STR_LIT>' ) ] ) <EOL> ) <EOL> sess . commit ( ) <EOL> a1 = sess . query ( A ) . first ( ) <EOL> b2 = B ( id = <NUM_LIT:1> ) <EOL> assert b2 . data is None <EOL> b2 . data = None <EOL> a1 . bs = [ b2 ] <EOL> sess . commit ( ) <EOL> assert a1 . bs [ <NUM_LIT:0> ] . data is None <EOL> class BasicStaleChecksTest ( fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:data>' , Integer ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , ForeignKey ( '<STR_LIT>' ) , primary_key = True ) , <EOL> Column ( '<STR_LIT:data>' , Integer ) <EOL> ) <EOL> def _fixture ( self , confirm_deleted_rows = True ) : <EOL> parent , child = self . tables . parent , self . tables . child <EOL> class Parent ( fixtures . BasicEntity ) : <EOL> pass <EOL> class Child ( fixtures . BasicEntity ) : <EOL> pass <EOL> mapper ( Parent , parent , properties = { <EOL> '<STR_LIT>' : relationship ( Child , uselist = False , <EOL> cascade = "<STR_LIT>" , <EOL> backref = "<STR_LIT>" ) , <EOL> } , confirm_deleted_rows = confirm_deleted_rows ) <EOL> mapper ( Child , child ) <EOL> return Parent , Child <EOL> def test_update_single_missing ( self ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:2> ) <EOL> sess . add ( p1 ) <EOL> sess . flush ( ) <EOL> sess . execute ( self . tables . parent . delete ( ) ) <EOL> p1 . data = <NUM_LIT:3> <EOL> assert_raises_message ( <EOL> orm_exc . StaleDataError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . flush <EOL> ) <EOL> def test_update_single_missing_broken_multi_rowcount ( self ) : <EOL> @ util . memoized_property <EOL> def rowcount ( self ) : <EOL> if len ( self . context . compiled_parameters ) > <NUM_LIT:1> : <EOL> return - <NUM_LIT:1> <EOL> else : <EOL> return self . context . rowcount <EOL> with patch . object ( <EOL> config . db . dialect , "<STR_LIT>" , False ) : <EOL> with patch ( <EOL> "<STR_LIT>" , <EOL> rowcount ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:2> ) <EOL> sess . add ( p1 ) <EOL> sess . flush ( ) <EOL> sess . execute ( self . tables . parent . delete ( ) ) <EOL> p1 . data = <NUM_LIT:3> <EOL> assert_raises_message ( <EOL> orm_exc . StaleDataError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . flush <EOL> ) <EOL> def test_update_multi_missing_broken_multi_rowcount ( self ) : <EOL> @ util . memoized_property <EOL> def rowcount ( self ) : <EOL> if len ( self . context . compiled_parameters ) > <NUM_LIT:1> : <EOL> return - <NUM_LIT:1> <EOL> else : <EOL> return self . context . rowcount <EOL> with patch . object ( <EOL> config . db . dialect , "<STR_LIT>" , False ) : <EOL> with patch ( <EOL> "<STR_LIT>" , <EOL> rowcount ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:2> ) <EOL> p2 = Parent ( id = <NUM_LIT:2> , data = <NUM_LIT:3> ) <EOL> sess . add_all ( [ p1 , p2 ] ) <EOL> sess . flush ( ) <EOL> sess . execute ( self . tables . parent . delete ( ) . where ( Parent . id == <NUM_LIT:1> ) ) <EOL> p1 . data = <NUM_LIT:3> <EOL> p2 . data = <NUM_LIT:4> <EOL> sess . flush ( ) <EOL> eq_ ( <EOL> sess . query ( Parent . id , Parent . data ) . all ( ) , <EOL> [ ( <NUM_LIT:2> , <NUM_LIT:4> ) ] <EOL> ) <EOL> def test_update_value_missing_broken_multi_rowcount ( self ) : <EOL> @ util . memoized_property <EOL> def rowcount ( self ) : <EOL> if len ( self . context . compiled_parameters ) > <NUM_LIT:1> : <EOL> return - <NUM_LIT:1> <EOL> else : <EOL> return self . context . rowcount <EOL> with patch . object ( <EOL> config . db . dialect , "<STR_LIT>" , False ) : <EOL> with patch ( <EOL> "<STR_LIT>" , <EOL> rowcount ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:1> ) <EOL> sess . add ( p1 ) <EOL> sess . flush ( ) <EOL> sess . execute ( self . tables . parent . delete ( ) ) <EOL> p1 . data = literal ( <NUM_LIT:1> ) <EOL> assert_raises_message ( <EOL> orm_exc . StaleDataError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . flush <EOL> ) <EOL> @ testing . requires . sane_multi_rowcount <EOL> def test_delete_multi_missing_warning ( self ) : <EOL> Parent , Child = self . _fixture ( ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:2> , child = None ) <EOL> p2 = Parent ( id = <NUM_LIT:2> , data = <NUM_LIT:3> , child = None ) <EOL> sess . add_all ( [ p1 , p2 ] ) <EOL> sess . flush ( ) <EOL> sess . execute ( self . tables . parent . delete ( ) ) <EOL> sess . delete ( p1 ) <EOL> sess . delete ( p2 ) <EOL> assert_raises_message ( <EOL> exc . SAWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> sess . flush <EOL> ) <EOL> def test_delete_multi_missing_allow ( self ) : <EOL> Parent , Child = self . _fixture ( confirm_deleted_rows = False ) <EOL> sess = Session ( ) <EOL> p1 = Parent ( id = <NUM_LIT:1> , data = <NUM_LIT:2> , child = None ) <EOL> p2 = Parent ( id = <NUM_LIT:2> , data = <NUM_LIT:3> , child = None ) <EOL> sess . add_all ( [ p1 , p2 ] ) <EOL> sess . flush ( ) <EOL> sess . execute ( self . tables . parent . delete ( ) ) <EOL> sess . delete ( p1 ) <EOL> sess . delete ( p2 ) <EOL> sess . flush ( ) <EOL> class BatchInsertsTest ( fixtures . MappedTest , testing . AssertsExecutionResults ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT:t>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:50> ) ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) , server_default = '<STR_LIT>' ) <EOL> ) <EOL> def test_batch_interaction ( self ) : <EOL> """<STR_LIT>""" <EOL> t = self . tables . t <EOL> class T ( fixtures . ComparableEntity ) : <EOL> pass <EOL> mapper ( T , t ) <EOL> sess = Session ( ) <EOL> sess . add_all ( [ <EOL> T ( data = '<STR_LIT>' ) , <EOL> T ( data = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:3> , data = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:4> , data = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:5> , data = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:6> , data = func . lower ( '<STR_LIT>' ) ) , <EOL> T ( id = <NUM_LIT:7> , data = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:8> , data = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:9> , data = '<STR_LIT>' , def_ = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:10> , data = '<STR_LIT>' , def_ = '<STR_LIT>' ) , <EOL> T ( id = <NUM_LIT:11> , data = '<STR_LIT>' ) , <EOL> ] ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> sess . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:3> } , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:4> } , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:5> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:6> } <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:7> } , { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:8> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:9> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:10> , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:11> } <EOL> ) , <EOL> ) <EOL> class LoadersUsingCommittedTest ( UOWTest ) : <EOL> """<STR_LIT>""" <EOL> def _mapper_setup ( self , passive_updates = True ) : <EOL> users , Address , addresses , User = ( self . tables . users , <EOL> self . classes . Address , <EOL> self . tables . addresses , <EOL> self . classes . User ) <EOL> mapper ( User , users , properties = { <EOL> '<STR_LIT>' : relationship ( Address , <EOL> order_by = addresses . c . email_address , <EOL> passive_updates = passive_updates , <EOL> backref = '<STR_LIT:user>' ) <EOL> } ) <EOL> mapper ( Address , addresses ) <EOL> return create_session ( autocommit = False ) <EOL> def test_before_update_m2o ( self ) : <EOL> """<STR_LIT>""" <EOL> sess = self . _mapper_setup ( ) <EOL> Address , User = self . classes . Address , self . classes . User <EOL> def before_update ( mapper , connection , target ) : <EOL> assert target . user . id == target . user_id == u2 . id <EOL> from sqlalchemy import event <EOL> event . listen ( Address , '<STR_LIT>' , before_update ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 ] ) <EOL> sess . add ( u1 ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u2 ) <EOL> sess . commit ( ) <EOL> sess . expunge_all ( ) <EOL> a1 = sess . query ( Address ) . get ( a1 . id ) <EOL> assert a1 . user_id == u1 . id <EOL> a1 . user_id = u2 . id <EOL> sess . flush ( ) <EOL> def test_before_update_o2m_passive ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_before_update_o2m ( True ) <EOL> def test_before_update_o2m_notpassive ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_before_update_o2m ( False ) <EOL> def _test_before_update_o2m ( self , passive_updates ) : <EOL> sess = self . _mapper_setup ( passive_updates = passive_updates ) <EOL> Address , User = self . classes . Address , self . classes . User <EOL> class AvoidReferencialError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def before_update ( mapper , connection , target ) : <EOL> if passive_updates : <EOL> assert '<STR_LIT>' not in target . __dict__ <EOL> eq_ ( target . addresses , [ ] ) <EOL> else : <EOL> assert '<STR_LIT>' in target . __dict__ <EOL> eq_ ( [ a . id for a in target . addresses ] , <EOL> [ a . id for a in [ a1 , a2 ] ] ) <EOL> raise AvoidReferencialError ( ) <EOL> from sqlalchemy import event <EOL> event . listen ( User , '<STR_LIT>' , before_update ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> a2 = Address ( email_address = '<STR_LIT>' ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' , addresses = [ a1 , a2 ] ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> sess . expunge_all ( ) <EOL> u1 = sess . query ( User ) . get ( u1 . id ) <EOL> u1 . id = <NUM_LIT:2> <EOL> try : <EOL> sess . flush ( ) <EOL> except AvoidReferencialError : <EOL> pass <EOL> class NoAttrEventInFlushTest ( fixtures . MappedTest ) : <EOL> """<STR_LIT>""" <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT:test>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , Integer , default = <NUM_LIT:5> ) , <EOL> Column ( '<STR_LIT>' , Integer , server_default = "<STR_LIT:5>" ) <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class Thing ( cls . Basic ) : <EOL> pass <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Thing = cls . classes . Thing <EOL> mapper ( Thing , cls . tables . test , eager_defaults = True ) <EOL> def test_no_attr_events_flush ( self ) : <EOL> Thing = self . classes . Thing <EOL> mock = Mock ( ) <EOL> event . listen ( Thing . id , "<STR_LIT>" , mock . id ) <EOL> event . listen ( Thing . prefetch_val , "<STR_LIT>" , mock . prefetch_val ) <EOL> event . listen ( Thing . returning_val , "<STR_LIT>" , mock . prefetch_val ) <EOL> t1 = Thing ( ) <EOL> s = Session ( ) <EOL> s . add ( t1 ) <EOL> s . flush ( ) <EOL> eq_ ( len ( mock . mock_calls ) , <NUM_LIT:0> ) <EOL> eq_ ( t1 . id , <NUM_LIT:1> ) <EOL> eq_ ( t1 . prefetch_val , <NUM_LIT:5> ) <EOL> eq_ ( t1 . returning_val , <NUM_LIT:5> ) <EOL> class EagerDefaultsTest ( fixtures . MappedTest ) : <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT:test>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:foo>' , Integer , server_default = "<STR_LIT:3>" ) <EOL> ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:foo>' , Integer ) , <EOL> Column ( '<STR_LIT:bar>' , Integer , server_onupdate = FetchedValue ( ) ) <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class Thing ( cls . Basic ) : <EOL> pass <EOL> class Thing2 ( cls . Basic ) : <EOL> pass <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Thing = cls . classes . Thing <EOL> mapper ( Thing , cls . tables . test , eager_defaults = True ) <EOL> Thing2 = cls . classes . Thing2 <EOL> mapper ( Thing2 , cls . tables . test2 , eager_defaults = True ) <EOL> def test_insert_defaults_present ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> t1 , t2 = ( <EOL> Thing ( id = <NUM_LIT:1> , foo = <NUM_LIT:5> ) , <EOL> Thing ( id = <NUM_LIT:2> , foo = <NUM_LIT:10> ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 ] ) <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:5> , '<STR_LIT:id>' : <NUM_LIT:1> } , { '<STR_LIT:foo>' : <NUM_LIT:10> , '<STR_LIT:id>' : <NUM_LIT:2> } ] <EOL> ) , <EOL> ) <EOL> def go ( ) : <EOL> eq_ ( t1 . foo , <NUM_LIT:5> ) <EOL> eq_ ( t2 . foo , <NUM_LIT:10> ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_insert_defaults_present_as_expr ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> t1 , t2 = ( <EOL> Thing ( id = <NUM_LIT:1> , foo = text ( "<STR_LIT>" ) ) , <EOL> Thing ( id = <NUM_LIT:2> , foo = text ( "<STR_LIT>" ) ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 ] ) <EOL> if testing . db . dialect . implicit_returning : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:2> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) <EOL> ) <EOL> else : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:2> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:2> } ] <EOL> ) , <EOL> ) <EOL> def go ( ) : <EOL> eq_ ( t1 . foo , <NUM_LIT:7> ) <EOL> eq_ ( t2 . foo , <NUM_LIT:10> ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_insert_defaults_nonpresent ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> t1 , t2 = ( <EOL> Thing ( id = <NUM_LIT:1> ) , <EOL> Thing ( id = <NUM_LIT:2> ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 ] ) <EOL> if testing . db . dialect . implicit_returning : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . commit , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:2> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> ) <EOL> else : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . commit , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } , { '<STR_LIT:id>' : <NUM_LIT:2> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:2> } ] <EOL> ) <EOL> ) <EOL> def test_update_defaults_nonpresent ( self ) : <EOL> Thing2 = self . classes . Thing2 <EOL> s = Session ( ) <EOL> t1 , t2 , t3 , t4 = ( <EOL> Thing2 ( id = <NUM_LIT:1> , foo = <NUM_LIT:1> , bar = <NUM_LIT:2> ) , <EOL> Thing2 ( id = <NUM_LIT:2> , foo = <NUM_LIT:2> , bar = <NUM_LIT:3> ) , <EOL> Thing2 ( id = <NUM_LIT:3> , foo = <NUM_LIT:3> , bar = <NUM_LIT:4> ) , <EOL> Thing2 ( id = <NUM_LIT:4> , foo = <NUM_LIT:4> , bar = <NUM_LIT:5> ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 , t3 , t4 ] ) <EOL> s . flush ( ) <EOL> t1 . foo = <NUM_LIT:5> <EOL> t2 . foo = <NUM_LIT:6> <EOL> t2 . bar = <NUM_LIT:10> <EOL> t3 . foo = <NUM_LIT:7> <EOL> t4 . foo = <NUM_LIT:8> <EOL> t4 . bar = <NUM_LIT:12> <EOL> if testing . db . dialect . implicit_returning : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:1> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:6> , '<STR_LIT:bar>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:2> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:7> , '<STR_LIT>' : <NUM_LIT:3> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:8> , '<STR_LIT:bar>' : <NUM_LIT:12> , '<STR_LIT>' : <NUM_LIT:4> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> ) <EOL> else : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:6> , '<STR_LIT:bar>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:2> } ] , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:7> , '<STR_LIT>' : <NUM_LIT:3> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:8> , '<STR_LIT:bar>' : <NUM_LIT:12> , '<STR_LIT>' : <NUM_LIT:4> } ] , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:3> } ] <EOL> ) <EOL> ) <EOL> def go ( ) : <EOL> eq_ ( t1 . bar , <NUM_LIT:2> ) <EOL> eq_ ( t2 . bar , <NUM_LIT:10> ) <EOL> eq_ ( t3 . bar , <NUM_LIT:4> ) <EOL> eq_ ( t4 . bar , <NUM_LIT:12> ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_update_defaults_present_as_expr ( self ) : <EOL> Thing2 = self . classes . Thing2 <EOL> s = Session ( ) <EOL> t1 , t2 , t3 , t4 = ( <EOL> Thing2 ( id = <NUM_LIT:1> , foo = <NUM_LIT:1> , bar = <NUM_LIT:2> ) , <EOL> Thing2 ( id = <NUM_LIT:2> , foo = <NUM_LIT:2> , bar = <NUM_LIT:3> ) , <EOL> Thing2 ( id = <NUM_LIT:3> , foo = <NUM_LIT:3> , bar = <NUM_LIT:4> ) , <EOL> Thing2 ( id = <NUM_LIT:4> , foo = <NUM_LIT:4> , bar = <NUM_LIT:5> ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 , t3 , t4 ] ) <EOL> s . flush ( ) <EOL> t1 . foo = <NUM_LIT:5> <EOL> t1 . bar = text ( "<STR_LIT>" ) <EOL> t2 . foo = <NUM_LIT:6> <EOL> t2 . bar = <NUM_LIT:10> <EOL> t3 . foo = <NUM_LIT:7> <EOL> t4 . foo = <NUM_LIT:8> <EOL> t4 . bar = text ( "<STR_LIT>" ) <EOL> if testing . db . dialect . implicit_returning : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:1> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:6> , '<STR_LIT:bar>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:2> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:7> , '<STR_LIT>' : <NUM_LIT:3> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:8> , '<STR_LIT>' : <NUM_LIT:4> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) , <EOL> ) <EOL> else : <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . flush , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:6> , '<STR_LIT:bar>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:2> } ] , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:7> , '<STR_LIT>' : <NUM_LIT:3> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:8> , '<STR_LIT>' : <NUM_LIT:4> } ] , <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:3> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:4> } ] <EOL> ) <EOL> ) <EOL> def go ( ) : <EOL> eq_ ( t1 . bar , <NUM_LIT:2> ) <EOL> eq_ ( t2 . bar , <NUM_LIT:10> ) <EOL> eq_ ( t3 . bar , <NUM_LIT:4> ) <EOL> eq_ ( t4 . bar , <NUM_LIT:12> ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_insert_defaults_bulk_insert ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> mappings = [ <EOL> { "<STR_LIT:id>" : <NUM_LIT:1> } , <EOL> { "<STR_LIT:id>" : <NUM_LIT:2> } <EOL> ] <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> lambda : s . bulk_insert_mappings ( Thing , mappings ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:id>' : <NUM_LIT:1> } , { '<STR_LIT:id>' : <NUM_LIT:2> } ] <EOL> ) <EOL> ) <EOL> def test_update_defaults_bulk_update ( self ) : <EOL> Thing2 = self . classes . Thing2 <EOL> s = Session ( ) <EOL> t1 , t2 , t3 , t4 = ( <EOL> Thing2 ( id = <NUM_LIT:1> , foo = <NUM_LIT:1> , bar = <NUM_LIT:2> ) , <EOL> Thing2 ( id = <NUM_LIT:2> , foo = <NUM_LIT:2> , bar = <NUM_LIT:3> ) , <EOL> Thing2 ( id = <NUM_LIT:3> , foo = <NUM_LIT:3> , bar = <NUM_LIT:4> ) , <EOL> Thing2 ( id = <NUM_LIT:4> , foo = <NUM_LIT:4> , bar = <NUM_LIT:5> ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 , t3 , t4 ] ) <EOL> s . flush ( ) <EOL> mappings = [ <EOL> { "<STR_LIT:id>" : <NUM_LIT:1> , "<STR_LIT:foo>" : <NUM_LIT:5> } , <EOL> { "<STR_LIT:id>" : <NUM_LIT:2> , "<STR_LIT:foo>" : <NUM_LIT:6> , "<STR_LIT:bar>" : <NUM_LIT:10> } , <EOL> { "<STR_LIT:id>" : <NUM_LIT:3> , "<STR_LIT:foo>" : <NUM_LIT:7> } , <EOL> { "<STR_LIT:id>" : <NUM_LIT:4> , "<STR_LIT:foo>" : <NUM_LIT:8> } <EOL> ] <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> lambda : s . bulk_update_mappings ( Thing2 , mappings ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:1> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:6> , '<STR_LIT:bar>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:2> } ] <EOL> ) , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:foo>' : <NUM_LIT:7> , '<STR_LIT>' : <NUM_LIT:3> } , { '<STR_LIT:foo>' : <NUM_LIT:8> , '<STR_LIT>' : <NUM_LIT:4> } ] <EOL> ) <EOL> ) <EOL> def test_update_defaults_present ( self ) : <EOL> Thing2 = self . classes . Thing2 <EOL> s = Session ( ) <EOL> t1 , t2 = ( <EOL> Thing2 ( id = <NUM_LIT:1> , foo = <NUM_LIT:1> , bar = <NUM_LIT:2> ) , <EOL> Thing2 ( id = <NUM_LIT:2> , foo = <NUM_LIT:2> , bar = <NUM_LIT:3> ) <EOL> ) <EOL> s . add_all ( [ t1 , t2 ] ) <EOL> s . flush ( ) <EOL> t1 . bar = <NUM_LIT:5> <EOL> t2 . bar = <NUM_LIT:10> <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . commit , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT:bar>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:1> } , { '<STR_LIT:bar>' : <NUM_LIT:10> , '<STR_LIT>' : <NUM_LIT:2> } ] , <EOL> dialect = '<STR_LIT>' <EOL> ) <EOL> ) <EOL> class TypeWoBoolTest ( fixtures . MappedTest , testing . AssertsExecutionResults ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> from sqlalchemy import TypeDecorator <EOL> class NoBool ( object ) : <EOL> def __nonzero__ ( self ) : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> class MyWidget ( object ) : <EOL> def __init__ ( self , text ) : <EOL> self . text = text <EOL> def __eq__ ( self , other ) : <EOL> return NoBool ( ) <EOL> cls . MyWidget = MyWidget <EOL> class MyType ( TypeDecorator ) : <EOL> impl = String ( <NUM_LIT:50> ) <EOL> def process_bind_param ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = value . text <EOL> return value <EOL> def process_result_value ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = MyWidget ( value ) <EOL> return value <EOL> Table ( <EOL> '<STR_LIT:test>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:value>' , MyType ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class Thing ( cls . Basic ) : <EOL> pass <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Thing = cls . classes . Thing <EOL> mapper ( Thing , cls . tables . test ) <EOL> def test_update_against_none ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> s . add ( Thing ( value = self . MyWidget ( "<STR_LIT:foo>" ) ) ) <EOL> s . commit ( ) <EOL> t1 = s . query ( Thing ) . first ( ) <EOL> t1 . value = None <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( Thing . value ) . scalar ( ) , None <EOL> ) <EOL> def test_update_against_something_else ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> s . add ( Thing ( value = self . MyWidget ( "<STR_LIT:foo>" ) ) ) <EOL> s . commit ( ) <EOL> t1 = s . query ( Thing ) . first ( ) <EOL> t1 . value = self . MyWidget ( "<STR_LIT:bar>" ) <EOL> s . commit ( ) <EOL> eq_ ( <EOL> s . query ( Thing . value ) . scalar ( ) . text , "<STR_LIT:bar>" <EOL> ) <EOL> def test_no_update_no_change ( self ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> s . add ( Thing ( value = self . MyWidget ( "<STR_LIT:foo>" ) , unrelated = '<STR_LIT>' ) ) <EOL> s . commit ( ) <EOL> t1 = s . query ( Thing ) . first ( ) <EOL> t1 . unrelated = '<STR_LIT>' <EOL> self . assert_sql_execution ( <EOL> testing . db , <EOL> s . commit , <EOL> CompiledSQL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> [ { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : '<STR_LIT>' } ] <EOL> ) , <EOL> ) <EOL> eq_ ( <EOL> s . query ( Thing . value ) . scalar ( ) . text , "<STR_LIT:foo>" <EOL> ) <EOL> class NullEvaluatingTest ( fixtures . MappedTest , testing . AssertsExecutionResults ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> from sqlalchemy import TypeDecorator <EOL> class EvalsNull ( TypeDecorator ) : <EOL> impl = String ( <NUM_LIT:50> ) <EOL> should_evaluate_none = True <EOL> def process_bind_param ( self , value , dialect ) : <EOL> if value is None : <EOL> value = '<STR_LIT>' <EOL> return value <EOL> Table ( <EOL> '<STR_LIT:test>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , EvalsNull ( ) ) , <EOL> Column ( '<STR_LIT>' , EvalsNull ( ) , default = '<STR_LIT>' ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) , default = '<STR_LIT>' ) , <EOL> Column ( <EOL> '<STR_LIT>' , String ( <NUM_LIT:50> ) . evaluates_none ( ) ) , <EOL> Column ( <EOL> '<STR_LIT>' , <EOL> String ( <NUM_LIT:50> ) . evaluates_none ( ) , default = '<STR_LIT>' ) , <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class Thing ( cls . Basic ) : <EOL> pass <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Thing = cls . classes . Thing <EOL> mapper ( Thing , cls . tables . test ) <EOL> def _assert_col ( self , name , value ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> col = getattr ( Thing , name ) <EOL> obj = s . query ( col ) . filter ( col == value ) . one ( ) <EOL> eq_ ( obj [ <NUM_LIT:0> ] , value ) <EOL> def _test_insert ( self , attr , expected ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> t1 = Thing ( ** { attr : None } ) <EOL> s . add ( t1 ) <EOL> s . commit ( ) <EOL> self . _assert_col ( attr , expected ) <EOL> def _test_bulk_insert ( self , attr , expected ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> s . bulk_insert_mappings ( <EOL> Thing , [ { attr : None } ] <EOL> ) <EOL> s . commit ( ) <EOL> self . _assert_col ( attr , expected ) <EOL> def _test_insert_novalue ( self , attr , expected ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> t1 = Thing ( ) <EOL> s . add ( t1 ) <EOL> s . commit ( ) <EOL> self . _assert_col ( attr , expected ) <EOL> def _test_bulk_insert_novalue ( self , attr , expected ) : <EOL> Thing = self . classes . Thing <EOL> s = Session ( ) <EOL> s . bulk_insert_mappings ( <EOL> Thing , [ { } ] <EOL> ) <EOL> s . commit ( ) <EOL> self . _assert_col ( attr , expected ) <EOL> def test_evalnull_nodefault_insert ( self ) : <EOL> self . _test_insert ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_evalnull_nodefault_bulk_insert ( self ) : <EOL> self . _test_bulk_insert ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_evalnull_nodefault_insert_novalue ( self ) : <EOL> self . _test_insert_novalue ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_evalnull_nodefault_bulk_insert_novalue ( self ) : <EOL> self . _test_bulk_insert_novalue ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_evalnull_default_insert ( self ) : <EOL> self . _test_insert ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_evalnull_default_bulk_insert ( self ) : <EOL> self . _test_bulk_insert ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_evalnull_default_insert_novalue ( self ) : <EOL> self . _test_insert_novalue ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_evalnull_default_bulk_insert_novalue ( self ) : <EOL> self . _test_bulk_insert_novalue ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_no_evalnull_nodefault_insert ( self ) : <EOL> self . _test_insert ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_no_evalnull_nodefault_bulk_insert ( self ) : <EOL> self . _test_bulk_insert ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_no_evalnull_nodefault_insert_novalue ( self ) : <EOL> self . _test_insert_novalue ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_no_evalnull_nodefault_bulk_insert_novalue ( self ) : <EOL> self . _test_bulk_insert_novalue ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_no_evalnull_default_insert ( self ) : <EOL> self . _test_insert ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_no_evalnull_default_bulk_insert ( self ) : <EOL> self . _test_bulk_insert ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_no_evalnull_default_insert_novalue ( self ) : <EOL> self . _test_insert_novalue ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_no_evalnull_default_bulk_insert_novalue ( self ) : <EOL> self . _test_bulk_insert_novalue ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_builtin_evalnull_nodefault_insert ( self ) : <EOL> self . _test_insert ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_builtin_evalnull_nodefault_bulk_insert ( self ) : <EOL> self . _test_bulk_insert ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_builtin_evalnull_nodefault_insert_novalue ( self ) : <EOL> self . _test_insert_novalue ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_builtin_evalnull_nodefault_bulk_insert_novalue ( self ) : <EOL> self . _test_bulk_insert_novalue ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_builtin_evalnull_default_insert ( self ) : <EOL> self . _test_insert ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_builtin_evalnull_default_bulk_insert ( self ) : <EOL> self . _test_bulk_insert ( <EOL> "<STR_LIT>" , None <EOL> ) <EOL> def test_builtin_evalnull_default_insert_novalue ( self ) : <EOL> self . _test_insert_novalue ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) <EOL> def test_builtin_evalnull_default_bulk_insert_novalue ( self ) : <EOL> self . _test_bulk_insert_novalue ( <EOL> "<STR_LIT>" , '<STR_LIT>' <EOL> ) </s>
