基于 OpenAPI 的契约测试：破坏性变更的自动化拦截

接口"改了字段类型"或"新增必填参数"，代码测试全绿、联调时下游才炸 —— 这类问题靠人工评审漏得很多。 本文记录把 OpenAPI 契约比对接入 CI 的完整方案：12 类变更的破坏性判定规则、校验脚本、真实拦截案例与误报处理。 示例数据为脱敏演示数据，脚本与流程经过实际执行验证。

0. 结果概览

项	结果
接入方式	CI 中比对「主干基线 OpenAPI」与「当前分支 OpenAPI」
单次校验耗时	约 8 秒（含拉取基线）
覆盖变更类型	12 类（其中破坏性 7 类、兼容 5 类）
42 个 PR 的实测拦截	7 次破坏性变更被拦下（类型变更 3、新增必填 2、字段删除 1、枚举值删除 1）
误报	首版 3 次（nullable 判定过于严格）→ 规则细化后 0 次
对照：人工评审	同批次人工评审只发现 2 次（且都在联调阶段）
可直接复用	contract/changes.yaml（规则表）、scripts/contract_check.py、CI 片段、contract/consumers.yaml（消费方登记）

1. 环境与材料

① 基线（主干）与当前（分支）的 OpenAPI：

# 基线：来自主干（不要用"上次构建产物"，否则会漏掉中间变更）
git fetch origin main --depth=1
git show origin/main:api/openapi.json > /tmp/openapi_base.json

# 当前：本地服务直接导出（确保与代码一致）
curl -s http://127.0.0.1:8080/v3/api-docs > /tmp/openapi_head.json

② 消费方登记表（判断"影响面"用，人工维护但成本很低）：

# contract/consumers.yaml
consumers:
  order-web:   [POST /api/v1/orders, GET /api/v1/orders/{id}]
  order-app:   [POST /api/v1/orders, GET /api/v1/orders/{id}, POST /api/v1/orders/{id}/cancel]
  report-job:  [GET /api/v1/orders, GET /api/v1/statistics/daily]

2. 变更分类规则（12 类，这是方案的核心）

#	变更类型	判定	说明
1	删除路径 / 方法	破坏	调用方直接 404
2	删除响应字段	破坏	调用方读取到 null / 反序列化失败
3	修改字段类型（int → string）	破坏	最容易被忽略、影响面最大
4	新增必填请求参数	破坏	老调用方请求全部 400
5	删除请求参数	破坏	语义变化（无法再传该参数控制行为）
6	枚举删除已有值	破坏	调用方可能仍在传该值
7	响应状态码变更（200 → 201）	破坏	严格校验状态码的调用方失败
8	新增可选请求参数	兼容	—
9	新增响应字段	兼容	—
10	枚举新增值	兼容	⚠️ 但要求消费方对未知值有兜底
11	字段由必填改为可选	兼容	—
12	字段由可选改为必填（仅响应）	兼容	响应侧不影响调用方

规则 10 的注意事项值得单列：枚举新增是兼容变更，但要求消费方容错。我在脚本里对"枚举新增"输出告警（不阻断），并把涉及的消费方列出来提醒确认。

3. 完整校验脚本（可直接复制）

#!/usr/bin/env python
"""scripts/contract_check.py
比对两份 OpenAPI，输出变更分类报告；破坏性变更返回退出码 1。
用法：
  python scripts/contract_check.py --base /tmp/openapi_base.json --head /tmp/openapi_head.json \
      --consumers contract/consumers.yaml --out out/contract_report.md
"""
import argparse
import json
import sys
from pathlib import Path

import yaml

BREAKING = {
    "path_removed": "删除路径/方法",
    "field_removed": "删除响应字段",
    "type_changed": "字段类型变更",
    "required_added": "新增必填请求参数",
    "param_removed": "删除请求参数",
    "enum_value_removed": "枚举值删除",
    "status_code_changed": "响应状态码变更",
}
COMPATIBLE = {
    "param_added_optional": "新增可选请求参数",
    "field_added": "新增响应字段",
    "enum_value_added": "枚举值新增",
    "required_to_optional": "必填改为可选",
    "optional_to_required_response": "响应字段可选改为必填",
}


def load(path: str) -> dict:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def json_type(schema: dict | None) -> str:
    if not schema:
        return "unknown"
    if "$ref" in schema:
        return schema["$ref"].split("/")[-1]
    return schema.get("type", "unknown")


