Транзакционное юнит-тестирование приложений с БД

В современном мире множество приложений используют трехуровневую архитектуру с базой данных в слоях данных. Наличие юнит-тестов обычно упрощает поддержку продукта, но присутствие базы данных в архитектуре заставляет разработчиков применять смекалку.

В этой статье я хочу провести обзор разных способов юнит-тестирования приложения с БД и рассказать о способе, который я не видел в русскоязычном сегменте интернета. Статья будет посвящена Python 3, pytest и ORM-фреймворку SQLAlchemy, но методы переносимы на другие инструменты.

Окружение

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

from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()


class Server(Base):
    __tablename__ = 'example_server'

    id = Column(Integer, primary_key=True)

    ip = Column(String, nullable=False)
    hostname = Column(String, nullable=False)
    power_on = Column(Boolean, server_default='False')

В качестве тестируемой функции предложим ту, что выключает сервер. Сделаем ее простой и странной:

from sqlalchemy.orm import Session

def power_off(session: Session, server: Server) -> bool:
    """
    Method tries to power off server
    :param server:
    :return: True if success, False otherwise
    """
    if server.id % 2 != 0:
        success = True
    else:
        success = False
        
    if success:
        server.power_on = False
        
    session.commit()
    return success

Итак, у нас есть небольшой отрывок приложения, которое совершает какую-то полезную работу и использует данные из базы. Как его можно протестировать?

Отсутствие тестирования

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

Достоинства:

• не требует умений;
• экономит время на старте проекта.

Недостатки:

• негативные последствия в долгосрочной перспективе.

Не стоит принимать этот пункт как решение.

Имитация базы данных

Приведенный пример достаточно простой, и обращение к ORM-фреймворку только одно при коммите у объекта сессии, так как информация о сервере передается в виде объекта. При такой архитектуре можно передать модифицированный объект Session. Например:

class MockSession(Session):
    def commit(self):
        pass


def test_mock():
    mock = MockSession()
    server = Server()
    server.id = 1
    server.power_on = True

    assert power_off(mock, server) is True
    assert server.power_on is False

Такой подход позволяет не использовать базу данных, а значит, тесты будут работать быстрее.

Достоинства:

• требует минимальной настройки;
• наиболее быстрое выполнение тестов.

Недостатки:

• неприменим или крайне сложен для некоторых архитектурных подходов;
• код инициализации начальных данных растет с наличием связей в объекте;
• запись в БД не производится, следовательно ORM не проверяет ограничения (constraint), заданные в модели.

Последнее не является критичной проблемой, но, если проверка будет производиться, это повысит доверие к результатам тестов.

Резидентная база данных

Когда нужна временная база данных, вспоминается несколько решений, умеющих хранить данные в памяти. Мне приходят на ум минимум два решения:

• встраиваемая БД — SQLite;
• легковесная БД, о которой я слышал от Java-разработчиков, — H2.

Очевидно, что SQLite во многом уступает «полноценным» базам данных, но это самый простой в настройке вариант, поэтому начнем с него. Теперь в тестах необходимо создать подключение к БД и сессию. Создаем соответствующие фикстуры.

HABR_TEST_DB_URL="HABR_TEST_DB_URL"

@pytest.fixture(scope="function")
def engine():
    if HABR_TEST_DB_URL not in os.environ:
        skip_reason_message: str = (
            f"Environment var with name {HABR_TEST_DB_URL!r} is not provided. "
            "Set this with a path to the real test database to run skipped tests."
        )
        pytest.skip(msg=skip_reason_message)

    engine = create_engine(
        os.environ[HABR_TEST_DB_URL],
        echo=False
    )
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    try:
        yield engine
    finally:
        Base.metadata.drop_all(engine, checkfirst=True)


@pytest.fixture
def session(engine):
    session = Session(engine)
    yield session

Фикстура engine принудительно сбрасывает информацию в БД, которая может помешать тесту, и создает «чистую» схему в соответствии с описанием ORM-моделей. По завершении тестирования схема сбрасывается.

Обратите внимание, что схема подключения к БД передается через переменную окружения HABR_TEST_DB_URL. Фикстура engine предусматривает отсутствие данной переменной окружения и корректно обрабатывает эту ситуацию: отмечает тесты как пропущенные с говорящим сообщением об ошибке. Таким образом все тесты, использующие базу данных, будут пропускаться при ее отсутствии.

Теперь создаем фикстуру, которая представляет сервер.

