Тестування коду Airflow

Створення конвеєрів даних з Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Три рівні тестування

Піраміда тестування Dag

Інтегриті (цілісність): чи Dag завантажується без помилок?
Юніт: чи бізнес-логіка дає коректні результати?
Інтеграція: чи весь Dag працює end-to-end?

Тести цілісності з DagBag

from airflow.models import DagBag


dag_bag = DagBag(include_examples=False)


def test_no_import_errors():
    assert len(dag_bag.import_errors) == 0


def test_dag_loaded():
    assert "daily_etl" in dag_bag.dags

Чому Dags ламаються під час імпорту

ModuleNotFoundError

ModuleNotFoundError: відсутній пакет провайдера або хибний шлях імпорту
NameError: перейменована змінна/функція або описка
ImportError: циклічні імпорти між файлами Dag
Тести цілісності допомагають уникати цих проблем

Юніт-тестування функцій завдань

У файлі Dag:

def clean_record(record):
    return {
        "name": record["name"].strip(),
        "email": record["email"].lower(),
    }

@task
def transform(records):
    return [clean_record(r)
            for r in records]

Винесіть бізнес-логіку

У тестовому файлі:

from dags.data_cleaning import (
    clean_record,
)

def test_strips_whitespace():
    result = clean_record(
      {"name": "  Alice  ",
       "email": "[email protected]"}
    )
    assert result["name"] == "Alice"

Зосередьте юніт-тести на бізнес-логіці

Інтеграційні тести з dag.test()

import pytest
from airflow.models import DagBag
from pendulum import datetime

dag_bag = DagBag(include_examples=False)


def test_etl_pipeline():
    dag = dag_bag.get_dag("etl_output")
    assert dag is not None

    dag.test(logical_date=datetime(2026, 1, 15))

    output = Path("/tmp/etl_results.json")
    assert output.exists()
    results = json.loads(output.read_text())
    assert len(results) == 2

Запустіть реальний Dag з контрольним вводом і перевірте вивід

Тестування в CI

CI pipeline

Цілісність + юніт: кожен коміт (швидко)
Інтеграція: pull request або щоночі (повільніше)
Жоден Dag не потрапляє у прод без проходження всіх трьох

Давайте потренуємось!

Створення конвеєрів даних з Airflow