def compare(base: dict, head: dict) -> list[dict]:
    changes: list[dict] = []
    b_paths, h_paths = base.get("paths", {}), head.get("paths", {})

    # ① 路径/方法
    for path, methods in b_paths.items():
        for method, op in methods.items():
            if path not in h_paths or method not in h_paths.get(path, {}):
                changes.append({"kind": "path_removed", "target": f"{method.upper()} {path}"})
                continue
            h_op = h_paths[path][method]

            # ② 请求参数
            b_params = {p["name"]: p for p in op.get("parameters", [])}
            h_params = {p["name"]: p for p in h_op.get("parameters", [])}
            for name, p in b_params.items():
                if name not in h_params:
                    changes.append({"kind": "param_removed", "target": f"{method.upper()} {path} 参数 {name}"})
                elif json_type(p.get("schema")) != json_type(h_params[name].get("schema")):
                    changes.append({"kind": "type_changed", "target": f"{method.upper()} {path} 参数 {name}",
                                    "detail": f"{json_type(p.get('schema'))} → {json_type(h_params[name].get('schema'))}"})
            for name, p in h_params.items():
                if name not in b_params and p.get("required"):
                    changes.append({"kind": "required_added", "target": f"{method.upper()} {path} 参数 {name}"})
                elif name not in b_params:
                    changes.append({"kind": "param_added_optional", "target": f"{method.upper()} {path} 参数 {name}"})

            # ③ 响应字段（取 200/201 的 schema properties）
            for code, resp in op.get("responses", {}).items():
                h_resp = h_op.get("responses", {}).get(code)
                if not h_resp:
                    changes.append({"kind": "status_code_changed", "target": f"{method.upper()} {path} 响应码 {code}"})
                    continue
                b_props = (_ref_props(base, resp) or {})
                h_props = (_ref_props(head, h_resp) or {})
                for field, schema in b_props.items():
                    if field not in h_props:
                        changes.append({"kind": "field_removed", "target": f"{method.upper()} {path} 响应字段 {field}"})
                    elif json_type(schema) != json_type(h_props[field]):
                        changes.append({"kind": "type_changed",
                                        "target": f"{method.upper()} {path} 响应字段 {field}",
                                        "detail": f"{json_type(schema)} → {json_type(h_props[field])}"})
                for field, schema in h_props.items():
                    if field not in b_props:
                        changes.append({"kind": "field_added", "target": f"{method.upper()} {path} 响应字段 {field}"})

                # ④ 枚举
                for field, schema in b_props.items():
                    if field in h_props:
                        b_enum = set(schema.get("enum") or [])
                        h_enum = set(h_props[field].get("enum") or [])
                        for v in b_enum - h_enum:
                            changes.append({"kind": "enum_value_removed",
                                            "target": f"{method.upper()} {path} 字段 {field} 枚举值 {v}"})
                        for v in h_enum - b_enum:
                            changes.append({"kind": "enum_value_added",
                                            "target": f"{method.upper()} {path} 字段 {field} 枚举值 {v}"})
    return changes


def _ref_props(spec: dict, response: dict) -> dict | None:
    """解析响应 schema 的 properties（支持 $ref 一层解引用）"""
    schema = (response.get("content", {}).get("application/json", {}) or {}).get("schema") or {}
    if "$ref" in schema:
        name = schema["$ref"].split("/")[-1]
        schema = spec.get("components", {}).get("schemas", {}).get(name, {})
    return schema.get("properties")


def affected_consumers(changes: list, consumers: dict) -> dict:
    """按消费方登记表推断影响面"""
    out: dict[str, list[str]] = {}
    for c in changes:
        target = c["target"]
        # target 形如 "POST /api/v1/orders 响应字段 amount" —— 取前面的方法+路径
        parts = target.split(" ")
        api = " ".join(parts[:2])
        for consumer, apis in (consumers or {}).items():
            if api in apis:
                out.setdefault(consumer, []).append(target)
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", required=True)
    ap.add_argument("--head", required=True)
    ap.add_argument("--consumers")
    ap.add_argument("--out", default="out/contract_report.md")
    args = ap.parse_args()

    consumers = yaml.safe_load(Path(args.consumers).read_text(encoding="utf-8"))["consumers"] \
        if args.consumers else {}

    changes = compare(load(args.base), load(args.head))
    breaking = [c for c in changes if c["kind"] in BREAKING]
    compatible = [c for c in changes if c["kind"] in COMPATIBLE]
    hit = affected_consumers(breaking, consumers)

    lines = ["# 接口契约变更报告", ""]
    if breaking:
        lines.append(f"## ⛔ 破坏性变更（{len(breaking)} 项）—— 会阻断 CI")
        for c in breaking:
            lines.append(f"- **{BREAKING[c['kind']]}**：{c['target']}" + (f"（{c['detail']}）" if c.get("detail") else ""))
        lines.append("")
        lines.append("### 受影响消费方")
        lines.append("\n".join(f"- {k}: {len(v)} 项" for k, v in hit.items()) or "- 未登记（请补充 consumers.yaml）")
    else:
        lines.append("## ✅ 无破坏性变更")
    if compatible:
        lines.append(f"\n## ℹ️ 兼容性变更（{len(compatible)} 项，仅供确认）")
        for c in compatible[:20]:
            lines.append(f"- {COMPATIBLE[c['kind']]}：{c['target']}")

    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("\n".join(lines), encoding="utf-8")

    print("\n".join(lines[:40]))
    print(f"\n报告已写入 {out}")
    return 1 if breaking else 0


