为Snowflake数据库操作写单元测试：Mock还是真实连接？

结论先行 #

Mock单元测试和真实连接测试都需要，二者各司其职，并非二选一的关系。

为什么需要两种测试？ #

• Mock单元测试：聚焦代码逻辑本身，验证get_database_connection方法是否正确调用snowflake.connector.connect、传递参数、处理返回值。它速度快、不依赖外部资源，适合开发阶段快速迭代、CI/CD流水线频繁运行，保障核心逻辑正确性。你当前的Mock测试很有价值，必须保留。
• 真实连接测试：属于集成测试范畴，用来验证代码与真实Snowflake数据库的交互是否正常——比如参数格式是否符合要求、网络权限是否通畅、数据库实例能否正确响应连接请求。这类测试能发现Mock覆盖不到的问题，比如真实环境中的认证错误、网络超时、数据库配置不匹配等。

如何推进？ #

1. 保留并完善现有Mock测试 #

可以补充测试场景，验证传入database和schema参数时，是否正确传递给connect方法：

@patch('services.SnowflakeAPI.snowflake.connector')
def test_get_database_connection_with_db_schema(self, mock_connection):
    mock_connection.connect.return_value = "dbconnection"
    
    SnowflakeApi.get_database_connection(
        "user", "password", "account", "warehouse",
        database="test_db", schema="test_schema"
    )
    
    # 验证connect方法是否收到正确参数
    mock_connection.connect.assert_called_once_with(
        user="user",
        password="password",
        account="account",
        warehouse="warehouse",
        database="test_db",
        schema="test_schema"
    )

2. 添加真实连接的集成测试 #

这类测试需注意以下几点：

• 不要在CI流水线中频繁运行，可手动触发或定时执行
• 使用专门的测试数据库实例，绝对禁止连接生产环境
• 测试完成后及时清理资源（如关闭连接、游标）

示例代码：

import os
from unittest import TestCase
from services import SnowflakeApi

class SnowflakeDbIntegrationTest(TestCase):
    def setUp(self):
        # 从环境变量读取测试数据库凭据，避免硬编码
        self.test_user = os.getenv("SNOWFLAKE_TEST_USER")
        self.test_password = os.getenv("SNOWFLAKE_TEST_PASSWORD")
        self.test_account = os.getenv("SNOWFLAKE_TEST_ACCOUNT")
        self.test_warehouse = os.getenv("SNOWFLAKE_TEST_WAREHOUSE")
        self.test_database = os.getenv("SNOWFLAKE_TEST_DB")

    def test_real_database_connection(self):
        # 建立真实连接
        conn = SnowflakeApi.get_database_connection(
            self.test_user,
            self.test_password,
            self.test_account,
            self.test_warehouse,
            database=self.test_database
        )
        
        # 验证连接有效性，执行简单查询
        cursor = conn.cursor()
        cursor.execute("SELECT CURRENT_DATABASE()")
        result = cursor.fetchone()
        
        self.assertEqual(result[0], self.test_database)
        
        # 清理资源
        cursor.close()
        conn.close()

最佳实践总结 #

• 单元测试（Mock）：覆盖所有代码逻辑分支，保证快速反馈，作为日常开发的核心测试手段
• 集成测试（真实连接）：验证与外部系统的交互，定期运行，确保生产环境兼容性
• 隔离测试环境：真实测试使用专门的测试库，避免影响生产数据

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