@pytest.fixture
def server(session):
    s = Server()
    s.ip = '127.0.0.1'
    s.hostname = 'home'
    s.power_on = True
    session.add(s)
    session.commit()
    return s

В отличие от предыдущего пункта, здесь необходимо указать все поля, которые не могут быть null. Иначе база данных просто не примет наш запрос. Эта фикстура однажды определяется и может быть переиспользована, например, в тестах редактирования записи. Напишем два простейших теста:

def test_presence(server):
    assert server.ip == '127.0.0.1'


def test_embedded_db(session, server):
    assert power_off(session, server) is True
    assert server.power_on is False

Данный способ не оптимальный с точки зрения производительности: фикстура engine имеет область видимости function. Это значит, что подключение к БД будет создаваться перед началом теста и уничтожаться после его завершения.

Однако этот способ не ограничен SQLite. Схема подключения указывается в переменной окружения. Если есть нужный драйвер и правильные настройки для подключения, то можно использовать другую БД.

Достоинства:

• использование ORM-фреймворка предоставляет больше возможностей для регрессионного тестирования;
• не требует изменений в логике приложения, даже при наличии наследников от класса Session;
• перед началом теста в БД лежит минимально необходимый набор данных.

Недостатки:

• каждый тест вызывает создание и удаление всей схемы в базе данных;
• если проект использует особенности конкретной БД, то тесты могут сломаться.

Но если мы уже используем базу данных, то почему бы не взять для тестов БД, аналогичную используемой в продакшене?

Транзакционное тестирование

Правильнее всего использовать для тестирования ту базу данных, под которую разрабатывалось приложение. Это уменьшит разницу между окружением разработчика и тестовым окружением и позволит использовать особенности конкретной БД. Именно этот способ мы используем в Selectel для тестирования систем продукта «Выделенные серверы».

Пересоздавать всю схему в БД — это долгое действие, особенно в больших приложениях. Сократить количество ненужных действий можно с помощью вложенных транзакций:

• создаем схему в БД;
• начинаем транзакцию;
• вносим в базу исходные данные;
• запускаем тест и получаем результат;
• откатываем транзакцию;
• повторяем пункты 2-5 для оставшихся тестов;
• удаляем все данные.

При запуске теста SQLAlchemy начинает вложенную транзакцию, которая откатится при завершении теста. Эта особенность ограничивает круг доступных баз данных, так как SQLite, например, вложенные транзакции не поддерживает.

Для работы данного способа необходимо определить собственную сессию и фабрику сессий.

class TestSession(SessionBase):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.begin_nested()

        @event.listens_for(self, "after_transaction_end")
        def restart_savepoint(session, transaction):
            if transaction.nested and not transaction._parent.nested:
                session.expire_all()
                session.begin_nested()


Session = scoped_session(sessionmaker(autoflush=False, class_=TestSession))

Эта «магия» запускает вложенную транзакцию, позволяя приложению свободно выполнять любые запросы, в том числе коммиты. Но когда придет время, «родительская» транзакция откатится, как и изменения, произведенные тестом. Для этого фикстуру engine() нужно использовать из предыдущего примера с единственной поправкой: область видимости изменяется с function на session. А вот фикстура session претерпевает значительные изменения.

@pytest.fixture
def session(engine):
    connection = engine.connect()
    transaction = connection.begin()

    Session.configure(bind=engine)
    session = Session()

    try:
        yield session
    finally:
        Session.remove()
        transaction.rollback()
        connection.close()

Фикстура создает сессию и запускает «глобальную» транзакцию, прежде чем передать себя тестам. При этом код тестов не отличается от предыдущего примера со встраиваемыми базами данных.

Единственное различие заключается в том, что в этом способе идентификаторы объектов могут быть разными, а в предыдущем способе они чаще всего начинаются с единицы, так как схема и все последовательности (Sequence) для автоинкремента создаются и удаляются на каждый тест.

Достоинства:

• позволяет тестировать приложения на «родной» базе данных с использованием ее особенностей;
• не требует изменения кода приложения.

Недостатки:

• несовместим с некоторыми базами данных, например, для SQLite нужны «костыли».

Этот способ также описан в секции о транзакциях и подключениях в документации SQLAlchemy.

Заключение

В данной статье мы рассмотрели разные способы тестирования приложений, которые используют базу данных. Несмотря на то, что пример достаточно синтетический, код тестов можно переносить в собственные проекты. Для удобства я выложил все на GitHub.

Помните, что тестирование не дает 100% гарантии отсутствия ошибок, но значительно повышает как качество продукта, так и доверие пользователя.