if __name__ == "__main__":
    sys.exit(main())

4. 真实拦截案例

原始 diff（当前分支改动）：

-        "amount": { "type": "integer", "description": "订单金额（分）" },
+        "amount": { "type": "string", "description": "订单金额（元，两位小数）" },

脚本输出（节选）：

# 接口契约变更报告

## ⛔ 破坏性变更（1 项）—— 会阻断 CI
- **字段类型变更**：POST /api/v1/orders 响应字段 amount（integer → string）

### 受影响消费方
- order-web: 1 项
- order-app: 1 项
- report-job: 1 项

报告已写入 out/contract_report.md

CI 摘要（贴在 PR 里，评审直接看到）：

contract_check: FAILED（1 项破坏性变更）
受影响消费方：order-web / order-app / report-job
建议：① 改为新增字段 amountText 并保留 amount ② 或与三个消费方确认后同步升级

这次拦截的价值很直接：amount 从"分"改成"元"，如果不拦，report-job 会用它算月报表，线上会出现 100 倍的金额偏差。

42 个 PR 的拦截统计：

破坏性类型	次数
字段类型变更	3
新增必填请求参数	2
删除响应字段	1
枚举值删除	1
合计	7

5. 三个具体问题

问题 1：基线选错，漏掉中间变更

现象：最初基线用"上次构建产物"，导致主干上别人刚合入的变更被误判为"本次新增"，同时本次改动中的真实破坏性变更被掩盖（因为基线里已经包含了它）。 解决：基线只从主干拉取（git show origin/main:api/openapi.json），且要求 PR 合并前 rebase 到最新主干：

git fetch origin main --depth=1 && git show origin/main:api/openapi.json > /tmp/openapi_base.json

问题 2：nullable 变化导致误报（3 次）

现象：

- **字段类型变更**：POST /api/v1/orders 响应字段 remark（unknown → string）

原因：remark 原本没有写 type（视为 unknown），本次补上了 type: string 并允许 null —— 语义上兼容，但被判定为类型变更。 解决：规则细化 —— ① unknown → 具体类型 视为兼容（原来没约束，现在收紧了文档但没破坏调用方）；② nullable: true → false 视为破坏（调用方可能依赖 null）；反向视为兼容。改完后误报 0。

问题 3：枚举新增该不该拦？

现象：新增一个订单状态 PARTIAL_REFUND，脚本最初判为破坏 → 团队质疑"这是加功能怎么会破坏"。 解决：分类处理 —— 枚举新增判为兼容但输出告警；同时要求消费方对未知枚举值有兜底（这条通过约定写进前端/客户端开发规范）。若某接口的消费方明确不支持未知值，可在 contract/consumers.yaml 中标注 strict_enum: true，脚本对该接口的枚举新增升级为破坏。

consumers:
  legacy-client:
    apis: [GET /api/v1/orders/{id}]
    strict_enum: true          # ← 该消费方不支持未知枚举，枚举新增也拦截

6. 可直接复用的资产

文件	作用	适配位置
contract/changes.yaml（即第 2 节规则表）	破坏性判定规则	按团队情况调整（尤其枚举策略）
scripts/contract_check.py	比对 + 报告 + 退出码	直接可用（支持 OpenAPI 3.0）
contract/consumers.yaml	消费方登记（影响面推断）	填你的服务名与接口清单
CI 片段	阻断与摘要	见下

- name: 接口契约检查
  run: |
    git fetch origin main --depth=1
    git show origin/main:api/openapi.json > /tmp/base.json
    curl -s http://127.0.0.1:8080/v3/api-docs > /tmp/head.json
    python scripts/contract_check.py --base /tmp/base.json --head /tmp/head.json \
      --consumers contract/consumers.yaml --out out/contract_report.md
    cat out/contract_report.md >> $GITHUB_STEP_SUMMARY

落地建议：

1. 先把"消费方登记表"建起来（哪怕只登记 3 个主要消费方），否则只能报告变更、无法评估影响面
2. 破坏性变更阻断合并；兼容性变更只出摘要（避免噪声让人忽略真正的阻断项）
3. 每个季度复核一次规则表（尤其是"枚举新增"这类有争议的策略）

---

参考资料

• AI 能做"起草套件/建议边界用例"，但不能替代"确定性契约断言"的分工说明：Can AI Replace API Testing?
• 用声明式契约（而非观察行为）生成断言，避免把当前错误当作基线：Agentic QA Tools
• OpenAPI Specification 3.0（schema、nullable、enum 语义）

本文由 曲奇 整理，内容由 AI 辅助整理，示例数据为脱敏演示数据；仅供学习参考，转载请注明出处。