单元测试如何mock snowflake-connector-python及游标执行查询

完全可以实现，直接用Python标准库自带的unittest.mock模块就能完成snowflake-connector-python的Mock，不需要真实发起网络连接，也不需要依赖测试环境的Snowflake服务。

核心思路 #

所有Snowflake操作的入口都是snowflake.connector.connect方法，测试时只要把这个方法替换成Mock对象，让它返回自定义的Mock连接，再让Mock连接的cursor()方法返回Mock游标，提前预设游标执行方法的返回值、异常，就可以覆盖所有查询逻辑的测试，还能校验传入的连接参数、SQL语句、查询参数是否符合预期。

代码示例 #

先假设你现有的业务类实现如下：

# 业务代码文件: sf_client.py
import snowflake.connector

class SnowflakeQueryClient:
    def __init__(self, credential: dict):
        # 传入凭证建立连接
        self.conn = snowflake.connector.connect(**credential)
    
    def get_query_result(self, sql: str, query_params: tuple = ()):
        # 生成游标执行查询
        with self.conn.cursor() as cur:
            cur.execute(sql, query_params)
            return cur.fetchall()

对应的单元测试实现（以pytest为例，unittest写法逻辑完全一致）：

# 测试文件: test_sf_client.py
from unittest.mock import patch, MagicMock
from sf_client import SnowflakeQueryClient

def test_query_logic():
    # 提前定义测试用的Mock数据
    expect_result = [
        (1, "product_a", 29.9),
        (2, "product_b", 59.9)
    ]
    test_credential = {
        "account": "test_acc",
        "user": "test_user",
        "password": "test_pwd",
        "warehouse": "test_wh",
        "database": "test_db",
        "schema": "public"
    }
    test_sql = "SELECT id, name, price FROM product WHERE price > %s"
    test_params = (10,)

    # 注意patch路径必须是业务代码中引用connect的路径
    with patch("sf_client.snowflake.connector.connect") as mock_connect:
        # 构造Mock游标
        mock_cursor = MagicMock()
        # 兼容with上下文管理器写法
        mock_cursor.__enter__.return_value = mock_cursor
        # 预设fetchall返回测试数据
        mock_cursor.fetchall.return_value = expect_result

        # 构造Mock连接
        mock_conn = MagicMock()
        mock_conn.cursor.return_value = mock_cursor

        # 让connect方法返回Mock连接
        mock_connect.return_value = mock_conn

        # 初始化客户端，不会发起真实连接
        client = SnowflakeQueryClient(test_credential)
        # 执行查询
        actual_result = client.get_query_result(test_sql, test_params)

        # 断言结果符合预期
        assert actual_result == expect_result
        # 校验连接凭证被正确传入
        mock_connect.assert_called_once_with(**test_credential)
        # 校验SQL和参数被正确传入execute方法
        mock_cursor.execute.assert_called_once_with(test_sql, test_params)

常见场景适配 #

• 如果要模拟查询报错：直接设置mock_cursor.execute.side_effect = Exception("mock query failed")，即可测试异常捕获、重试等分支逻辑。
• 如果用了fetchone()、fetch_pandas_all()等其他游标方法：和fetchall的处理逻辑一致，直接给对应Mock方法设置return_value即可。
• 如果要校验连接关闭、游标关闭等逻辑：直接对Mock对象调用assert_called()系列断言方法即可，不需要额外配置。

注意事项 #

patch路径不要写错：不要直接patchsnowflake.connector.connect，必须patch你业务代码中导入该方法的路径，比如业务代码在sf_client.py中导入了snowflake.connector，patch路径就写sf_client.snowflake.connector.connect，否则Mock不会生效。

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