📚 ProTestingInfo 🔷 Канал по… on Telegram: “Как тестировать SQL-функции и как в будущем это…”

Как тестировать SQL-функции и как в будущем это автоматизировать?

Такой вопрос мне переадресовала Надя.

Я работал в финтехе на легаси-проекте, где вся логика была построена на хранимых процедурах. Самый большой минус в том, что тестирование на этом уровне не развито в отрасли и представляет довольно часто ряд костылей, одним из которых я поделюсь. Я пробовал писать тесты прямо на синтаксисе SQL, где результаты просто выводились обычными принтами True или False (прошел/не прошел). Это было интересно, но помогало только решению конкретной задачи, и данный подход был нерасширяемым.

Ну ладно, хватит слов, давайте расскажу свой костыль. На мой взгляд, удобнее всего это сделать на Python, на примере MS SQL Server.

Погнали!
С помощью pyodbc сначала я пишу простой клиент, который будет логгировать каждый запрос и преобразовывать его в список словарей для удобства работы с данными по колонкам.

import pyodbc
import os
import structlog

class BaseClient:
    def __init__(self, server, database, user, password, autocommit=False, logging_level=logging.DEBUG):
        self.log = structlog.get_logger(self.__class__.__name__).bind(service='db')
        system_name = os.name
        if system_name == 'posix':
            driver = 'ODBC Driver 17 for SQL Server'
        else:
            driver = 'SQL Server'

        connection_string = f'DRIVER={{{driver}}};' \
                            f'SERVER={server};' \
                            f'DATABASE={database};' \
                            f'UID={user};' \
                            f'PWD={password};'\
                            f'Trusted_Connection={trusted_connection}'

        with pyodbc.connect(connection_string, autocommit=self.autocommit, timeout=30) as connect:
            self.connect = connect
            self.cursor = connect.cursor()

    def logged_query(self, query):
        log = self.log.bind(request_id=str(uuid.uuid4()))
        print(f'\n{query}')
        log.msg(
            'request',
            caller=inspect.stack()[1][3],
            func_name=inspect.stack()[0][3]
        )
        result = self._to_dict(self.cursor.execute(query))
        if result:
            log.msg(
                'response',
                result=result,
            )
        return result

    @staticmethod
    def _to_dict(cursor):
        try:
            columns = [column[0] for column in cursor.description]
        except TypeError:
            return []
        else:
            dataset = [dict(zip(columns, row)) for row in cursor.fetchall()]
        return dataset

После того как базовый клиент написан, можно уже реализовать класс с вызовами хранимых процедур. Можно отнаследовать его от базового класса, чтобы использовать его методы и реализовать вызов.

class ApiTestProcedures(BaseClient):
    def some_procedure_execute(self, param1, param2):
        """
        SET NOCOUNT ON - это писать обязательно, иначе не будет работать))
        """
        query = f"""
        SET NOCOUNT ON;
        EXEC [test].[SomeProcedure]
        @Parameter1 = {param1}
       ,@Parameter2 = {param2}"""
        postings = self.logged_query(query)
        return postings

Вся подготовка сделана теперь можно писать тест, тут уже как обычно.

@pytest.fixture(scope="session")
def db_client():
    client = ApiTestProcedures("тут параметры подключения")
    yield client
    client.close()


def test_some_procedure(db_client):
    result = db_client.some_procedure_execute(param1=1, param2=2)
    for row in result:
        assert row["column_name"] == "column_value"

Вы можете использовать другие библиотеки для сериализации данных, например, SQLAlchemy или Pandas для удобства работы и анализа данных. Для этого нужно вместо метода _to_dict реализовать свои функции упаковки данных.

Пишите вопросы в тред, если возникнут интересные темы, постараюсь ответить в новом посте!