Apache Airflow 单元测试 DAG 目录详解:test_mode 机制与基于 DagBag 的测试 DAG 访问方式【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowairflow-core/tests/unit/dags/是 Apache Airflow 核心仓库中专为单元测试准备的 DAG 资源目录。本文以 目录说明文档 为核心,结合配置加载源码与测试配置模板,讲清楚该目录的定位与文件组织、单元测试中通过DagBag访问测试 DAG 的标准方式,以及unit_test_mode如何改变 Airflow 的dags_folder指向,使你在阅读或维护 Airflow 单元测试时能够准确定位测试 DAG 并理解其加载机制。单元测试 DAG 目录的定位该目录存放的是 Airflow 单元测试所依赖的 DAG 文件,与生产环境的DAGS_FOLDER是完全分离的一套资源。目录中的每一个test_*.py文件本身就是一个可被 Airflow 解析的 DAG 定义文件,同时它们又被测试代码当作“被测对象”来引用。从目录命名与文件清单可以推断出这些 DAG 覆盖了单元测试中需要模拟的各类 DAG 形态,主要包括:参数校验类:test_valid_param.py、test_valid_param2.py、test_invalid_param.py至test_invalid_param4.py,用于验证 DAGparams的合法与非法场景;非法/边界定义类:test_invalid_cron.py(非法 cron 表达式)、test_invalid_dup_task.py(重复 task_id)、test_future_start_date.py、test_missing_owner.py、test_with_non_default_owner.py等,用于覆盖 DAG 解析时的校验逻辑;运行与调度场景类:test_backfill_with_upstream_failed_task.py(backfill 时上游失败)、test_cli_triggered_dags.py(CLI 触发)、test_double_trigger.py、test_on_failure_callback.py、test_heartbeat_failed_fast.py(心跳失败快速失败)、test_dag_version_inflation_check.py、test_external_task_sensor_check_existense.py等;触发器场景类:test_dag_test_with_trigger.py、test_dag_test_with_dynamic_trigger.py,用于airflow dags test命令下的触发器行为验证;映射任务类:test_mapped_classic.py、test_mapped_taskflow.py,覆盖 Mapped Task 与 TaskFlow API 的组合;模拟登录(Impersonation)类:test_default_impersonation.py、test_impersonation.py、test_no_impersonation.py;序列化与集成能力类:test_dag_reserialize.py、test_dag_xcom_openlineage.py、test_dag_parsing_context.py、test_parsing_context.py;目录结构类:subdir1/、subdir2/子目录、test_zip/打包目录以及no_dags.py空 DAG 文件,用于模拟 DAG 文件分布在不同层级目录、目录中无 DAG 等场景。以 test_miscellaneous.py 为例,它是一个典型的测试 DAG:定义了 dag_id 为miscellaneous_test_dag的 DAG,配置了schedule0 0 * * *、start_date、dagrun_timeout、tags与params,并组合了EmptyOperator与BashOperator任务。这类文件被测试代码引用时,测试断言的就是其解析结果、任务依赖与运行行为。顶层测试配置 tests/conftest.py 中也对该目录下的个别 DAG 文件(如tests/unit/dags/subdir1/test_ignore_this.py、tests/unit/dags/test_invalid_dup_task.py)做了单独点名,说明 pytest 收集阶段对这些特殊文件(例如会触发解析警告或重复 task_id 的文件)存在针对性的处理策略。单元测试中访问测试 DAG 的标准方式README 给出的核心用法非常简洁:在单元测试内部,通过DagBag实例按dag_id取出目标 DAG:dagbag DagBag() dag dagbag.get_dag(dag_id)这里有两个关键前提:DagBag()默认解析的是当前dags_folder指向的目录。在单元测试环境中,该目录已被切换为本仓库的tests/unit/dags(见下一节的配置机制),因此DagBag()无需传参即可解析到这里的测试 DAG;只有test_mode打开时该方式才成立。README 明确强调:“Note this only works whentest_modeis on; otherwise the normal AirflowDAGS_FOLDERwill take precedence”——如果单元测试模式未开启,DagBag解析的将是用户环境中的常规DAGS_FOLDER,get_dag(dag_id)自然取不到这里定义的 DAG。这种访问方式的价值在于:测试代码不需要把 DAG 定义 import 进测试模块,而是以“Airflow 运行时真实解析 DAG 文件”的方式获取 DAG 对象,从而覆盖 DagBag 文件解析、任务构建、依赖链接等完整链路,而不仅仅是对象构造。test_mode:unit_test_mode 如何重定向 dags_folder“test_mode打开”在源码中对应[core]段的unit_test_mode配置项。其定义位于 config.yml:unit_test_mode: description: | Turn unit test mode on (overwrites many configuration options with test values at runtime) default: False即:开启后会在运行时用测试值覆盖大量默认配置。配置加载入口在 configuration.py 的 initialize_config:def initialize_config() - AirflowConfigParser: airflow_config_parser AirflowConfigParser() if airflow_config_parser.getboolean(core, unit_test_mode): airflow_config_parser.load_test_config() else: load_standard_airflow_configuration(airflow_config_parser) # 如果用户在 airflow.cfg 中显式设置了 unit_test_mode, # 仍然会尊重该设置并叠加加载测试配置文件 if airflow_config_parser.getboolean(core, unit_test_mode): airflow_config_parser.load_test_config() return airflow_config_parserload_test_config 的实现说明了测试配置的加载顺序:读取airflow/config_templates/unit_tests.cfg的内容;调用remove_all_read_configurations()清空已读取的配置,只保留默认值;将测试配置写入 parser,并动态生成随机的Fernetkey 与 JWT secret,再执行expand_all_configuration_values()完成{FERNET_KEY}、{TEST_DAGS_FOLDER}等模板变量的展开。测试配置文件 unit_tests.cfg 的文件头注释进一步说明:测试配置通过 pytest fixture 设置AIRFLOW__CORE__UNIT_TEST_MODETrue来加载(见 tests/conftest.py),其目的是让测试无论跑在 Breeze、本地虚拟环境还是 CI 中,都获得一致的配置环境。dags_folder的重定向正是其中的第一项:[core] # We want to read dags from the test dags folder dags_folder {TEST_DAGS_FOLDER} # we want to read plugins from example_dags/plugins folder plugins_folder {TEST_PLUGINS_FOLDER} # we want to read fernet key generated dynamically in load_test_config fernet_key {FERNET_KEY} # for tests we use local executor by default executor LocalExecutor # We do not want dags to be paused at creation dags_are_paused_at_creation False # We want to load examples load_examples True # No default impersonation in tests default_impersonation # We want to use unit test mode unit_test_mode True # We want to use a shorter timeout for task cleanup killed_task_cleanup_time 5 # We only allow our own classes to be deserialized in tests allowed_deserialization_classes airflow.* tests.*其中TEST_DAGS_FOLDER全局变量在 configuration.py 中定义:当仓库内的单元测试 DAG 目录存在时,TEST_DAGS_FOLDER指向该目录;否则回退到{AIRFLOW_HOME}/dags。这保证了开发场景下测试 DAG 目录始终可用。除dags_folder外,以下测试环境配置项值得注意(均取自 unit_tests.cfg):配置项测试环境值对单元测试的影响dags_folder{TEST_DAGS_FOLDER}将 DAG 解析目录切换到tests/unit/dagsexecutorLocalExecutor默认使用本地执行器,避免测试依赖外部队列dags_are_paused_at_creationFalseDAG 创建时不处于 paused 状态,便于直接创建 DagRunload_examplesTrue允许测试加载示例 DAG 作为对照unit_test_modeTrue保持单元测试模式开关为开killed_task_cleanup_time5缩短任务清理超时,加快测试节奏allowed_deserialization_classesairflow.* tests.*反序列化白名单收窄到 airflow 与测试代码fernet_key/plugins_folder动态生成 / 测试目录隔离加密密钥与插件加载来源unit_test_mode的影响不止于配置加载。从源码结构看,它还参与运行期行为分支,例如 scheduler_job_runner.py 中is_unit_test: bool conf.getboolean(core, unit_test_mode)用于调度器判断测试环境,dagcode.py 中同样读取该开关调整 DAG 代码处理逻辑。可以推断:该开关是贯穿“配置层 运行层”的单元测试总开关。在测试中组合使用:从 DagBag 取 DAG 到发起 DagRun把上述机制组合起来,一个典型的单元测试流程是:pytest fixture 设置AIRFLOW__CORE__UNIT_TEST_MODETrue,initialize_config()检测到该开关后调用load_test_config(),dags_folder被重定向到本仓库的tests/unit/dags;测试代码执行dagbag DagBag(); dag dagbag.get_dag(miscellaneous_test_dag),按 dag_id 取出 test_miscellaneous.py 中定义的 DAG;基于取到的dag对象创建 DagRun、任务实例,断言解析结果、依赖关系或运行状态。由于dags_are_paused_at_creation False,取出的 DAG 默认可直接调度;由于executor LocalExecutor,不需要 Redis/RabbitMQ 等外部组件,单测即可独立运行。维护该目录时的注意事项结合目录现状与 unit_tests.cfg 中的贡献者注释,可以归纳出维护该目录的几条实践约定:命名约定:测试 DAG 文件统一以test_前缀命名并直接放在本目录(或subdir1/、subdir2/、test_zip/子目录),命名即语义(如test_invalid_cron.py),便于测试代码按文件名定位 dag_id;特殊文件需同步 conftest:会触发解析异常或重复 task_id 的文件(如test_invalid_dup_task.py、subdir1/test_ignore_this.py)在 tests/conftest.py 中被单独列出处理,新增此类“异常样本” DAG 时需要考虑其对 pytest 收集与 DagBag 警告的影响;空目录场景:保留no_dags.py这类不含 DAG 的文件,用于覆盖“目录中不存在可解析 DAG”的边界路径;配置隔离:任何希望在所有单测中生效的默认值,应写入unit_tests.cfg而非散落在各测试中;针对个别测试的配置覆盖,推荐使用conf_vars上下文管理器或AIRFLOW__SECTION__KEY环境变量——这正是unit_tests.cfg文件头注释明确给出的建议。小结airflow-core/tests/unit/dags/通过“独立 DAG 资源目录 unit_test_mode配置重定向 DagBag运行时解析”三者配合,构成了 Airflow 单元测试的 DAG 基础设施:README 给出的DagBag().get_dag(dag_id)是最小可用模式,而其成立依赖于 configuration.py 中的配置加载分支与 unit_tests.cfg 对dags_folder的重定向。理解这条链路,是阅读 Airflow 单测中一切“从 DagBag 取 DAG 再断言”的测试代码的前提。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
