--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/python/testing.md - https://ytsaurus.tech/docs/ru/flow/python/testing.md - href: https://ytsaurus.tech/docs/ru/flow/python/testing.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Тестирование в YTsaurus Flow (Python) {% note info %} Данная страница описывает **юнит-тестирование** компонентов Python-[пайплайна](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline), а также **интеграционное тестирование** полного пайплайна через `FlowTestPythonBase`. {% endnote %} ## Общая архитектура тестирования {#architecture} В продакшене C++ [воркер](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#worker) отправляет gRPC-запросы [компаньону](https://ytsaurus.tech/docs/ru/flow/concepts/companion.md), передавая [сообщения](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#message), таймеры, [стейты](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state) и [watermark-и](https://ytsaurus.tech/docs/ru/flow/concepts/watermarks.md). Компаньон вызывает `Computation.do_process()`, который делегирует обработку в `ProcessFunction`. В юнит-тестах компоненты пайплайна тестируются прямым вызовом `do_process()` на объекте `Computation` с подготовленным `RequestContext`. Все тесты используют стандартный pytest. ## Зависимости {#dependencies} Для юнит-тестирования компаньон-библиотеки необходимо добавить зависимости в `ya.make`: ``` PEERDIR( yt/yt/flow/library/python/companion ) ``` При использовании Protobuf-стейтов также добавьте зависимость на proto-библиотеку. ## Тестирование ProcessFunction {#testing-process} [Исходный код тестов](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/test/test_computation.py) Для тестирования `RowFunction` / `BatchFunction` создаётся объект `Computation` и вызывается `do_process()` с подготовленным `RequestContext`. ### Создание RequestContext `RequestContext` — это dataclass, содержащий входные данные для обработки: ```python from yt.yt.flow.library.python.companion.context import RequestContext from yt.yt.flow.library.python.companion.job import Job from yt.yt.flow.library.python.companion.stream import ( StreamIdsMapping, StreamSpecs, RawStream, ) from yt.yt.flow.library.python.companion.row import ExtendedMessage def make_request_ctx(messages=None, timers=None): """Создание минимального RequestContext для тестов.""" mapping = StreamIdsMapping({"input": 0, "output": 1}) specs = StreamSpecs(mapping, [RawStream("input"), RawStream("output")]) job = Job( job_id="test-job", computation_id="test-comp", stream_specs=specs, static_spec={}, ) return RequestContext( job_id="test-job", request_id="test-req", computation_id="test-comp", messages=messages or [], timers=timers or [], stream_specs=specs, job=job, ) ``` ### Тестирование RowFunction ```python from yt.yt.flow.library.python.companion.computation import ( Computation, RowFunction, ) from yt.yt.flow.library.python.companion.row import ( ExtendedMessage, Message, ) class PassthroughFunction(RowFunction): def on_message(self, message, output, ctx): output.add_message( Message(message_id=message.message_id, stream_id="output") ) def test_passthrough(): comp = Computation( computation_id="test", process_function=PassthroughFunction(), ) messages = [ ExtendedMessage(message_id="m1", stream_id="input"), ExtendedMessage(message_id="m2", stream_id="input"), ] ctx = make_request_ctx(messages=messages) response = comp.do_process(ctx) assert len(response.transform_results) == 2 assert response.transform_results[0].messages[0].message_id == "m1" assert response.transform_results[1].messages[0].message_id == "m2" ``` ### Тестирование таймеров ```python from yt.yt.flow.library.python.companion.row import Timer class TimerFunction(RowFunction): def on_message(self, message, output, ctx): output.add_timer(trigger_timestamp=1000, event_timestamp=500) def on_timer(self, timer, output, ctx): output.add_message( Message(message_id="from-timer", stream_id="output") ) def test_timer_roundtrip(): comp = Computation( computation_id="test", process_function=TimerFunction(), ) messages = [ExtendedMessage(message_id="m1")] timers = [Timer(message_id="t1")] ctx = make_request_ctx(messages=messages, timers=timers) response = comp.do_process(ctx) results = response.transform_results # Сообщение создаёт таймер, таймер создаёт сообщение assert any(r.timers for r in results) assert any(r.messages for r in results) ``` ## Тестирование стейтов {#testing-states} [Исходный код тестов](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/test/test_context.py) ### YSON State ```python import yt.type_info as ti from yt.yt.flow.library.python.companion.context import DefaultRuntimeContext from yt.yt.flow.library.python.companion.row import ( ColumnSchema, ExtendedMessage, Payload, TableSchema, ) from yt.yt.flow.library.python.companion.stream import ( StreamIdsMapping, StreamSpecs, RawStream, ) from yt.yt.flow.library.python.companion.wire_protocol import ( ColumnValueType, UnversionedRow, UnversionedValue, ) KEY_SCHEMA = TableSchema([ColumnSchema("id", ti.String)]) def make_key_payload(): row = UnversionedRow(values=[ UnversionedValue(column_id=0, type=ColumnValueType.STRING, value=b"test-key"), ]) return Payload(row, KEY_SCHEMA) def make_ctx(internal_state_names=None, **kwargs): streams = [RawStream("input"), RawStream("output")] mapping = StreamIdsMapping({s.stream_id: i for i, s in enumerate(streams)}) specs = StreamSpecs(mapping, streams) return DefaultRuntimeContext( internal_state_names=internal_state_names or set(), stream_specs=specs, internal_states=kwargs.get("internal_states", {}), external_states=kwargs.get("external_states", {}), watermarks={}, min_watermark=0, computation_parameters={}, key_schema=KEY_SCHEMA, ) def test_yson_state_roundtrip(): ctx = make_ctx(internal_state_names={"word-state"}) message = ExtendedMessage(message_id="m1", key=make_key_payload()) accessor = ctx.state("word-state", message) accessor.set({"word": "hello", "count": 1}) accessor2 = ctx.state("word-state", message) result = accessor2.get() assert result["word"] == "hello" assert result["count"] == 1 ``` ### Proto State ```python from google.protobuf.wrappers_pb2 import Int64Value def test_proto_state_roundtrip(): ctx = make_ctx(internal_state_names={"join-state"}) message = ExtendedMessage(message_id="m1", key=make_key_payload()) accessor = ctx.proto_state("join-state", message, Int64Value) state = Int64Value(value=42) accessor.set(state) accessor2 = ctx.proto_state("join-state", message, Int64Value) assert accessor2.get().value == 42 ``` ### External State ```python from yt.yt.flow.library.python.companion.state import StatesHolder from yt.yt.flow.library.python.companion.row import PayloadBuilder STATE_SCHEMA = TableSchema([ ColumnSchema("count", ti.Int64), ColumnSchema("name", ti.String), ]) def test_external_state_roundtrip(): ext_holder = StatesHolder("ext", KEY_SCHEMA, STATE_SCHEMA) ctx = make_ctx(external_states={"/shuffle-state": ext_holder}) message = ExtendedMessage(message_id="m1", key=make_key_payload()) state = ctx.external_state("/shuffle-state", message) builder = state.to_builder() builder.set("count", 99) state.set(builder.finish()) state2 = ctx.external_state("/shuffle-state", message) assert state2.get("count") == 99 ``` ## Анализ результатов {#analyzing-response} Объект `ResponseContext`, возвращаемый `do_process()`, содержит: | Поле | Тип | Описание | |------|-----|----------| | `transform_results` | `List[TransformResult]` | Список результатов обработки | | `internal_states` | `Dict[str, StatesHolder]` | Внутренние стейты после обработки | | `external_states` | `Dict[str, StatesHolder]` | Внешние стейты после обработки | Каждый `TransformResult` содержит: | Поле | Тип | Описание | |------|-----|----------| | `parent_ids` | `List[str]` | Идентификаторы родительских сообщений | | `messages` | `List[Message]` | Выходные сообщения | | `timers` | `List[NewTimer]` | Созданные таймеры | ## Интеграционное тестирование с FlowTestPythonBase {#e2e-tests} Для полного интеграционного тестирования пайплайна (с реальными C++ воркерами, очередями и стримами) используется базовый класс `FlowTestPythonBase`. ### Зависимости {#integration-dependencies} Помимо `PEERDIR` на `integration_test_base`, интеграционному тесту нужны рецепт кластера, `DEPENDS` на бинарь пайплайна и `flow_server`, а также `DATA` со спекой. Полный `ya.make` теста из [WordCount](https://ytsaurus.tech/docs/ru/flow/python/examples/wordcount.md): ```text PY3TEST() INCLUDE(${ARCADIA_ROOT}/yt/yt/flow/library/python/integration_test_base/recipe.inc) TEST_SRCS( test_wordcount.py yt_sync.py ) PEERDIR( yt/yt/flow/library/python/queue ) DEPENDS( ${MODDIR}/.. yt/yt/flow/bin/flow_server ) DATA(arcadia/${MODDIR}/pipeline.yson) REQUIREMENTS( cpu:4 ram:32 ) TAG(ya:huge_logs) SIZE(MEDIUM) END() ``` ### Настройка {#python-test-setup} Тест наследуется от `FlowTestPythonBase` и задаёт атрибут `PYTHON_COMPANION_BINARY`: ```python class Test(FlowTestPythonBase): PYTHON_COMPANION_BINARY = yatest.common.binary_path("yt/yt/flow/examples/python/word_count/word_count") ``` | Атрибут | Описание | |---------|----------| | `PYTHON_COMPANION_BINARY` | Путь к бинарю Python-компаньона | [Пример E2E-теста WordCount](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/python/word_count/test/test_wordcount.py) {% note warning %} Интеграционные тесты требуют развёрнутого кластера YTsaurus и запускаются через `ya make -tt`. Для быстрой итерации используйте юнит-тесты, описанные выше. {% endnote %} Общие принципы написания интеграционных тестов на пайплайны: * Тестируем конечные сценарии. То есть: * Записываем входные/исходные данные в локальный YT. * В спеке пайплайна отмечаем источники как `finite=%true`. * Для [Key-visitor-стрима](https://ytsaurus.tech/docs/ru/flow/concepts/key_visitor.md), который должен работать во время теста, задаём `finite=%false`. * Запускаем пайплайн. * Когда визитор должен завершиться, вызываем метод базового класса интеграционного теста [`self.ask_key_visitor_to_complete("", "")`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/integration_test_base/yt_flow_base.py): он переключает [динамический параметр `finite`](https://ytsaurus.tech/docs/ru/flow/concepts/key_visitor.md#dynamic-params) этого стрима на `%true`. * Ждём, пока пайплайн завершится. * Проверяем выходные данные. * Готовим окружение тем же кодом, что и в проде: * Тестируем тот же бинарь пайплайна что будет работать в проде. * Генерируем спеки пайплайна тем же кодом, что будет генерировать их для прода. * Если входные/выходные данные нетривиально сериализуются/парсятся, то делаем эту работу общим с продом кодом. * Используем [общий тестовый фреймворк](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/integration_test_base) (есть `README.md`). * Делаем failover тест. * Многие ошибки вскрываются на выпадении воркеров и переподхвате их работы другими воркерами. Поэтому делаем тест с `problems=True` и с более чем 1 воркером. * Пишем стабильные тесты. * Помним, что в CI любая часть теста может работать неожиданно долго. * Идеальное время работы одного теста — 20 секунд при любом типе сборки. * Для санитайзерных сборок уменьшаем количество входных данных. * Все локальные таймауты выставляем с кратным запасом. * Логика выполнения теста не должна значимо зависеть от текущего времени. Например, тест не должен падать, если запустился до полуночи, а завершился — после. ## Отладка тестов {#debug} ### Логи {#debug-logs} После завершения работы теста его логи можно найти в `test-results/py3test/testing_out_stuff` в директории теста. Основные важные: * `run.log` — логи python-теста. * `//Controller_...` — логи контроллеров (`.err` — это stderr процесса, `.log` — обычные логи, которые пишутся через `YT_LOG_...`). * `//Worker_...` — логи воркеров. * `//Runner...` — логи раннера. Если что-то не работает, то имеет смысл поискать ошибки во всех этих логах. Принцип изучения логов такой же, как в рабочем пайплайне; см. [логи Vanilla-операции](https://ytsaurus.tech/docs/ru/flow/devops/vanilla/diagnostics/logs.md). Логи можно посмотреть ещё до завершения теста, для этого нужно найти временную директорию, в которой тест работает. Самый простой способ — запускать тест с флагом `--keep-temps`: `ya make --keep-temps -ttt `. В этом случае `ya make` не удалит временную директорию по окончании теста и напечатает ссылку на неё в выводе. Также можно воспользоваться такими bash-алиасами: ```bash alias curtestdir="ps -f -u $USER | python3 -c \"import sys, re; drs = set(e for e in re.findall(r'[\s=](/\S*testing' + r'_out_stuff)\b', sys.stdin.read())); print('' if len(drs) == 1 else 'Select first from ' + repr(drs), file=sys.stderr); print(list(drs)[0])\"" alias cdcurtestdir='cd $(curtestdir)' # Достать из логов ссылку на UI локального YTsaurus. alias curlocalyt='cat $(curtestdir)/stderr 2>/dev/null | grep YT' ``` ### Тестовый фреймворк {#debug-test-framework} Как настроить окружение для лучшей работы тестового фреймворка и как можно влиять на параметры тестирования можно прочитать в [README.md](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/integration_test_base) фреймворка. Поведение интеграционных тестов настраивается через `--test-param NAME=VAL`: #| || **Параметр** | **По умолчанию** | **Значения** | **Действие** || || `RUNNER_LOG_LEVEL` | | `Error`, `Info`, `Debug`, … | Уровень логирования процесса, запускающего пайплайн (runner). || || `PAUSE_BEFORE_FLOW_PROCESS_FEDERATION_TEARDOWN` | `0` | `0`, `1` | Зависнуть в тесте перед остановкой процессов Flow. В комбинации с `--test-disable-timeout` позволяет надолго оставить работающий локальный YTsaurus и федерацию процессов Flow для неспешного изучения через UI. || || `EXTERNAL_YT_CONFIG` | (не задан) | yson — см. ниже | Запускать пайплайн на реальных внешних кластерах YTsaurus вместо локального рецепта. || |# Примеры: ```bash ya make -A --test-param RUNNER_LOG_LEVEL=Debug ya make -A --test-disable-timeout --test-param PAUSE_BEFORE_FLOW_PROCESS_FEDERATION_TEARDOWN=1 ``` ### `EXTERNAL_YT_CONFIG` {#external-yt-config} Локальный YTsaurus всё ещё стартует рецептом, но тест его игнорирует. Обязательные поля, общие для всех кластеров: - `path` — базовая директория. - `tablet_cell_bundle` — bundle создаваемых динамических таблиц. - `proxy_role` — RPC proxy role. Опционально: `primary_medium` (по умолчанию `"default"`). Список `clusters` — первый элемент primary. Поля записи: `cluster_name` (обязательно), `proxy_url` (по умолчанию равен `cluster_name`). Авторизация: при внешнем YTsaurus `YT_TOKEN`/`YT_USER` из локального рецепта чистятся, yt-wrapper подхватывает токен из `~/.yt/token`. Изоляция: `work_yt_path = path//`; директория `path/` удаляется и пересоздаётся один раз на класс в `setup_class`. Пример: ```bash ya make -A --test-param 'EXTERNAL_YT_CONFIG={path="//tmp/yt_flow";tablet_cell_bundle=default;proxy_role=default;clusters=[{cluster_name=};];}' ``` ## Ускорение итерации: `--ext-py` {#ext-py} Флаг `--ext-py` избавляет от линковки бинаря на каждый запуск `ya make`, если изменения с предыдущего запуска касались только `*.py` файлов: ```bash ya make -A --ext-py ``` ## См. также - [Computation (Python)](https://ytsaurus.tech/docs/ru/flow/python/computation.md) - [Работа со стейтами (Python)](https://ytsaurus.tech/docs/ru/flow/python/state.md) - [Флаг distribute (Python)](https://ytsaurus.tech/docs/ru/flow/python/distribute.md) - [Тестирование (Java)](https://ytsaurus.tech/docs/ru/flow/java/testing.md) - Если дорабатываете сам Flow — [Фреймворк для тестирования пайплайнов](https://ytsaurus.tech/docs/ru/flow/contributor/testing-framework.md).