DAG 测试与 CI 校验是什么
把 DAG 当代码来测：CI 里用 DagBag 解析整个 dags 目录，收集导入错误；再用 pytest 对关键任务与依赖结构做断言。它拦住的是最常见也最廉价可查的一类问题——语法错误、变量名拼错、循环依赖、dag_id 重复、导入时就连外部系统。这些错误一旦上线，DAG 会直接从界面消失或整条链路不可调度，而 CI 里几秒钟就能发现。
基本信息
| 冒烟测试 | DagBag 解析 dags 目录并收集 import_errors，应为空 |
|---|---|
| 结构断言 | 断言任务存在、依赖关系正确、关键参数符合预期 |
| 环境隔离 | CI 使用独立元数据库或 mock，避免污染生产元数据 |
| 执行命令 | pytest 直接跑；也可用 airflow dags list-import-errors 手工核对 |
| 卡点设置 | 解析失败的 DAG 禁止合并，作为合并前的必过检查 |
| 注意点 | DAG 文件顶层不要执行重活，解析期的外部调用会让测试变慢变脆 |
典型使用场景
- 合并前检查：防止语法错误或拼写错误进入主分支
- 重构保护：调整任务名或依赖后用断言确认结构未被意外改动
- 升级回归：Airflow 版本升级后批量验证所有 DAG 仍可解析
- 发布前验证：新 DAG 上线前跑一次冒烟测试，避免上线即消失
- 规范巡检：检查 DAG 命名、调度表达式与默认参数是否符合团队约定
DagBag 冒烟测试
DagBag 冒烟测试
import pytest
from airflow.models import DagBag
@pytest.fixture()
def dagbag():
# CI 环境用独立元数据库，避免写入生产元数据
return DagBag(dag_folder="dags", include_examples=False)
def test_no_import_errors(dagbag):
assert not dagbag.import_errors, 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_ids
def test_no_duplicate_dag_ids(dagbag):
assert len(dagbag.dags) == len({d.dag_id for d in dagbag.dags.values()})命令行核对
命令行核对
# 本地快速查看哪些 DAG 解析失败及原因
airflow dags list-import-errors
# 确认 DAG 是否被正确收集（数量与名称）
airflow dags list | head -20
# 检查单个 DAG 的任务结构
airflow tasks list daily_orders --treeCI 检查项分层
| 检查项 | 能拦住什么 | 成本 |
|---|---|---|
| DagBag 解析 | 语法错误、导入失败、DAG 消失 | 秒级，收益最高 |
| dag_id 唯一性 | 重复定义导致覆盖或解析报错 | 秒级 |
| 结构断言 | 任务缺失、依赖被误改 | 秒级 |
| 默认参数校验 | 重试、超时、owner 等未按规范配置 | 秒级 |
| 任务级单测 | 业务函数逻辑错误 | 较高，按需覆盖 |
CI 里跑 DagBag 是成本最低的一道防线：语法错误、循环依赖与重复 dag_id 都能在合并前拦住，比上线后从调度界面发现便宜得多。
测试环境不要连生产元数据库跑 DagBag：解析过程可能写入元数据，污染生产环境的 DAG 与任务记录，排查起来非常麻烦。