如何修补/模拟导入？

Question

我正在为气流 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？

Answer 1

您需要推迟导入文件，直到可以Variable在测试数据库中设置值。方法startTestRun将是完美的地方。