pytest测试Airflow DAG完整性触发ModuleNotFoundError问题求助

问题根因 #

Airflow 运行环境默认会将 ${AIRFLOW_HOME}/dags 目录加入 Python 模块搜索路径，因此 DAG 文件中 from common.helper 的导入方式在容器内可正常运行。但本地测试时 dags 目录未加入 Python 搜索路径，同时你的测试代码存在路径拼接错误，共同导致了 ModuleNotFoundError。

步骤1：修正测试代码的路径错误 #

test_dag_integrity.py 中存在错误的路径拼接逻辑，DAG_PATH 是 glob 匹配模式而非目录，不能用来拼接模块路径，修改对应代码即可：

@pytest.mark.parametrize("dag_file", DAG_FILES)
def test_dag_integrity(dag_file):
    # Load file
    module_name, _ = os.path.splitext(dag_file)
    # 删掉错误的路径拼接，直接使用dag_file的完整路径
    # 原错误写法：module_path = os.path.join(DAG_PATH, dag_file)
    module_path = dag_file
    mod_spec = importlib.util.spec_from_file_location(
        module_name,
        module_path,
    )
    # 后续代码保持不变

步骤2：配置Python模块搜索路径 #

二选一即可：

方案A：临时指定路径（无需修改配置） #

执行测试命令时将 dags 目录加入搜索路径：

PYTHONPATH="./dags" python -m pytest tests/

方案B：pytest永久配置 #

在项目根目录新建 pytest.ini 文件，写入以下配置：

[pytest]
pythonpath = . dags

配置完成后直接执行 pytest tests/ 即可正常运行测试。

内容的提问来源于stack exchange，提问作者yudhiesh