Airflow DAG完整性测试报sqlite3无variable表错误的处理方案咨询

• 方案1：使用Airflow官方提供的mock装饰器
  Airflow自带mock_variables、mock_connections装饰器，不需要依赖真实元数据库即可模拟变量/连接返回值，是单元测试的首选方案：

import pytest
from airflow.utils.decorators import mock_variables
from dags import your_target_dag

@mock_variables(variables={
    "AIRFLOW_VAR_BLOB_CONTAINER": "test_blob_container",
    "OTHER_REQUIRED_VAR": "test_value"
})
def test_dag_integrity():
    # 此时导入DAG不会触发元数据库查询
    assert your_target_dag.dag is not None
    # 可追加DAG结构校验、任务依赖校验等逻辑

对应处理Connection的场景可以用@mock_connections装饰器，用法一致。

• 方案2：直接mock Variable/Connection的get方法
  如果需要更灵活的返回逻辑，可直接通过pytest-mock mock对应类的get方法：

def test_dag_load(mocker):
    # 模拟Variable.get按key返回测试值
    mock_var_get = mocker.patch("airflow.models.Variable.get")
    mock_var_get.side_effect = lambda key: {
        "AIRFLOW_VAR_BLOB_CONTAINER": "test_container",
        "DB_NAME": "test_db"
    }[key]
    
    # 模拟Connection.get返回测试连接
    mock_conn_get = mocker.patch("airflow.models.Connection.get")
    from airflow.models import Connection
    mock_conn_get.return_value = Connection(
        conn_id="test_blob_conn",
        conn_type="azure_blob",
        login="test_account",
        password="test_key"
    )
    
    from dags.your_dag import dag
    assert dag.dag_id == "expected_dag_id"

• 方案3：通过环境变量传递变量
  Airflow内置支持AIRFLOW_VAR_<变量名>格式的环境变量，优先级高于元数据库存储的Variable，不需要连接数据库即可被识别。你可以在测试启动前提前注入对应环境变量：
  可以在pytest的conftest.py中配置自动生效的fixture：

import os
import pytest

@pytest.fixture(autouse=True)
def inject_test_airflow_vars():
    # 提前设置测试用变量
    os.environ["AIRFLOW_VAR_AIRFLOW_VAR_BLOB_CONTAINER"] = "test_container"
    os.environ["AIRFLOW_VAR_CUSTOM_VAR"] = "custom_test_val"
    yield
    # 测试结束后清理环境变量
    for key in ["AIRFLOW_VAR_AIRFLOW_VAR_BLOB_CONTAINER", "AIRFLOW_VAR_CUSTOM_VAR"]:
        if key in os.environ:
            del os.environ[key]

该方案无需修改DAG代码和mock逻辑，和生产环境的变量读取行为完全一致。

• 方案4：初始化临时元数据库
  如果你的测试场景需要真实的元数据库交互（比如集成测试），可以在运行pytest前先执行airflow db init初始化临时SQLite数据库，再通过airflow variables set、airflow connections add提前写入需要的测试数据即可。

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