| 冒烟测试 | DagBag 解析 dags 目录并收集 import_errors |
|---|---|
| 结构断言 | 断言任务存在、依赖正确 |
| 隔离 | CI 用独立元数据库或 mock |
| 卡点 | 解析失败的 DAG 禁止合并 |
DagBag 冒烟测试
DagBag 冒烟测试
import pytest
from airflow.models import DagBag
@pytest.fixture()
def dagbag():
return DagBag(dag_folder="dags", include_examples=False)
def test_no_import_errors(dagbag):
assert not dagbag.import_errors
def test_daily_orders_has_load(dagbag):
dag = dagbag.get_dag("daily_orders")
assert dag is not None and "load" in dag.task_idsCI 里跑 DagBag 能拦住语法错误、循环依赖与重复 dag_id，是成本较低的防线。