測試 Airflow 程式碼

使用 Airflow 建置資料管線

Volker Janz

Senior Developer Advocate at Astronomer

三層級測試

Dag testing pyramid

完整性：Dag 是否可無錯載入？
單元：商業邏輯是否產生正確結果？
整合：完整 Dag 是否可端到端執行？

使用 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

為什麼 Dag 在匯入時會壞掉

ModuleNotFoundError

ModuleNotFoundError：遺漏 provider 套件或匯入路徑錯誤
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

完整性＋單元：每次 commit（快速）
整合：pull request 或 nightly（較慢）
未通過三者者，Dag 不得上線

一起來練習吧！

使用 Airflow 建置資料管線