<s> """<STR_LIT>""" <EOL> from sqlalchemy . testing import assert_raises , assert_raises_message <EOL> from sqlalchemy import Integer , String , ForeignKey , Sequence , exc as sa_exc <EOL> from sqlalchemy . testing . schema import Table <EOL> from sqlalchemy . testing . schema import Column <EOL> from sqlalchemy . orm import mapper , relationship , create_session , class_mapper , backref , sessionmaker , Session <EOL> from sqlalchemy . orm import attributes , exc as orm_exc <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import eq_ <EOL> from sqlalchemy . testing import fixtures <EOL> from test . orm import _fixtures <EOL> class O2MCollectionTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Address , addresses , users , User = ( cls . classes . Address , <EOL> cls . tables . addresses , <EOL> cls . tables . users , <EOL> cls . classes . User ) <EOL> mapper ( Address , addresses ) <EOL> mapper ( User , users , properties = dict ( <EOL> addresses = relationship ( Address , backref = "<STR_LIT:user>" ) , <EOL> ) ) <EOL> def test_collection_move_hitslazy ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> a2 = Address ( email_address = "<STR_LIT>" ) <EOL> a3 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 , a2 , a3 ] ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , a1 , a2 , a3 ] ) <EOL> sess . commit ( ) <EOL> def go ( ) : <EOL> u2 . addresses . append ( a1 ) <EOL> u2 . addresses . append ( a2 ) <EOL> u2 . addresses . append ( a3 ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> def test_collection_move_preloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 ] ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u1 . addresses <EOL> u2 . addresses . append ( a1 ) <EOL> assert a1 . user is u2 <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> def test_collection_move_notloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 ] ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u2 . addresses . append ( a1 ) <EOL> assert a1 . user is u2 <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> def test_collection_move_commitfirst ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , addresses = [ a1 ] ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u1 . addresses <EOL> u2 . addresses . append ( a1 ) <EOL> assert a1 . user is u2 <EOL> sess . commit ( ) <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> def test_scalar_move_preloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> a1 . user = u1 <EOL> sess . add_all ( [ u1 , u2 , a1 ] ) <EOL> sess . commit ( ) <EOL> u1 . addresses <EOL> a1 . user = u2 <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> def test_plain_load_passive ( self ) : <EOL> """<STR_LIT>""" <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> a1 . user = u1 <EOL> sess . add_all ( [ u1 , u2 , a1 ] ) <EOL> sess . commit ( ) <EOL> def go ( ) : <EOL> a1 . user = u2 <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> def test_set_none ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> a1 . user = u1 <EOL> sess . add_all ( [ u1 , a1 ] ) <EOL> sess . commit ( ) <EOL> def go ( ) : <EOL> a1 . user = None <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:0> ) <EOL> assert a1 not in u1 . addresses <EOL> def test_scalar_move_notloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> a1 . user = u1 <EOL> sess . add_all ( [ u1 , u2 , a1 ] ) <EOL> sess . commit ( ) <EOL> a1 . user = u2 <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> def test_scalar_move_commitfirst ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> u1 = User ( name = '<STR_LIT>' ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> a1 = Address ( email_address = '<STR_LIT>' ) <EOL> a1 . user = u1 <EOL> sess . add_all ( [ u1 , u2 , a1 ] ) <EOL> sess . commit ( ) <EOL> u1 . addresses <EOL> a1 . user = u2 <EOL> sess . commit ( ) <EOL> assert a1 not in u1 . addresses <EOL> assert a1 in u2 . addresses <EOL> class O2OScalarBackrefMoveTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Address , addresses , users , User = ( cls . classes . Address , <EOL> cls . tables . addresses , <EOL> cls . tables . users , <EOL> cls . classes . User ) <EOL> mapper ( Address , addresses ) <EOL> mapper ( User , users , properties = { <EOL> '<STR_LIT:address>' : relationship ( Address , backref = backref ( "<STR_LIT:user>" ) , uselist = False ) <EOL> } ) <EOL> def test_collection_move_preloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u1 . address <EOL> u2 . address = a1 <EOL> assert u2 . address is a1 <EOL> assert a1 . user is u2 <EOL> assert u1 . address is a1 <EOL> assert u2 . address is a1 <EOL> def test_scalar_move_preloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> a2 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> sess . add_all ( [ u1 , a1 , a2 ] ) <EOL> sess . commit ( ) <EOL> a1 . user <EOL> a2 . user = u1 <EOL> assert u1 . address is a2 <EOL> assert a1 . user is u1 <EOL> assert a2 . user is u1 <EOL> def test_collection_move_notloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u2 . address = a1 <EOL> assert u2 . address is a1 <EOL> assert a1 . user is u2 <EOL> assert u1 . address is None <EOL> assert u2 . address is a1 <EOL> def test_scalar_move_notloaded ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> a2 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> sess . add_all ( [ u1 , a1 , a2 ] ) <EOL> sess . commit ( ) <EOL> a2 . user = u1 <EOL> assert u1 . address is a2 <EOL> assert a1 . user is u1 <EOL> assert a2 . user is u1 <EOL> def test_collection_move_commitfirst ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u1 . address <EOL> u2 . address = a1 <EOL> assert u2 . address is a1 <EOL> assert a1 . user is u2 <EOL> sess . commit ( ) <EOL> assert u1 . address is None <EOL> assert u2 . address is a1 <EOL> def test_scalar_move_commitfirst ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> a2 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> sess . add_all ( [ u1 , a1 , a2 ] ) <EOL> sess . commit ( ) <EOL> assert a1 . user is u1 <EOL> a2 . user = u1 <EOL> assert u1 . address is a2 <EOL> assert a1 . user is u1 <EOL> sess . commit ( ) <EOL> assert u1 . address is a2 <EOL> assert a1 . user is None <EOL> assert a2 . user is u1 <EOL> class O2OScalarMoveTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Address , addresses , users , User = ( cls . classes . Address , <EOL> cls . tables . addresses , <EOL> cls . tables . users , <EOL> cls . classes . User ) <EOL> mapper ( Address , addresses ) <EOL> mapper ( User , users , properties = { <EOL> '<STR_LIT:address>' : relationship ( Address , uselist = False ) <EOL> } ) <EOL> def test_collection_move_commitfirst ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> sess . add_all ( [ u1 , u2 ] ) <EOL> sess . commit ( ) <EOL> u1 . address <EOL> u2 . address = a1 <EOL> assert u2 . address is a1 <EOL> sess . commit ( ) <EOL> assert u1 . address is None <EOL> assert u2 . address is a1 <EOL> class O2OScalarOrphanTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Address , addresses , users , User = ( cls . classes . Address , <EOL> cls . tables . addresses , <EOL> cls . tables . users , <EOL> cls . classes . User ) <EOL> mapper ( Address , addresses ) <EOL> mapper ( User , users , properties = { <EOL> '<STR_LIT:address>' : relationship ( Address , uselist = False , <EOL> backref = backref ( '<STR_LIT:user>' , single_parent = True , <EOL> cascade = "<STR_LIT>" ) ) <EOL> } ) <EOL> def test_m2o_event ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> sess = sessionmaker ( ) ( ) <EOL> a1 = Address ( email_address = "<STR_LIT>" ) <EOL> u1 = User ( name = '<STR_LIT>' , address = a1 ) <EOL> sess . add ( u1 ) <EOL> sess . commit ( ) <EOL> sess . expunge ( u1 ) <EOL> u2 = User ( name = '<STR_LIT>' ) <EOL> u2 . address = a1 <EOL> sess . commit ( ) <EOL> assert sess . query ( User ) . count ( ) == <NUM_LIT:1> <EOL> class M2MCollectionMoveTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> keywords , items , item_keywords , Keyword , Item = ( cls . tables . keywords , <EOL> cls . tables . items , <EOL> cls . tables . item_keywords , <EOL> cls . classes . Keyword , <EOL> cls . classes . Item ) <EOL> mapper ( Item , items , properties = { <EOL> '<STR_LIT>' : relationship ( Keyword , secondary = item_keywords , <EOL> backref = '<STR_LIT>' ) <EOL> } ) <EOL> mapper ( Keyword , keywords ) <EOL> def test_add_remove_pending_backref ( self ) : <EOL> """<STR_LIT>""" <EOL> Item , Keyword = ( self . classes . Item , self . classes . Keyword ) <EOL> session = Session ( autoflush = False ) <EOL> i1 = Item ( description = '<STR_LIT>' ) <EOL> session . add ( i1 ) <EOL> session . commit ( ) <EOL> session . expire ( i1 , [ '<STR_LIT>' ] ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> k1 . items . append ( i1 ) <EOL> k1 . items . remove ( i1 ) <EOL> eq_ ( i1 . keywords , [ ] ) <EOL> def test_remove_add_pending_backref ( self ) : <EOL> """<STR_LIT>""" <EOL> Item , Keyword = ( self . classes . Item , self . classes . Keyword ) <EOL> session = Session ( autoflush = False ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keywords = [ k1 ] ) <EOL> session . add ( i1 ) <EOL> session . commit ( ) <EOL> session . expire ( i1 , [ '<STR_LIT>' ] ) <EOL> k1 . items . remove ( i1 ) <EOL> k1 . items . append ( i1 ) <EOL> eq_ ( i1 . keywords , [ k1 ] ) <EOL> def test_pending_combines_with_flushed ( self ) : <EOL> """<STR_LIT>""" <EOL> Item , Keyword = ( self . classes . Item , self . classes . Keyword ) <EOL> session = Session ( testing . db , autoflush = False ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> k2 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keywords = [ k1 ] ) <EOL> session . add ( i1 ) <EOL> session . add ( k2 ) <EOL> session . commit ( ) <EOL> k2 . items . append ( i1 ) <EOL> eq_ ( <EOL> set ( attributes . instance_state ( i1 ) . <EOL> _pending_mutations [ '<STR_LIT>' ] . added_items ) , <EOL> set ( [ k2 ] ) <EOL> ) <EOL> eq_ ( i1 . keywords , [ k1 , k2 ] ) <EOL> eq_ ( session . scalar ( "<STR_LIT>" ) , <NUM_LIT:1> ) <EOL> assert '<STR_LIT>' not in attributes . instance_state ( i1 ) . _pending_mutations <EOL> def test_duplicate_adds ( self ) : <EOL> Item , Keyword = ( self . classes . Item , self . classes . Keyword ) <EOL> session = Session ( testing . db , autoflush = False ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keywords = [ k1 ] ) <EOL> session . add ( i1 ) <EOL> session . commit ( ) <EOL> k1 . items . append ( i1 ) <EOL> eq_ ( i1 . keywords , [ k1 , k1 ] ) <EOL> session . expire ( i1 , [ '<STR_LIT>' ] ) <EOL> k1 . items . append ( i1 ) <EOL> eq_ ( i1 . keywords , [ k1 , k1 ] ) <EOL> session . expire ( i1 , [ '<STR_LIT>' ] ) <EOL> k1 . items . append ( i1 ) <EOL> eq_ ( i1 . keywords , [ k1 , k1 ] ) <EOL> eq_ ( k1 . items , [ i1 , i1 , i1 , i1 ] ) <EOL> session . commit ( ) <EOL> eq_ ( k1 . items , [ i1 ] ) <EOL> class M2MScalarMoveTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> keywords , items , item_keywords , Keyword , Item = ( cls . tables . keywords , <EOL> cls . tables . items , <EOL> cls . tables . item_keywords , <EOL> cls . classes . Keyword , <EOL> cls . classes . Item ) <EOL> mapper ( Item , items , properties = { <EOL> '<STR_LIT>' : relationship ( Keyword , secondary = item_keywords , <EOL> uselist = False , <EOL> backref = backref ( "<STR_LIT>" , uselist = False ) ) <EOL> } ) <EOL> mapper ( Keyword , keywords ) <EOL> def test_collection_move_preloaded ( self ) : <EOL> Item , Keyword = self . classes . Item , self . classes . Keyword <EOL> sess = sessionmaker ( ) ( ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keyword = k1 ) <EOL> i2 = Item ( description = '<STR_LIT>' ) <EOL> sess . add_all ( [ i1 , i2 , k1 ] ) <EOL> sess . commit ( ) <EOL> assert i1 . keyword is k1 <EOL> i2 . keyword = k1 <EOL> assert k1 . item is i2 <EOL> assert i1 . keyword is k1 <EOL> assert i2 . keyword is k1 <EOL> def test_collection_move_notloaded ( self ) : <EOL> Item , Keyword = self . classes . Item , self . classes . Keyword <EOL> sess = sessionmaker ( ) ( ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keyword = k1 ) <EOL> i2 = Item ( description = '<STR_LIT>' ) <EOL> sess . add_all ( [ i1 , i2 , k1 ] ) <EOL> sess . commit ( ) <EOL> i2 . keyword = k1 <EOL> assert k1 . item is i2 <EOL> assert i1 . keyword is None <EOL> assert i2 . keyword is k1 <EOL> def test_collection_move_commit ( self ) : <EOL> Item , Keyword = self . classes . Item , self . classes . Keyword <EOL> sess = sessionmaker ( ) ( ) <EOL> k1 = Keyword ( name = '<STR_LIT>' ) <EOL> i1 = Item ( description = '<STR_LIT>' , keyword = k1 ) <EOL> i2 = Item ( description = '<STR_LIT>' ) <EOL> sess . add_all ( [ i1 , i2 , k1 ] ) <EOL> sess . commit ( ) <EOL> assert i1 . keyword is k1 <EOL> i2 . keyword = k1 <EOL> assert k1 . item is i2 <EOL> sess . commit ( ) <EOL> assert i1 . keyword is None <EOL> assert i2 . keyword is k1 <EOL> class O2MStaleBackrefTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> Address , addresses , users , User = ( cls . classes . Address , <EOL> cls . tables . addresses , <EOL> cls . tables . users , <EOL> cls . classes . User ) <EOL> mapper ( Address , addresses ) <EOL> mapper ( User , users , properties = dict ( <EOL> addresses = relationship ( Address , backref = "<STR_LIT:user>" ) , <EOL> ) ) <EOL> def test_backref_pop_m2o ( self ) : <EOL> User , Address = self . classes . User , self . classes . Address <EOL> u1 = User ( ) <EOL> u2 = User ( ) <EOL> a1 = Address ( ) <EOL> u1 . addresses . append ( a1 ) <EOL> u2 . addresses . append ( a1 ) <EOL> assert a1 not in u1 . addresses <EOL> assert a1 . user is u2 <EOL> assert a1 in u2 . addresses <EOL> class M2MStaleBackrefTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> keywords , items , item_keywords , Keyword , Item = ( cls . tables . keywords , <EOL> cls . tables . items , <EOL> cls . tables . item_keywords , <EOL> cls . classes . Keyword , <EOL> cls . classes . Item ) <EOL> mapper ( Item , items , properties = { <EOL> '<STR_LIT>' : relationship ( Keyword , secondary = item_keywords , <EOL> backref = '<STR_LIT>' ) <EOL> } ) <EOL> mapper ( Keyword , keywords ) <EOL> def test_backref_pop_m2m ( self ) : <EOL> Keyword , Item = self . classes . Keyword , self . classes . Item <EOL> k1 = Keyword ( ) <EOL> k2 = Keyword ( ) <EOL> i1 = Item ( ) <EOL> k1 . items . append ( i1 ) <EOL> k2 . items . append ( i1 ) <EOL> k2 . items . append ( i1 ) <EOL> i1 . keywords = [ ] <EOL> k2 . items . remove ( i1 ) <EOL> assert len ( k2 . items ) == <NUM_LIT:0> </s>
