Модульное тестирование функции, которая зависит от базы данных

Я запускаю тесты по некоторым функциям. У меня есть функция, которая использует запросы к базе данных. Итак, я просмотрел блоги и документы, в которых говорится, что мы должны создать базу данных в памяти или тестовую базу данных, чтобы использовать такие функции. Ниже моя функция,

def already_exists(story_data,c):
    # TODO(salmanhaseeb): Implement de-dupe functionality by checking if it already
    # exists in the DB.
    c.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
    (number_of_rows,)=c.fetchone()
    if number_of_rows > 0:
        return True
    return False

Эта функция попадает в производственную базу данных. Мой вопрос заключается в том, что во время тестирования я создаю базу данных в памяти и заполняю там свои значения, я буду запрашивать эту базу данных (тестовую базу данных). Но я хочу проверить свои already_exists() функция, после вызова моего already_exists функция из теста, моя производственная база данных будет поражена. Как сделать так, чтобы моя тестовая БД работала при тестировании этой функции?

Источник

2 ответа

Решение

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

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

def already_exists(story_data):
    # Here `connection` is a singleton which returns the database connection.
    connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (story_data.post_id,))
    (number_of_rows,) = connection.fetchone()
    if number_of_rows > 0:
        return True
    return False

Или сделать connection метод на каждый класс и ход already_exists в метод. Вероятно, это должен быть метод независимо.

def already_exists(self):
    # Here the connection is associated with the object.
    self.connection.execute("""SELECT COUNT(*) from posts where post_id = ?""", (self.post_id,))
    (number_of_rows,) = self.connection.fetchone()
    if number_of_rows > 0:
        return True
    return False

Но на самом деле вы не должны запускать этот код самостоятельно. Вместо этого вы должны использовать ORM, такой как SQLAlchemy, который позаботится об основных запросах и управлении соединениями, как это для вас. У него одно соединение, "сессия".

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from sqlalchemy_declarative import Address, Base, Person

engine = create_engine('sqlite:///sqlalchemy_example.db')
Base.metadata.bind = engine

DBSession = sessionmaker(bind=engine)
session = DBSession()

Затем вы используете это, чтобы сделать запросы. Например, у него есть exists метод.

session.query(Post.id).filter(q.exists()).scalar()

Использование ORM значительно упростит ваш код. Вот краткое руководство по основам и более длинное и полное руководство.

Источник

Есть два пути, которые вы можете выбрать для решения этой проблемы:

  1. Сделайте интеграционный тест вместо юнит-теста и просто используйте копию реальной базы данных.
  2. Предоставить подделку для метода вместо фактического объекта подключения.

Что вы должны сделать, зависит от того, чего вы пытаетесь достичь.

Если вы хотите проверить, работает ли сам запрос, вам следует использовать интеграционный тест. Полная остановка. Единственный способ убедиться в правильности запроса - запустить его с тестовыми данными, уже имеющимися в копии базы данных. Запуск его с другой технологией баз данных (например, запуск с SQLite, когда ваша производственная база данных в PostgreSQL) не гарантирует его работоспособность. Потребность в копии базы данных означает, что вам потребуется некоторый автоматизированный процесс ее развертывания, который можно легко вызвать для отдельной базы данных. В любом случае у вас должен быть такой автоматизированный процесс, так как он помогает обеспечить согласованность ваших развертываний в разных средах, позволяет тестировать их перед выпуском и "документировать" процесс обновления базы данных. Стандартные решения для этого - инструменты миграции, написанные на вашем языке программирования, такие как albemic, или инструменты для выполнения необработанного SQL, такие как yoyo или Flyway. Вам нужно будет вызвать развертывание и заполнить его тестовыми данными до запуска теста, затем запустить тест и подтвердить вывод, который вы ожидаете получить.

Если вы хотите проверить код вокруг запроса, а не сам запрос, то вы можете использовать подделку для объекта подключения. Наиболее распространенным решением этого является макет. Макеты предоставляют подставки, которые можно настроить для приема вызовов функций и входов и возврата некоторого вывода вместо реального объекта. Это позволит вам проверить, что логика метода работает правильно, предполагая, что запрос возвращает ожидаемые вами результаты. Для вашего метода такой тест может выглядеть примерно так:

from unittest.mock import Mock

...

def test_already_exists_returns_true_for_positive_count():
    mockConn = Mock(
        execute=Mock(),
        fetchone=Mock(return_value=(5,)),
    )
    story = Story(post_id=10) # Making some assumptions about what your object might look like.

    result = already_exists(story, mockConn)

    assert result

    # Possibly assert calls on the mock. Value of these asserts is debatable.
    mockConn.execute.assert_called("""SELECT COUNT(*) from posts where post_id = ?""", (story.post_id,))
    mockConn.fetchone.assert_called()
Источник
Другие вопросы по тегам