Тестирование миграции данных на python с pytest-bdd и testcontainers

В рамках проекта цифровой модернизации для одного из наших клиентов возникла задача миграции данных из одной модели хранения в другую. Для тестирования такого решения мы обратились к практикам BDD (Behaviour Driven Development) и виртуализации зависимостей с помощью docker контейнеров. В данном посте рассмотрен рецепт как можно организовать тестирование подобного решения с помощью pytest-bdd и testcontainers на python. Весь исходный код доступен по ссылке.

Об инфраструктуре

Как видно из диаграммы выше, источник данных живет в Azure. Миграция подразумевает перенос (копирование) актуальных данных из базы в Google Cloud сервис Datastore. API развертывается через Cloud Build как Cloud Run сервис.

Что мигрируем?

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

В нашем примере будет использоваться домен Posts. Информация о постах разбросана по следующим таблицам:

• post (id, title, summary) - общие сведения о постах

• category (id, title) - общие сведения о категориях

• post_category (id, postId, categoryId) - связующая таблица для постов и категорий

• post_meta (id, postId, key, content) - метаданные относящиеся к посту

В Google Datastore всё это будем хранить в сущностях вида Posts. Следует отметить, что категории и метаданные относящиеся к посту в MSSQL хранятся в различных таблицах, но в Datastore они будут храниться в определенных полях - meta и categories

Можно оценить, что разработка такого решения сводится к написанию SQL запросов различного уровня сложности, каких-то дополнительных преобразований (случай метаданных) и корректному отображению данных.

Каким образом это можно тестировать?

В качестве одного из вариантов можно было бы рассмотреть покрытие миграции smoke тестами. Для этого пришлось бы на каждое изменение SQL запроса или другой логики развёртывать новую версию API, прогонять эти тесты, и, в случае падения, искать причины в логах. От инфраструктуры клиента в Azure (VPN + MSSQL) можно было бы отвязаться использованием Google Cloud SQL. Но, в любом случае, этот feedback loop не то, чего нам хотелось бы в силу его полной зависимости от Cloud-провайдера, что несёт за собой:

• временные затраты на развёртывание

• дополнительные ресурсы на поддержку инфраструктурного кода

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

Проблему зависимости от Cloud можно решить, развернув setup локально. По сути, у нас здесь два ключевых компонента - это MSSQL и Google Datastore. Давайте рассмотрим как запустить эти вещи локально.

Локальный setup

На помощь приходит библиотека testcontainers. Это небольшая библиотека, позволяющая запускать docker контейнеры для тестирования. В состав уже входит набор некоторых готовых классов-обёрток для популярных решений. А так же имеется удобный API для собственных нужд. Взглянем на следующий код:

@fixture(scope="session", autouse=True)
def mssql_connection() -> Connection:
    with SqlServerContainer(
        "mcr.microsoft.com/mssql/server:2017-latest"
    ) as mssql:
        engine = create_engine(mssql.get_connection_url())
        db.schema.metadata.create_all(engine)
        yield engine.connect()

Здесь, мы подготавливаем контейнер с MSSQL. Обратим внимание, что для fixture используется scope=session. По умолчанию, pytest использует scope=function, что предполагает вызов функции на каждый тест. Это можно было бы использовать, но запускать контейнер под каждый тест весьма расточительно. Хотя, в этом случае, можно не беспокоиться об изоляции тестовых данных друг от друга, т.к. каждый раз у нас будет свежий контейнер без данных. Но так как мы заинтересованы в том, чтобы сократить время выполнения всех тестов (а чем их больше, тем это чувствительнее), то будем использовать session. В результате контейнер будет создан лишь один раз перед запуском всех тестов. Но в таком случае возникает проблема изоляции данных. Оставлять данные после отработки теста нам не интересно, поэтому решать эту проблему будем удалением - запомним таблицы, которые заполнялись перед запуском теста и очистим после.

Для Datastore контейнера напишем свой класс. Отмечу, что Google предоставляет набор эмуляторов для некоторых своих сервисов. В список поддерживаемых сервисов входит и Datastore. Наш класс базируется на image от Google со всеми необходимыми зависимостями. Подготовим контейнер:

@fixture(scope="session", autouse=True)
def gds_client():
    with DatastoreContainer() as ds:
        yield ds.get_client()

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

Тестирование

Рассмотрим подход к написанию тестов. BDD - это эволюция практики TDD, где вводится новый концепт behaviour. Мы не будем в деталях останавливаться здесь на особенностях этой практики, отметим лишь два важных момента:

