夜雨聆风学习资料网

ARTICLE · 1277022

AI 尝试从接口文档生成 JMeter 脚本

AI 尝试从接口文档生成 JMeter 脚本
适合读者:用过 JMeter写过接口压测脚本的测试同学。文中所运行都是本机小并发调试(JMeter 5.6.3,被压的是一个 Python 写的示例下单接口,源码见文末附录),用来说明脚本和团队约定在执行时的效果。

一、我们的做法

交给 AI 来做:把接口路径、请求方法、请求体字段从接口文档抄进 JMeter,每个请求挂断言、配请求头。最早我们用 Cursor,配一份规则文件(jmeter-generation-rules.mdc)和一个模板脚本(test.jmx)让 AI 照着生成。后来把规则收进了 Claude Code 的 skill /performance-test。

skill 需要这几样输入:

输入
是否必需
作用
性能测试场景文档
必需
写清每个场景调哪些接口、并发多少、跑多久、目标指标、哪些参数要换着用。由写测试用例那一步产出
接口文档
必需
Swagger/OpenAPI 的 JSON 地址(如 /v2/api-docs)或导出文件,Claude Code 会自己下载读取
真实请求示例
推荐
从浏览器或微信开发者工具复制的 curl。接口文档只有字段结构,没有能用的值,AI 照着示例填
开发设计文档
可选
补充鉴权方式、接口之间的依赖关系
测试环境地址、鉴权方式
必需
登录接口是哪个、token 怎么拿

skill 不执行压测,并发、时长、循环次数怎么组合,由性能测试工程师执行时决定。

场景文档大致是这样(示意):

## 场景 1:下单流程- 涉及接口:GET /api/products → POST /api/orders → GET /api/orders/{id}- 并发用户数:20;Ramp-up:5 秒;持续时间:60 秒- 性能指标:下单 P99 ≤ 500ms,成功率 ≥ 99%- 参数传递:商品列表取 productId 传给下单;下单取 orderId 传给订单详情- 参数化:username、password(data/users.txt);quantity(data/orders.txt)

调用时把三样输入给到 skill:

/performance-test 场景文档 test/testcases/下单-性能测试场景-01.md,接口文档 openapi.json,请求示例 order_request.txt

skill 里的规则沿用了最早那份规则文件的四部分:脚本模板(现在写成了结构规范)、核心组件清单、脚本要求、输出格式。

二、脚本模板

最早给 AI 一个能跑的模板 test.jmx 照着写。用 skill 后不再读模板文件,而是把模板的结构直接写成规范:测试计划由哪些组件组成、线程组参数怎么写、事务控制器怎么勾。线程组和事务控制器这几项是这样规定的:

ThreadGroup.num_threads        ${__P(NumberofThreads,场景并发数)}ThreadGroup.ramp_time          ${__P(Ramp-Up,场景加压时间)}ThreadGroup.duration          ${__P(Duration,场景持续时间)}LoopController.loops          ${__P(Loop,1)}TransactionController.parent  true(勾选 Generate parent sample)

事务控制器勾选 Generate parent sample 后,一个事务里的几个请求合并成一行统计,报告里能直接看到“下单流程”整体的耗时。

三、核心组件清单

skill 要求每个生成的脚本必须满足这些组件约束:

#
组件
要求
作用
1
线程组
线程数、Ramp-Up、持续时间、循环次数都用 __P 读,默认值取场景文档
执行时用 -J 覆盖
2
HTTP 请求默认值
协议、域名、端口取自接口文档,禁止编造;单个请求上不再填服务器名称或 IP
所有请求共用一套地址
3
HTTP 信息头管理器
Content-Type
、Accept-Encoding、Authorization(值为 ${token})
统一请求头和鉴权
4
事务控制器
每个业务事务一个,勾选 Generate parent sample
报告里看整笔业务耗时
5
CSV 数据文件设置
用到 ${变量} 就必须加;数据文件用 .txt,变量名和表头一致;token 不放进数据文件
参数化,不同线程用不同数据
6
JSON 提取器
前后接口传参必须用,禁止写死;提取不到时给默认值 -1
前后接口传参
7
断言
每个请求挂响应断言(校验 HTTP 状态码)和 JSR223 断言(校验业务成功标识,只记日志,不判失败)
出问题时日志里有完整响应
8
监听器
聚合报告启用;察看结果树、Graphite 后端监听器加上但默认禁用
看结果

(「CSV 数据文件设置」是 JMeter 组件的名字,它读 .txt 也一样。)

