Python中使用pytest-mock模拟SQLAlchemy数据库调用的单元测试实现

核心测试思路 #

单元测试需要严格划定方法边界：针对get_cust_id的测试，所有涉及真实数据库连接、SQL执行、ORM查询的逻辑都属于外部依赖，需要全部模拟替换，全程不触碰真实数据库实例，仅验证当前方法本身的分支逻辑正确性。

源码前置提醒：你提供的代码里有两处容易引发运行错误的问题，测试前需要先修正：

• get_customer_id_sqlalchemy的else分支中，__res_id[0]是变量名拼写错误，前文定义的变量为_res_id，直接运行会抛出NameError
• 方法入参_session_id没有被加入查询过滤条件，目前SQL查询逻辑和入参无关，需要和业务侧确认是否为遗漏的逻辑

前置准备 #

安装pytest配套的mock插件pytest-mock，安装完成后可以直接使用内置的mocker fixture完成所有模拟逻辑，无需手动编写patch上下文。

具体实现 #

1. 基础场景测试（仅验证get_cust_id本身逻辑） #

这部分测试完全隔离下层ORM实现，不模拟具体的SQL查询细节，仅替换self.sqlalchemy_orm.get_customer_id_sqlalchemy方法的返回值，覆盖get_cust_id的两个核心分支：

# test_customer_service.py
import pytest

class TestGetCustId:
    def setup_method(self):
        # 直接创建被测类实例，绕过__init__初始化，避免触发真实数据库连接
        from your_module import CustomerService  # 替换为你实际的业务类导入路径
        self.customer_service = CustomerService.__new__(CustomerService)

    def test_return_session_id_when_orm_returns_none(self, mocker):
        # 模拟下游ORM方法返回空值
        mock_orm_call = mocker.patch.object(
            self.customer_service.sqlalchemy_orm,
            "get_customer_id_sqlalchemy",
            return_value=None
        )
        test_session_id = "test_session_123"
        # 执行被测方法
        result = self.customer_service.get_cust_id(test_session_id)
        # 断言返回值为传入的原始session_id
        assert result == test_session_id
        # 验证下游ORM方法确实收到了传入的session_id参数
        mock_orm_call.assert_called_once_with(test_session_id)

    def test_return_cust_id_when_orm_returns_valid_value(self, mocker):
        mock_real_cust_id = "real_cust_456"
        # 模拟下游ORM方法返回有效的客户ID
        mocker.patch.object(
            self.customer_service.sqlalchemy_orm,
            "get_customer_id_sqlalchemy",
            return_value=mock_real_cust_id
        )
        test_session_id = "test_session_123"
        result = self.customer_service.get_cust_id(test_session_id)
        # 断言返回值为查询到的客户ID，而非原始session_id
        assert result == mock_real_cust_id

2. 扩展测试（同步验证下层ORM调用逻辑，零真实DB交互） #

如果需要连带验证get_customer_id_sqlalchemy的SQL调用是否符合预期，也不需要连接真实数据库，通过链式mock完全模拟ORM的查询链即可：

def test_orm_query_logic_without_real_db(self, mocker):
        # 直接创建ORM类实例，绕过初始化逻辑
        from your_module import CustomerOrm  # 替换为你实际的ORM类导入路径
        orm_instance = CustomerOrm.__new__(CustomerOrm)
        # 预设查询返回值
        mock_query_row = ("cust_789",)
        # 模拟数据库会话对象
        mock_session = mocker.MagicMock()
        orm_instance.session = mock_session
        # 链式模拟完整ORM调用链：query -> with_entities -> filter -> order_by -> one_or_none
        mock_session.query.return_value.with_entities.return_value.filter.return_value.order_by.return_value.one_or_none.return_value = mock_query_row
        # 标记为非持久连接，验证session关闭逻辑
        orm_instance.persistent_connection = False
        # 执行ORM方法
        result = orm_instance.get_customer_id_sqlalchemy(123)
        # 断言返回值正确
        assert result == "cust_789"
        # 验证查询流程的核心调用
        mock_session.query.assert_called_once()
        mock_session.query.return_value.with_entities.assert_called_once()
        # 验证非持久连接场景下，会话被正确关闭
        mock_session.close.assert_called_once()

关键注意事项 #

• 模拟时尽量绕过被测类的__init__初始化流程，避免初始化过程中意外建立真实数据库连接，用类的__new__方法直接创建实例、手动赋值测试需要的属性即可
• 断言模拟方法的调用参数时，要和实际业务参数完全匹配，避免模拟逻辑和真实业务流程脱节
• 如果需要覆盖查询返回空值的分支，可以把链式mock的one_or_none返回值设为None，验证方法内的空值判断逻辑是否生效
• 测试前一定要修正源码里的变量名笔误，否则即使mock逻辑完全正确，也会因为代码本身的语法错误抛出异常

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