FastAPI + httpx 接口自动化怎么设计?从单用例执行到套件变量传递(实战代码)

发布时间:2026/6/24 10:20:10

FastAPI + httpx 接口自动化怎么设计?从单用例执行到套件变量传递(实战代码) 做接口自动化Postman pytest 脚本能跑通第一条链路但接口一多就会遇到几个问题登录 token 怎么在步骤间传递同一条接口测试/预发环境怎么切换断言逻辑写在哪、报告怎么落库FastAPI 项目里执行引擎要不要和 Web 服务共用异步事件循环下面结合我在开源项目BrickCore里的后端实现把接口自动化模块的核心执行链路拆开讲。你可以直接借鉴到自己的 FastAPI 项目里不一定非要用完整平台。一、分层模型先想清楚数据怎么流不要把「发 HTTP」和「管理用例」揉在一个函数里。推荐四层ApiDefinition接口元数据 ↓ 一对多 ApiTestCase用例覆盖参数 断言 提取规则 ↓ 多对多 ApiTestSuite套件顺序/并行、失败即停 ↓ ApiTestPlan计划定时、跨套件、报告汇总设计原则接口定义与用例分离——接口改 path 只改一处用例只声明「覆盖哪些字段」执行结果必须落库——request_detail/assertions_result/extracted_vars分开存报告和排错才有着力点变量池贯穿执行上下文——套件内后序用例读取前序提取值路由按职责拆分apis.py管理接口、cases.py管理用例、suites.py/exec.py负责执行、records.py查报告。二、单用例执行一条完整的流水线run_single_case()是整套引擎的核心。一次执行大致经过这些阶段按实际代码顺序加载用例/接口/环境 → 合并变量项目 环境 传入变量 → Token 授权注入 → 前置脚本 → 替换 URL/Header/Body 中的 ${{var}} → httpx 发请求支持重试 → 断言评估 → 变量提取 → 后置脚本 → 写入 ApiRunRecord2.1 变量合并优先级怎么定执行前先把多层变量合成一个 dict后面所有替换都基于它# 项目全局变量 环境变量 套件传入的 accumulated_varsproj_varsproject.global_varsor{}env_varsenv.global_varsor{}all_variables{**proj_vars,**env_vars,**(variablesor{})}# Token 授权登录接口自动刷新后注入auth_vars,auth_errawaitinject_auth_variables(project_id,env_id,all_variables)ifauth_vars:all_variables.update(auth_vars)踩坑点环境变量里如果也配置了token要和授权模块约定优先级否则会出现「刚刷新的 token 被旧值覆盖」。2.2 URL 拼接base_url pathdefbuild_api_url(base_url:str,path:str)-str:ifpath.startswith((http://,https://)):returnpath# 允许用例写绝对地址base_urlbase_url.rstrip(/)pathpathifpath.startswith(/)else/pathreturnbase_urlpath环境表存base_url如https://api.test.com接口定义只存/v1/users/login。2.3 变量替换递归处理 dict/list/str请求里任意字段都可能有${{token}}需要递归替换defreplace_variables(data,variables):支持 ${{var_name}}、${var}、{{var}} 三种写法ifisinstance(data,str):returnresolver.replace_in_string(data)ifisinstance(data,dict):return{k:replace_variables(v,variables)fork,vindata.items()}ifisinstance(data,list):return[replace_variables(item,variables)foritemindata]returndata建议替换时同时记录replacements明细原值 → 替换后写入request_detail报告里能展开「当时到底发了什么」排错效率差一个数量级。2.4 httpx 发请求和 FastAPI 共用异步asyncwithhttpx.AsyncClient(timeoutcase.timeout,follow_redirectsTrue)asclient:responseawaitclient.request(methodapi.method.upper(),urlurl,headersprepare_httpx_headers(headers),paramsparams,jsonbodyifbody_typejsonelseNone,dataform_dataifbody_typeformelseNone,)prepare_httpx_headers()单独抽出来是因为 httpx 对非 ASCII header 需要encodingutf-8中文 Cookie 等场景否则会炸。为什么不用 requestsFastAPI 路由本身 async执行引擎若用同步 requests高并发时要线程池包一层计划里 50 条用例串行还能忍并行或定时任务一多就堵事件循环。2.5 断言配置驱动别写死在代码里用例上的assertions是 JSON 数组执行时逐条评估defevaluate_assertion(response,response_body,assertion):tassertion[type]# status_code / json_path / header / containsoperatorassertion[operator]# equals / gt / contains ...expectedassertion[expected]iftstatus_code:actualresponse.status_codeeliftjson_path:actualget_json_path_value(response_body,assertion[target])eliftcontains:textjson.dumps(response_body,ensure_asciiFalse)returnstr(expected)intext,text# ... 再按 operator 比较returncompare(actual,expected,operator),actual前端配置示例[{type:status_code,operator:equals,expected:200},{type:json_path,target:$.data.token,operator:not_equals,expected:},{type:json_path,target:$.data.userId,operator:gt,expected:0}]2.6 变量提取给下一步用extractors:[{name:token,source:json,path:$.data.token},{name:order_id,source:json,path:$.data.id}]提取结果写入extracted_vars并 merge 进当前上下文供套件内下一条用例使用。三、套件执行变量如何在用例间传递单用例跑通不够回归要跑一串。套件顺序执行时关键是accumulated_vars累积传递accumulated_varsawaitmerge_exec_variables(env,project_global_vars,{})forcase_idincase_ids:resultawaitrun_single_case(case_id,env.id,suite_record_id,username,variablesaccumulated_vars,# 传入当前变量池project_global_varsproject_global_vars,)# 本用例提取的变量合并给下一条accumulated_vars.update(result.extracted_varsor{})ifsuite.stop_on_failureandresult.status!success:break# 失败即停顺序 vs 并行的差异模式变量传递适用场景顺序前序extracted_vars传给后序登录→下单→查询并行各用例共享 setup 后快照互不传无依赖的读接口压测并行实现里我们对每个用例copy一份变量快照避免协程间互相污染。四、实战示例登录 → 创建订单 → 查订单不用看 UI只看数据就能理解整条链路。用例 1登录{name:登录获取token,api_path:/api/login,method:POST,body:{username:test,password:123456},assertions:[{type:status_code,operator:equals,expected:200}],extractors:[{name:token,source:json,path:$.data.token}]}用例 2创建订单Header 引用变量{name:创建订单,request_headers:[{key:Authorization,value:Bearer ${{token}}}],body:{product_id:1001,count:2},extractors:[{name:order_id,source:json,path:$.data.id}]}用例 3查询订单{name:查询订单详情,request_headers:[{key:Authorization,value:Bearer ${{token}}}],path:/api/orders/${{order_id}},assertions:[{type:json_path,target:$.data.id,operator:equals,expected:${{order_id}}}]}三条用例加入同一套件顺序执行无需手写 Python 脚本串流程。五、Swagger 导入把「录接口」自动化手工录入 50 个 REST 接口很折磨。导入接口本质是解析 OpenAPI JSONrouter.post(/import/swagger)asyncdefimport_swagger(project_id:int,file:UploadFile):datajson.loads(awaitfile.read())swaggerSwaggerDoc(**data)forpath,methodsinswagger.paths.items():formethod,operationinmethods.items():ifmethod.startswith(x-):continue# 解析 parameters、requestBody、responses# 处理 components/schemas 里的 $refawaitApiDefinition.create(project_idproject_id,methodmethod.upper(),pathpath,nameoperation.get(summary)orpath,sourceswagger,# headers / params / body 从 schema 展开)两个实现细节$ref要递归解析——#/components/schemas/User不展开body 结构就是空的responses优先取 200/201——顺便存响应 schema后面可做 JSON Schema 校验导入后base_url仍放环境配置不要写死在每个接口上。六、执行记录报告和排错靠它一次用例执行至少落这些字段awaitApiRunRecord.create(case_idcase.id,statussuccessifall_passedelsefailed,request_urlurl,request_headerssafe_headers,request_bodybody,response_statusresponse.status_code,response_bodyresponse_body,response_timetotal_ms,# 含断言、脚本的总耗时assertions_result[...],# 每条断言的 expected/actual/passedextracted_vars{token:xxx},request_detail{# 变量替换明细、重试记录、脚本日志replacements:[...],retry_info:{...},stage_timings:{...},},)计划级报告 汇总多条ApiRunRecord按suite_run_record_id分组即可。七、自己实现最小版MVP 只要 4 个函数如果想先做个 Demo不必一上来做完整平台函数职责merge_variables()合并环境/项目/上下文变量replace_variables()递归替换${{}}run_single_case()httpx 请求 断言 提取run_suite()for 循环 accumulated_vars.update()一个 FastAPI 路由POST /run-suite入参case_ids env_id就能跑通 80% 的场景。定时任务、权限、报告页面可以后补。八、和 pytest requests 怎么选维度pytest 脚本平台化引擎上手低中步骤间传参自己写 fixture配置 extractors 变量池非开发维护难测试可在页面改断言报告靠 allure 等插件内置落库 导出适合单项目 CI多项目、多人协作个人练手继续 pytest团队要 Web 管理、定时回归、权限隔离值得做平台化。九、Docker 部署附录若使用 BrickCore 开源版自建根目录docker-compose.yml可拉起 MySQL / Redis / RabbitMQ / MinIO / 前后端。速查gitclone https://gitee.com/BanZhuanKeOrz/BrickCore.gitcdBrickCorecdfrontendnpminstallnpmrun buildcd..dockercompose up-d--builddockerexec-ifastapi_mysql mysql --default-character-setutf8mb4\-uadmin-pBrickCore123456fastapidatabase.sql完整步骤见仓库docs-site/guide/docker-deploy.md。部署不是本文重点需要时再跟文档即可。十、小结接口自动化后端的核心不是「能发 HTTP」而是这四件事配置化——断言、提取、覆盖参数都是 JSON别硬编码变量池——合并优先级清晰替换过程可追踪套件编排——顺序执行时accumulated_vars累积传递可观测——请求/响应/断言/替换明细落库以上逻辑来自 BrickCore AI测试平台的后端实现FastAPI httpx Tortoise ORM开源地址https://gitee.com/BanZhuanKeOrz/BrickCore在线体验http://43.142.83.156/admin / BrickCore123456下一篇介绍平台侧 AI 生用例、UI 自动化等功能本篇专注后端执行引擎。觉得有帮助欢迎 Star有问题评论区留言。

相关新闻