AI 生成时容易漏组件,列成清单逐项打勾,漏掉的概率小很多。

约定 1:业务失败只记日志,报告错误率不含业务失败

响应断言只看 HTTP 状态码,业务码由 JSR223 断言检查但只打日志、不判失败,方便排查。压测时有些业务失败是并发下的正常结果(库存不足、已领取、未中奖),算成错误反而会让错误率失真。

接口返回 HTTP 200、但业务码报错时,报告里会统计成成功。 示例服务里,下单数量超过 5 件会返回 HTTP 200 + {"code": 1001, "msg": "单笔限购 5 件"},测试数据里每 10 笔有 1 笔超限:

下单错误率
失败信息在哪
按约定(业务失败只记日志)
0.0%
只在 jmeter.log 里(20 笔里的 2 笔)
换成业务失败也判失败
10.0%(634 / 6338)
报告里直接显示 业务失败: {"code": 1001, ...}

按这个约定执行,看完报告还要去 jmeter.log 统计 JSR223 断言记下的业务失败。按断言脚本里 log.info 打的关键字统计就行,比如:

grep -c "业务未按预期成功" jmeter.log

再按业务码分类,区分哪些是正常的业务拒绝、哪些是真出错。

另外,JSR223 断言里的业务成功标识要按每个接口的实际返回写,比如本文的 '"code": 0'。如果写成“响应里出现 data 就算成功”这种宽松判断,报错响应带 "data": null 时,日志里也不会记下来。检查生成结果时要确认这一点。

约定 2:每个环境一份脚本

HTTP 请求默认值,让 AI 把接口文档里的协议、域名、端口直接写。

<stringProp name="HTTPSampler.domain">127.0.0.1</stringProp><stringProp name="HTTPSampler.port">8080</stringProp>

补充 1:token 不能写死

HTTP 信息头管理器: Authorization 用 ${token},token 通过登录请求获取。

在skill 里补了一句:真实请求示例里的 token、Cookie 只作参考,禁止写进脚本。

补充 2:察看结果树和 Graphite 后端监听器默认禁用

  • Graphite 后端监听器
    :最早的规则要求必须加。如果本机没有 Graphite 服务,压测时 jmeter.log 每秒报一次 Error writing to Graphite: Connection refused,不影响结果,但会把真正的错误淹没。现在生成时加上、默认禁用,有 Graphite 服务再启用。
  • 察看结果树
    :调试时有用,正式压测时会占用大量内存,JMeter 官方也建议压测时不要开。现在生成时加上、默认禁用。

四、脚本要求

skill 对生成内容的要求(节选):

- 目标版本:JMeter 5.6.3- 脚本中的域名必须来自接口文档,禁止臆造- 请求体字段必须基于接口文档的 DTO 定义,禁止凭空发明字段;必填字段必须赋值- 有真实请求示例时,字段值优先照示例填;示例里没有的字段按接口文档的类型给默认值- 示例请求头里的 token / Cookie 只作参考,禁止写进脚本- JSON 请求体写在 Body Data 中,参数化字段用 ${变量名}- 需要唯一的数据用 __UUID、__RandomString 生成

接口文档和请求示例要一起给。DTO 是接口文档里定义的请求数据结构,只看 curl 示例,AI 会漏掉示例里恰好没传的必填字段;只看接口文档,又没有能用的字段值。

五、输出格式

skill 规定的产出:

test/performance-test/{需求名称}/├── {需求名称}-性能测试.jmx└── data/    ├── users.txt    └── {场景名}_data.txt
场景较多时,可以按场景拆成多个 .jmx,放在同一目录。合并交给人在 GUI 里做。

生成后检查与交付。 脚本由 AI 生成,由性能测试工程师执行,中间要有一次交接。skill 在生成后做三步:

  1. XML 完整性
    :xmllint --noout 生成的脚本.jmx,没有 xmllint 就用 JMeter 打开确认不报错。
  2. 组件清单核对
    :对照第三节的清单逐条列出满足情况,不满足的写明原因。
  3. 交付说明
    :给执行的QA,至少写清——
    • 脚本文件清单(按事务拆分时,每个事务对应哪个 .jmx);
    • 脚本里所有 __P 变量及默认值(NumberofThreads、Ramp-Up、Loop、Duration 等);
    • 需要的数据文件(.txt)、列名、建议行数;
    • 鉴权方式(登录接口、token 从哪个响应字段提取);
    • 已禁用的监听器(察看结果树、Graphite),需要时手动启用。

