def mock_engine(dialect_name=None):
 from sqlalchemy import create_engine
 if (not dialect_name):
  dialect_name = config.db.name
 buffer = []
 def executor(sql, *a, **kw):
  buffer.append(sql)
 def assert_sql(stmts):
  recv = [re.sub('[\\n\\t]', '', str(s)) for s in buffer]
  assert (recv == stmts), recv
 def print_sql():
  d = engine.dialect
  return '\n'.join((str(s.compile(dialect=d)) for s in engine.mock))
 engine = create_engine((dialect_name + '://'), strategy='mock', executor=executor)
 assert (not hasattr(engine, 'mock'))
 engine.mock = buffer
 engine.assert_sql = assert_sql
 engine.print_sql = print_sql
 return engine

def mock_engine ( dialect_name = None ) : from sqlalchemy import create_engine if ( not dialect_name ) : dialect_name = config . db . name buffer = [ ] def executor ( sql , * a , ** kw ) : buffer . append ( sql ) def assert_sql ( stmts ) : recv = [ re . sub ( '[\\n\\t]' , '' , str ( s ) ) for s in buffer ] assert ( recv == stmts ) , recv def print_sql ( ) : d = engine . dialect return '\n' . join ( ( str ( s . compile ( dialect = d ) ) for s in engine . mock ) ) engine = create_engine ( ( dialect_name + '://' ) , strategy = 'mock' , executor = executor ) assert ( not hasattr ( engine , 'mock' ) ) engine . mock = buffer engine . assert_sql = assert_sql engine . print_sql = print_sql return engine
provides a mocking engine based on the current testing .