pytest - Airflow - DAG 完整性测试 - sqlalchemy.exc.OperationalError : (sqlite3. OperationalError)没有这样的表:变量

Question

我目前正在第一次编写一些 Airflow DAG 完整性测试。我遇到了一个错误，我的一些运算符(operator)/任务引用了 Airflow 变量，例如:

test_var= Variable.get("AIRFLOW_VAR_TEST_VAR")

当我使用 pytest 运行完整性测试时，出现以下错误:
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) 没有这样的表:变量
我可以通过用硬编码值替换 Variable.get 代码来解决这个问题，但想知道是否有更好的方法来处理这个错误？
谢谢，

Answer 1

你应该跑AIRFLOW__CORE__UNIT_TEST_MODE=True airflow db reset这将从头开始初始化并重新创建单元测试 sqlite 数据库。
或者，您可以使用特定于 Airflow 的开关运行 pytest --with-db-init开关做同样的事情。