六、正式压测前还要准备什么

脚本调通只是第一步。正式执行性能测试,还要另外准备这些,本文不展开:

  • 独立的性能测试环境
    :配置接近生产,和功能测试环境隔开,避免压测影响别人测功能,也避免别人的操作干扰压测数据。压测机和被压服务分开部署。
  • 专门准备的测试数据
    :账号数量够用,数据量级接近生产,压完能清理。不要直接从生产导数据,里面有用户隐私。
  • 外部依赖用挡板
    :短信、支付、第三方开放平台这类接口,先跟后端确认用假接口替代,否则会产生真实费用,或者触发对方限流封号。

附录:示例服务源码

order_api.py,只用 Python 标准库,python3 order_api.py 8080 启动。下单接口里模拟了一个 20 连接的数据库连接池,本文的调试并发用不满,可以忽略。

import jsonimport randomimport sysimport threadingimport timeimport uuidfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom urllib.parse import parse_qs, urlparseUSERS = {f"user{i:03d}": "Test@123" for i in range(1, 201)}PRODUCTS = [{"id": i, "name": f"商品{i}", "price": 10 * i} for i in range(1, 21)]TOKENS, ORDERS = {}, {}lock = threading.Lock()# 模拟数据库连接池只有 20 个连接,下单要拿到连接才能写库db_pool = threading.BoundedSemaphore(20)class Handler(BaseHTTPRequestHandler):    protocol_version = "HTTP/1.1"    def log_message(self, *args):        pass    def reply(self, status, body):        data = json.dumps(body, ensure_ascii=False).encode()        self.send_response(status)        self.send_header("Content-Type", "application/json;charset=UTF-8")        self.send_header("Content-Length", str(len(data)))        self.end_headers()        self.wfile.write(data)    def read_json(self):        n = int(self.headers.get("Content-Length", 0))        return json.loads(self.rfile.read(n) or b"{}")    def current_user(self):        token = self.headers.get("Authorization", "").removeprefix("Bearer ")        return TOKENS.get(token)    def do_POST(self):        path = urlparse(self.path).path        if path == "/api/login":            time.sleep(random.uniform(0.02, 0.06))            body = self.read_json()            if USERS.get(body.get("username")) != body.get("password"):                return self.reply(200, {"code": 1002, "msg": "用户名或密码错误"})            token = uuid.uuid4().hex            TOKENS[token] = body["username"]            return self.reply(200, {"code": 0, "data": {"token": token}})        if path == "/api/orders":            # 先读完请求体再判鉴权,否则 keep-alive 连接里残留的 body 会被当成下一个请求            body = self.read_json()            if not self.current_user():                return self.reply(401, {"code": 401, "msg": "未登录"})            # 业务失败照样回 HTTP 200,不少项目是这么设计的            if body.get("quantity", 0) > 5:                return self.reply(200, {"code": 1001, "msg": "单笔限购 5 件"})            order = {"id": uuid.uuid4().hex[:12], "productId": body["productId"],                     "quantity": body["quantity"], "status": "CREATED"}            with db_pool:                time.sleep(random.uniform(0.02, 0.06))                with lock:                    ORDERS[order["id"]] = order            return self.reply(200, {"code": 0, "data": order})        self.reply(404, {"code": 404, "msg": "not found"})    def do_GET(self):        time.sleep(random.uniform(0.01, 0.03))        url = urlparse(self.path)        if not self.current_user():            return self.reply(401, {"code": 401, "msg": "未登录"})        if url.path == "/api/products":            q = parse_qs(url.query)            page, size = int(q.get("page", [1])[0]), int(q.get("pageSize", [10])[0])            items = PRODUCTS[(page - 1) * size: page * size]            return self.reply(200, {"code": 0, "data": {"list": items, "total": len(PRODUCTS)}})        if url.path.startswith("/api/orders/"):            order = ORDERS.get(url.path.rsplit("/", 1)[1])            if not order:                return self.reply(200, {"code": 1004, "msg": "订单不存在"})            return self.reply(200, {"code": 0, "data": order})        self.reply(404, {"code": 404, "msg": "not found"})class Server(ThreadingHTTPServer):    request_queue_size = 256    daemon_threads = Trueif __name__ == "__main__":    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080    print(f"order api listening on 127.0.0.1:{port}")    Server(("127.0.0.1", port), Handler).serve_forever()

相关学习资料