我正在为 Airflow dag 编写测试并遇到模拟/修补 dag 的问题。
# dag.py
from airflow.models import Variable
ENVIRONMENT = Variable.get("environment")
# test_dag.py
import dag
class TestDAG(TestCase):
def test_something(self):
pass
因为我只是在函数或类之外设置变量,所以它在导入期间运行 Variable.get()。这会给我一个 SQLAlchemy 错误,因为它试图连接到一个数据库并获取变量。
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: variable
[SQL: SELECT variable.val AS variable_val, variable.id AS variable_id, variable."key" AS variable_key, variable.is_encrypted AS variable_is_encrypted
FROM variable
WHERE variable."key" = ?
LIMIT ? OFFSET ?]
[parameters: ('environment', 1, 0)]
有没有办法在导入之前修补/模拟airflow.models.Variable?
最佳答案
您需要推迟导入文件,直到您可以设置 Variable
值到测试数据库中。一个 startTestRun
方法将是完美的地方。
关于python - 如何修补/模拟导入?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/62314746/