Перейти к содержанию

Тесты для etl-блоков

Введение

Разработка etl-блоков выполняется в локальном режиме без подключения к экземпляру Analytic Workspace.

Для проверки корректности блока используются автоматические тестирующие функции, описываемые разработчиком параллельно с созданием исходных текстов блока.

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

Инструменты

pytest

Для создания и запуска тестов etl-блоков в AW используется фреймворк pytest. Он устанавливается в виртуальное окружение вместе с клиентской библиотекой analytic-workspace-client

Visual Studio Code

Если для разработки блоков вы используете Visual Studio Code, то внутри этой IDE есть возможность настроить запуск тестов из графического интерфейса.

Для этого откройте раздел "Testing" и нажмите кнопку "Configure Python Tests"

Далее выберите пункт "pytest".

В следующем вопросе рекомендуется выбрать "Root directory", но в целом вы можете указать папку конкретного теста.

В завершении настройки вы увидете структуру ваших автотестов.

Расположение тестов

Код тестовых функций рекомендуется размещать в файле с названием test_%uid блока%.py .

Файлы блока aw_sql_block
.
├── block_code.py
├── block_meta.json
├── test_aw_sql_block.py
└── README.md

При большом количестве тестов, тестирующие функции можно разбить на несколько файлов.

Тесты располагаются в двух файлах
.
├── block_code.py
├── block_meta.json
├── test_aw_sql_block_meta.py
├── test_aw_sql_block_data.py
└── README.md

План тестирования

План тестирования etl-блока должен включать в себя следующие проверки:

  1. Проверка правильности заполнения метаданных;
  2. Проверка функции получения схемы данных;
  3. Проверка функции обработки данных;
  4. Проверка обработчиков действий (при их наличии в блоке).

Кроме этого, рекомендуется добавить в тесты не только позитивные сценарии (когда пользователь вложил в блок все нужные объекты и корректно заполнил параметры блока), но и негативные:

  1. Пользователь не добавил внутрь блока ни одного объекта;
  2. Пользователь добавил внутрь блока объекты, над которыми блок не может выполнить корректную обработку данных;
  3. Пользователь не указал (или указал, но неправильно) параметры блока и так далее.

С проверками негативных сценариев вы можете ознакомиться в блоке Декоратор из раздела Примеры.

Здесь, вы можете увидеть как проверяется поведение функций block_schema и block_data, когда пользователь не указал имя функции (тесты test_block_schema_empty_function_name и test_block_data_empty_function_name).

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

Для работы большинства etl-блоков требуются данные для вложенных в блок объектов. Поэтому, в тестовом модуле нужно указать, с какими данными вложенных объектов будут запускать тесты блока.

При описании тестовых данных используются объекты класса ModelObjectTestData.

Объявление объекта с тестовыми данными
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from aw_client.etl_blocks import ModelObjectTestData

TEST_TABLE_DATA = ModelObjectTestData(
    model_name='test_data',
    rows=[
        {'id': 1, 'name': 'Название 1', 'date': datetime(2024, 9, 21, 9, 0, 0)},
        {'id': 2, 'name': 'Название 2', 'date': datetime(2024, 9, 22, 12, 1, 1)}
    ],
    schema=[
        {'model_name': 'id', 'simple_type': 'number'}, 
        {'model_name': 'name', 'simple_type': 'string'},
        {'model_name': 'date', 'simple_type': 'date'},
    ]
)

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

Использование тестовых данных в тестах
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from aw_client.etl_blocks import (  
    ModelObjectTestData,
    get_etl_block_schema,  # (1)!
    get_etl_block_data,  # (2)!
    run_block_action  # (3)!
)


TEST_TABLE_DATA = ModelObjectTestData( ... )


def test_schema():
    """ 
    Этот тест использует тестовые данные TEST_OBJECT 
    """
    block_params = { ... }

    block_schema = get_etl_block_schema(
        block_path=Path(__file__).parent, 
        test_data=[TEST_OBJECT],
        params=block_params
    )


def test_data():
    """ 
    И этот тест тоже использует TEST_OBJECT 
    """
    block_params = { ... }

    block_schema = get_etl_block_data(
        block_path=Path(__file__).parent, 
        test_data=[TEST_OBJECT],
        params=block_params
    )


