Тесты кодовых моделей

Кодовую модель удобно проверять локально, не публикуя изменения в репозиторий: для этого рядом с model.py и model.yml размещается файл тестов, который запускает модель на локальном Spark с тестовыми данными.

Подготовка

Для локального запуска нужны Java и pyspark — они устанавливаются вместе с клиентской библиотекой при настройке рабочего места:

(.venv) $ pip install "analytic-workspace-client[dev]~=1.38.*"

Файл тестов

Файл тестов называется test_<имя модели>.py и располагается в папке модели. В нём принято описывать два теста: проверку метаданных и проверку самого кода модели.

import datetime
from pathlib import Path

import pytest

from aw_client.data_source import DataSourceTable
from aw_client.models.code import DataSourceBindings, DataSourceBindingToTestTable
from aw_client.models.code.runtime import run_code_model, read_code_model_meta


@pytest.fixture
def data_sources():
    """Привязка источников к тестовым таблицам"""
    return DataSourceBindings(
        bindings={
            "products": DataSourceBindingToTestTable(
                table=DataSourceTable(
                    rows=[
                        {"id": 1, "name": "Товар 1"},
                    ]
                )
            ),
            "customers": DataSourceBindingToTestTable(
                table=DataSourceTable(
                    rows=[
                        {"id": 1, "name": "Покупатель 1"},
                    ]
                )
            ),
            "orders": DataSourceBindingToTestTable(
                table=DataSourceTable(
                    rows=[
                        {
                            "id": 1,
                            "product_id": 1,
                            "customer_id": 1,
                            "summa": 1000,
                            "created_at": datetime.datetime(2025, 1, 1, 0, 0, 0),
                        }
                    ]
                )
            ),
        }
    )


def test_meta():
    """Проверка метаданных модели"""
    read_code_model_meta(Path(__file__).parent)


def test_model(data_sources):
    """Проверка модели на тестовых данных"""
    model_root = run_code_model(Path(__file__).parent, data_sources=data_sources)

    assert set(model_root.df.columns).issuperset(
        {
            "id",
            "product_id",
            "customer_id",
            "summa",
            "name",
            "name_1",
            "order_day",
            "order_month",
            "order_year",
            "summa_with_discount",
        }
    )
    assert model_root.df.count() == 1

1. Фикстура описывает привязку источников: для каждого источника из model.yml задаётся тестовая таблица с данными.
2. Тест проверяет, что файл model.yml корректен.
3. Тест запускает модель и проверяет результат: состав колонок и число строк.
4. Ассерты формулируются под вашу предметную область — здесь проверяется состав колонок итогового датафрейма.

Тест метаданных: test_meta

Тест вызывает read_code_model_meta(Path(__file__).parent) — эта функция читает model.yml и валидирует его по схеме: обязательные поля, допустимые значения type и т.д. Если метаданные некорректны, тест упадёт с понятным сообщением.

Тест модели: test_model

Основной тест запускает модель функцией run_code_model():

Параметр	Описание
model_path	Путь к папке модели — обычно Path(__file__).parent
data_sources	Привязки источников: объект DataSourceBindings с тестовыми таблицами

Функция:

1. читает метаданные из model.yml;
2. создаёт локальную сессию Spark;
3. импортирует model.py и вызывает main() с заявленными в сигнатуре параметрами;
4. вычисляет итоговый датафрейм и возвращает корневой узел графа — у него доступны df (итоговый датафрейм) и обычные методы Spark для проверок.

Тестовые данные: DataSourceBindingToTestTable

Каждый источник из model.yml привязывается к тестовой таблице DataSourceBindingToTestTable:

Параметр	Описание
table	Объект DataSourceTable со списком строк rows — словари «имя колонки: значение»
data_source_id	Необязательно. Идентификатор источника в AW — для сценария запуска на реальном сервере

Тестовые строки должны покрывать ключевые сценарии вашей трансформации: граничные значения, отсутствующие связи (для left join), пустые значения полей.

Запуск тестов

Тесты запускаются стандартным pytest из папки репозитория:

(.venv) $ pytest model_shop/test_model_shop.py -v

В VS Code тесты можно запускать из боковой панели Testing — подробности см. в разделе тестов etl-блоков — механика запуска идентична.

Первый запуск медленный

При первом запуске Spark инициализирует локальное окружение — это занимает заметное время. Последующие запуски в той же сессии выполняются быстрее.

Что проверять в тестах

Рекомендуемый минимальный набор проверок:

• метаданные — test_meta всегда включайте в файл тестов;
• структура результата — состав колонок итогового датафрейма соответствует списку полей из model.yml;
• бизнес-логика — конкретные значения: суммы, агрегаты, результаты соединений;
• крайние случаи — пустые источники, отсутствующие ключи соединения.

Публикация

Тесты — часть репозитория: они помогают не только вам, но и коллегам при code review. Рабочий процесс с ветками и merge в production-ветку описан в разделе Git-репозитории.