<s> from __future__ import absolute_import <EOL> from . import util as testutil <EOL> from sqlalchemy import pool , orm , util <EOL> from sqlalchemy . engine import default , url <EOL> from sqlalchemy . util import decorator <EOL> from sqlalchemy import types as sqltypes , schema , exc as sa_exc <EOL> import warnings <EOL> import re <EOL> from . exclusions import db_spec , _is_excluded <EOL> from . import assertsql <EOL> from . import config <EOL> from . util import fail <EOL> import contextlib <EOL> from . import mock <EOL> def expect_warnings ( * messages , ** kw ) : <EOL> """<STR_LIT>""" <EOL> return _expect_warnings ( sa_exc . SAWarning , messages , ** kw ) <EOL> @ contextlib . contextmanager <EOL> def expect_warnings_on ( db , * messages , ** kw ) : <EOL> """<STR_LIT>""" <EOL> spec = db_spec ( db ) <EOL> if isinstance ( db , util . string_types ) and not spec ( config . _current ) : <EOL> yield <EOL> else : <EOL> with expect_warnings ( * messages , ** kw ) : <EOL> yield <EOL> def emits_warning ( * messages ) : <EOL> """<STR_LIT>""" <EOL> @ decorator <EOL> def decorate ( fn , * args , ** kw ) : <EOL> with expect_warnings ( assert_ = False , * messages ) : <EOL> return fn ( * args , ** kw ) <EOL> return decorate <EOL> def expect_deprecated ( * messages , ** kw ) : <EOL> return _expect_warnings ( sa_exc . SADeprecationWarning , messages , ** kw ) <EOL> def emits_warning_on ( db , * messages ) : <EOL> """<STR_LIT>""" <EOL> @ decorator <EOL> def decorate ( fn , * args , ** kw ) : <EOL> with expect_warnings_on ( db , assert_ = False , * messages ) : <EOL> return fn ( * args , ** kw ) <EOL> return decorate <EOL> def uses_deprecated ( * messages ) : <EOL> """<STR_LIT>""" <EOL> @ decorator <EOL> def decorate ( fn , * args , ** kw ) : <EOL> with expect_deprecated ( * messages , assert_ = False ) : <EOL> return fn ( * args , ** kw ) <EOL> return decorate <EOL> @ contextlib . contextmanager <EOL> def _expect_warnings ( exc_cls , messages , regex = True , assert_ = True ) : <EOL> if regex : <EOL> filters = [ re . compile ( msg , re . I | re . S ) for msg in messages ] <EOL> else : <EOL> filters = messages <EOL> seen = set ( filters ) <EOL> real_warn = warnings . warn <EOL> def our_warn ( msg , exception , * arg , ** kw ) : <EOL> if not issubclass ( exception , exc_cls ) : <EOL> return real_warn ( msg , exception , * arg , ** kw ) <EOL> if not filters : <EOL> return <EOL> for filter_ in filters : <EOL> if ( regex and filter_ . match ( msg ) ) or ( not regex and filter_ == msg ) : <EOL> seen . discard ( filter_ ) <EOL> break <EOL> else : <EOL> real_warn ( msg , exception , * arg , ** kw ) <EOL> with mock . patch ( "<STR_LIT>" , our_warn ) : <EOL> yield <EOL> if assert_ : <EOL> assert not seen , "<STR_LIT>" % "<STR_LIT:U+002CU+0020>" . join ( "<STR_LIT>" % ( s . pattern if regex else s ) for s in seen ) <EOL> def global_cleanup_assertions ( ) : <EOL> """<STR_LIT>""" <EOL> _assert_no_stray_pool_connections ( ) <EOL> _STRAY_CONNECTION_FAILURES = <NUM_LIT:0> <EOL> def _assert_no_stray_pool_connections ( ) : <EOL> global _STRAY_CONNECTION_FAILURES <EOL> testutil . lazy_gc ( ) <EOL> if pool . _refs : <EOL> _STRAY_CONNECTION_FAILURES += <NUM_LIT:1> <EOL> print ( "<STR_LIT>" <EOL> % str ( pool . _refs ) ) <EOL> testutil . gc_collect ( ) <EOL> if pool . _refs : <EOL> err = str ( pool . _refs ) <EOL> pool . _refs . clear ( ) <EOL> _STRAY_CONNECTION_FAILURES = <NUM_LIT:0> <EOL> assert False , "<STR_LIT>" "<STR_LIT>" % err <EOL> elif _STRAY_CONNECTION_FAILURES > <NUM_LIT:10> : <EOL> assert False , "<STR_LIT>" <EOL> _STRAY_CONNECTION_FAILURES = <NUM_LIT:0> <EOL> def eq_regex ( a , b , msg = None ) : <EOL> assert re . match ( b , a ) , msg or "<STR_LIT>" % ( a , b ) <EOL> def eq_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a == b , msg or "<STR_LIT>" % ( a , b ) <EOL> def ne_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a != b , msg or "<STR_LIT>" % ( a , b ) <EOL> def le_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a <= b , msg or "<STR_LIT>" % ( a , b ) <EOL> def is_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a is b , msg or "<STR_LIT>" % ( a , b ) <EOL> def is_not_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a is not b , msg or "<STR_LIT>" % ( a , b ) <EOL> def in_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a in b , msg or "<STR_LIT>" % ( a , b ) <EOL> def not_in_ ( a , b , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a not in b , msg or "<STR_LIT>" % ( a , b ) <EOL> def startswith_ ( a , fragment , msg = None ) : <EOL> """<STR_LIT>""" <EOL> assert a . startswith ( fragment ) , msg or "<STR_LIT>" % ( <EOL> a , fragment ) <EOL> def eq_ignore_whitespace ( a , b , msg = None ) : <EOL> a = re . sub ( r'<STR_LIT>' , "<STR_LIT>" , a ) <EOL> a = re . sub ( r'<STR_LIT>' , "<STR_LIT:U+0020>" , a ) <EOL> b = re . sub ( r'<STR_LIT>' , "<STR_LIT>" , b ) <EOL> b = re . sub ( r'<STR_LIT>' , "<STR_LIT:U+0020>" , b ) <EOL> assert a == b , msg or "<STR_LIT>" % ( a , b ) <EOL> def assert_raises ( except_cls , callable_ , * args , ** kw ) : <EOL> try : <EOL> callable_ ( * args , ** kw ) <EOL> success = False <EOL> except except_cls : <EOL> success = True <EOL> assert success , "<STR_LIT>" <EOL> def assert_raises_message ( except_cls , msg , callable_ , * args , ** kwargs ) : <EOL> try : <EOL> callable_ ( * args , ** kwargs ) <EOL> assert False , "<STR_LIT>" <EOL> except except_cls as e : <EOL> assert re . search ( <EOL> msg , util . text_type ( e ) , re . UNICODE ) , "<STR_LIT>" % ( msg , e ) <EOL> print ( util . text_type ( e ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> class AssertsCompiledSQL ( object ) : <EOL> def assert_compile ( self , clause , result , params = None , <EOL> checkparams = None , dialect = None , <EOL> checkpositional = None , <EOL> check_prefetch = None , <EOL> use_default_dialect = False , <EOL> allow_dialect_select = False , <EOL> literal_binds = False , <EOL> schema_translate_map = None ) : <EOL> if use_default_dialect : <EOL> dialect = default . DefaultDialect ( ) <EOL> elif allow_dialect_select : <EOL> dialect = None <EOL> else : <EOL> if dialect is None : <EOL> dialect = getattr ( self , '<STR_LIT>' , None ) <EOL> if dialect is None : <EOL> dialect = config . db . dialect <EOL> elif dialect == '<STR_LIT:default>' : <EOL> dialect = default . DefaultDialect ( ) <EOL> elif dialect == '<STR_LIT>' : <EOL> dialect = default . StrCompileDialect ( ) <EOL> elif isinstance ( dialect , util . string_types ) : <EOL> dialect = url . URL ( dialect ) . get_dialect ( ) ( ) <EOL> kw = { } <EOL> compile_kwargs = { } <EOL> if schema_translate_map : <EOL> kw [ '<STR_LIT>' ] = schema_translate_map <EOL> if params is not None : <EOL> kw [ '<STR_LIT>' ] = list ( params ) <EOL> if literal_binds : <EOL> compile_kwargs [ '<STR_LIT>' ] = True <EOL> if isinstance ( clause , orm . Query ) : <EOL> context = clause . _compile_context ( ) <EOL> context . statement . use_labels = True <EOL> clause = context . statement <EOL> if compile_kwargs : <EOL> kw [ '<STR_LIT>' ] = compile_kwargs <EOL> c = clause . compile ( dialect = dialect , ** kw ) <EOL> param_str = repr ( getattr ( c , '<STR_LIT>' , { } ) ) <EOL> if util . py3k : <EOL> param_str = param_str . encode ( '<STR_LIT:utf-8>' ) . decode ( '<STR_LIT:ascii>' , '<STR_LIT:ignore>' ) <EOL> print ( <EOL> ( "<STR_LIT>" + <EOL> util . text_type ( c ) + <EOL> param_str ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> print ( <EOL> "<STR_LIT>" + <EOL> util . text_type ( c ) . encode ( '<STR_LIT:utf-8>' ) + <EOL> param_str ) <EOL> cc = re . sub ( r'<STR_LIT>' , '<STR_LIT>' , util . text_type ( c ) ) <EOL> eq_ ( cc , result , "<STR_LIT>" % ( cc , result , dialect ) ) <EOL> if checkparams is not None : <EOL> eq_ ( c . construct_params ( params ) , checkparams ) <EOL> if checkpositional is not None : <EOL> p = c . construct_params ( params ) <EOL> eq_ ( tuple ( [ p [ x ] for x in c . positiontup ] ) , checkpositional ) <EOL> if check_prefetch is not None : <EOL> eq_ ( c . prefetch , check_prefetch ) <EOL> class ComparesTables ( object ) : <EOL> def assert_tables_equal ( self , table , reflected_table , strict_types = False ) : <EOL> assert len ( table . c ) == len ( reflected_table . c ) <EOL> for c , reflected_c in zip ( table . c , reflected_table . c ) : <EOL> eq_ ( c . name , reflected_c . name ) <EOL> assert reflected_c is reflected_table . c [ c . name ] <EOL> eq_ ( c . primary_key , reflected_c . primary_key ) <EOL> eq_ ( c . nullable , reflected_c . nullable ) <EOL> if strict_types : <EOL> msg = "<STR_LIT>" <EOL> assert isinstance ( reflected_c . type , type ( c . type ) ) , msg % ( reflected_c . type , c . type ) <EOL> else : <EOL> self . assert_types_base ( reflected_c , c ) <EOL> if isinstance ( c . type , sqltypes . String ) : <EOL> eq_ ( c . type . length , reflected_c . type . length ) <EOL> eq_ ( <EOL> set ( [ f . column . name for f in c . foreign_keys ] ) , <EOL> set ( [ f . column . name for f in reflected_c . foreign_keys ] ) <EOL> ) <EOL> if c . server_default : <EOL> assert isinstance ( reflected_c . server_default , <EOL> schema . FetchedValue ) <EOL> assert len ( table . primary_key ) == len ( reflected_table . primary_key ) <EOL> for c in table . primary_key : <EOL> assert reflected_table . primary_key . columns [ c . name ] is not None <EOL> def assert_types_base ( self , c1 , c2 ) : <EOL> assert c1 . type . _compare_type_affinity ( c2 . type ) , "<STR_LIT>" % ( c1 . name , c1 . type , c2 . type ) <EOL> class AssertsExecutionResults ( object ) : <EOL> def assert_result ( self , result , class_ , * objects ) : <EOL> result = list ( result ) <EOL> print ( repr ( result ) ) <EOL> self . assert_list ( result , class_ , objects ) <EOL> def assert_list ( self , result , class_ , list ) : <EOL> self . assert_ ( len ( result ) == len ( list ) , <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + class_ . __name__ ) <EOL> for i in range ( <NUM_LIT:0> , len ( list ) ) : <EOL> self . assert_row ( class_ , result [ i ] , list [ i ] ) <EOL> def assert_row ( self , class_ , rowobj , desc ) : <EOL> self . assert_ ( rowobj . __class__ is class_ , <EOL> "<STR_LIT>" + repr ( class_ ) ) <EOL> for key , value in desc . items ( ) : <EOL> if isinstance ( value , tuple ) : <EOL> if isinstance ( value [ <NUM_LIT:1> ] , list ) : <EOL> self . assert_list ( getattr ( rowobj , key ) , value [ <NUM_LIT:0> ] , value [ <NUM_LIT:1> ] ) <EOL> else : <EOL> self . assert_row ( value [ <NUM_LIT:0> ] , getattr ( rowobj , key ) , value [ <NUM_LIT:1> ] ) <EOL> else : <EOL> self . assert_ ( getattr ( rowobj , key ) == value , <EOL> "<STR_LIT>" % ( <EOL> key , getattr ( rowobj , key ) , value ) ) <EOL> def assert_unordered_result ( self , result , cls , * expected ) : <EOL> """<STR_LIT>""" <EOL> class immutabledict ( dict ) : <EOL> def __hash__ ( self ) : <EOL> return id ( self ) <EOL> found = util . IdentitySet ( result ) <EOL> expected = set ( [ immutabledict ( e ) for e in expected ] ) <EOL> for wrong in util . itertools_filterfalse ( lambda o : <EOL> isinstance ( o , cls ) , found ) : <EOL> fail ( '<STR_LIT>' % ( <EOL> type ( wrong ) . __name__ , cls . __name__ ) ) <EOL> if len ( found ) != len ( expected ) : <EOL> fail ( '<STR_LIT>' % ( <EOL> len ( found ) , len ( expected ) ) ) <EOL> NOVALUE = object ( ) <EOL> def _compare_item ( obj , spec ) : <EOL> for key , value in spec . items ( ) : <EOL> if isinstance ( value , tuple ) : <EOL> try : <EOL> self . assert_unordered_result ( <EOL> getattr ( obj , key ) , value [ <NUM_LIT:0> ] , * value [ <NUM_LIT:1> ] ) <EOL> except AssertionError : <EOL> return False <EOL> else : <EOL> if getattr ( obj , key , NOVALUE ) != value : <EOL> return False <EOL> return True <EOL> for expected_item in expected : <EOL> for found_item in found : <EOL> if _compare_item ( found_item , expected_item ) : <EOL> found . remove ( found_item ) <EOL> break <EOL> else : <EOL> fail ( <EOL> "<STR_LIT>" % ( <EOL> cls . __name__ , repr ( expected_item ) ) ) <EOL> return True <EOL> def sql_execution_asserter ( self , db = None ) : <EOL> if db is None : <EOL> from . import db as db <EOL> return assertsql . assert_engine ( db ) <EOL> def assert_sql_execution ( self , db , callable_ , * rules ) : <EOL> with self . sql_execution_asserter ( db ) as asserter : <EOL> callable_ ( ) <EOL> asserter . assert_ ( * rules ) <EOL> def assert_sql ( self , db , callable_ , rules ) : <EOL> newrules = [ ] <EOL> for rule in rules : <EOL> if isinstance ( rule , dict ) : <EOL> newrule = assertsql . AllOf ( * [ <EOL> assertsql . CompiledSQL ( k , v ) for k , v in rule . items ( ) <EOL> ] ) <EOL> else : <EOL> newrule = assertsql . CompiledSQL ( * rule ) <EOL> newrules . append ( newrule ) <EOL> self . assert_sql_execution ( db , callable_ , * newrules ) <EOL> def assert_sql_count ( self , db , callable_ , count ) : <EOL> self . assert_sql_execution ( <EOL> db , callable_ , assertsql . CountStatements ( count ) ) <EOL> @ contextlib . contextmanager <EOL> def assert_execution ( self , * rules ) : <EOL> assertsql . asserter . add_rules ( rules ) <EOL> try : <EOL> yield <EOL> assertsql . asserter . statement_complete ( ) <EOL> finally : <EOL> assertsql . asserter . clear_rules ( ) <EOL> def assert_statement_count ( self , count ) : <EOL> return self . assert_execution ( assertsql . CountStatements ( count ) ) </s>
