首页 > 解决方案 > 模拟全局变量,在 Python 中连接到数据库

问题描述

我正在为 Airflow DAG 准备单元测试,但遇到了以下问题。

我有一个具有通用功能的模块,它将气流变量导入通用全局变量:

"""
Script for common functions used in airflow DAGs
"""
[...]
from airflow.models import Variable
[...]
config = Variable.get("some_dag_details")
[...]

我有导入 DAG 文件以进行测试验证的代码。

import unittest
from unittest.mock import patch
from airflow.models import DagBag
from some_package import common_functions


class DAGValidationTest(unittest.TestCase):
    DETAILS = {
        "some important data"
    }

    def setUp(self):
        with patch.object(common_functions.Variable, 'get', return_value=self.DETAILS) as mock_get_variable:
[...]

问题是在我能够模拟类变量或变量配置之前调用了连接数据库的代码。Python 将第 4 行(from some_package import common_functions)显示为错误源。在调用这个脚本之前如何模拟这个对象?

标签: pythonpython-3.xairflowpython-unittest

解决方案


推荐阅读