Как протестировать MySQL-запросы, используя sqlalchemy и sqlite?

Question

У меня есть следующая структура кода, написанная на Python3.6, которую мне нужно протестировать с использованием sqlite3 (из-за стандартов, определенных в моем проекте):

class BigSecretService:
    """ Class designed to make calculations based on data stored in MySQL. """
    def load_data(self):
        # load some data using sqlalchemy ORM

    def get_values_from_fields(self, fields):
        # here's getting values via sqlalchemy execute with raw query:
        self.sql_service.execute(SOME_QUERY)

    def process_data(self, data, values):
        # again execute some raw query
        # process data and put into result list
        return reuslt_list

    def make_calculations(self, params):
        data = self.load_data()
        values = self.get_values_from_fields(fields)
        result_vector = process_data(data, values)

SOME_QUERY находится в отдельном модуле, и его формат выглядит следующим образом:

"SELECT SUM(some_field) FROM some_table WHERE col1 = :col1 AND col2 = :col2"

Чтобы покрыть make_calculations в моем тесте компонентов, я разработал ужасные патчи:

class PatchedConnection:                                                                                                   
""" Class is used to transform queries to sqlite format before executing. """                                          
def __init__(self, connection, engine):                                                                                
    self.connection = connection                                                                                       
    self.engine = engine                                                                                               

def __call__(self):                                                                                                    
    conn = self.connection()                                                                                           
    conn.execute = self.patched_execute(conn.execute)                                                                  
    return conn                                                                                                        

def transform_date(self, date):                                                                                        
    try:                                                                                                               
        # quick check just for testing                                                                                 
        if '+00:00' in date:                                                                                           
            date = date.replace('T', ' ').replace('+00:00', '.000000')                                                 
    finally:                                                                                                           
        return date                                                                                                    

def patched_execute(self, f_execute):                                                                                  
    def prepare_args_for_sqlite(query, *args):                                                                         
        # check if query is in sqlite format                                                                           
        if args:                                                                                                       
            if '?' in str(query):                                                                                      
                args = list(map(self.transform_date, list(args[0].values())))                                          
                return self.engine.execute(str(query), args)                                                           
            return f_execute(query, args[0])                                                                           
        else:                                                                                                          
            return f_execute(query)                                                                                    
    return prepare_args_for_sqlite

Тогда в тесте это выглядит так:

   QUERY_TEMPLATE_SQLITE = 'SELECT SUM(some_field) FROM some_table WHERE col1 = ? AND col2 = ?'
   with mock.patch('path_to_my_service.SOME_QUERY', QUERY_TEMPLATE_SQLITE):           
      self.sql_service.get_connection = PatchedConnection(self.sql_service.get_connection, self.engine)               
      response = self.client.simulate_post("/v1/secret_service/make_calculations",                      
                                           headers=self.auth_header,                                          
                                           body=json.dumps(payload))                                          

      self.assertEqual(response.status_code, 200)    
      # then check response.text

Пока это работает, но я считаю, что должно быть гораздо лучшее решение. Более того, в patched_execute аргументы из dict конвертируются в список, и кто знает, будет ли порядок значений dict все время одинаковым. Итак, мой вопрос, как правильно выполнить такое тестирование с помощью данных инструментов?

Answer 1

Если вам нужно перехватить и обработать SQL, отправляемый в базу данных, то использование основных событий https://docs.sqlalchemy.org/en/13/core/events.html будет самым простым способом сделать это.Событие before_cursor_execute будет соответствовать вашим целям, как описано в следующем примере из документации по SQLAlchemy.

@event.listens_for(engine, "before_cursor_execute", retval=True)
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    # do something with statement, parameters
    return statement, parameters

Однако из приведенного вами примера я не уверен, что это необходимо.Запрос MySQL, который вы перечислили, также является допустимым запросом SQLite и не требует никаких манипуляций.Кроме того, если вы передадите свои параметры как объекты python, а не как строки, то опять-таки не нужно никаких манипуляций, так как SQLAlchemy правильно отобразит их в бэкэнд.