• мы по-прежнему остаёмся в рамках парадигмы Test First

• тесты становятся своего рода документацией, к которой могут обращаться и не технические специалисты (в нашем проекте это было важно)

Два зайца одним выстрелом: тестируем код и пишем документацию.

Подход к реализации behaviours будет зависеть от выбора конкретной библиотеки. В целом, эти подходы похожи. Наш выбор упал на плагин для pytest который реализует подмножество языка Gherkin - pytest-bdd. Этот плагин позволяет описывать тесты на human-friendly языке с помощью зарезервированных синтаксических конструкций Given/When/Then и использовать pytest runner для запуска.

Тестирование нашей миграции подразумевает следующее:

• наличие определенного набора данных в MSSQL (Given data in MSSQL)

• непосредственный запуск миграции (When migration triggered)

• валидация новых данных в Datastore (Then check data exists in Datastore)

Рассмотрим наш feature-файл с тестами:

Feature: Posts migration path

  Scenario: Test that a selected post can be migrated with categories
    Given Table post has { "id": 45, "title": "Welcome to BDD" }
    And Table category has { "id": 1, "title": "python" }
    And Table category has { "id": 2, "title": "pytest-bdd" }
    And Table post_category has { "postId": 45, "categoryId": 1 }
    And Table post_category has { "postId": 45, "categoryId": 2 }
    When the post migration is triggered for postId=45
    Then post migrated as { "title": "Welcome to BDD", "legacy_id": 45, "categories": ["python", "pytest-bdd"], "meta": {} }
    And 1 post migrated

  Scenario: Test that a selected post can be migrated with meta
    Given Table post has { "id": 47, "title": "Welcome to testcontainers" }
    And Table post_meta has { "postId": 47, "key": "author", "content": "Vitalii Karniushin" }
    And Table post_meta has { "postId": 47, "key": "company", "content": "CTS" }
    When the post migration is triggered for postId=47
    Then post migrated as { "title": "Welcome to testcontainers", "legacy_id": 47, "categories": [], "meta": { "author": "Vitalii Karniushin", "company": "CTS" } }
    And 1 post migrated

Здесь у нас два сценария:

В первом сценарии мы мигрируем пост с которым связаны две категории. Эти данные разбросаны по трём разным таблицам. В конструкции Given мы указываем имя таблицы и используем JSON формат для описания данных. Этого достаточно для того, чтобы вставить запись в нужную таблицу:

@given(parsers.parse("table {table_name} has {payload}"))
def table_insert(
    mssql_connection: Connection, context: Any, table_name: str, payload: str
) -> None:
    obj = json.loads(payload)
    mssql_connection.execute(table_refs[table_name].insert().values(obj))

Заполняем таблицы и запускаем миграцию. После чего, ожидаем создание сущности в Datastore - для этого необходимо указать содержимое сущности, для описания которого снова используется JSON:

@then(parsers.parse("post migrated as {payload}"))
def check_post_as(gds_client: datastore.Client, payload: str) -> None:
    expected = json.loads(payload)
    q = gds_client.query(
        kind="Posts",
        filters=[("legacy_id", "=", expected["legacy_id"])],
    )
    post = list(q.fetch())[0]
    embedded = { 
        p[0]: dict(p[1].items()) for p in post.items() 
        if isinstance(p[1], datastore.Entity) 
    }
    actual = dict(post.items()) | embedded
    assert actual == expected

Как было отмечено выше, мы хотим зачищать данные после выполнения каждого теста. Для этого можно воспользоваться механизмом hooks которые добавлены в pytest-bdd - pytest_bdd_after_scenario, или добавить финальную фразу непосредственно в сценарий. Дело вкуса. Мы выбрали второй вариант, где перед зачисткой дополнительно проверяем количество перенесённых сущностей. В примере используется одна, но в реальности может быть и набор из нескольких, поэтому финальная конструкция параметризована количеством сущностей:

@then(parsers.parse("{count:d} posts migrated"))
@then(parsers.parse("{count:d} post migrated"))
def check_posts(
    mssql_connection: Connection,
    gds_client: datastore.Client,
    context: Any,
    count: int,
) -> None:
    query = gds_client.query(kind="Posts")
    res = list(query.fetch())
    assert len(res) == count
    cleanup_sql(mssql_connection, context)
    cleanup_datastore()

Во втором сценарии мигрируем пост с метаданными. В MSSQL это фиксируется в другой таблице - post_meta. В перенесённой сущности - в свойстве meta.

Итоги

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