如何测试async with代码块中的异常以补全测试覆盖?
测试异步commit异常分支的方案
要覆盖conn.commit()抛出异常后的代码块,关键是正确模拟异步会话的commit方法抛出异常,之前patch失败大概率是patch的目标不对,或者没处理异步方法的异常抛出。以下是具体实现步骤:
核心思路
因为代码里用async with get_session() as conn获取会话,所以需要:
- 让
get_session()返回一个Mock异步上下文管理器 - 给这个Mock的
commit()方法设置抛出异常的逻辑 - 用
pytest-asyncio运行异步测试用例
具体测试代码示例
假设你用pytest和unittest.mock来写测试:
import pytest from unittest.mock import patch, MagicMock import re from your_module import YourClass # 替换成你的类名 @pytest.mark.asyncio async def test_update_eda_configuration_commit_exception(): # 1. 构造测试用的ConfigurationInputOnUpdate对象 test_config_input = MagicMock(id=1, attribute="test_attr", value="test_val") # 2. 模拟数据库查询结果:找到要更新的配置 mock_config = MagicMock() mock_result = MagicMock() mock_result.scalars().unique().first.return_value = mock_config # 3. 构造Mock会话:让commit()抛出带orig属性的异常 mock_conn = MagicMock() mock_conn.execute.return_value = mock_result # 模拟commit抛出异常,异常对象要有orig属性,且转成字符串包含DETAIL字段 mock_exception = Exception("test error") mock_exception.orig = "ERROR: something wrong\nDETAIL: duplicate value" mock_conn.commit.side_effect = mock_exception # 4. 模拟get_session()返回的异步上下文管理器 async def mock_get_session(): yield mock_conn # 5. patch get_session,运行测试方法 with patch("your_module.get_session", side_effect=mock_get_session): result = await YourClass.update_eda_configuration(test_config_input) # 6. 验证异常分支的逻辑是否执行 # 检查re.findall是否按预期执行 assert re.findall(pattern='DETAIL:.*', string=str(mock_exception.orig)) == ["DETAIL: duplicate value"] # 验证commit方法被调用过 mock_conn.commit.assert_awaited_once()
关键注意点
- patch路径必须准确:
patch("your_module.get_session")中的路径,是update_eda_configuration所在模块里导入get_session的路径,不能直接patchget_session的原始定义路径(除非二者在同一模块)。 - 异步方法的异常模拟:
mock_conn.commit是异步方法,直接给side_effect赋值异常即可,pytest-asyncio会正确处理异步方法的异常抛出。 - 异常对象需包含orig属性:你的代码中用到了
e.orig,所以必须给模拟的异常对象添加该属性,否则会触发AttributeError。
如果不用pytest,也可以用asyncio.run()来执行异步测试逻辑,但pytest-asyncio会让异步测试的编写和运行更顺畅。
内容的提问来源于stack exchange,提问作者Plaoo
相关产品推荐
相关产品推荐

