<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import sys <EOL> import re <EOL> py3k = sys . version_info >= ( <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> if py3k : <EOL> import configparser <EOL> else : <EOL> import ConfigParser as configparser <EOL> fixtures = None <EOL> engines = None <EOL> exclusions = None <EOL> warnings = None <EOL> profiling = None <EOL> assertions = None <EOL> requirements = None <EOL> config = None <EOL> testing = None <EOL> util = None <EOL> file_config = None <EOL> logging = None <EOL> include_tags = set ( ) <EOL> exclude_tags = set ( ) <EOL> options = None <EOL> def setup_options ( make_option ) : <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , type = "<STR_LIT:string>" , callback = _log , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , <EOL> type = "<STR_LIT:string>" , callback = _log , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , type = "<STR_LIT:string>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> make_option ( '<STR_LIT>' , action = '<STR_LIT>' , callback = _list_dbs , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , type = "<STR_LIT:string>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , <EOL> dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , type = "<STR_LIT:string>" , <EOL> callback = _requirements_opt , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , <EOL> dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , callback = _include_tag , <EOL> type = "<STR_LIT:string>" , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT>" , callback = _exclude_tag , <EOL> type = "<STR_LIT:string>" , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , <EOL> dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" ) <EOL> make_option ( "<STR_LIT>" , action = "<STR_LIT:store_true>" , <EOL> dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" ) <EOL> def configure_follower ( follower_ident ) : <EOL> """<STR_LIT>""" <EOL> from sqlalchemy . testing import provision <EOL> provision . FOLLOWER_IDENT = follower_ident <EOL> def memoize_important_follower_config ( dict_ ) : <EOL> """<STR_LIT>""" <EOL> dict_ [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : include_tags , <EOL> '<STR_LIT>' : exclude_tags <EOL> } <EOL> def restore_important_follower_config ( dict_ ) : <EOL> """<STR_LIT>""" <EOL> global include_tags , exclude_tags <EOL> include_tags . update ( dict_ [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> exclude_tags . update ( dict_ [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def read_config ( ) : <EOL> global file_config <EOL> file_config = configparser . ConfigParser ( ) <EOL> file_config . read ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def pre_begin ( opt ) : <EOL> """<STR_LIT>""" <EOL> global options <EOL> options = opt <EOL> for fn in pre_configure : <EOL> fn ( options , file_config ) <EOL> def set_coverage_flag ( value ) : <EOL> options . has_coverage = value <EOL> _skip_test_exception = None <EOL> def set_skip_test ( exc ) : <EOL> global _skip_test_exception <EOL> _skip_test_exception = exc <EOL> def post_begin ( ) : <EOL> """<STR_LIT>""" <EOL> for fn in post_configure : <EOL> fn ( options , file_config ) <EOL> global util , fixtures , engines , exclusions , assertions , warnings , profiling , config , testing <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import fixtures , engines , exclusions <EOL> from sqlalchemy . testing import assertions , warnings , profiling <EOL> from sqlalchemy . testing import config <EOL> from sqlalchemy import util <EOL> warnings . setup_filters ( ) <EOL> def _log ( opt_str , value , parser ) : <EOL> global logging <EOL> if not logging : <EOL> import logging <EOL> logging . basicConfig ( ) <EOL> if opt_str . endswith ( '<STR_LIT>' ) : <EOL> logging . getLogger ( value ) . setLevel ( logging . INFO ) <EOL> elif opt_str . endswith ( '<STR_LIT>' ) : <EOL> logging . getLogger ( value ) . setLevel ( logging . DEBUG ) <EOL> def _list_dbs ( * args ) : <EOL> print ( "<STR_LIT>" ) <EOL> for macro in sorted ( file_config . options ( '<STR_LIT>' ) ) : <EOL> print ( "<STR_LIT>" % ( macro , file_config . get ( '<STR_LIT>' , macro ) ) ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> def _requirements_opt ( opt_str , value , parser ) : <EOL> _setup_requirements ( value ) <EOL> def _exclude_tag ( opt_str , value , parser ) : <EOL> exclude_tags . add ( value . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) ) <EOL> def _include_tag ( opt_str , value , parser ) : <EOL> include_tags . add ( value . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) ) <EOL> pre_configure = [ ] <EOL> post_configure = [ ] <EOL> def pre ( fn ) : <EOL> pre_configure . append ( fn ) <EOL> return fn <EOL> def post ( fn ) : <EOL> post_configure . append ( fn ) <EOL> return fn <EOL> @ pre <EOL> def _setup_options ( opt , file_config ) : <EOL> global options <EOL> options = opt <EOL> @ pre <EOL> def _monkeypatch_cdecimal ( options , file_config ) : <EOL> if options . cdecimal : <EOL> import cdecimal <EOL> sys . modules [ '<STR_LIT>' ] = cdecimal <EOL> @ post <EOL> def _init_skiptest ( options , file_config ) : <EOL> from sqlalchemy . testing import config <EOL> config . _skip_test_exception = _skip_test_exception <EOL> @ post <EOL> def _engine_uri ( options , file_config ) : <EOL> from sqlalchemy . testing import config <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import provision <EOL> if options . dburi : <EOL> db_urls = list ( options . dburi ) <EOL> else : <EOL> db_urls = [ ] <EOL> if options . db : <EOL> for db_token in options . db : <EOL> for db in re . split ( r'<STR_LIT>' , db_token ) : <EOL> if db not in file_config . options ( '<STR_LIT>' ) : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % db ) <EOL> else : <EOL> db_urls . append ( file_config . get ( '<STR_LIT>' , db ) ) <EOL> if not db_urls : <EOL> db_urls . append ( file_config . get ( '<STR_LIT>' , '<STR_LIT:default>' ) ) <EOL> for db_url in db_urls : <EOL> cfg = provision . setup_config ( <EOL> db_url , options , file_config , provision . FOLLOWER_IDENT ) <EOL> if not config . _current : <EOL> cfg . set_as_current ( cfg , testing ) <EOL> @ post <EOL> def _requirements ( options , file_config ) : <EOL> requirement_cls = file_config . get ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> _setup_requirements ( requirement_cls ) <EOL> def _setup_requirements ( argument ) : <EOL> from sqlalchemy . testing import config <EOL> from sqlalchemy import testing <EOL> if config . requirements is not None : <EOL> return <EOL> modname , clsname = argument . split ( "<STR_LIT::>" ) <EOL> mod = __import__ ( modname ) <EOL> for component in modname . split ( "<STR_LIT:.>" ) [ <NUM_LIT:1> : ] : <EOL> mod = getattr ( mod , component ) <EOL> req_cls = getattr ( mod , clsname ) <EOL> config . requirements = testing . requires = req_cls ( ) <EOL> @ post <EOL> def _prep_testing_database ( options , file_config ) : <EOL> from sqlalchemy . testing import config , util <EOL> from sqlalchemy . testing . exclusions import against <EOL> from sqlalchemy import schema , inspect <EOL> if options . dropfirst : <EOL> for cfg in config . Config . all_configs ( ) : <EOL> e = cfg . db <EOL> inspector = inspect ( e ) <EOL> try : <EOL> view_names = inspector . get_view_names ( ) <EOL> except NotImplementedError : <EOL> pass <EOL> else : <EOL> for vname in view_names : <EOL> e . execute ( schema . _DropView ( <EOL> schema . Table ( vname , schema . MetaData ( ) ) <EOL> ) ) <EOL> if config . requirements . schemas . enabled_for_config ( cfg ) : <EOL> try : <EOL> view_names = inspector . get_view_names ( <EOL> schema = "<STR_LIT>" ) <EOL> except NotImplementedError : <EOL> pass <EOL> else : <EOL> for vname in view_names : <EOL> e . execute ( schema . _DropView ( <EOL> schema . Table ( vname , schema . MetaData ( ) , <EOL> schema = "<STR_LIT>" ) <EOL> ) ) <EOL> util . drop_all_tables ( e , inspector ) <EOL> if config . requirements . schemas . enabled_for_config ( cfg ) : <EOL> util . drop_all_tables ( e , inspector , schema = cfg . test_schema ) <EOL> if against ( cfg , "<STR_LIT>" ) : <EOL> from sqlalchemy . dialects import postgresql <EOL> for enum in inspector . get_enums ( "<STR_LIT:*>" ) : <EOL> e . execute ( postgresql . DropEnumType ( <EOL> postgresql . ENUM ( <EOL> name = enum [ '<STR_LIT:name>' ] , <EOL> schema = enum [ '<STR_LIT>' ] ) ) ) <EOL> @ post <EOL> def _reverse_topological ( options , file_config ) : <EOL> if options . reversetop : <EOL> from sqlalchemy . orm . util import randomize_unitofwork <EOL> randomize_unitofwork ( ) <EOL> @ post <EOL> def _post_setup_options ( opt , file_config ) : <EOL> from sqlalchemy . testing import config <EOL> config . options = options <EOL> config . file_config = file_config <EOL> @ post <EOL> def _setup_profiling ( options , file_config ) : <EOL> from sqlalchemy . testing import profiling <EOL> profiling . _profile_stats = profiling . ProfileStatsFile ( <EOL> file_config . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def want_class ( cls ) : <EOL> if not issubclass ( cls , fixtures . TestBase ) : <EOL> return False <EOL> elif cls . __name__ . startswith ( '<STR_LIT:_>' ) : <EOL> return False <EOL> elif config . options . backend_only and not getattr ( cls , '<STR_LIT>' , <EOL> False ) : <EOL> return False <EOL> else : <EOL> return True <EOL> def want_method ( cls , fn ) : <EOL> if not fn . __name__ . startswith ( "<STR_LIT>" ) : <EOL> return False <EOL> elif fn . __module__ is None : <EOL> return False <EOL> elif include_tags : <EOL> return ( <EOL> hasattr ( cls , '<STR_LIT>' ) and <EOL> exclusions . tags ( cls . __tags__ ) . include_test ( <EOL> include_tags , exclude_tags ) <EOL> ) or ( <EOL> hasattr ( fn , '<STR_LIT>' ) and <EOL> fn . _sa_exclusion_extend . include_test ( <EOL> include_tags , exclude_tags ) <EOL> ) <EOL> elif exclude_tags and hasattr ( cls , '<STR_LIT>' ) : <EOL> return exclusions . tags ( cls . __tags__ ) . include_test ( <EOL> include_tags , exclude_tags ) <EOL> elif exclude_tags and hasattr ( fn , '<STR_LIT>' ) : <EOL> return fn . _sa_exclusion_extend . include_test ( include_tags , exclude_tags ) <EOL> else : <EOL> return True <EOL> def generate_sub_tests ( cls , module ) : <EOL> if getattr ( cls , '<STR_LIT>' , False ) : <EOL> for cfg in _possible_configs_for_cls ( cls ) : <EOL> name = "<STR_LIT>" % ( cls . __name__ , cfg . db . name , cfg . db . driver ) <EOL> subcls = type ( <EOL> name , <EOL> ( cls , ) , <EOL> { <EOL> "<STR_LIT>" : ( "<STR_LIT>" % ( cfg . db . name , cfg . db . driver ) ) , <EOL> } <EOL> ) <EOL> setattr ( module , name , subcls ) <EOL> yield subcls <EOL> else : <EOL> yield cls <EOL> def start_test_class ( cls ) : <EOL> _do_skips ( cls ) <EOL> _setup_engine ( cls ) <EOL> def stop_test_class ( cls ) : <EOL> engines . testing_reaper . _stop_test_ctx ( ) <EOL> try : <EOL> if not options . low_connections : <EOL> assertions . global_cleanup_assertions ( ) <EOL> finally : <EOL> _restore_engine ( ) <EOL> def _restore_engine ( ) : <EOL> config . _current . reset ( testing ) <EOL> def final_process_cleanup ( ) : <EOL> engines . testing_reaper . _stop_test_ctx_aggressive ( ) <EOL> assertions . global_cleanup_assertions ( ) <EOL> _restore_engine ( ) <EOL> def _setup_engine ( cls ) : <EOL> if getattr ( cls , '<STR_LIT>' , None ) : <EOL> eng = engines . testing_engine ( options = cls . __engine_options__ ) <EOL> config . _current . push_engine ( eng , testing ) <EOL> def before_test ( test , test_module_name , test_class , test_name ) : <EOL> name = test_class . __name__ <EOL> suffix = "<STR_LIT>" % ( config . db . name , config . db . driver ) <EOL> if name . endswith ( suffix ) : <EOL> name = name [ <NUM_LIT:0> : - ( len ( suffix ) ) ] <EOL> id_ = "<STR_LIT>" % ( test_module_name , name , test_name ) <EOL> profiling . _current_test = id_ <EOL> def after_test ( test ) : <EOL> engines . testing_reaper . _after_test_ctx ( ) <EOL> def _possible_configs_for_cls ( cls , reasons = None ) : <EOL> all_configs = set ( config . Config . all_configs ( ) ) <EOL> if cls . __unsupported_on__ : <EOL> spec = exclusions . db_spec ( * cls . __unsupported_on__ ) <EOL> for config_obj in list ( all_configs ) : <EOL> if spec ( config_obj ) : <EOL> all_configs . remove ( config_obj ) <EOL> if getattr ( cls , '<STR_LIT>' , None ) : <EOL> spec = exclusions . db_spec ( * util . to_list ( cls . __only_on__ ) ) <EOL> for config_obj in list ( all_configs ) : <EOL> if not spec ( config_obj ) : <EOL> all_configs . remove ( config_obj ) <EOL> if hasattr ( cls , '<STR_LIT>' ) : <EOL> requirements = config . requirements <EOL> for config_obj in list ( all_configs ) : <EOL> for requirement in cls . __requires__ : <EOL> check = getattr ( requirements , requirement ) <EOL> skip_reasons = check . matching_config_reasons ( config_obj ) <EOL> if skip_reasons : <EOL> all_configs . remove ( config_obj ) <EOL> if reasons is not None : <EOL> reasons . extend ( skip_reasons ) <EOL> break <EOL> if hasattr ( cls , '<STR_LIT>' ) : <EOL> non_preferred = set ( ) <EOL> requirements = config . requirements <EOL> for config_obj in list ( all_configs ) : <EOL> for requirement in cls . __prefer_requires__ : <EOL> check = getattr ( requirements , requirement ) <EOL> if not check . enabled_for_config ( config_obj ) : <EOL> non_preferred . add ( config_obj ) <EOL> if all_configs . difference ( non_preferred ) : <EOL> all_configs . difference_update ( non_preferred ) <EOL> return all_configs <EOL> def _do_skips ( cls ) : <EOL> reasons = [ ] <EOL> all_configs = _possible_configs_for_cls ( cls , reasons ) <EOL> if getattr ( cls , '<STR_LIT>' , False ) : <EOL> for c in getattr ( cls , '<STR_LIT>' ) : <EOL> if c ( ) : <EOL> config . skip_test ( "<STR_LIT>" % ( <EOL> cls . __name__ , c . __name__ ) <EOL> ) <EOL> if not all_configs : <EOL> if getattr ( cls , '<STR_LIT>' , False ) : <EOL> msg = "<STR_LIT>" % ( <EOL> cls . __name__ , cls . __only_on__ ) <EOL> else : <EOL> msg = "<STR_LIT>" % ( <EOL> cls . __name__ , <EOL> "<STR_LIT:U+002CU+0020>" . join ( <EOL> "<STR_LIT>" % ( <EOL> config_obj . db . name , <EOL> "<STR_LIT:.>" . join ( <EOL> str ( dig ) for dig in <EOL> config_obj . db . dialect . server_version_info ) , <EOL> config_obj . db . driver <EOL> ) <EOL> for config_obj in config . Config . all_configs ( ) <EOL> ) , <EOL> "<STR_LIT:U+002CU+0020>" . join ( reasons ) <EOL> ) <EOL> config . skip_test ( msg ) <EOL> elif hasattr ( cls , '<STR_LIT>' ) : <EOL> non_preferred = set ( ) <EOL> spec = exclusions . db_spec ( * util . to_list ( cls . __prefer_backends__ ) ) <EOL> for config_obj in all_configs : <EOL> if not spec ( config_obj ) : <EOL> non_preferred . add ( config_obj ) <EOL> if all_configs . difference ( non_preferred ) : <EOL> all_configs . difference_update ( non_preferred ) <EOL> if config . _current not in all_configs : <EOL> _setup_config ( all_configs . pop ( ) , cls ) <EOL> def _setup_config ( config_obj , ctx ) : <EOL> config . _current . push ( config_obj , testing ) </s>
