Kiểm thử mã Airflow

Xây dựng Data Pipeline với Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Ba cấp độ kiểm thử

Tháp kiểm thử Dag

Integrity: Dag có tải lên mà không lỗi không?
Unit: Logic nghiệp vụ cho kết quả đúng không?
Integration: Toàn bộ Dag chạy end-to-end chứ?

Kiểm thử integrity với 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

Vì sao Dag lỗi khi import

ModuleNotFoundError

ModuleNotFoundError: thiếu gói provider hoặc sai đường dẫn import
NameError: đổi tên biến/hàm hoặc lỗi gõ
ImportError: import vòng giữa các file Dag
Kiểm thử integrity giúp tránh các lỗi này

Unit test cho hàm task

Trong file 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]

Tách logic nghiệp vụ

Trong file kiểm thử:

from dags.data_cleaning import (
    clean_record,
)

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

Tập trung unit test vào logic nghiệp vụ

Kiểm thử tích hợp với 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

Chạy Dag thật với đầu vào kiểm soát và kiểm tra đầu ra

Kiểm thử trong CI

CI pipeline

Integrity + unit: mọi commit (nhanh)
Integration: pull request hoặc nightly (chậm hơn)
Không Dag nào lên production nếu chưa qua cả ba

Cùng luyện tập nào!

Xây dựng Data Pipeline với Airflow