<s> from sqlalchemy . engine import reflection <EOL> from sqlalchemy . testing . assertions import eq_ , assert_raises , AssertsExecutionResults <EOL> from sqlalchemy . testing import fixtures <EOL> from sqlalchemy import testing <EOL> from sqlalchemy import inspect <EOL> from sqlalchemy import Table , Column , MetaData , Integer , String , PrimaryKeyConstraint , ForeignKey , join , Sequence , UniqueConstraint , Index <EOL> from sqlalchemy import exc <EOL> import sqlalchemy as sa <EOL> from sqlalchemy . dialects . postgresql import base as postgresql <EOL> from sqlalchemy . dialects . postgresql import ARRAY <EOL> import re <EOL> class ForeignTableReflectionTest ( fixtures . TablesTest , AssertsExecutionResults ) : <EOL> """<STR_LIT>""" <EOL> __requires__ = '<STR_LIT>' , <EOL> __only_on__ = '<STR_LIT>' <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> from sqlalchemy . testing import config <EOL> dblink = config . file_config . get ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> testtable = Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:30> ) ) ) <EOL> for ddl in [ <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % dblink , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> ] : <EOL> sa . event . listen ( metadata , "<STR_LIT>" , sa . DDL ( ddl ) ) <EOL> for ddl in [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> "<STR_LIT>" <EOL> ] : <EOL> sa . event . listen ( metadata , "<STR_LIT>" , sa . DDL ( ddl ) ) <EOL> def test_foreign_table_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( set ( table . columns . keys ( ) ) , set ( [ '<STR_LIT:id>' , '<STR_LIT:data>' ] ) , <EOL> "<STR_LIT>" ) <EOL> def test_get_foreign_table_names ( self ) : <EOL> inspector = inspect ( testing . db ) <EOL> with testing . db . connect ( ) as conn : <EOL> ft_names = inspector . get_foreign_table_names ( ) <EOL> eq_ ( ft_names , [ '<STR_LIT>' ] ) <EOL> def test_get_table_names_no_foreign ( self ) : <EOL> inspector = inspect ( testing . db ) <EOL> with testing . db . connect ( ) as conn : <EOL> names = inspector . get_table_names ( ) <EOL> eq_ ( names , [ '<STR_LIT>' ] ) <EOL> class MaterializedViewReflectionTest ( <EOL> fixtures . TablesTest , AssertsExecutionResults ) : <EOL> """<STR_LIT>""" <EOL> __only_on__ = '<STR_LIT>' <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> testtable = Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:30> ) ) ) <EOL> @ sa . event . listens_for ( testtable , "<STR_LIT>" ) <EOL> def insert_data ( target , connection , ** kw ) : <EOL> connection . execute ( <EOL> target . insert ( ) , <EOL> { "<STR_LIT:id>" : <NUM_LIT> , "<STR_LIT:data>" : '<STR_LIT>' } <EOL> ) <EOL> materialized_view = sa . DDL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> plain_view = sa . DDL ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> sa . event . listen ( testtable , '<STR_LIT>' , plain_view ) <EOL> sa . event . listen ( testtable , '<STR_LIT>' , materialized_view ) <EOL> sa . event . listen ( <EOL> testtable , '<STR_LIT>' , <EOL> sa . DDL ( "<STR_LIT>" ) <EOL> ) <EOL> sa . event . listen ( <EOL> testtable , '<STR_LIT>' , <EOL> sa . DDL ( "<STR_LIT>" ) <EOL> ) <EOL> def test_mview_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( set ( table . columns . keys ( ) ) , set ( [ '<STR_LIT:id>' , '<STR_LIT:data>' ] ) , <EOL> "<STR_LIT>" ) <EOL> def test_mview_select ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( <EOL> table . select ( ) . execute ( ) . fetchall ( ) , <EOL> [ ( <NUM_LIT> , '<STR_LIT>' , ) ] <EOL> ) <EOL> def test_get_view_names ( self ) : <EOL> insp = inspect ( testing . db ) <EOL> eq_ ( set ( insp . get_view_names ( ) ) , set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> def test_get_view_definition ( self ) : <EOL> insp = inspect ( testing . db ) <EOL> eq_ ( <EOL> re . sub ( <EOL> r'<STR_LIT>' , '<STR_LIT:U+0020>' , <EOL> insp . get_view_definition ( "<STR_LIT>" ) . strip ( ) ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> class DomainReflectionTest ( fixtures . TestBase , AssertsExecutionResults ) : <EOL> """<STR_LIT>""" <EOL> __only_on__ = '<STR_LIT>' <EOL> __backend__ = True <EOL> @ classmethod <EOL> def setup_class ( cls ) : <EOL> con = testing . db . connect ( ) <EOL> for ddl in '<STR_LIT>' , '<STR_LIT>' , "<STR_LIT>" , '<STR_LIT>' : <EOL> try : <EOL> con . execute ( ddl ) <EOL> except exc . DBAPIError as e : <EOL> if '<STR_LIT>' not in str ( e ) : <EOL> raise e <EOL> con . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' ) <EOL> @ classmethod <EOL> def teardown_class ( cls ) : <EOL> con = testing . db . connect ( ) <EOL> con . execute ( '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' ) <EOL> con . execute ( '<STR_LIT>' ) <EOL> con . execute ( "<STR_LIT>" ) <EOL> con . execute ( "<STR_LIT>" ) <EOL> con . execute ( "<STR_LIT>" ) <EOL> def test_table_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( set ( table . columns . keys ( ) ) , set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> "<STR_LIT>" ) <EOL> assert isinstance ( table . c . answer . type , Integer ) <EOL> def test_domain_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( str ( table . columns . answer . server_default . arg ) , '<STR_LIT>' , <EOL> "<STR_LIT>" ) <EOL> assert not table . columns . answer . nullable , '<STR_LIT>' <EOL> def test_enum_domain_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( <EOL> table . c . data . type . enums , <EOL> [ '<STR_LIT:test>' ] <EOL> ) <EOL> def test_table_is_reflected_test_schema ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True , <EOL> schema = '<STR_LIT>' ) <EOL> eq_ ( set ( table . columns . keys ( ) ) , set ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> "<STR_LIT>" ) <EOL> assert isinstance ( table . c . anything . type , Integer ) <EOL> def test_schema_domain_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True , <EOL> schema = '<STR_LIT>' ) <EOL> eq_ ( str ( table . columns . answer . server_default . arg ) , '<STR_LIT:0>' , <EOL> "<STR_LIT>" ) <EOL> assert table . columns . answer . nullable , '<STR_LIT>' <EOL> def test_crosschema_domain_is_reflected ( self ) : <EOL> metadata = MetaData ( testing . db ) <EOL> table = Table ( '<STR_LIT>' , metadata , autoload = True ) <EOL> eq_ ( str ( table . columns . answer . server_default . arg ) , '<STR_LIT:0>' , <EOL> "<STR_LIT>" ) <EOL> assert table . columns . answer . nullable , '<STR_LIT>' <EOL> def test_unknown_types ( self ) : <EOL> from sqlalchemy . databases import postgresql <EOL> ischema_names = postgresql . PGDialect . ischema_names <EOL> postgresql . PGDialect . ischema_names = { } <EOL> try : <EOL> m2 = MetaData ( testing . db ) <EOL> assert_raises ( exc . SAWarning , Table , '<STR_LIT>' , m2 , <EOL> autoload = True ) <EOL> @ testing . emits_warning ( '<STR_LIT>' ) <EOL> def warns ( ) : <EOL> m3 = MetaData ( testing . db ) <EOL> t3 = Table ( '<STR_LIT>' , m3 , autoload = True ) <EOL> assert t3 . c . answer . type . __class__ == sa . types . NullType <EOL> finally : <EOL> postgresql . PGDialect . ischema_names = ischema_names <EOL> class ReflectionTest ( fixtures . TestBase ) : <EOL> __only_on__ = '<STR_LIT>' <EOL> __backend__ = True <EOL> @ testing . fails_if ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> @ testing . provide_metadata <EOL> def test_reflected_primary_key_order ( self ) : <EOL> meta1 = self . metadata <EOL> subject = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , primary_key = True ) , <EOL> PrimaryKeyConstraint ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> meta1 . create_all ( ) <EOL> meta2 = MetaData ( testing . db ) <EOL> subject = Table ( '<STR_LIT>' , meta2 , autoload = True ) <EOL> eq_ ( subject . primary_key . columns . keys ( ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> @ testing . provide_metadata <EOL> def test_pg_weirdchar_reflection ( self ) : <EOL> meta1 = self . metadata <EOL> subject = Table ( '<STR_LIT>' , meta1 , Column ( '<STR_LIT>' , Integer , <EOL> primary_key = True ) ) <EOL> referer = Table ( <EOL> '<STR_LIT>' , meta1 , <EOL> Column ( <EOL> '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) ) <EOL> meta1 . create_all ( ) <EOL> meta2 = MetaData ( testing . db ) <EOL> subject = Table ( '<STR_LIT>' , meta2 , autoload = True ) <EOL> referer = Table ( '<STR_LIT>' , meta2 , autoload = True ) <EOL> self . assert_ ( ( subject . c [ '<STR_LIT>' ] <EOL> == referer . c . ref ) . compare ( <EOL> subject . join ( referer ) . onclause ) ) <EOL> @ testing . provide_metadata <EOL> def test_reflect_default_over_128_chars ( self ) : <EOL> Table ( '<STR_LIT:t>' , self . metadata , <EOL> Column ( '<STR_LIT:x>' , String ( <NUM_LIT:200> ) , server_default = "<STR_LIT>" * <NUM_LIT> ) <EOL> ) . create ( testing . db ) <EOL> m = MetaData ( ) <EOL> t = Table ( '<STR_LIT:t>' , m , autoload = True , autoload_with = testing . db ) <EOL> eq_ ( <EOL> t . c . x . server_default . arg . text , "<STR_LIT>" % ( <EOL> "<STR_LIT>" * <NUM_LIT> ) <EOL> ) <EOL> @ testing . fails_if ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> @ testing . provide_metadata <EOL> def test_renamed_sequence_reflection ( self ) : <EOL> metadata = self . metadata <EOL> t = Table ( '<STR_LIT:t>' , metadata , Column ( '<STR_LIT:id>' , Integer , primary_key = True ) ) <EOL> metadata . create_all ( ) <EOL> m2 = MetaData ( testing . db ) <EOL> t2 = Table ( '<STR_LIT:t>' , m2 , autoload = True , implicit_returning = False ) <EOL> eq_ ( t2 . c . id . server_default . arg . text , <EOL> "<STR_LIT>" ) <EOL> r = t2 . insert ( ) . execute ( ) <EOL> eq_ ( r . inserted_primary_key , [ <NUM_LIT:1> ] ) <EOL> testing . db . connect ( ) . execution_options ( autocommit = True ) . execute ( '<STR_LIT>' <EOL> ) <EOL> m3 = MetaData ( testing . db ) <EOL> t3 = Table ( '<STR_LIT:t>' , m3 , autoload = True , implicit_returning = False ) <EOL> eq_ ( t3 . c . id . server_default . arg . text , <EOL> "<STR_LIT>" ) <EOL> r = t3 . insert ( ) . execute ( ) <EOL> eq_ ( r . inserted_primary_key , [ <NUM_LIT:2> ] ) <EOL> @ testing . provide_metadata <EOL> def test_renamed_pk_reflection ( self ) : <EOL> metadata = self . metadata <EOL> t = Table ( '<STR_LIT:t>' , metadata , Column ( '<STR_LIT:id>' , Integer , primary_key = True ) ) <EOL> metadata . create_all ( ) <EOL> testing . db . connect ( ) . execution_options ( autocommit = True ) . execute ( '<STR_LIT>' ) <EOL> m2 = MetaData ( testing . db ) <EOL> t2 = Table ( '<STR_LIT:t>' , m2 , autoload = True ) <EOL> eq_ ( [ c . name for c in t2 . primary_key ] , [ '<STR_LIT>' ] ) <EOL> @ testing . provide_metadata <EOL> def test_has_temporary_table ( self ) : <EOL> assert not testing . db . has_table ( "<STR_LIT>" ) <EOL> user_tmp = Table ( <EOL> "<STR_LIT>" , self . metadata , <EOL> Column ( "<STR_LIT:id>" , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( <NUM_LIT:50> ) ) , <EOL> prefixes = [ '<STR_LIT>' ] <EOL> ) <EOL> user_tmp . create ( testing . db ) <EOL> assert testing . db . has_table ( "<STR_LIT>" ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_one ( self ) : <EOL> meta1 = self . metadata <EOL> users = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:30> ) , nullable = False ) , <EOL> schema = '<STR_LIT>' ) <EOL> addresses = Table ( <EOL> '<STR_LIT>' , meta1 , <EOL> Column ( <EOL> '<STR_LIT>' , Integer , primary_key = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , Integer , ForeignKey ( <EOL> users . c . user_id ) ) , <EOL> Column ( <EOL> '<STR_LIT>' , String ( <NUM_LIT:20> ) ) , schema = '<STR_LIT>' ) <EOL> meta1 . create_all ( ) <EOL> meta2 = MetaData ( testing . db ) <EOL> addresses = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = '<STR_LIT>' ) <EOL> users = Table ( '<STR_LIT>' , meta2 , mustexist = True , <EOL> schema = '<STR_LIT>' ) <EOL> j = join ( users , addresses ) <EOL> self . assert_ ( ( users . c . user_id <EOL> == addresses . c . remote_user_id ) . compare ( j . onclause ) ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_two ( self ) : <EOL> meta1 = self . metadata <EOL> subject = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) ) <EOL> referer = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> schema = '<STR_LIT>' ) <EOL> meta1 . create_all ( ) <EOL> meta2 = MetaData ( testing . db ) <EOL> subject = Table ( '<STR_LIT>' , meta2 , autoload = True ) <EOL> referer = Table ( '<STR_LIT>' , meta2 , schema = '<STR_LIT>' , <EOL> autoload = True ) <EOL> self . assert_ ( ( subject . c . id <EOL> == referer . c . ref ) . compare ( <EOL> subject . join ( referer ) . onclause ) ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_three ( self ) : <EOL> meta1 = self . metadata <EOL> subject = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> schema = '<STR_LIT>' ) <EOL> referer = Table ( <EOL> '<STR_LIT>' , <EOL> meta1 , <EOL> Column ( <EOL> '<STR_LIT:id>' , <EOL> Integer , <EOL> primary_key = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , <EOL> Integer , <EOL> ForeignKey ( '<STR_LIT>' ) ) , <EOL> schema = '<STR_LIT>' ) <EOL> meta1 . create_all ( ) <EOL> meta2 = MetaData ( testing . db ) <EOL> subject = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = '<STR_LIT>' ) <EOL> referer = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = '<STR_LIT>' ) <EOL> self . assert_ ( ( subject . c . id <EOL> == referer . c . ref ) . compare ( <EOL> subject . join ( referer ) . onclause ) ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_four ( self ) : <EOL> meta1 = self . metadata <EOL> subject = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> schema = '<STR_LIT>' ) <EOL> referer = Table ( <EOL> '<STR_LIT>' , <EOL> meta1 , <EOL> Column ( <EOL> '<STR_LIT:id>' , <EOL> Integer , <EOL> primary_key = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , <EOL> Integer , <EOL> ForeignKey ( '<STR_LIT>' ) ) , <EOL> schema = '<STR_LIT>' ) <EOL> meta1 . create_all ( ) <EOL> conn = testing . db . connect ( ) <EOL> conn . detach ( ) <EOL> conn . execute ( "<STR_LIT>" ) <EOL> meta2 = MetaData ( bind = conn ) <EOL> subject = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = '<STR_LIT>' , <EOL> postgresql_ignore_search_path = True ) <EOL> referer = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = '<STR_LIT>' , <EOL> postgresql_ignore_search_path = True ) <EOL> self . assert_ ( ( subject . c . id <EOL> == referer . c . ref ) . compare ( <EOL> subject . join ( referer ) . onclause ) ) <EOL> conn . close ( ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_five ( self ) : <EOL> meta1 = self . metadata <EOL> default_schema = testing . db . dialect . default_schema_name <EOL> subject = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) ) <EOL> referer = Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) ) <EOL> meta1 . create_all ( ) <EOL> meta2 = MetaData ( testing . db ) <EOL> subject = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = default_schema , <EOL> postgresql_ignore_search_path = True <EOL> ) <EOL> referer = Table ( '<STR_LIT>' , meta2 , autoload = True , <EOL> schema = default_schema , <EOL> postgresql_ignore_search_path = True <EOL> ) <EOL> assert subject . schema == default_schema <EOL> self . assert_ ( ( subject . c . id <EOL> == referer . c . ref ) . compare ( <EOL> subject . join ( referer ) . onclause ) ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_six ( self ) : <EOL> meta1 = self . metadata <EOL> Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> schema = '<STR_LIT>' <EOL> ) <EOL> Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> schema = '<STR_LIT>' <EOL> ) <EOL> meta1 . create_all ( ) <EOL> with testing . db . connect ( ) as conn : <EOL> conn . detach ( ) <EOL> conn . execute ( <EOL> "<STR_LIT>" ) <EOL> m1 = MetaData ( conn ) <EOL> t1_schema = Table ( '<STR_LIT>' , <EOL> m1 , <EOL> schema = "<STR_LIT>" , <EOL> autoload = True ) <EOL> t2_schema = Table ( '<STR_LIT>' , <EOL> m1 , <EOL> schema = "<STR_LIT>" , <EOL> autoload = True ) <EOL> t2_no_schema = Table ( '<STR_LIT>' , <EOL> m1 , <EOL> autoload = True ) <EOL> t1_no_schema = Table ( '<STR_LIT>' , <EOL> m1 , <EOL> autoload = True ) <EOL> m2 = MetaData ( conn ) <EOL> t1_schema_isp = Table ( '<STR_LIT>' , <EOL> m2 , <EOL> schema = "<STR_LIT>" , <EOL> autoload = True , <EOL> postgresql_ignore_search_path = True ) <EOL> t2_schema_isp = Table ( '<STR_LIT>' , <EOL> m2 , <EOL> schema = "<STR_LIT>" , <EOL> autoload = True , <EOL> postgresql_ignore_search_path = True ) <EOL> assert t2_schema . c . sid . references ( <EOL> t1_no_schema . c . id ) <EOL> assert t2_no_schema . c . sid . references ( <EOL> t1_no_schema . c . id ) <EOL> assert t2_schema_isp . c . sid . references ( t1_schema_isp . c . id ) <EOL> @ testing . provide_metadata <EOL> def test_cross_schema_reflection_seven ( self ) : <EOL> meta1 = self . metadata <EOL> Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> schema = '<STR_LIT>' <EOL> ) <EOL> Table ( '<STR_LIT>' , meta1 , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> schema = '<STR_LIT>' <EOL> ) <EOL> meta1 . create_all ( ) <EOL> with testing . db . connect ( ) as conn : <EOL> conn . detach ( ) <EOL> conn . execute ( <EOL> "<STR_LIT>" ) <EOL> meta2 = MetaData ( conn ) <EOL> meta2 . reflect ( schema = "<STR_LIT>" ) <EOL> eq_ ( set ( meta2 . tables ) , set ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> meta3 = MetaData ( conn ) <EOL> meta3 . reflect ( <EOL> schema = "<STR_LIT>" , postgresql_ignore_search_path = True ) <EOL> eq_ ( set ( meta3 . tables ) , set ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> @ testing . provide_metadata <EOL> def test_uppercase_lowercase_table ( self ) : <EOL> metadata = self . metadata <EOL> a_table = Table ( '<STR_LIT:a>' , metadata , Column ( '<STR_LIT:x>' , Integer ) ) <EOL> A_table = Table ( '<STR_LIT:A>' , metadata , Column ( '<STR_LIT:x>' , Integer ) ) <EOL> a_table . create ( ) <EOL> assert testing . db . has_table ( "<STR_LIT:a>" ) <EOL> assert not testing . db . has_table ( "<STR_LIT:A>" ) <EOL> A_table . create ( checkfirst = True ) <EOL> assert testing . db . has_table ( "<STR_LIT:A>" ) <EOL> def test_uppercase_lowercase_sequence ( self ) : <EOL> a_seq = Sequence ( '<STR_LIT:a>' ) <EOL> A_seq = Sequence ( '<STR_LIT:A>' ) <EOL> a_seq . create ( testing . db ) <EOL> assert testing . db . dialect . has_sequence ( testing . db , "<STR_LIT:a>" ) <EOL> assert not testing . db . dialect . has_sequence ( testing . db , "<STR_LIT:A>" ) <EOL> A_seq . create ( testing . db , checkfirst = True ) <EOL> assert testing . db . dialect . has_sequence ( testing . db , "<STR_LIT:A>" ) <EOL> a_seq . drop ( testing . db ) <EOL> A_seq . drop ( testing . db ) <EOL> @ testing . provide_metadata <EOL> def test_index_reflection ( self ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . metadata <EOL> t1 = Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( <EOL> '<STR_LIT:id>' , String ( <NUM_LIT:10> ) , nullable = False ) , <EOL> Column ( <EOL> '<STR_LIT:name>' , String ( <NUM_LIT:20> ) , index = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , String ( <NUM_LIT:20> ) ) ) <EOL> metadata . create_all ( ) <EOL> testing . db . execute ( """<STR_LIT>""" ) <EOL> testing . db . execute ( """<STR_LIT>""" ) <EOL> testing . db . execute ( """<STR_LIT>""" ) <EOL> def go ( ) : <EOL> m2 = MetaData ( testing . db ) <EOL> t2 = Table ( '<STR_LIT>' , m2 , autoload = True ) <EOL> assert len ( t2 . indexes ) == <NUM_LIT:2> <EOL> tmp = [ ( idx . name , idx ) for idx in t2 . indexes ] <EOL> tmp . sort ( ) <EOL> r1 , r2 = [ idx [ <NUM_LIT:1> ] for idx in tmp ] <EOL> assert r1 . name == '<STR_LIT>' <EOL> assert r1 . unique == True <EOL> assert r2 . unique == False <EOL> assert [ t2 . c . id ] == r1 . columns <EOL> assert [ t2 . c . name ] == r2 . columns <EOL> testing . assert_warnings ( <EOL> go , <EOL> [ '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] ) <EOL> @ testing . provide_metadata <EOL> def test_index_reflection_modified ( self ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . metadata <EOL> t1 = Table ( '<STR_LIT:t>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:x>' , Integer ) <EOL> ) <EOL> metadata . create_all ( ) <EOL> conn = testing . db . connect ( ) . execution_options ( autocommit = True ) <EOL> conn . execute ( "<STR_LIT>" ) <EOL> conn . execute ( "<STR_LIT>" ) <EOL> ind = testing . db . dialect . get_indexes ( conn , "<STR_LIT:t>" , None ) <EOL> eq_ ( ind , [ { '<STR_LIT>' : False , '<STR_LIT>' : [ '<STR_LIT:y>' ] , '<STR_LIT:name>' : '<STR_LIT>' } ] ) <EOL> conn . close ( ) <EOL> @ testing . fails_if ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> @ testing . provide_metadata <EOL> def test_index_reflection_with_storage_options ( self ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . metadata <EOL> Table ( <EOL> '<STR_LIT:t>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:x>' , Integer ) <EOL> ) <EOL> metadata . create_all ( ) <EOL> with testing . db . connect ( ) . execution_options ( autocommit = True ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> ind = testing . db . dialect . get_indexes ( conn , "<STR_LIT:t>" , None ) <EOL> eq_ ( ind , [ { '<STR_LIT>' : False , '<STR_LIT>' : [ '<STR_LIT:x>' ] , '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } } ] ) <EOL> m = MetaData ( ) <EOL> t1 = Table ( '<STR_LIT:t>' , m , autoload_with = conn ) <EOL> eq_ ( <EOL> list ( t1 . indexes ) [ <NUM_LIT:0> ] . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> { "<STR_LIT>" : "<STR_LIT>" } <EOL> ) <EOL> @ testing . provide_metadata <EOL> def test_index_reflection_with_access_method ( self ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . metadata <EOL> Table ( <EOL> '<STR_LIT:t>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:x>' , ARRAY ( Integer ) ) <EOL> ) <EOL> metadata . create_all ( ) <EOL> with testing . db . connect ( ) . execution_options ( autocommit = True ) as conn : <EOL> conn . execute ( "<STR_LIT>" ) <EOL> ind = testing . db . dialect . get_indexes ( conn , "<STR_LIT:t>" , None ) <EOL> eq_ ( ind , [ { '<STR_LIT>' : False , '<STR_LIT>' : [ '<STR_LIT:x>' ] , '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } ] ) <EOL> m = MetaData ( ) <EOL> t1 = Table ( '<STR_LIT:t>' , m , autoload_with = conn ) <EOL> eq_ ( <EOL> list ( t1 . indexes ) [ <NUM_LIT:0> ] . dialect_options [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' <EOL> ) <EOL> @ testing . provide_metadata <EOL> def test_foreign_key_option_inspection ( self ) : <EOL> metadata = self . metadata <EOL> Table ( <EOL> '<STR_LIT>' , <EOL> metadata , <EOL> Column ( <EOL> '<STR_LIT:id>' , <EOL> String ( <EOL> length = <NUM_LIT:32> ) , <EOL> nullable = False , <EOL> primary_key = True ) , <EOL> Column ( <EOL> '<STR_LIT>' , <EOL> ForeignKey ( <EOL> '<STR_LIT>' , <EOL> name = '<STR_LIT>' , <EOL> match = '<STR_LIT>' , <EOL> onupdate = '<STR_LIT>' , <EOL> ondelete = '<STR_LIT>' , <EOL> deferrable = True , <EOL> initially = '<STR_LIT>' ) ) ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , String ( length = <NUM_LIT:32> ) , nullable = False , primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( length = <NUM_LIT:255> ) ) , <EOL> Column ( <EOL> '<STR_LIT>' , <EOL> ForeignKey ( <EOL> '<STR_LIT>' , <EOL> name = '<STR_LIT>' , <EOL> onupdate = '<STR_LIT>' , ondelete = '<STR_LIT>' , <EOL> deferrable = False , <EOL> initially = '<STR_LIT>' <EOL> ) <EOL> ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer ( ) , nullable = False , primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( length = <NUM_LIT:255> ) ) <EOL> ) <EOL> fk_ref = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None <EOL> } <EOL> } <EOL> } <EOL> metadata . create_all ( ) <EOL> inspector = inspect ( testing . db ) <EOL> fks = inspector . get_foreign_keys ( '<STR_LIT>' ) + inspector . get_foreign_keys ( '<STR_LIT>' ) <EOL> for fk in fks : <EOL> eq_ ( fk , fk_ref [ fk [ '<STR_LIT:name>' ] ] ) <EOL> @ testing . provide_metadata <EOL> def test_inspect_enums_schema ( self ) : <EOL> conn = testing . db . connect ( ) <EOL> enum_type = postgresql . ENUM ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , name = '<STR_LIT>' , <EOL> schema = '<STR_LIT>' , <EOL> metadata = self . metadata ) <EOL> enum_type . create ( conn ) <EOL> inspector = reflection . Inspector . from_engine ( conn . engine ) <EOL> eq_ ( <EOL> inspector . get_enums ( '<STR_LIT>' ) , [ { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> } ] ) <EOL> @ testing . provide_metadata <EOL> def test_inspect_enums ( self ) : <EOL> enum_type = postgresql . ENUM ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , name = '<STR_LIT>' , metadata = self . metadata ) <EOL> enum_type . create ( testing . db ) <EOL> inspector = reflection . Inspector . from_engine ( testing . db ) <EOL> eq_ ( inspector . get_enums ( ) , [ <EOL> { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ] ) <EOL> @ testing . provide_metadata <EOL> def test_inspect_enums_star ( self ) : <EOL> enum_type = postgresql . ENUM ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , name = '<STR_LIT>' , metadata = self . metadata ) <EOL> schema_enum_type = postgresql . ENUM ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , name = '<STR_LIT>' , <EOL> schema = '<STR_LIT>' , <EOL> metadata = self . metadata ) <EOL> enum_type . create ( testing . db ) <EOL> schema_enum_type . create ( testing . db ) <EOL> inspector = reflection . Inspector . from_engine ( testing . db ) <EOL> eq_ ( inspector . get_enums ( ) , [ <EOL> { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ] ) <EOL> eq_ ( inspector . get_enums ( '<STR_LIT:*>' ) , [ <EOL> { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> } ] ) <EOL> @ testing . provide_metadata <EOL> @ testing . only_on ( "<STR_LIT>" ) <EOL> def test_reflection_with_unique_constraint ( self ) : <EOL> insp = inspect ( testing . db ) <EOL> meta = self . metadata <EOL> uc_table = Table ( '<STR_LIT>' , meta , <EOL> Column ( '<STR_LIT:a>' , String ( <NUM_LIT:10> ) ) , <EOL> UniqueConstraint ( '<STR_LIT:a>' , name = '<STR_LIT>' ) ) <EOL> uc_table . create ( ) <EOL> indexes = set ( i [ '<STR_LIT:name>' ] for i in insp . get_indexes ( '<STR_LIT>' ) ) <EOL> constraints = set ( i [ '<STR_LIT:name>' ] <EOL> for i in insp . get_unique_constraints ( '<STR_LIT>' ) ) <EOL> self . assert_ ( '<STR_LIT>' in indexes ) <EOL> self . assert_ ( '<STR_LIT>' in constraints ) <EOL> reflected = Table ( '<STR_LIT>' , MetaData ( testing . db ) , autoload = True ) <EOL> indexes = set ( i . name for i in reflected . indexes ) <EOL> constraints = set ( uc . name for uc in reflected . constraints ) <EOL> self . assert_ ( '<STR_LIT>' not in indexes ) <EOL> self . assert_ ( '<STR_LIT>' in constraints ) <EOL> @ testing . provide_metadata <EOL> def test_reflect_unique_index ( self ) : <EOL> insp = inspect ( testing . db ) <EOL> meta = self . metadata <EOL> uc_table = Table ( '<STR_LIT>' , meta , <EOL> Column ( '<STR_LIT:a>' , String ( <NUM_LIT:10> ) ) , <EOL> Index ( '<STR_LIT>' , '<STR_LIT:a>' , unique = True ) ) <EOL> uc_table . create ( ) <EOL> indexes = dict ( ( i [ '<STR_LIT:name>' ] , i ) for i in insp . get_indexes ( '<STR_LIT>' ) ) <EOL> constraints = set ( i [ '<STR_LIT:name>' ] <EOL> for i in insp . get_unique_constraints ( '<STR_LIT>' ) ) <EOL> self . assert_ ( '<STR_LIT>' in indexes ) <EOL> assert indexes [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assert_ ( '<STR_LIT>' not in constraints ) <EOL> reflected = Table ( '<STR_LIT>' , MetaData ( testing . db ) , autoload = True ) <EOL> indexes = dict ( ( i . name , i ) for i in reflected . indexes ) <EOL> constraints = set ( uc . name for uc in reflected . constraints ) <EOL> self . assert_ ( '<STR_LIT>' in indexes ) <EOL> assert indexes [ '<STR_LIT>' ] . unique <EOL> self . assert_ ( '<STR_LIT>' not in constraints ) <EOL> class CustomTypeReflectionTest ( fixtures . TestBase ) : <EOL> class CustomType ( object ) : <EOL> def __init__ ( self , arg1 = None , arg2 = None ) : <EOL> self . arg1 = arg1 <EOL> self . arg2 = arg2 <EOL> ischema_names = None <EOL> def setup ( self ) : <EOL> ischema_names = postgresql . PGDialect . ischema_names <EOL> postgresql . PGDialect . ischema_names = ischema_names . copy ( ) <EOL> self . ischema_names = ischema_names <EOL> def teardown ( self ) : <EOL> postgresql . PGDialect . ischema_names = self . ischema_names <EOL> self . ischema_names = None <EOL> def _assert_reflected ( self , dialect ) : <EOL> for sch , args in [ <EOL> ( '<STR_LIT>' , ( None , None ) ) , <EOL> ( '<STR_LIT>' , ( None , None ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , None ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ] : <EOL> column_info = dialect . _get_column_info ( <EOL> '<STR_LIT>' , sch , None , False , <EOL> { } , { } , '<STR_LIT>' ) <EOL> assert isinstance ( column_info [ '<STR_LIT:type>' ] , self . CustomType ) <EOL> eq_ ( column_info [ '<STR_LIT:type>' ] . arg1 , args [ <NUM_LIT:0> ] ) <EOL> eq_ ( column_info [ '<STR_LIT:type>' ] . arg2 , args [ <NUM_LIT:1> ] ) <EOL> def test_clslevel ( self ) : <EOL> postgresql . PGDialect . ischema_names [ '<STR_LIT>' ] = self . CustomType <EOL> dialect = postgresql . PGDialect ( ) <EOL> self . _assert_reflected ( dialect ) <EOL> def test_instancelevel ( self ) : <EOL> dialect = postgresql . PGDialect ( ) <EOL> dialect . ischema_names = dialect . ischema_names . copy ( ) <EOL> dialect . ischema_names [ '<STR_LIT>' ] = self . CustomType <EOL> self . _assert_reflected ( dialect ) </s>
