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

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

0 голосов
/ 12 сентября 2018

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

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 ]

0 голосов
/ 16 сентября 2018

Существует два способа решения этой проблемы:

  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()
0 голосов
/ 12 сентября 2018

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

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

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 значительно упростит ваш код. Вот краткое руководство по основам и длиннее и более полное руководство .

...