You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何使用assert_called_with断言对象实例?(Pytest异步测试场景)

解决Pytest断言SQLModel Select对象失败的问题

你遇到的问题是因为两次创建的SelectOfScalar是不同实例,即使逻辑完全一致,内存地址差异也会导致断言失败。以下是几种可行的解决方法:

方法1:比较编译后的SQL语句和参数

通过将预期与实际的Select对象编译为SQL字符串及参数,验证二者逻辑等价,而非直接比较实例本身。

@pytest.mark.asyncio
async def test_get_user_by_login(mocker):
    # Arrange
    mock_exec_result = AsyncMock(return_value=None)
    mock_session = mocker.AsyncMock()
    mock_session.exec.return_value = mock_exec_result
    expected_login = LOGIN
    expected_sql = select(User).where(User.login == expected_login)

    # Act
    await UserService.get_user_by_login(mock_session, expected_login)

    # Assert
    # 获取实际调用的SQL参数
    actual_sql = mock_session.exec.call_args[0][0]
    # 编译为可比较的SQL字符串和参数
    expected_compiled = expected_sql.compile()
    actual_compiled = actual_sql.compile()
    
    # 验证SQL语句和参数完全一致
    assert str(expected_compiled) == str(actual_compiled)
    assert expected_compiled.params == actual_compiled.params
    mock_exec_result.first.assert_called_once_with()

方法2:使用unittest.mock.ANY宽松匹配

如果不需要严格验证SQL的具体内容,仅需确认exec方法被调用,可以用ANY匹配任意Select对象:

from unittest.mock import ANY

@pytest.mark.asyncio
async def test_get_user_by_login(mocker):
    # Arrange
    mock_exec_result = AsyncMock(return_value=None)
    mock_session = mocker.AsyncMock()
    mock_session.exec.return_value = mock_exec_result

    # Act
    await UserService.get_user_by_login(mock_session, LOGIN)

    # Assert
    mock_session.exec.assert_called_once_with(ANY)
    mock_exec_result.first.assert_called_once_with()

方法3:自定义匹配器

编写自定义匹配函数,封装SQL等价性校验逻辑,让断言更简洁:

def matches_select(expected):
    def matcher(actual):
        expected_compiled = expected.compile()
        actual_compiled = actual.compile()
        return str(expected_compiled) == str(actual_compiled) and expected_compiled.params == actual_compiled.params
    return matcher

@pytest.mark.asyncio
async def test_get_user_by_login(mocker):
    # Arrange
    mock_exec_result = AsyncMock(return_value=None)
    mock_session = mocker.AsyncMock()
    mock_session.exec.return_value = mock_exec_result
    expected_sql = select(User).where(User.login == LOGIN)

    # Act
    await UserService.get_user_by_login(mock_session, LOGIN)

    # Assert
    mock_session.exec.assert_called_once_with(matches_select(expected_sql))
    mock_exec_result.first.assert_called_once_with()

内容的提问来源于stack exchange,提问作者Mister Balise

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.12 22:42:46