<s> from sqlalchemy . ext import serializer <EOL> from sqlalchemy import testing <EOL> from sqlalchemy import Integer , String , ForeignKey , select , desc , func , util , MetaData , literal_column <EOL> from sqlalchemy . testing . schema import Table <EOL> from sqlalchemy . testing . schema import Column <EOL> from sqlalchemy . orm import relationship , sessionmaker , scoped_session , class_mapper , mapper , joinedload , configure_mappers , aliased <EOL> from sqlalchemy . testing import eq_ , AssertsCompiledSQL <EOL> from sqlalchemy . util import u , ue <EOL> from sqlalchemy . testing import fixtures <EOL> class User ( fixtures . ComparableEntity ) : <EOL> pass <EOL> class Address ( fixtures . ComparableEntity ) : <EOL> pass <EOL> users = addresses = Session = None <EOL> class SerializeTest ( AssertsCompiledSQL , fixtures . MappedTest ) : <EOL> run_setup_mappers = '<STR_LIT>' <EOL> run_inserts = '<STR_LIT>' <EOL> run_deletes = None <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> global users , addresses <EOL> users = Table ( '<STR_LIT>' , metadata , Column ( '<STR_LIT:id>' , Integer , <EOL> primary_key = True ) , Column ( '<STR_LIT:name>' , String ( <NUM_LIT:50> ) ) ) <EOL> addresses = Table ( '<STR_LIT>' , metadata , Column ( '<STR_LIT:id>' , Integer , <EOL> primary_key = True ) , Column ( '<STR_LIT:email>' , <EOL> String ( <NUM_LIT:50> ) ) , Column ( '<STR_LIT>' , Integer , <EOL> ForeignKey ( '<STR_LIT>' ) ) ) <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> global Session <EOL> Session = scoped_session ( sessionmaker ( ) ) <EOL> mapper ( User , users , properties = { '<STR_LIT>' <EOL> : relationship ( Address , backref = '<STR_LIT:user>' , <EOL> order_by = addresses . c . id ) } ) <EOL> mapper ( Address , addresses ) <EOL> configure_mappers ( ) <EOL> @ classmethod <EOL> def insert_data ( cls ) : <EOL> params = [ dict ( list ( zip ( ( '<STR_LIT:id>' , '<STR_LIT:name>' ) , column_values ) ) ) <EOL> for column_values in [ ( <NUM_LIT:7> , '<STR_LIT>' ) , ( <NUM_LIT:8> , '<STR_LIT>' ) , ( <NUM_LIT:9> , <EOL> '<STR_LIT>' ) , ( <NUM_LIT:10> , '<STR_LIT>' ) ] ] <EOL> users . insert ( ) . execute ( params ) <EOL> addresses . insert ( ) . execute ( [ dict ( list ( zip ( ( '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT:email>' <EOL> ) , column_values ) ) ) <EOL> for column_values in [ ( <NUM_LIT:1> , <NUM_LIT:7> , <EOL> '<STR_LIT>' ) , ( <NUM_LIT:2> , <NUM_LIT:8> , <EOL> '<STR_LIT>' ) , ( <NUM_LIT:3> , <NUM_LIT:8> , <EOL> '<STR_LIT>' ) , ( <NUM_LIT:4> , <NUM_LIT:8> , <EOL> '<STR_LIT>' ) , ( <NUM_LIT:5> , <NUM_LIT:9> , <EOL> '<STR_LIT>' ) ] ] ) <EOL> def test_tables ( self ) : <EOL> assert serializer . loads ( serializer . dumps ( users , - <NUM_LIT:1> ) , <EOL> users . metadata , Session ) is users <EOL> def test_columns ( self ) : <EOL> assert serializer . loads ( serializer . dumps ( users . c . name , - <NUM_LIT:1> ) , <EOL> users . metadata , Session ) is users . c . name <EOL> def test_mapper ( self ) : <EOL> user_mapper = class_mapper ( User ) <EOL> assert serializer . loads ( serializer . dumps ( user_mapper , - <NUM_LIT:1> ) , <EOL> None , None ) is user_mapper <EOL> def test_attribute ( self ) : <EOL> assert serializer . loads ( serializer . dumps ( User . name , - <NUM_LIT:1> ) , None , <EOL> None ) is User . name <EOL> def test_expression ( self ) : <EOL> expr = select ( [ users ] ) . select_from ( users . join ( addresses ) ) . limit ( <NUM_LIT:5> ) <EOL> re_expr = serializer . loads ( serializer . dumps ( expr , - <NUM_LIT:1> ) , <EOL> users . metadata , None ) <EOL> eq_ ( str ( expr ) , str ( re_expr ) ) <EOL> assert re_expr . bind is testing . db <EOL> eq_ ( re_expr . execute ( ) . fetchall ( ) , [ ( <NUM_LIT:7> , '<STR_LIT>' ) , ( <NUM_LIT:8> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:8> , '<STR_LIT>' ) , ( <NUM_LIT:8> , '<STR_LIT>' ) , ( <NUM_LIT:9> , '<STR_LIT>' ) ] ) <EOL> def test_query_one ( self ) : <EOL> q = Session . query ( User ) . filter ( User . name == '<STR_LIT>' ) . options ( joinedload ( User . addresses ) ) <EOL> q2 = serializer . loads ( <EOL> serializer . dumps ( q , - <NUM_LIT:1> ) , <EOL> users . metadata , Session ) <EOL> def go ( ) : <EOL> eq_ ( q2 . all ( ) , [ <EOL> User ( name = '<STR_LIT>' , addresses = [ Address ( id = <NUM_LIT:2> ) , <EOL> Address ( id = <NUM_LIT:3> ) , Address ( id = <NUM_LIT:4> ) ] ) ] ) <EOL> self . assert_sql_count ( testing . db , go , <NUM_LIT:1> ) <EOL> eq_ ( q2 . join ( User . addresses ) . filter ( Address . email <EOL> == '<STR_LIT>' ) . value ( func . count ( literal_column ( '<STR_LIT:*>' ) ) ) , <NUM_LIT:1> ) <EOL> u1 = Session . query ( User ) . get ( <NUM_LIT:8> ) <EOL> q = Session . query ( Address ) . filter ( Address . user <EOL> == u1 ) . order_by ( desc ( Address . email ) ) <EOL> q2 = serializer . loads ( serializer . dumps ( q , - <NUM_LIT:1> ) , users . metadata , <EOL> Session ) <EOL> eq_ ( q2 . all ( ) , [ Address ( email = '<STR_LIT>' ) , <EOL> Address ( email = '<STR_LIT>' ) , <EOL> Address ( email = '<STR_LIT>' ) ] ) <EOL> @ testing . requires . non_broken_pickle <EOL> def test_query_two ( self ) : <EOL> q = Session . query ( User ) . join ( User . addresses ) . filter ( Address . email . like ( '<STR_LIT>' ) ) <EOL> q2 = serializer . loads ( serializer . dumps ( q , - <NUM_LIT:1> ) , users . metadata , <EOL> Session ) <EOL> eq_ ( q2 . all ( ) , [ User ( name = '<STR_LIT>' ) ] ) <EOL> eq_ ( list ( q2 . values ( User . id , User . name ) ) , [ ( <NUM_LIT:9> , '<STR_LIT>' ) ] ) <EOL> @ testing . requires . non_broken_pickle <EOL> def test_orm_join ( self ) : <EOL> from sqlalchemy . orm . util import join <EOL> j = join ( User , Address , User . addresses ) <EOL> j2 = serializer . loads ( serializer . dumps ( j , - <NUM_LIT:1> ) , users . metadata ) <EOL> assert j2 . left is j . left <EOL> assert j2 . right is j . right <EOL> assert j2 . _target_adapter . _next <EOL> @ testing . exclude ( '<STR_LIT>' , '<STR_LIT>' , ( <NUM_LIT:3> , <NUM_LIT:5> , <NUM_LIT:9> ) , <EOL> '<STR_LIT>' ) <EOL> def test_aliases ( self ) : <EOL> u7 , u8 , u9 , u10 = Session . query ( User ) . order_by ( User . id ) . all ( ) <EOL> ualias = aliased ( User ) <EOL> q = Session . query ( User , ualias ) . join ( ualias , User . id <EOL> < ualias . id ) . filter ( User . id < <NUM_LIT:9> ) . order_by ( User . id , <EOL> ualias . id ) <EOL> eq_ ( list ( q . all ( ) ) , [ ( u7 , u8 ) , ( u7 , u9 ) , ( u7 , u10 ) , ( u8 , u9 ) , <EOL> ( u8 , u10 ) ] ) <EOL> q2 = serializer . loads ( serializer . dumps ( q , - <NUM_LIT:1> ) , users . metadata , <EOL> Session ) <EOL> eq_ ( list ( q2 . all ( ) ) , [ ( u7 , u8 ) , ( u7 , u9 ) , ( u7 , u10 ) , ( u8 , u9 ) , <EOL> ( u8 , u10 ) ] ) <EOL> @ testing . requires . non_broken_pickle <EOL> def test_any ( self ) : <EOL> r = User . addresses . any ( Address . email == '<STR_LIT:x>' ) <EOL> ser = serializer . dumps ( r , - <NUM_LIT:1> ) <EOL> x = serializer . loads ( ser , users . metadata ) <EOL> eq_ ( str ( r ) , str ( x ) ) <EOL> def test_unicode ( self ) : <EOL> m = MetaData ( ) <EOL> t = Table ( ue ( '<STR_LIT>' ) , m , <EOL> Column ( ue ( '<STR_LIT>' ) , Integer ) ) <EOL> expr = select ( [ t ] ) . where ( t . c [ ue ( '<STR_LIT>' ) ] == <NUM_LIT:5> ) <EOL> expr2 = serializer . loads ( serializer . dumps ( expr , - <NUM_LIT:1> ) , m ) <EOL> self . assert_compile ( <EOL> expr2 , <EOL> ue ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> dialect = "<STR_LIT:default>" <EOL> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> testing . main ( ) </s>
