<s> from sqlalchemy . testing import assert_raises_message <EOL> from sqlalchemy import MetaData , Integer , ForeignKey <EOL> from sqlalchemy . testing . schema import Table <EOL> from sqlalchemy . testing . schema import Column <EOL> from sqlalchemy . orm import mapper , create_session <EOL> import sqlalchemy as sa <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import fixtures , eq_ , engines , is_ <EOL> from sqlalchemy . orm import relationship , Session , backref , sessionmaker <EOL> from test . orm import _fixtures <EOL> from sqlalchemy . testing . mock import Mock <EOL> class BindIntegrationTest ( _fixtures . FixtureTest ) : <EOL> run_inserts = None <EOL> def test_mapped_binds ( self ) : <EOL> Address , addresses , users , User = ( self . classes . Address , <EOL> self . tables . addresses , <EOL> self . tables . users , <EOL> self . classes . User ) <EOL> m2 = sa . MetaData ( ) <EOL> users_unbound = users . tometadata ( m2 ) <EOL> addresses_unbound = addresses . tometadata ( m2 ) <EOL> mapper ( Address , addresses_unbound ) <EOL> mapper ( User , users_unbound , properties = { <EOL> '<STR_LIT>' : relationship ( Address , <EOL> backref = backref ( "<STR_LIT:user>" , cascade = "<STR_LIT:all>" ) , <EOL> cascade = "<STR_LIT:all>" ) } ) <EOL> sess = Session ( binds = { User : self . metadata . bind , <EOL> Address : self . metadata . bind } ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> eq_ ( sess . query ( User ) . filter ( User . id == <NUM_LIT:1> ) . all ( ) , <EOL> [ User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) ] ) <EOL> sess . execute ( users_unbound . insert ( ) , params = dict ( id = <NUM_LIT:2> , <EOL> name = '<STR_LIT>' ) ) <EOL> eq_ ( sess . execute ( users_unbound . select ( users_unbound . c . id <EOL> == <NUM_LIT:2> ) ) . fetchall ( ) , [ ( <NUM_LIT:2> , '<STR_LIT>' ) ] ) <EOL> eq_ ( sess . execute ( users_unbound . select ( User . id == <NUM_LIT:2> ) ) . fetchall ( ) , <EOL> [ ( <NUM_LIT:2> , '<STR_LIT>' ) ] ) <EOL> sess . execute ( users_unbound . delete ( ) ) <EOL> eq_ ( sess . execute ( users_unbound . select ( ) ) . fetchall ( ) , [ ] ) <EOL> sess . close ( ) <EOL> def test_table_binds ( self ) : <EOL> Address , addresses , users , User = ( self . classes . Address , <EOL> self . tables . addresses , <EOL> self . tables . users , <EOL> self . classes . User ) <EOL> m2 = sa . MetaData ( ) <EOL> users_unbound = users . tometadata ( m2 ) <EOL> addresses_unbound = addresses . tometadata ( m2 ) <EOL> mapper ( Address , addresses_unbound ) <EOL> mapper ( User , users_unbound , properties = { <EOL> '<STR_LIT>' : relationship ( Address , <EOL> backref = backref ( "<STR_LIT:user>" , cascade = "<STR_LIT:all>" ) , <EOL> cascade = "<STR_LIT:all>" ) } ) <EOL> Session = sessionmaker ( binds = { users_unbound : self . metadata . bind , <EOL> addresses_unbound : self . metadata . bind } ) <EOL> sess = Session ( ) <EOL> u1 = User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) <EOL> sess . add ( u1 ) <EOL> eq_ ( sess . query ( User ) . filter ( User . id == <NUM_LIT:1> ) . all ( ) , <EOL> [ User ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) ] ) <EOL> sess . execute ( users_unbound . insert ( ) , params = dict ( id = <NUM_LIT:2> , name = '<STR_LIT>' ) ) <EOL> eq_ ( sess . execute ( users_unbound . select ( users_unbound . c . id <EOL> == <NUM_LIT:2> ) ) . fetchall ( ) , [ ( <NUM_LIT:2> , '<STR_LIT>' ) ] ) <EOL> eq_ ( sess . execute ( users_unbound . select ( User . id == <NUM_LIT:2> ) ) . fetchall ( ) , <EOL> [ ( <NUM_LIT:2> , '<STR_LIT>' ) ] ) <EOL> sess . execute ( users_unbound . delete ( ) ) <EOL> eq_ ( sess . execute ( users_unbound . select ( ) ) . fetchall ( ) , [ ] ) <EOL> sess . close ( ) <EOL> def test_bind_from_metadata ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> session = create_session ( ) <EOL> session . execute ( users . insert ( ) , dict ( name = '<STR_LIT>' ) ) <EOL> assert len ( session . query ( User ) . filter_by ( name = '<STR_LIT>' ) . all ( ) ) == <NUM_LIT:1> <EOL> session . execute ( users . delete ( ) ) <EOL> assert len ( session . query ( User ) . filter_by ( name = '<STR_LIT>' ) . all ( ) ) == <NUM_LIT:0> <EOL> session . close ( ) <EOL> def test_bind_arguments ( self ) : <EOL> users , Address , addresses , User = ( self . tables . users , <EOL> self . classes . Address , <EOL> self . tables . addresses , <EOL> self . classes . User ) <EOL> mapper ( User , users ) <EOL> mapper ( Address , addresses ) <EOL> e1 = engines . testing_engine ( ) <EOL> e2 = engines . testing_engine ( ) <EOL> e3 = engines . testing_engine ( ) <EOL> sess = Session ( e3 ) <EOL> sess . bind_mapper ( User , e1 ) <EOL> sess . bind_mapper ( Address , e2 ) <EOL> assert sess . connection ( ) . engine is e3 <EOL> assert sess . connection ( bind = e1 ) . engine is e1 <EOL> assert sess . connection ( mapper = Address , bind = e1 ) . engine is e1 <EOL> assert sess . connection ( mapper = Address ) . engine is e2 <EOL> assert sess . connection ( clause = addresses . select ( ) ) . engine is e2 <EOL> assert sess . connection ( mapper = User , <EOL> clause = addresses . select ( ) ) . engine is e1 <EOL> assert sess . connection ( mapper = User , <EOL> clause = addresses . select ( ) , <EOL> bind = e2 ) . engine is e2 <EOL> sess . close ( ) <EOL> @ engines . close_open_connections <EOL> def test_bound_connection ( self ) : <EOL> users , User = self . tables . users , self . classes . User <EOL> mapper ( User , users ) <EOL> c = testing . db . connect ( ) <EOL> sess = create_session ( bind = c ) <EOL> sess . begin ( ) <EOL> transaction = sess . transaction <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> assert transaction . _connection_for_bind ( testing . db , None ) is transaction . _connection_for_bind ( c , None ) is c <EOL> assert_raises_message ( sa . exc . InvalidRequestError , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> transaction . _connection_for_bind , <EOL> testing . db . connect ( ) , None ) <EOL> transaction . rollback ( ) <EOL> assert len ( sess . query ( User ) . all ( ) ) == <NUM_LIT:0> <EOL> sess . close ( ) <EOL> def test_bound_connection_transactional ( self ) : <EOL> User , users = self . classes . User , self . tables . users <EOL> mapper ( User , users ) <EOL> c = testing . db . connect ( ) <EOL> sess = create_session ( bind = c , autocommit = False ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> sess . close ( ) <EOL> assert not c . in_transaction ( ) <EOL> assert c . scalar ( "<STR_LIT>" ) == <NUM_LIT:0> <EOL> sess = create_session ( bind = c , autocommit = False ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> sess . commit ( ) <EOL> assert not c . in_transaction ( ) <EOL> assert c . scalar ( "<STR_LIT>" ) == <NUM_LIT:1> <EOL> c . execute ( "<STR_LIT>" ) <EOL> assert c . scalar ( "<STR_LIT>" ) == <NUM_LIT:0> <EOL> c = testing . db . connect ( ) <EOL> trans = c . begin ( ) <EOL> sess = create_session ( bind = c , autocommit = True ) <EOL> u = User ( name = '<STR_LIT>' ) <EOL> sess . add ( u ) <EOL> sess . flush ( ) <EOL> assert c . in_transaction ( ) <EOL> trans . commit ( ) <EOL> assert not c . in_transaction ( ) <EOL> assert c . scalar ( "<STR_LIT>" ) == <NUM_LIT:1> <EOL> class SessionBindTest ( 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> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , Integer ) ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class Foo ( cls . Basic ) : <EOL> pass <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> test_table , Foo = cls . tables . test_table , cls . classes . Foo <EOL> meta = MetaData ( ) <EOL> test_table . tometadata ( meta ) <EOL> assert meta . tables [ '<STR_LIT>' ] . bind is None <EOL> mapper ( Foo , meta . tables [ '<STR_LIT>' ] ) <EOL> def test_session_bind ( self ) : <EOL> Foo = self . classes . Foo <EOL> engine = self . metadata . bind <EOL> for bind in ( engine , engine . connect ( ) ) : <EOL> try : <EOL> sess = create_session ( bind = bind ) <EOL> assert sess . bind is bind <EOL> f = Foo ( ) <EOL> sess . add ( f ) <EOL> sess . flush ( ) <EOL> assert sess . query ( Foo ) . get ( f . id ) is f <EOL> finally : <EOL> if hasattr ( bind , '<STR_LIT>' ) : <EOL> bind . close ( ) <EOL> def test_session_unbound ( self ) : <EOL> Foo = self . classes . Foo <EOL> sess = create_session ( ) <EOL> sess . add ( Foo ( ) ) <EOL> assert_raises_message ( <EOL> sa . exc . UnboundExecutionError , <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> sess . flush ) <EOL> class GetBindTest ( fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) <EOL> ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) <EOL> ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , ForeignKey ( '<STR_LIT>' ) , primary_key = True ) <EOL> ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) <EOL> ) <EOL> @ classmethod <EOL> def setup_classes ( cls ) : <EOL> class MixinOne ( cls . Basic ) : <EOL> pass <EOL> class BaseClass ( cls . Basic ) : <EOL> pass <EOL> class ClassWMixin ( MixinOne , cls . Basic ) : <EOL> pass <EOL> class JoinedSubClass ( BaseClass ) : <EOL> pass <EOL> class ConcreteSubClass ( BaseClass ) : <EOL> pass <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> mapper ( cls . classes . ClassWMixin , cls . tables . w_mixin_table ) <EOL> mapper ( cls . classes . BaseClass , cls . tables . base_table ) <EOL> mapper ( <EOL> cls . classes . JoinedSubClass , <EOL> cls . tables . joined_sub_table , inherits = cls . classes . BaseClass ) <EOL> mapper ( <EOL> cls . classes . ConcreteSubClass , <EOL> cls . tables . concrete_sub_table , inherits = cls . classes . BaseClass , <EOL> concrete = True ) <EOL> def _fixture ( self , binds ) : <EOL> return Session ( binds = binds ) <EOL> def test_fallback_table_metadata ( self ) : <EOL> session = self . _fixture ( { } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> testing . db <EOL> ) <EOL> def test_bind_base_table_base_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . tables . base_table : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_base_table_joined_sub_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . tables . base_table : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . JoinedSubClass ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_joined_sub_table_joined_sub_class ( self ) : <EOL> base_class_bind = Mock ( name = '<STR_LIT>' ) <EOL> joined_class_bind = Mock ( name = '<STR_LIT>' ) <EOL> session = self . _fixture ( { <EOL> self . tables . base_table : base_class_bind , <EOL> self . tables . joined_sub_table : joined_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . JoinedSubClass ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_base_table_concrete_sub_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . tables . base_table : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . ConcreteSubClass ) , <EOL> testing . db <EOL> ) <EOL> def test_bind_sub_table_concrete_sub_class ( self ) : <EOL> base_class_bind = Mock ( name = '<STR_LIT>' ) <EOL> concrete_sub_bind = Mock ( name = '<STR_LIT>' ) <EOL> session = self . _fixture ( { <EOL> self . tables . base_table : base_class_bind , <EOL> self . tables . concrete_sub_table : concrete_sub_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . ConcreteSubClass ) , <EOL> concrete_sub_bind <EOL> ) <EOL> def test_bind_base_class_base_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . classes . BaseClass : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_mixin_class_simple_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . classes . MixinOne : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . ClassWMixin ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_base_class_joined_sub_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . classes . BaseClass : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . JoinedSubClass ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_joined_sub_class_joined_sub_class ( self ) : <EOL> base_class_bind = Mock ( name = '<STR_LIT>' ) <EOL> joined_class_bind = Mock ( name = '<STR_LIT>' ) <EOL> session = self . _fixture ( { <EOL> self . classes . BaseClass : base_class_bind , <EOL> self . classes . JoinedSubClass : joined_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . JoinedSubClass ) , <EOL> joined_class_bind <EOL> ) <EOL> def test_bind_base_class_concrete_sub_class ( self ) : <EOL> base_class_bind = Mock ( ) <EOL> session = self . _fixture ( { <EOL> self . classes . BaseClass : base_class_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . ConcreteSubClass ) , <EOL> base_class_bind <EOL> ) <EOL> def test_bind_sub_class_concrete_sub_class ( self ) : <EOL> base_class_bind = Mock ( name = '<STR_LIT>' ) <EOL> concrete_sub_bind = Mock ( name = '<STR_LIT>' ) <EOL> session = self . _fixture ( { <EOL> self . classes . BaseClass : base_class_bind , <EOL> self . classes . ConcreteSubClass : concrete_sub_bind <EOL> } ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . BaseClass ) , <EOL> base_class_bind <EOL> ) <EOL> is_ ( <EOL> session . get_bind ( self . classes . ConcreteSubClass ) , <EOL> concrete_sub_bind <EOL> ) </s>
