#NEWLINE# def testing_engine( url = None , options = None ) : #NEWLINE# #INDENT# 'Produce an engine configured by --options with optional overrides.' #NEWLINE# from sqlalchemy import create_engine #NEWLINE# from . assertsql import asserter #NEWLINE# if ( not options ) : #NEWLINE# #INDENT# use_reaper = True #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# use_reaper = options . pop ( 'use_reaper' , True ) #NEWLINE# #UNINDENT# url = ( url or config . db . url ) #NEWLINE# if ( options is None ) : #NEWLINE# #INDENT# options = config . db_opts #NEWLINE# #UNINDENT# engine = create_engine ( url , ** options ) #NEWLINE# if isinstance ( engine . pool , pool . QueuePool ) : #NEWLINE# #INDENT# engine . pool . _timeout = 0 #NEWLINE# engine . pool . _max_overflow = 0 #NEWLINE# #UNINDENT# event . listen ( engine , 'after_execute' , asserter . execute ) #NEWLINE# event . listen ( engine , 'after_cursor_execute' , asserter . cursor_execute ) #NEWLINE# if use_reaper : #NEWLINE# #INDENT# event . listen ( engine . pool , 'connect' , testing_reaper . connect ) #NEWLINE# event . listen ( engine . pool , 'checkout' , testing_reaper . checkout ) #NEWLINE# event . listen ( engine . pool , 'invalidate' , testing_reaper . invalidate ) #NEWLINE# testing_reaper . add_engine ( engine ) #NEWLINE# #UNINDENT# return engine