014、测试之道：使用Pytest进行单元测试、集成测试与异步测试

014、测试之道使用Pytest进行单元测试、集成测试与异步测试一、从线上一个诡异bug说起上周排查一个生产环境问题用户上传文件偶尔会超时日志里没有任何异常堆栈。最初怀疑是Nginx配置问题折腾半天无果。后来在本地用curl循环测试了上百次终于复现了一次——服务端日志显示请求进来了但业务逻辑根本没执行。问题出在依赖注入的一个异步函数里某个条件分支下忘了写await。这种问题在开发时很难发现因为大多数测试用例都走了正常流程。这件事让我重新审视了测试策略光有单元测试不够集成测试覆盖不全更危险而异步代码的测试必须用对工具和方法。今天我们就用Pytest这把瑞士军刀把FastAPI的测试体系彻底讲透。二、Pytest基础别再用unittest那套了先看一个典型错误示范# test_wrong.pyimportunittestfromfastapi.testclientimportTestClientfrommainimportappclassTestApp(unittest.TestCase):defsetUp(self):self.clientTestClient(app)deftest_read_item(self):responseself.client.get(/items/42)self.assertEqual(response.status_code,200)# 太啰嗦了Pytest的写法清爽得多# test_right.pyfromfastapi.testclientimportTestClientfrommainimportapp clientTestClient(app)deftest_read_item():responseclient.get(/items/42)assertresponse.status_code200# 直接用assert多自然assertresponse.json()[item_id]42# 链式断言关键优势不用写类函数就是测试用例assert后面可以接任何表达式失败时会自动输出详细信息夹具fixture系统比setUp/tearDown灵活十倍三、夹具Fixture测试的共享工具箱夹具是Pytest的灵魂很多人只用了皮毛。看个实际场景# conftest.pyimportpytestfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmakerfrommainimportapp,get_dbfromdatabaseimportBase# 数据库夹具 - 每个测试用独立的数据库pytest.fixture(scopefunction)deftest_db():enginecreate_engine(sqlite:///./test.db)TestingSessionLocalsessionmaker(bindengine)# 创建表Base.metadata.create_all(bindengine)dbTestingSessionLocal()try:yielddbfinally:db.close()# 清理表避免测试间相互影响Base.metadata.drop_all(bindengine)# 客户端夹具 - 依赖数据库夹具pytest.fixturedefclient(test_db):defoverride_get_db():try:yieldtest_dbfinally:passapp.dependency_overrides[get_db]override_get_dbwithTestClient(app)asc:yieldc app.dependency_overrides.clear()# 重要一定要清理覆盖使用起来极其顺手deftest_create_item(client):responseclient.post(/items/,json{name:测试物品})assertresponse.status_code201dataresponse.json()assertdata[name]测试物品assertidindata# 验证生成了ID经验之谈conftest.py里的夹具对整个目录生效作用域scope选function最安全module级可能引入测试污染夹具可以嵌套形成依赖链记得清理dependency_overrides我在这栽过跟头四、单元测试专注单个组件单元测试要快、要隔离。FastAPI的依赖注入系统让这变得简单# 测试纯函数deftest_password_hash():fromauthimporthash_password hashedhash_password(mypassword)asserthashed!mypassword# 确实加密了assertlen(hashed)64# SHA256输出长度# 测试依赖项deftest_get_current_user():fromauthimportget_current_userfromunittest.mockimportMock mock_requestMock()mock_request.headers{Authorization:Bearer valid_token}# 模拟验证逻辑userget_current_user(mock_request,tokenvalid_token)assertuser.usernametestuser关键点用unittest.mock模拟外部依赖每个测试只关注一个函数或类测试边界条件和异常分支五、集成测试验证组件协作集成测试要模拟真实调用链但不用启动完整服务deftest_full_flow(client,test_db):# 1. 创建用户user_respclient.post(/users/,json{username:test,password:secret})user_iduser_resp.json()[id]# 2. 登录获取tokenlogin_respclient.post(/login,data{username:test,password:secret})tokenlogin_resp.json()[access_token]# 3. 用token创建物品item_respclient.post(/items/,json{title:我的物品},headers{Authorization:fBearer{token}})assertitem_resp.status_code201# 验证数据库确实写入了frommodelsimportItem db_itemtest_db.query(Item).first()assertdb_item.title我的物品assertdb_item.owner_iduser_id# 关键验证关联关系踩坑提醒测试顺序很重要上面的测试依赖下面的夹具用test_db直接查库验证比只检查API响应更可靠集成测试可以暴露接口设计缺陷比如缺少必要的关联查询六、异步测试FastAPI的核心难点异步测试最容易出错重点看三种写法importpytestfromhttpximportAsyncClient# 方法1使用AsyncClient - 推荐pytest.mark.asyncioasyncdeftest_async_endpoint():asyncwithAsyncClient(appapp,base_urlhttp://test)asac:responseawaitac.get(/async-endpoint)assertresponse.status_code200# 方法2测试异步依赖pytest.mark.asyncioasyncdeftest_async_dependency():fromdependenciesimportasync_redis_client# 直接调用异步依赖函数valueawaitasync_redis_client.get(key)assertvalueisNone# 或具体的预期值# 方法3测试后台任务importasynciofromtasksimportprocess_uploadpytest.mark.asyncioasyncdeftest_background_task():# 模拟文件上传mock_fileMock()mock_file.read.return_valuebtest content# 直接调用任务函数resultawaitprocess_upload(mock_file)assertresult[status]processed# 验证副作用比如数据库写入frommodelsimportTaskLog logtest_db.query(TaskLog).first()assertlog.filenameupload.txt特别注意事项一定要加pytest.mark.asyncio装饰器TestClient不支持异步必须用AsyncClient异步夹具要用pytest_asyncio.fixture测试BackgroundTasks时最好直接测试任务函数本身七、测试覆盖率与实战技巧光写测试不够要知道覆盖了哪些代码# 安装覆盖率工具pipinstallpytest-cov# 运行测试并生成报告pytest--covapp --cov-reporthtml tests/# 查看哪些行没覆盖到pytest--covapp --cov-reportterm-missing tests/我的经验规则控制器层路由重点测状态码和响应格式业务逻辑层测所有分支包括异常路径数据访问层测CRUD和查询边界工具函数要求100%覆盖率一个实用技巧——用monkeypatch模拟环境变量deftest_config_with_env(monkeypatch):monkeypatch.setenv(DATABASE_URL,sqlite:///test.db)fromconfigimportsettingsassertsettings.database_urlsqlite:///test.db八、个人建议测试不是写完就完事的要让它成为开发流程的一部分。我在团队里推行这些实践本地提交前跑一遍相关模块的测试用pytest -xvs tests/test_specific.py快速验证CI流水线配置pytest --cov --cov-fail-under80覆盖率不达标就失败遇到bug时先写一个复现bug的测试用例修复后再确保它通过异步代码用asyncio.run()在同步环境里调试但测试一定要用异步方式最后说个真事我们有个服务曾经因为测试没覆盖timeout参数上线后在高并发下频繁超时。后来补了个压力测试才发现问题。所以记住——测试不仅要覆盖“应该发生什么”还要覆盖“可能发生什么”。好的测试像安全网让你敢在代码高空作业。开始可能觉得麻烦等它救过你几次命你就离不开了。