def test_action_autofill():
    """ 
    Обработчики действий тоже могут использовать тестовые данные 
    """
    block_params = { ... }

    action_result = run_block_action(
        block_path=Path(__file__).parent,
        action_name='autofill',
        test_data=[TEST_OBJECT],
        params=block_params
    )
  1. Функция get_etl_block_schema используется запуска механизма получения схемы данных блока.
  2. Функция get_etl_block_data используется запуска механизма обработки данных блока.
  3. Функция run_block_action используется запуска обработчиков действий.

Где указывать тестовые данные

Так как тестовые данные могут встречаться сразу в нескольких тестовых функциях, то имеет смысл размещать объявление объектов с тестовыми данными на самом верхнем уровне модуля с тестами block_test.py.

В примере выше так и сделано.

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

Запуск тестов выполнятся командой pytest. Для этого, активируйте виртуальное окружение перейдите в папку блока и выполните её.

(.venv) $ cd my_block
(.venv) $ pytest

Можно запускать тесты сразу для нескольких блоков. Для этого перейдите в папку с вашими тестами и запустите команду pytest там.

(.venv) $ cd folder_with_blocks
(.venv) $ pytest

Если для разработки тестов вы используете Visual Studio Code и настроили запуск тестов, то можете запускать тесты (по одной функции, по одному блоку, по всем блокам сразу) прямо из IDE.

API

Метаданные блока ETLBlockMeta

Объект для хранения метаданных блока.

Атрибут Тип Описание
uid string Уникальный идентификатор блока
name string Название блока
description string Описание блока
author string Строка с указанием автора
version string Строка с указанием версии
updated_at datetime Строка с временной меткой последнего изменения блока
params list Список параметров блока

Тестовые данные ModelObjectTestData

За объявление тестовых данных отвечает класс ModelObjectTestData из модуля aw_client.etl_blocks.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from aw_client.etl_blocks import ModelObjectTestData

TEST_TABLE_DATA = ModelObjectTestData(
    model_name='test_data',
    rows=[
        {'id': 1, 'name': 'Название 1', 'date': datetime(2024, 9, 21, 9, 0, 0)},
        {'id': 2, 'name': 'Название 2', 'date': datetime(2024, 9, 22, 12, 1, 1)}
    ],
    schema=[
        {'model_name': 'id', 'simple_type': 'number'}, 
        {'model_name': 'name', 'simple_type': 'string'},
        {'model_name': 'date', 'simple_type': 'date'},
    ]
)
Атрибут Тип Обязательно Описание
model_name string Да Название объекта с тестовыми данными в модели. По этому названию код блока обращается к вложенным в него датафреймам и структуре.
rows list[dict] Да Список словарей со строками тестовых данных
schema list[dict] Нет Описание схемы тестовых данных

Про model_name

Значение model_name соответствует названию объекта при просмотре структуры вложенных объектов на форме настройки блока.

Типы в схеме тестовых данных

В описании схемы данных schema тестового объекта указываются названия ключей из словарей rows и тип данных simple_type.

Возможные типы данных:

  • string - строка;
  • number - число (целое или с плавающей запятой);
  • bool - логическое (булево) значение;
  • date - дата и время.

Если в параметре simple_type указано значение не из этого типа, то считается, что оно равно string.

Значение параметра schema не является обязательным. Если не указано (или указано schema=None), то система попытается автоматически вывести схему данных на основе значений первого словаря данных из rows.

Загрузка метаданных блока

Выполняется функцией get_etl_block_meta из модуля aw_client.etl_blocks.

Аргументы

Аргумент Тип Обязательно Описание
block_path Path Да Путь к папке, в которой хранятся исходные файлы. В указанной папке должен находиться файл block_meta.json

Возвращаемое значение

Функция возвращает объект класса ETLBlockMeta с метаданными блока.

Исключительные ситуации

Исключение Когда возникает
Exception В папке block_path не найден файл block_meta.json
VaildationError Содержимое файла block_meta.json не прошло форматно-логический контроль

Получение схемы данных блока

Выполняется функцией get_etl_block_schema из модуля aw_client.etl_blocks.

Аргументы

