Pytest测试数据库CRUD返回函数对象而非返回值如何解决

问题1：test_read报错拿到函数对象的修复 #

根因 #

• fixture返回的是封装好的可调用函数，测试用例中未传入参数调用该函数，直接对比函数对象和预期值导致报错
• @pytest.mark.parametrize 传入的是字符串形式的入参，无法直接用于函数调用
• pytest默认按用例名字典序执行，test_read 会比 test_update 先执行，导致读取时还没有目标数据

修复步骤 #

1. 安装pytest-ordering控制用例执行顺序：
   pip install pytest-ordering
2. 调整测试用例参数写法，改为结构化参数直接传递，调用fixture返回的函数拿到实际返回值再做断言，标记执行顺序确保update先执行。

问题2：表删除不生效的修复 #

根因 #

• delete_collection 中异步run_sync用法错误：run_sync要求第一个参数为可调用对象，后续参数自动传给该可调用对象，原代码直接执行了drop_all并将返回值传给run_sync，没有在异步连接上下文中执行删表操作
• 测试用例中用于校验的collection_names是预加载的缓存值，删表后没有重新获取最新的表元数据，导致校验逻辑失效

修复代码 #

/src 主脚本delete_collection函数修正

async def delete_collection(self, table_name, syncro_engine=sync_engine):
    table = self.retrieve_table_obj(table_name)
    async with self._engine.begin() as conn:
        # 修正run_sync用法，传入可调用对象和对应参数，不要直接执行drop_all
        await conn.run_sync(Base.metadata.drop_all, tables=[table], checkfirst=True)

/test/test_db.py 测试用例修正

import pytest

# 标记执行顺序为1，优先执行update写入测试数据
@pytest.mark.run(order=1)
@pytest.mark.parametrize('collection, query, data',
                         [('user', {'name': 'Alice'}, {'org': 'Wonderland'})])
def test_update(test_update, collection, query, data):
    # 调用函数实际执行更新操作
    test_update(collection, query, data)

# 标记执行顺序为2，update执行完成后再执行read校验
@pytest.mark.run(order=2)
@pytest.mark.parametrize('collection, query, mods, expected',
                         [('user', {'name': 'Alice'}, {'fields': ['org']}, [('Wonderland',)])])
def test_read(test_read, collection, query, mods, expected):
    # 调用read函数拿到实际返回值再断言
    actual_result = test_read(collection, query, mods)
    assert actual_result == expected

# 标记执行顺序为3，最后执行删表测试
@pytest.mark.run(order=3)
@pytest.mark.parametrize('table_name, sync_engine, expected_table',
                         [('image', sync_engine, 'image')])
def test_drop_table(drop_table, table_name, sync_engine, expected_table):
    # 执行删表操作
    drop_table(table_name, sync_engine)
    # 删表后重新获取最新的表列表，可替换为你实际获取表名的方法
    latest_collections = DB._cache.classes.keys()
    with pytest.raises(KeyError) as error_info:
        if expected_table in latest_collections:
            raise KeyError
    assert error_info.type is KeyError

补充说明：原read函数中循环匹配字段时每次都会覆盖result变量，如果mods['fields']传多个字段只会返回最后一个字段的查询结果，如果你需要查询多个字段需要调整该函数的逻辑。

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