Напишите тесты, основанные на контейнерных зависимостях, и организуйте их с помощью Tox.

При разработке нового приложения нам часто нужны внешние сервисы, такие как базы данных, брокеры сообщений, кэш-память или API для других микросервисов. Тестирование интерфейсов между этими компонентами — практика, которой часто пренебрегают. Некоторые полностью пропускают эти тесты, а другие издеваются над ответами. Это, однако, оставляет интеграционные тесты незавершенными, поскольку эти интерфейсы могут легко дать сбой из-за изменений в API-интерфейсах поставщиков, изменений схемы базы данных и проблем с конфигурацией.

Один из способов проведения этих тестов — запуск зависимостей в виде контейнеров, которые взаимодействуют с приложением в среде, аналогичной производственной. В этой статье я покажу, как запускать эти тесты в конвейере непрерывной интеграции с помощью действий GitHub. Для этого необходимо написать небольшое приложение на Python с доступом к базе данных postgres для создания и выбора пользователей.

Тесты будут написаны с использованием pytest, одной из ведущих сред тестирования для Python. База данных будет работать как контейнер с использованием Docker. Тесты будут координироваться с tox, оркестратором тестирования для Python, который изначально возник как инструмент для тестирования различных версий Python, а затем эволюционировал, чтобы позволить разработчикам изолировать тестирование от сред разработки, централизовать конфигурацию тестирования и даже координировать зависимости с помощью докера. контейнеры.

Давайте код!

Вы можете следовать инструкциям и фрагментам кода или проверить весь (но небольшой) проект прямо в моем репозитории GitHub. Файловая структура проекта такова:

.
├── README.md
├── migrations
│   └── schema.sql
├── requirements.txt
├── src
│   └── testing_containers
│       ├── __init__.py
│       ├── db
│       │   ├── __init__.py
│       │   └── db.py
│       └── model
│           ├── __init__.py
│           └── users.py
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   └── db
│       └── test_db.py
└── tox.ini

Поскольку мы используем базу данных Postgres в качестве нашей зависимости, давайте начнем с написания небольшой таблицы, представляющей Users в нашем приложении. Я назову его migrations/schema.sql. Вот как это выглядит:

CREATE TABLE users (
 email varchar(64) primary key,
 name varchar(64) not null
);

Для представления объектов в приложении я буду использовать pydantic. Это отличный способ представления объектов, получения подсказок по типам, и он особенно полезен для проверки данных, когда мы используем эти сущности в API. Итак, создайте новый файл Python, назовите его users.py и вставьте следующее:

from pydantic import BaseModel


class User(BaseModel):
    """Representation of User entity"""

    name: str
    email: str

Теперь мы можем написать некоторый шаблон для наших операций с базой данных. Давайте начнем с некоторых функций, которые оборачивают наш драйвер postgres (в данном случае psycopg2). Поскольку цель этой статьи не в том, чтобы научить пользоваться этим драйвером, я не буду подробно останавливаться на них, но рекомендую вам прочитать PYnative tutorial. Мы можем начать наш класс так:

from typing import List

import psycopg2
from psycopg2.extras import execute_values

from testing_containers.model.users import User


class Repo:
    # psycopg2 wrapper
    def __init__(self) -> None:
        self.conn = None
        try:
            self.connect()
            logging.info("db: database ready")
        except Exception as err:
            logging.error("db: failed to connect to database: %s", str(err))
            self.close()
            raise err

    def connect(self):
        """Stores a connection object `conn` of a postgres database"""
        logging.info("db: connecting to database")
        conn_str = os.environ.get("DB_DSN")
        self.conn = psycopg2.connect(conn_str)

    def close(self):
        """Closes the connection object `conn`"""
        if self.conn is not None:
            logging.info("db: closing database")
            self.conn.close()

    def execute_select_query(self, query: str, args: tuple = ()) -> List[tuple]:
        """Executes a read query and returns the result

        Args:
            query (str): the query to execute.
            args (tuple, optional): arguments to the select statement. Defaults to ().

        Returns:
            List[tuple]: result of the select statement. One element per record
        """
        with self.conn.cursor() as cur:
            cur.execute(query, args)
            return list(cur)

    def execute_multiple_insert_query(self, query: str, data: List[tuple], page_size: int = 100) -> None:
        """Execute a statement using :query:`VALUES` with a sequence of parameters.

        Args:
            query (str): the query to execute. It must contain a single ``%s``
                        placeholder, which will be replaced by a `VALUES list`__.
                        Example: ``"INSERT INTO table (id, f1, f2) VALUES %s"``.
            data (List[tuple]): sequence of sequences or dictionaries with the arguments to send to the query.
            page_size (int, optional): maximum number of *data* items to include in every statement.
                        If there are more items the function will execute more than one statement. Defaults to 100.
        """
        with self.conn.cursor() as cur:
            execute_values(cur, query, data, page_size=page_size)