Аргумент Тип Обязательно Описание
block_path Path Да Путь к папке, в которой хранятся исходные файлы. В указанной папке должен находиться файл block_meta.json
test_data Список объектов ModelObjectTestData Да Список тестовых данных для вложенных объектов. Если у блока нет вложенных объектов, то указывается пустой список []
params dict Да Словарь с параметрами, указанные пользователем на форме настройки блока
run_mode string Нет Режим выполнения блока
vault Vault Нет Хранилище чувствительных параметров модели
model_script_code string Нет Исходный код скрипта модели

Возвращаемое значение

Возвращает объект StructType со схемой данных блока.

Исключительные ситуации

Исключение Когда возникает
Exception 1. В папке block_path не найден файл block_code.py;
2. Файл block_code.py содержит синтаксические ошибки;
3. Скрипт модели из аргумента model_script_code содержит синтаксические ошибки.

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

Получение данных блока

Выполняется функцией get_etl_block_data из модуля aw_client.etl_blocks.

Аргументы

Аргумент Тип Обязательно Описание
block_path Path Да Путь к папке, в которой хранятся исходные файлы. В указанной папке должен находиться файл block_meta.json
test_data Список объектов ModelObjectTestData Да Список тестовых данных для вложенных объектов. Если у блока нет вложенных объектов, то указывается пустой список []
params dict Да Словарь с параметрами, указанные пользователем на форме настройки блока
run_mode string Нет Режим выполнения блока
vault Vault Нет Хранилище чувствительных параметров модели
model_script_code string Нет Исходный код скрипта модели

Возвращаемое значение

Возвращает датафрейм DataFrame с обработанными данными.

Исключительные ситуации

Исключение Когда возникает
Exception 1. В папке block_path не найден файл block_code.py;
2. Файл block_code.py содержит синтаксические ошибки;
3. Скрипт модели из аргумента model_script_code содержит синтаксические ошибки.

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

Выполнение действия

Выполняется функцией run_block_action из модуля aw_client.etl_blocks.

Аргументы

Аргумент Тип Обязательно Описание
block_path Path Да Путь к папке, в которой хранятся исходные файлы. В указанной папке должен находиться файл block_meta.json
action_name string Да Код действия, обработчик которого нужно запустить
test_data Список объектов ModelObjectTestData Да Список тестовых данных для вложенных объектов. Если у блока нет вложенных объектов, то указывается пустой список []
params dict Да Словарь с параметрами, указанные пользователем на форме настройки блока
run_mode string Нет Режим выполнения блока
vault Vault Нет Хранилище чувствительных параметров модели
model_script_code string Нет Исходный код скрипта модели

Возвращаемое значение

Возвращает объект класса ETLBlockActionResult с результатами запуска обработчика действия.

Объекты тестирования

Метаданные блока

Тест метаданных проверяет, что файл с метаданными блока block_meta.json проходит форматно-логический контроль.

Как тестировать:

  1. Загрузить метаданные тестируемого блока, используя функцию get_etl_block_meta;
  2. Если block_meta.json не проходит проверки, то будет выброшено исключение ValidationError.
Пример теста на проверку метаданных
1
2
3
4
5
6
7
8
9
from pathlib import Path
from aw_client.etl_blocks import get_etl_block_meta


def test_metadata():
    """ 
    Тест на проверку метаданных
    """
    get_etl_block_meta(block_path=Path(__file__).parent)

Построение схемы данных

Тест построения схемы блока проверяет, что функция block_meta() из модуля блока работает корректно.

Как тестировать:

  1. Заполнить словарь с входными параметрами блока так, как их указывает пользователь;
  2. Вызвать функцию get_etl_block_schema, передав в неё словарь параметров и список тестовых данных вложенных в блок объектов;
  3. Проверить с помощью выражений assert, что функция вернула ожидаемую схему данных блока.
Пример тест на построение схемы данных
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from pathlib import Path

from pyspark.sql.types import (
    LongType,
    StringType,
    BooleanType,
)

from aw_client.etl_blocks import (
    ModelObjectTestData,
    get_etl_block_schema
)


JSON_OBJECT_DATA = ModelObjectTestData(  
    model_name='json_object',
    rows=[
        {'id': 1, 'json_object': '{"x": 1, "y": "Название 1"}'},
        {'id': 2, 'json_object': '{"x": 2, "y": "Название 2"}'},
        {'id': 3, 'json_object': '{"x": 2, "z": true}'}
    ],
    schema=[
        {'model_name': 'id', 'simple_type': 'number'},
        {'model_name': 'json_object', 'simple_type': 'string'}
    ]
)


