<s> from sqlalchemy . testing import assert_raises , assert_raises_message , eq_ <EOL> import sqlalchemy as sa <EOL> from sqlalchemy import testing <EOL> from sqlalchemy import Integer , String , ForeignKey <EOL> from sqlalchemy . testing . schema import Table <EOL> from sqlalchemy . testing . schema import Column <EOL> from sqlalchemy . orm import mapper , relationship , Session , exc as orm_exc , sessionmaker , backref <EOL> from sqlalchemy . testing import fixtures <EOL> class M2MTest ( fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , test_needs_autoincrement = True , <EOL> primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( <NUM_LIT:30> ) , nullable = False ) , <EOL> test_needs_acid = True , <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , <EOL> test_needs_autoincrement = True , primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( <NUM_LIT:30> ) , nullable = False ) , <EOL> test_needs_acid = True , <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , test_needs_autoincrement = True , <EOL> primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) , <EOL> nullable = False ) , <EOL> Column ( '<STR_LIT:name>' , String ( <NUM_LIT:30> ) , nullable = False ) , <EOL> test_needs_acid = True , <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT>' , Integer , <EOL> ForeignKey ( '<STR_LIT>' ) ) , <EOL> test_needs_acid = True , <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT>' , Integer , <EOL> ForeignKey ( '<STR_LIT>' ) ) , <EOL> test_needs_acid = True , <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> test_needs_acid = True , <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class Place ( cls . Basic ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> class PlaceThingy ( cls . Basic ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> class Transition ( cls . Basic ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> def test_overlapping_attribute_error ( self ) : <EOL> place , Transition , place_input , Place , transition = ( self . tables . place , <EOL> self . classes . Transition , <EOL> self . tables . place_input , <EOL> self . classes . Place , <EOL> self . tables . transition ) <EOL> mapper ( Place , place , properties = { <EOL> '<STR_LIT>' : relationship ( Transition , <EOL> secondary = place_input , backref = '<STR_LIT>' ) <EOL> } ) <EOL> mapper ( Transition , transition , properties = { <EOL> '<STR_LIT>' : relationship ( Place , <EOL> secondary = place_input , backref = '<STR_LIT>' ) <EOL> } ) <EOL> assert_raises_message ( sa . exc . ArgumentError , <EOL> "<STR_LIT>" , <EOL> sa . orm . configure_mappers ) <EOL> def test_self_referential_roundtrip ( self ) : <EOL> place , Place , place_place = ( self . tables . place , <EOL> self . classes . Place , <EOL> self . tables . place_place ) <EOL> mapper ( Place , place , properties = { <EOL> '<STR_LIT>' : relationship ( <EOL> Place , <EOL> secondary = place_place , <EOL> primaryjoin = place . c . place_id == place_place . c . pl1_id , <EOL> secondaryjoin = place . c . place_id == place_place . c . pl2_id , <EOL> order_by = place_place . c . pl2_id <EOL> ) <EOL> } ) <EOL> sess = Session ( ) <EOL> p1 = Place ( '<STR_LIT>' ) <EOL> p2 = Place ( '<STR_LIT>' ) <EOL> p3 = Place ( '<STR_LIT>' ) <EOL> p4 = Place ( '<STR_LIT>' ) <EOL> p5 = Place ( '<STR_LIT>' ) <EOL> p6 = Place ( '<STR_LIT>' ) <EOL> p7 = Place ( '<STR_LIT>' ) <EOL> sess . add_all ( ( p1 , p2 , p3 , p4 , p5 , p6 , p7 ) ) <EOL> p1 . places . append ( p2 ) <EOL> p1 . places . append ( p3 ) <EOL> p5 . places . append ( p6 ) <EOL> p6 . places . append ( p1 ) <EOL> p7 . places . append ( p1 ) <EOL> p1 . places . append ( p5 ) <EOL> p4 . places . append ( p3 ) <EOL> p3 . places . append ( p4 ) <EOL> sess . commit ( ) <EOL> eq_ ( p1 . places , [ p2 , p3 , p5 ] ) <EOL> eq_ ( p5 . places , [ p6 ] ) <EOL> eq_ ( p7 . places , [ p1 ] ) <EOL> eq_ ( p6 . places , [ p1 ] ) <EOL> eq_ ( p4 . places , [ p3 ] ) <EOL> eq_ ( p3 . places , [ p4 ] ) <EOL> eq_ ( p2 . places , [ ] ) <EOL> def test_self_referential_bidirectional_mutation ( self ) : <EOL> place , Place , place_place = ( self . tables . place , <EOL> self . classes . Place , <EOL> self . tables . place_place ) <EOL> mapper ( Place , place , properties = { <EOL> '<STR_LIT>' : relationship ( <EOL> Place , <EOL> secondary = place_place , <EOL> primaryjoin = place . c . place_id == place_place . c . pl1_id , <EOL> secondaryjoin = place . c . place_id == place_place . c . pl2_id , <EOL> order_by = place_place . c . pl2_id , <EOL> backref = '<STR_LIT>' <EOL> ) <EOL> } ) <EOL> sess = Session ( ) <EOL> p1 = Place ( '<STR_LIT>' ) <EOL> p2 = Place ( '<STR_LIT>' ) <EOL> p2 . parent_places = [ p1 ] <EOL> sess . add_all ( [ p1 , p2 ] ) <EOL> p1 . parent_places . append ( p2 ) <EOL> sess . commit ( ) <EOL> assert p1 in p2 . parent_places <EOL> assert p2 in p1 . parent_places <EOL> def test_joinedload_on_double ( self ) : <EOL> """<STR_LIT>""" <EOL> place_input , transition , Transition , PlaceThingy , place , place_thingy , Place , place_output = ( self . tables . place_input , <EOL> self . tables . transition , <EOL> self . classes . Transition , <EOL> self . classes . PlaceThingy , <EOL> self . tables . place , <EOL> self . tables . place_thingy , <EOL> self . classes . Place , <EOL> self . tables . place_output ) <EOL> mapper ( PlaceThingy , place_thingy ) <EOL> mapper ( Place , place , properties = { <EOL> '<STR_LIT>' : relationship ( PlaceThingy , lazy = '<STR_LIT>' ) <EOL> } ) <EOL> mapper ( Transition , transition , properties = dict ( <EOL> inputs = relationship ( Place , place_output , lazy = '<STR_LIT>' ) , <EOL> outputs = relationship ( Place , place_input , lazy = '<STR_LIT>' ) , <EOL> ) <EOL> ) <EOL> tran = Transition ( '<STR_LIT>' ) <EOL> tran . inputs . append ( Place ( '<STR_LIT>' ) ) <EOL> tran . outputs . append ( Place ( '<STR_LIT>' ) ) <EOL> tran . outputs . append ( Place ( '<STR_LIT>' ) ) <EOL> sess = Session ( ) <EOL> sess . add ( tran ) <EOL> sess . commit ( ) <EOL> r = sess . query ( Transition ) . all ( ) <EOL> self . assert_unordered_result ( r , Transition , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : ( Place , [ { '<STR_LIT:name>' : '<STR_LIT>' } ] ) , <EOL> '<STR_LIT>' : ( Place , [ { '<STR_LIT:name>' : '<STR_LIT>' } , { '<STR_LIT:name>' : '<STR_LIT>' } ] ) <EOL> } ) <EOL> def test_bidirectional ( self ) : <EOL> place_input , transition , Transition , Place , place , place_output = ( <EOL> self . tables . place_input , <EOL> self . tables . transition , <EOL> self . classes . Transition , <EOL> self . classes . Place , <EOL> self . tables . place , <EOL> self . tables . place_output ) <EOL> mapper ( Place , place ) <EOL> mapper ( Transition , transition , properties = dict ( <EOL> inputs = relationship ( Place , place_output , <EOL> backref = backref ( '<STR_LIT>' , <EOL> order_by = transition . c . transition_id ) , <EOL> order_by = Place . place_id ) , <EOL> outputs = relationship ( Place , place_input , <EOL> backref = backref ( '<STR_LIT>' , <EOL> order_by = transition . c . transition_id ) , <EOL> order_by = Place . place_id ) , <EOL> ) <EOL> ) <EOL> t1 = Transition ( '<STR_LIT>' ) <EOL> t2 = Transition ( '<STR_LIT>' ) <EOL> t3 = Transition ( '<STR_LIT>' ) <EOL> p1 = Place ( '<STR_LIT>' ) <EOL> p2 = Place ( '<STR_LIT>' ) <EOL> p3 = Place ( '<STR_LIT>' ) <EOL> sess = Session ( ) <EOL> sess . add_all ( [ p3 , p1 , t1 , t2 , p2 , t3 ] ) <EOL> t1 . inputs . append ( p1 ) <EOL> t1 . inputs . append ( p2 ) <EOL> t1 . outputs . append ( p3 ) <EOL> t2 . inputs . append ( p1 ) <EOL> p2 . inputs . append ( t2 ) <EOL> p3 . inputs . append ( t2 ) <EOL> p1 . outputs . append ( t1 ) <EOL> sess . commit ( ) <EOL> self . assert_result ( [ t1 ] , <EOL> Transition , { '<STR_LIT>' : <EOL> ( Place , [ { '<STR_LIT:name>' : '<STR_LIT>' } , { '<STR_LIT:name>' : '<STR_LIT>' } ] ) } ) <EOL> self . assert_result ( [ p2 ] , <EOL> Place , { '<STR_LIT>' : <EOL> ( Transition , [ { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' } ] ) } ) <EOL> @ testing . requires . sane_multi_rowcount <EOL> def test_stale_conditions ( self ) : <EOL> Place , Transition , place_input , place , transition = ( <EOL> self . classes . Place , <EOL> self . classes . Transition , <EOL> self . tables . place_input , <EOL> self . tables . place , <EOL> self . tables . transition ) <EOL> mapper ( Place , place , properties = { <EOL> '<STR_LIT>' : relationship ( Transition , secondary = place_input , <EOL> passive_updates = False ) <EOL> } ) <EOL> mapper ( Transition , transition ) <EOL> p1 = Place ( '<STR_LIT>' ) <EOL> t1 = Transition ( '<STR_LIT>' ) <EOL> p1 . transitions . append ( t1 ) <EOL> sess = sessionmaker ( ) ( ) <EOL> sess . add_all ( [ p1 , t1 ] ) <EOL> sess . commit ( ) <EOL> p1 . place_id <EOL> p1 . transitions <EOL> sess . execute ( "<STR_LIT>" , mapper = Place ) <EOL> p1 . place_id = <NUM_LIT:7> <EOL> assert_raises_message ( <EOL> orm_exc . StaleDataError , <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> sess . commit <EOL> ) <EOL> sess . rollback ( ) <EOL> p1 . place_id <EOL> p1 . transitions <EOL> sess . execute ( "<STR_LIT>" , mapper = Place ) <EOL> p1 . transitions . remove ( t1 ) <EOL> assert_raises_message ( <EOL> orm_exc . StaleDataError , <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> sess . commit <EOL> ) <EOL> class AssortedPersistenceTests ( fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( "<STR_LIT:left>" , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:30> ) ) <EOL> ) <EOL> Table ( "<STR_LIT:right>" , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:30> ) ) , <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) , <EOL> primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) , <EOL> primary_key = True ) , <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class A ( cls . Comparable ) : <EOL> pass <EOL> class B ( cls . Comparable ) : <EOL> pass <EOL> def _standard_bidirectional_fixture ( self ) : <EOL> left , secondary , right = self . tables . left , self . tables . secondary , self . tables . right <EOL> A , B = self . classes . A , self . classes . B <EOL> mapper ( A , left , properties = { <EOL> '<STR_LIT>' : relationship ( B , secondary = secondary , <EOL> backref = '<STR_LIT>' , order_by = right . c . id ) <EOL> } ) <EOL> mapper ( B , right ) <EOL> def _bidirectional_onescalar_fixture ( self ) : <EOL> left , secondary , right = self . tables . left , self . tables . secondary , self . tables . right <EOL> A , B = self . classes . A , self . classes . B <EOL> mapper ( A , left , properties = { <EOL> '<STR_LIT>' : relationship ( B , secondary = secondary , <EOL> backref = backref ( '<STR_LIT:a>' , uselist = False ) , <EOL> order_by = right . c . id ) <EOL> } ) <EOL> mapper ( B , right ) <EOL> def test_session_delete ( self ) : <EOL> self . _standard_bidirectional_fixture ( ) <EOL> A , B = self . classes . A , self . classes . B <EOL> secondary = self . tables . secondary <EOL> sess = Session ( ) <EOL> sess . add_all ( [ <EOL> A ( data = '<STR_LIT>' , bs = [ B ( data = '<STR_LIT>' ) ] ) , <EOL> A ( data = '<STR_LIT>' , bs = [ B ( data = '<STR_LIT>' ) ] ) <EOL> ] ) <EOL> sess . commit ( ) <EOL> a1 = sess . query ( A ) . filter_by ( data = '<STR_LIT>' ) . one ( ) <EOL> sess . delete ( a1 ) <EOL> sess . flush ( ) <EOL> eq_ ( sess . query ( secondary ) . count ( ) , <NUM_LIT:1> ) <EOL> a2 = sess . query ( A ) . filter_by ( data = '<STR_LIT>' ) . one ( ) <EOL> sess . delete ( a2 ) <EOL> sess . flush ( ) <EOL> eq_ ( sess . query ( secondary ) . count ( ) , <NUM_LIT:0> ) <EOL> def test_remove_scalar ( self ) : <EOL> self . _bidirectional_onescalar_fixture ( ) <EOL> A , B = self . classes . A , self . classes . B <EOL> secondary = self . tables . secondary <EOL> sess = Session ( ) <EOL> sess . add_all ( [ <EOL> A ( data = '<STR_LIT>' , bs = [ B ( data = '<STR_LIT>' ) , B ( data = '<STR_LIT>' ) ] ) , <EOL> ] ) <EOL> sess . commit ( ) <EOL> a1 = sess . query ( A ) . filter_by ( data = '<STR_LIT>' ) . one ( ) <EOL> b2 = sess . query ( B ) . filter_by ( data = '<STR_LIT>' ) . one ( ) <EOL> assert b2 . a is a1 <EOL> b2 . a = None <EOL> sess . commit ( ) <EOL> eq_ ( a1 . bs , [ B ( data = '<STR_LIT>' ) ] ) <EOL> eq_ ( b2 . a , None ) <EOL> eq_ ( sess . query ( secondary ) . count ( ) , <NUM_LIT:1> ) </s>