Теперь напишем несколько функций для:

  1. Получить пользователя по имени
  2. Создать новых пользователей
def get_user(self, name: str) -> (User | None):
    """Retrieve a User by a name"""
    query = "SELECT name, email FROM users WHERE name = %s"
    res = self.execute_select_query(query, (name,))
    if len(res) == 0:
        return None
    record = res[0]
    return User(name=record[0], email=record[1])

def insert_users(self, users: List[User]) -> None:
    """Given a list of Users, insert them in the db"""
    query = "INSERT INTO users (name, email) VALUES %s"
    data: List[tuple] = [(u.name, u.email) for u in users]
    self.execute_multiple_insert_query(query, data)

Потрясающий! У нас есть некоторые функции, которые взаимодействуют без бизнес-сущностей.

Обратите внимание: если мы не тестируем эти функции, мы подвержены ошибкам при изменении схемы или нарушении ограничений базы данных.

Итак, приступим к тестированию!

import pytest
from psycopg2.errors import UniqueViolation


@pytest.mark.usefixtures("repo")
class TestRepo:
    def test_insert_users(self):
        repo: Repo = self.repo # repository instanced passed by the fixture
        # define some users
        alice = User(name="alice", email="[email protected]")
        bob = User(name="bob", email="[email protected]")
        robert = User(name="robert", email="[email protected]")
        
        # check that the users are not there at first
        result = repo.get_user("alice")
        assert result is None

        # check that the users are there after inserting
        users = [alice, bob]
        repo.insert_users(users)
        result = repo.get_user("alice")
        assert result == alice
        result = repo.get_user("bob")
        assert result == bob

        # check that pk fails
        with pytest.raises(UniqueViolation):
            repo.insert_users([robert])

Первая строка — это декоратор, указывающий, что наш тестовый класс TestRepo будет использовать фикстуру с именем repo. Фикстуры — это базовые функции, которые позволяют нам получать согласованные и воспроизводимые результаты тестирования.

Давайте разберемся в этой строке repo: Repo = self.repo. Мы создаем переменную с именем repo типа Repo, которая является нашим ранее определенным классом репозитория. Он переназначается с self.repo, который поступает от прибора, как я объясню позже.

Мы утверждаем пару вещей:

  1. В начале теста нет пользователей
  2. Alice и Bob пользователи извлекаются после вставки
  3. Существует исключение UniqueViolation, и оно возникает, когда мы пытаемся создать нового пользователя с уже существующим адресом электронной почты.

Давайте запишем наш прибор в новый файл conftest.py.

@pytest.fixture(scope="class", name="repo")
def repo(request):
    """Instantiates a database object"""
    db = Repo()
    try:
        request.cls.repo = db
        yield db
    finally:
        db.close()

В этом приспособлении мы делаем следующее:

  1. Создайте экземпляр нашего класса репозитория db = Repo .
  2. Внутри блока try (поскольку выполнение теста может вызвать исключение) мы используем фикстуру request pytest, чтобы наш тестовый класс мог получить доступ к репозиторию request.cls.repo = db. Я призываю вас узнать больше о светильнике request!
  3. Затем мы передаем экземпляр для использования в тестовых функциях.
    В блоке finally мы обязательно закрываем соединение с базой данных.

Большой! У нас есть наши тесты, но если вы уже хотите запустить pytest, вы увидите, что тесты не работают на фикстуре, так как мы не можем создать экземпляр репозитория без базы данных postgres!

Вот тут и приходит на помощь токсикоз. Создайте новый файл с именем tox.ini:

[tox]
envlist = py310

[testenv]
setenv =
    DB_DSN = postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable
deps =
    -r requirements.txt
commands = pytest

Этот минимальный файл конфигурации говорит нам, что мы будем проводить тестирование в среде Python 3.10. Он устанавливает переменную среды DB_DSN, указывает файл требований для установки в виртуальной среде и вызывает pytest. Мой файл требований выглядит так:

psycopg2-binary
pydantic
pytest

Кстати, я рекомендую использовать pip-toos для закрепления зависимостей. Это выходит за рамки данного руководства, но вы можете прочитать его здесь: https://github.com/jazzband/pip-tools.

Запустить ваши тесты теперь так же просто, как просто установить и запустить tox:

python -m pip install — user tox

tox

Конечно, это не удается, потому что я обещал, что tox будет координировать контейнер postgres, но я этого не сделал.

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

pip install tox-docker

Теперь мы можем расширить наш tox.ini. Вот как это сделать:

[tox]
envlist = py310

[testenv]
setenv =
    DB_DSN = postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable
deps =
    -r requirements.txt
commands = pytest
docker = postgres

[docker:postgres]
image = postgres:13.4-alpine
environment =
    POSTGRES_DB=postgres
    PGUSER=postgres
    POSTGRES_PASSWORD=postgres
    POSTGRES_HOST_AUTH_METHOD=trust
ports =
    5432:5432/tcp
healthcheck_cmd = pg_isready
healthcheck_timeout = 5
healthcheck_retries = 5
healthcheck_interval = 5
healthcheck_start_period = 1
volumes =
    bind:ro:{toxinidir}/migrations/schema.sql:/docker-entrypoint-initdb.d/0.init.sql

Обратите внимание, что в нашем разделе testenv мы указываем, что будем использовать док-контейнер с именем postgres, который мы сразу после определения. Мы устанавливаем образ докера, который он должен (вытягивать), использовать, переменные среды, порты для сопоставления, проверки работоспособности (полезно, чтобы убедиться, что наши тесты работают только тогда, когда наши контейнеры исправны) и тома (обратите внимание, я ссылаюсь на migrations/schema.sql, который содержит нашу таблицу SQL определение). Пожалуйста, ознакомьтесь с документацией tox-docker, если вам нужны подробности.

Теперь, запустив tox, мы проходим наши тесты!

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

Запуск наших тестов локально — отличная практика, но, чтобы быть очень тщательным, лучше, если мы также запускаем их в нашем инструменте контроля версий при выполнении запроса на включение. С помощью GitHub Actions мы можем создать небольшой конвейер CI для запуска tox каждый раз, когда фиксация отправляется в запрос на вытягивание. Просто создайте этот файл, /.github/worflows/pr-test.yaml :

name: PR Test

on:
  pull_request:
    branches:
      - main

jobs:   
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.10"]
    name: With Python ${{ matrix.python-version }}

    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Setup python
        uses: actions/setup-python@v2
        with:
          python-version: ${{ matrix.python-version }}
          architecture: x64

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          python -m pip install tox tox-gh-actions
      - name: Test with tox
        run: tox

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

Дальнейшее тестирование и альтернативы

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

[docker:redis]
image = bitnami/redis:latest
environment =
    ALLOW_EMPTY_PASSWORD=yes
    REDIS_PORT_NUMBER=7000
ports =
    7000:7000/tcp
healthcheck_cmd = redis-cli ping
healthcheck_timeout = 5
healthcheck_retries = 5
healthcheck_interval = 5
healthcheck_start_period = 1

Внешние зависимости, которые нельзя поместить в контейнер или которые представляют затраты или защищенные ресурсы, такие как частный API, лучше имитировать.
Если полезность не убеждает вас в токсичности, другой подход — установить зависимости в файл docker-compose, создайте и запустите службы, дождитесь их работоспособности, запустите pytests и корректно остановите и удалите контейнеры.

Ресурсы

  1. Репозиторий GitHub для этого проекта: https://github.com/vrgsdaniel/testing-containers
  2. Документация Pytest: https://docs.pytest.org/en/7.2.x/
  3. Фикстуры Pytest: https://docs.pytest.org/en/6.2.x/fixture.html
  4. Фиксатор запроса Pytest: https://docs.pytest.org/en/6.2.x/reference.html#std-fixture-request
  5. Токс: https://tox.wiki/ru/latest/
  6. Tox-docker: https://tox-docker.readthedocs.io/ru/latest/
  7. Действия GitHub: https://github.com/features/actions
  8. Руководство по PYnative postgres: https://pynative.com/python-postgresql-tutorial/
  9. Пип-инструменты: https://github.com/jazzband/pip-tools
Want to Connect?

If you found my post interesting, you can visit my LinkedIn or GitHub :).