def test_block_schema():
    """ 
    Тест на получение схемы данных блока
    """
    block_params = {
        'json_field': 'json_object',
        'json_schema': '{x: целое, y: строка, z: логическое}',
        'transform_type': 'object_to_columns',
        'rules_object_to_columns': [
            {'attr_name': 'x', 'column_name': 'column_1'},
            {'attr_name': 'y', 'column_name': 'column_2'},
            {'attr_name': 'z', 'column_name': 'column_3'}
        ]
    }

    block_schema = get_etl_block_schema(
        block_path=Path(__file__).parent,
        test_data=[JSON_OBJECT_DATA],
        params=block_params
    )

    assert len(block_schema.fields) == 5
    assert block_schema.fields[0].name == 'id'
    assert block_schema.fields[0].dataType == LongType()
    assert block_schema.fields[1].name == 'json_object'
    assert block_schema.fields[1].dataType == StringType()
    assert block_schema.fields[2].name == 'column_1'
    assert block_schema.fields[2].dataType == LongType()
    assert block_schema.fields[3].name == 'column_2'
    assert block_schema.fields[3].dataType == StringType()
    assert block_schema.fields[4].name == 'column_3'
    assert block_schema.fields[4].dataType == BooleanType()

Обработка данных

Тест обработки данных блока проверяет, что функция block_data() из модуля блока работает корректно.

Как тестировать:

  1. Заполнить словарь с входными параметрами блока так, как их указывает пользователь;
  2. Вызвать функцию get_etl_block_data, передав в неё словарь параметров и список тестовых данных вложенных в блок объектов;
  3. У результирующего датасета вызвать функцию collect() для преобразования его содержимого в обычный список записей;
  4. Проверить с помощью выражений assert, что функция вернула ожидаемые данные.
Пример теста на обработку данных
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from pathlib import Path
import pytest

from aw_client.etl_blocks import (
    ModelObjectTestData,
    get_etl_block_data
)

TEST_OBJECT = ModelObjectTestData(
    model_name='test_object',
    rows=[
        {'id': 1, 'name': 'Название 1'},
        {'id': 2, 'name': 'Название 2'}
    ],
    schema=[
        {'model_name': 'id', 'simple_type': 'number'}, 
        {'model_name': 'name', 'simple_type': 'string'}
    ]
)

def test_block_data():
    """
    Проверяем, как возвращаются данные SQL запроса
    """
    block_params = {
        'sql': 'select id, name from test_object'
    }

    df = get_etl_block_data(
        block_path=Path(__file__).parent, 
        test_data=[TEST_OBJECT],
        params=block_params
    )

    rows = df.collect()

    assert len(rows) == 2
    assert rows[0]['id'] == TEST_OBJECT.rows[0]['id']
    assert rows[0]['name'] == TEST_OBJECT.rows[0]['name']
    assert rows[1]['id'] == TEST_OBJECT.rows[1]['id']
    assert rows[1]['name'] == TEST_OBJECT.rows[1]['name']

Обработчики действий блоков

Тест на выполнение обработчика действия
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from pathlib import Path

from aw_client.etl_blocks import (
    ModelObjectTestData,
    run_block_action
)


JSON_OBJECT_DATA = ModelObjectTestData(  
    model_name='json_object',
    rows=[
        {'id': 1, 'json_object': '{"x": 1, "y": "Название 1"}'},
        {'id': 2, 'json_object': '{"x": 2, "y": "Название 2"}'},
        {'id': 3, 'json_object': '{"x": 2, "z": true}'}
    ],
    schema=[
        {'model_name': 'id', 'simple_type': 'number'},
        {'model_name': 'json_object', 'simple_type': 'string'}
    ]
)


def test_action_infer_json_schema():
    """ 
    Тест на выполнение обработчика действия "Заполнить JSON схему автоматически"
    """
    block_params = {
        'json_field': 'json_object'
    }

    action_result = run_block_action(
        block_path=Path(__file__).parent,
        action_name='autofill_schema',  
        params=block_params,
        test_data=[JSON_OBJECT_DATA]
    )

    json_schema = action_result.params_patch.get('json_schema')  

    assert json_schema is not None
    assert json_schema.replace('\n', '').replace(' ', '') == \
        '{x:целое,y:строка,z:логическое}'