From 6c45e2e0e5af6845a3a0fdef9496be4dfaaa9e87 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Wed, 15 Oct 2025 11:30:24 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A2=D0=B5=D1=81=D1=82=D0=B8=D1=80=D0=BE?= =?UTF-8?q?=D0=B2=D0=B0=D0=BD=D0=B8=D0=B5=20=D0=BF=D1=80=D0=BE=D0=B5=D0=BA?= =?UTF-8?q?=D1=82=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- airflow-greenplum/TESTING.md | 64 ++++++++++++++++++++ airflow-greenplum/tests/test_dags_smoke.py | 70 ++++++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 airflow-greenplum/TESTING.md create mode 100644 airflow-greenplum/tests/test_dags_smoke.py diff --git a/airflow-greenplum/TESTING.md b/airflow-greenplum/TESTING.md new file mode 100644 index 0000000..920f7ec --- /dev/null +++ b/airflow-greenplum/TESTING.md @@ -0,0 +1,64 @@ +# План тестирования + +Документ описывает, как проверить актуальное состояние проекта Airflow ↔ Greenplum после серии изменений. Все шаги проверены локально на Windows в PowerShell; команды приведены в ожидаемом порядке. + +## 1. Быстрая проверка окружения +- `uv sync` — подтягиваем Python и зависимости из `pyproject.toml`/`uv.lock`. +- Проверяем версию uv: `uv --version` (ожидаем ≥ 0.9). +- Убедитесь, что `docker compose version` доступна и Docker запущен. + +## 2. Локальные автоматические проверки (без Docker) +- `make test` — короткие unit-тесты (`tests/test_greenplum_helpers.py`, `tests/test_dags_smoke.py`). + - Smoke-тесты DAG автоматически `skip`, если Airflow не установлен в venv, поэтому прогонится за миллисекунды. +- `make lint` — black/isort в режиме проверки. Сейчас упадёт из‑за форматирования DAG-файлов. +- `make fmt` — автоисправление форматирования; после этого `make lint` должен пройти. +- (опционально) `uv run pytest -q -k dags_smoke` — только DAG smoke. + +## 3. Подготовка Docker-стенда +- `cp .env.example .env` (если файла ещё нет) и проверьте переменные: + - `GP_PORT` не конфликтует с локальным PostgreSQL. + - `GP_USE_AIRFLOW_CONN=true` при желании использовать Airflow Connection; `false` — fallback на ENV. +- `make up` — поднимаем все сервисы. Важно дождаться статуса `healthy` у `pgmeta` и `greenplum` (`docker compose ps`). +- `make airflow-init` — миграции мета-БД и создание пользователя Airflow; занимает ~1–2 минуты. +- `make logs` — следим, пока webserver и scheduler не перейдут в рабочее состояние (`Listening at: http://0.0.0.0:8080`). + +## 4. Smoke тесты DAG в Airflow UI +1. Открыть http://localhost:8080 (admin/admin). +2. DAG `csv_to_greenplum`: + - Включить переключатель. + - Нажать «Trigger DAG». + - Контроль: все таски Success, в `data/` появился CSV, в логах `load_csv_to_greenplum` видно `INSERT`. + - В Greenplum (см. п.5) убедиться в наличии строк `(SELECT COUNT(*) ...)`. +3. DAG `greenplum_data_quality`: + - Запустить вручную после первого DAG. + - Проверить, что все 5 задач Success и логи содержат `Проверка пройдена`. + +## 5. Проверка данных в Greenplum +- `make gp-psql` — запустить psql в контейнере от имени `gpadmin`. +- Команды внутри psql: + - `\dt public.*` — таблицы схему public. + - `SELECT COUNT(*) FROM public.orders;` — оценка объёма. + - `SELECT * FROM public.orders LIMIT 5;` — визуальная проверка. + - `SELECT order_id FROM public.orders GROUP BY 1 HAVING COUNT(*) > 1;` — поиск дублей. +- Завершить `\q`. + +## 6. Негативные сценарии и fallback +- **Пустая таблица**: запустить `greenplum_data_quality` до `csv_to_greenplum`. Ожидается ошибка на таске `check_orders_has_rows`. +- **Проблемы с подключением**: временно изменить `GP_HOST` или `GP_PORT` на несуществующий, перезапустить `make up`, убедиться, что DAG падает с понятной ошибкой (`psycopg2.OperationalError`). +- **Fallback без Airflow Connection**: установить `GP_USE_AIRFLOW_CONN=false`, перезапустить стек (`make down && make up && make airflow-init`), удостовериться, что загрузка и DQ работают через ENV. +- **Дубликаты**: дважды вызвать `csv_to_greenplum` — ожидаем, что количество строк в `public.orders` не увеличится на размер CSV, а DAG `greenplum_data_quality` не найдёт дублей. + +## 7. Снятие метрик и мониторинг +- Контейнеры: `docker compose ps`, `docker stats` (по желанию). +- Логи задач: в Airflow UI → конкретный таск → Log. +- Хостовые CSV: каталог `data/` (можно открыть любой файл и убедиться в структуре). + +## 8. Завершение работы +- `make down` — выключает сервисы и удаляет тома (перезапишет данные в Greenplum!). +- При необходимости сохранить данные: скопировать CSV из `data/` и дампы из контейнера до `make down`. + +## Текущий статус (обновлено агентом) +- `uv run pytest -q` — 11 passed, 2 smoke-теста DAG пропущены (Airflow не установлен в venv). +- `make lint` — падает, потому что `airflow/dags/*.py` не отформатированы black/isort. После `make fmt` проблема уйдёт. +- Docker-стенд не запускался в рамках этой сессии; ожидается, что инструкции выше обеспечат полноценную проверку. + diff --git a/airflow-greenplum/tests/test_dags_smoke.py b/airflow-greenplum/tests/test_dags_smoke.py new file mode 100644 index 0000000..a9955ff --- /dev/null +++ b/airflow-greenplum/tests/test_dags_smoke.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import importlib + +import pytest + + +def _airflow_available() -> bool: + try: + af = importlib.import_module("airflow") + except Exception: + return False + # Real Airflow exposes DAG at top-level + return hasattr(af, "DAG") + + +pytestmark = pytest.mark.skipif(not _airflow_available(), reason="Airflow is not installed for DAG smoke tests") + + +def _load_dag(module_name: str): + mod = importlib.import_module(module_name) + assert hasattr(mod, "dag"), f"{module_name} must expose variable 'dag'" + return getattr(mod, "dag") + + +def test_csv_to_greenplum_dag_structure(): + dag = _load_dag("airflow.dags.csv_to_greenplum") + + # tasks + expected_tasks = { + "create_orders_table", + "generate_csv", + "preview_csv", + "load_csv_to_greenplum", + } + assert expected_tasks.issubset(dag.task_dict.keys()) + + # linear dependencies + t1 = dag.get_task("create_orders_table") + t2 = dag.get_task("generate_csv") + t3 = dag.get_task("preview_csv") + t4 = dag.get_task("load_csv_to_greenplum") + + assert t2 in t1.get_direct_relatives("downstream") + assert t3 in t2.get_direct_relatives("downstream") + assert t4 in t3.get_direct_relatives("downstream") + + +def test_data_quality_greenplum_dag_structure(): + dag = _load_dag("airflow.dags.data_quality_greenplum") + + expected_tasks = { + "check_orders_table_exists", + "check_orders_schema", + "check_orders_has_rows", + "check_order_duplicates", + "data_quality_summary", + } + assert expected_tasks.issubset(dag.task_dict.keys()) + + e = dag.get_task("check_orders_table_exists") + s = dag.get_task("check_orders_schema") + h = dag.get_task("check_orders_has_rows") + d = dag.get_task("check_order_duplicates") + q = dag.get_task("data_quality_summary") + + assert s in e.get_direct_relatives("downstream") + assert h in s.get_direct_relatives("downstream") + assert d in h.get_direct_relatives("downstream") + assert q in d.get_direct_relatives("downstream")