drift-testing

Drift 技能

切勿修改正在测试的 OpenAPI 规范。

参考文件

• references/test-cases.md — 完整的测试用例 YAML 架构、所有模式、数据集和表达式
• references/auth.md — 身份验证配置、动态令牌、非标准方案、401/403 测试
• references/mock-server.md — 使用 Prism 进行本地测试：设置、Prefer 标头、规范质量问题
• references/lua-api.md — 完整的 Lua API：生命周期事件、http()、dbg()、导出函数
• references/cli-reference.md — 所有 CLI 命令/标志、配置、并行执行、退出代码
• references/pactflow-and-cicd.md — BDCT 发布工作流、GitHub Actions、GitLab CI

脚本

• scripts/extract_endpoints.py — 读取规范并输出所有操作和响应代码。 摘要模式会标记规范示例缺失的参数。脚手架模式（--scaffold）会生成可直接填写的 operations: YAML 区块，其中包含正确的身份验证模式、404 的 nil UUID、 4xx 的 ignore.schema 以及 FILL_IN 标记。使用 --only-missing <drift.yaml> 仅生成 现有测试文件尚未覆盖的缺失项。需要 pyyaml。
• scripts/extract_channels.py — extract_endpoints.py 的 AsyncAPI 对应版本。读取 AsyncAPI 3.x 规范并输出所有操作及其操作类型（send/receive）、执行模式（async-observe / async-inject / async-request-reply）和通道地址。脚手架模式（--scaffold）会生成可直接填写的 Drift operations: YAML 桩，并根据模式提供正确的触发器/探测器结构。使用 --only-missing <drift.yaml> 仅为尚未覆盖的操作生成脚手架。需要 pyyaml。仅支持 AsyncAPI 3.x。
• scripts/check_coverage.py — 覆盖率检查器：比较 OpenAPI 或 AsyncAPI 规范与 Drift 测试文件。 OpenAPI 覆盖率会跟踪操作和响应代码；AsyncAPI 覆盖率会跟踪每个操作/消息目标。 需要 pyyaml。
• scripts/run_loop.sh / scripts/run_loop.ps1 — 反馈循环运行器：反复运行 drift verify --failed，直到所有测试 通过，然后运行 check_coverage.py。两项检查都通过后，退出代码才为 0。依赖项会通过 uv 自动安装。 Windows 上请使用 .ps1 版本。
• scripts/start_mock.sh / scripts/start_mock.ps1 — 根据 OpenAPI 规范启动 Prism 模拟服务。如有需要，会安装 Prism。 支持 --port 和 --dynamic 标志。Windows 上请使用 .ps1 版本。

完整文档：https://support.smartbear.com/swagger/contract-testing/docs/en/drift.html 如需查找所有可用页面，请获取站点地图：https://support.smartbear.com/swagger/contract-testing/docs/sitemap.xml

---

安装

# Quickest — no install needed
npx @pactflow/drift --help

# Project-level (recommended for teams)
npm install --save-dev @pactflow/drift

# Global
npm install -g @pactflow/drift

# Verify
drift --version

---

项目设置

drift init   # interactive wizard — scaffolds all files below

drift init 是交互式命令 — 请让用户运行它。

drift/
├── drift.yaml              # Main config — sources, plugins, global settings
├── drift.lua               # Lifecycle hooks and helper functions
├── my-api.dataset.yaml     # Test data
└── my-api.tests.yaml       # Test cases

最简 drift.yaml：

# yaml-language-server: $schema=https://download.pactflow.io/drift/schemas/drift.testcases.v1.schema.json
drift-testcase-file: v1
title: "My API Tests"

sources:
  - name: source-oas # referenced in test targets
    path: ./openapi.yaml # or uri: https://... for remote specs
  - name: product-data
    path: ./product.dataset.yaml
  - name: functions
    path: ./product.lua

plugins:
  - name: oas # spec-first verification
  - name: json
  - name: data

global:
  auth:
    apply: true
    parameters:
      authentication:
        scheme: bearer # bearer | basic | api-key
        token: ${env:API_TOKEN}

operations:
  # test cases here — see references/test-cases.md

---

运行测试

# Basic run
drift verify --test-files drift.yaml --server-url https://api.example.com/v1

# Single operation (fast iteration)
drift verify --test-files drift.yaml --server-url https://api.example.com/v1 --operation getProductByID

# Re-run only failures
drift verify --test-files drift.yaml --server-url https://api.example.com/v1 --failed

# Filter by tags
drift verify --test-files drift.yaml --server-url https://api.example.com/v1 --tags smoke
drift verify --test-files drift.yaml --server-url https://api.example.com/v1 --tags '!destructive'

所有标志、并行执行、JUnit 输出和退出代码，请参阅 references/cli-reference.md。

---

全覆盖反馈循环

目标是实现端点全覆盖时：

注意 — 生产环境中的破坏性测试： 如果 --server-url 指向在线 生产 API，DELETE 和 POST 测试将产生永久影响。请务必使用专用测试账号， 并确认 DELETE 测试中使用的任何资源都可以安全删除。

复制此清单并跟踪进度：

Coverage Loop Progress:
- [ ] Step 0: Check current coverage (check_coverage.py)
- [ ] Step 1: Parse spec and collect operation list (openapi-parser skill or extract_endpoints.py)
- [ ] Step 2: Assemble initial test file
- [ ] Step 3: Run tests (run_loop.sh / run_loop.ps1)
- [ ] Step 4: Diagnose and fix each failure
- [ ] Step 5: Apply common fixes (hooks for state, data seeding)
- [ ] Step 6: Verify exit code 0 + full coverage

步骤 0 — 检查当前覆盖率

编写测试前或恢复现有测试套件时运行：

# Run against your spec and test file(s)
uv run path/to/scripts/check_coverage.py \
  --spec openapi.yaml \
  --test-files drift.yaml

# Multiple files / globs
uv run path/to/scripts/check_coverage.py \
  --spec openapi.yaml \
  --test-files "tests/*.yaml"

# Machine-readable output (for CI or scripting)
uv run path/to/scripts/check_coverage.py \
  --spec openapi.yaml \
  --test-files drift.yaml --json

输出显示：完全没有测试的操作、缺少特定响应代码测试的操作， 以及操作/代码的总体百分比。退出代码 0 = 全覆盖，1 = 仍有缺口。

默认情况下，脚本会排除 500/501/502/503（与下方步骤 1 中的规则相同）。传入 --exclude-codes 可自定义排除项。

步骤 1 — 解析规范

使用 extract_endpoints.py 收集完整的操作列表、每个操作中所有有文档记录的响应代码，以及可直接使用的 operations: YAML 存根。如果你的环境中有 openapi-parser skill，也可以改用它。

# See all operations + response codes, flagging params with no spec example
uv run scripts/extract_endpoints.py --spec openapi.yaml

# Generate skeleton stubs for every operation
uv run scripts/extract_endpoints.py --spec openapi.yaml \
  --scaffold --source my-oas > operations.yaml

# Generate ONLY the gaps not already in an existing test file
uv run scripts/extract_endpoints.py --spec openapi.yaml \
  --scaffold --only-missing drift.yaml --source my-oas >> drift.yaml

GET /products          → 200, 401, 404
POST /products         → 201, 400, 401
DELETE /products/{id}  → 204, 401, 403, 404

**关键：**任何没有规范级 example 的参数都会导致 Value for query parameter X is missing。请在 parameters.query/path/headers 中为每个参数提供明确的值。

全局必填查询参数（例如每个端点都需要的 ?version=YYYY-MM-DD）可以通过 http:request 钩子一次性注入，而不必在每个测试用例中重复：

["http:request"] = function(event, data)
  if data.query == nil then data.query = {} end
  data.query["version"] = "2024-01-04"
  return data   -- MUST return modified data
end

重复的 operationId 值 — 有些规范会对两个不同的路径重复使用相同的 operationId。请使用 method:path 定位受影响的操作：

target: source-oas:post:/orgs/{org_id}/apps/installs/{install_id}/secrets

覆盖率要求不包含 500 响应 — 500 响应需要服务器存在错误，因此无法确定性地触发。

步骤 2 — 组装初始测试文件

将 openapi-parser 生成的存根填入 drift.yaml。不必追求完美——循环会揭示缺失的内容。尽可能从简单的测试开始：

getProduct_Success:
  target: source-oas:getProductByID
  parameters:
    path:
      id: 10
  expected:
    response:
      statusCode: 200

为每个操作添加标签 — 标签可启用 --tags 筛选，并让测试套件更易管理：

getProduct_Success:
  target: source-oas:getProductByID
  tags: [smoke, read-only, products]
  ...

getProduct_Unauthorized:
  tags: [security, auth]
  ...

deleteProduct_Success:
  tags: [destructive, products]
  ...

常用标签：smoke、read-only、write、destructive、security、auth、regression。完整的标签说明请参阅 references/test-cases.md。

错误路径请参阅 references/test-cases.md 中关于 401、403、404 和 400 的模式。 模拟服务器设置请参阅 references/mock-server.md。

步骤 3 — 运行并捕获失败

run_loop.sh 脚本会自动完成步骤 3 到步骤 6 的全部工作：

# Runs drift --failed in a loop, then checks coverage. Exits 0 only when both pass.
path/to/scripts/run_loop.sh \
  --spec openapi.yaml \
  --test-files drift.yaml \
  --server-url https://api.example.com/v1

或者手动运行 drift 并迭代：

drift verify --test-files drift.yaml --server-url https://api.example.com/v1

# Re-run only failures to keep the loop fast
drift verify --test-files drift.yaml --server-url https://api.example.com/v1 --failed

使用模拟服务器进行本地测试时，请先启动 Prism：

path/to/scripts/start_mock.sh --spec openapi.yaml --port 4010
# then in another terminal:
path/to/scripts/run_loop.sh --spec openapi.yaml --test-files drift.yaml --server-url http://localhost:4010

步骤 4 — 诊断并修复每个失败

症状	可能原因	修复
收到 404，预期 200	测试数据不存在	添加 operation:started 钩子以创建资源
收到 200，预期 404	该 ID 恰好存在	使用 ${notIn(...)} 或 nil UUID 00000000-0000-0000-0000-000000000000
收到 401，预期 200	未配置身份验证	添加 global.auth 或检查令牌环境变量
收到 200，预期 401	未移除身份验证	添加 exclude: [auth] 和无效令牌
收到 403，预期 200	令牌缺少所需范围	使用权限充足的令牌
收到 200，预期 403	需要有效身份验证和禁止访问的资源	指向令牌无权访问的资源；参见 references/auth.md
响应上的架构验证错误	API 已偏离规范，或规范包含无效示例	检查规范示例是否有效 — Drift 正确报告规范错误的情况并非不可能
Value for query parameter X is missing	可选参数没有规范 example	为每个没有规范示例的参数提供明确的值
200 测试收到 400	缺少全局必需的查询参数	通过 http:request 钩子注入，或添加到每个测试用例中
收到 500	测试数据触发了服务器错误	修复数据

ignore: { schema: true } 只会抑制请求架构验证。对任何 4xx 场景都应使用它 — 尤其是在针对模拟服务器进行测试时，因为 Prism 不会强制执行身份验证，且可能返回不准确的错误正文。响应架构验证没有绕过方式；规范示例错误会表现为失败（参见 references/mock-server.md）。

多个 2xx 状态码： 为每个文档中记录的状态码编写一个测试 — 不支持 statusCode: [200, 204] 数组语法。

动态 ID 和钩子时机： 数据集表达式会在 operation:started 之前解析。请使用预先创建的静态 ID，或通过 http:request 重写 URL。

步骤 5 — 常见修复

测试前数据必须存在（DELETE、PUT、PATCH）：

["operation:started"] = function(event, data)
  if data.operation == "deleteProduct_Success" then
    http({ url = server_url .. "/products", method = "POST",
           body = { id = 10, name = "test", price = 9.99 } })
  end
end,
["operation:finished"] = function(event, data)
  http({ url = server_url .. "/products/10", method = "DELETE" })
end,

完整的 Lua API 和 data 对象结构请参见 references/lua-api.md。

步骤 6 — 确认退出代码为 0 且覆盖率完整

drift verify --test-files drift.yaml --server-url https://api.example.com/v1
echo "Exit code: $?"

宣布完成前，请确认覆盖率完整：

uv run path/to/scripts/check_coverage.py \
  --spec openapi.yaml --test-files drift.yaml
echo "Coverage exit: $?"

两个命令均以 0 退出即表示完成。

---

AsyncAPI 测试

AsyncAPI 测试会根据 AsyncAPI 3.x 文档验证事件驱动服务（Kafka、SNS、SQS）。它需要一个正在运行的真实代理 — 没有对应的模拟服务器。

支持的版本： 仅支持 AsyncAPI 3.x。Drift 不支持 AsyncAPI 2.x。

插件栈（每个 AsyncAPI 测试文件都必须包含）

# yaml-language-server: $schema=https://download.pactflow.io/drift/schemas/drift.testcases.v1.schema.json
drift-testcase-file: v1
title: "<service> AsyncAPI verification"

sources:
  - name: <source-name>               # referenced in target fields
    path: ./asyncapi/<service>.asyncapi.yaml

plugins:
  - name: asyncapi      # reads the AsyncAPI document
  - name: kafka         # transport; use aws-messaging for SNS/SQS
  - name: json          # payload validation

operations:
  # test cases go here

AsyncAPI 完整覆盖率反馈循环

AsyncAPI Coverage Loop Progress:
- [ ] Step 0: Check current coverage (check_coverage.py auto-detects AsyncAPI)
- [ ] Step 1: Parse spec and collect operation list (asyncapi-parser skill or extract_channels.py)
- [ ] Step 2: Assemble test file with correct plugins and sources
- [ ] Step 3: Start broker (Kafka/SNS/SQS) and run drift verify
- [ ] Step 4: Diagnose and fix failures
- [ ] Step 5: Verify full coverage

步骤 0 — 检查当前 AsyncAPI 覆盖率

check_coverage.py 会自动检测 AsyncAPI 规范，并切换为操作/消息级覆盖率（不含状态码）。 为兼容性，未限定的 <source>:<operationId> 目标会覆盖整个操作；如需精确指定多消息覆盖率，请使用 <source>:<operationId>:<messageId>。

uv run path/to/scripts/check_coverage.py \
  --spec service.asyncapi.yaml \
  --test-files "tests/*.yaml"
# Exit 0 = all operation/message targets have at least one test; 1 = gaps remain

步骤 1 — 解析规范

使用 extract_channels.py 查看所有操作并生成脚手架桩。对于复杂的消息架构（载荷中的 oneOf/anyOf/allOf/discriminator），请改用 asyncapi-parser 技能。

# Summary: see all operations + execution modes
uv run scripts/extract_channels.py --spec service.asyncapi.yaml

# Scaffold stubs for all operations
uv run scripts/extract_channels.py --spec service.asyncapi.yaml \
  --scaffold --source async-svc > drift/tests.yaml

# Scaffold only gaps in an existing test file
uv run scripts/extract_channels.py --spec service.asyncapi.yaml \
  --scaffold --only-missing drift/tests.yaml --fragments >> drift/tests.yaml

--fragments 会省略根 operations: 键，这样追加内容时就不会创建重复的 YAML 映射。 如果不使用该选项，脚手架输出将是完整的独立 operations: 块，可重定向到新文件。

步骤 2 — 组装测试文件

将脚手架桩与正确的插件/来源标头连接起来。每个操作都需要：

• parameters 中的 correlation-id — Drift 用它来匹配正确的消息
• timeout-ms — 等待消息的时长（默认值：5000ms）
• 用于 async-observe 的 trigger 钩子，或用于 async-inject、async-inject-capture 和 async-request-reply 的 payload
• 接收操作需要 probe 钩子（async-inject）或 probe-topic（async-inject-capture）

对于 trigger 钩子：trigger 命令必须将 ${parameters.correlation-id} 传递到已发布消息的标头中。correlation ID 不匹配会导致捕获超时。

步骤 3 — 运行测试

AsyncAPI 测试无需 --server-url — Drift 会连接到 AsyncAPI servers 块中声明的消息代理。

# Run all tests
drift verify --test-files drift/tests.yaml

# Re-run only failures
drift verify --test-files drift/tests.yaml --failed

# Single operation (fast iteration)
drift verify --test-files drift/tests.yaml --operation SendOrderCreated_Observe

注意： run_loop.sh 是为 OpenAPI 设计的（它要求使用 --server-url）。对于 AsyncAPI，请手动循环运行 drift verify --failed，直到所有测试通过，然后运行 check_coverage.py 检查覆盖率。

步骤 4 — 诊断 AsyncAPI 故障

症状	可能原因	修复方法
捕获超时，trigger 已成功	Trigger 发布了不同的 correlation ID	确认 trigger 将 ${parameters.correlation-id} 作为标头值传递
捕获超时，trigger 未运行	Trigger 脚本路径错误	使用相对于测试用例 YAML 文件所在位置的路径
捕获超时，未涉及 trigger	通道地址错误或消息代理未运行	检查 AsyncAPI 规范中的 address: 是否与实际主题名称一致
Probe 误报（不应通过的测试却通过了）	未清除先前测试的状态	在 probe 开始时添加清除状态的逻辑
Probe 间歇性失败	Probe 在服务完成处理前运行	在 probe 脚本中添加重试循环或延迟
发布时架构验证失败	载荷缺少必需字段	检查 components.messages.<name>.payload.required
预期请求-响应时使用了 async-observe 模式	AsyncAPI 操作中没有 reply 块	在 parameters 中添加 reply-channel: <topic>，或修正 AsyncAPI 规范
SNS/SQS：请求-响应无法正常工作	SNS/SQS 不支持请求-响应	请求-响应交互使用 Kafka

用于开发的本地 Kafka

使用 Docker 启动本地消息代理：

# Start Kafka (KRaft mode, no Zookeeper)
docker run -d --name drift-kafka -p 9092:9092 \
  -e KAFKA_ENABLE_KRAFT=yes \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENERS="PLAINTEXT://:9092,CONTROLLER://:9093" \
  -e KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://localhost:9092" \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP="PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT" \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS="1@localhost:9093" \
  -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=true \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  confluentinc/cp-kafka:latest

# Stop when done
docker stop drift-kafka && docker rm drift-kafka

AsyncAPI servers 块声明了消息代理连接；除非通过插件配置覆盖，否则 Drift 会使用该连接。

---

快速参考

场景	方法
无状态只读端点	声明式测试，不使用钩子
稳定的测试数据	数据集表达式
在测试前创建数据	operation:started 钩子
在测试后清理数据	operation:finished 钩子
动态值（UUID、时间戳）	在 Lua 中使用 exported_functions
确保返回 404	${notIn(...)} 或 nil UUID
在模拟服务上强制指定错误代码	Prefer: code=X 标头
不连接实时后端进行测试	Prism 模拟 — 参见 references/mock-server.md
非标准身份验证前缀	http:request 钩子 — 参见 references/auth.md
仅重新运行失败的测试	--failed 标志
发布到 PactFlow	--generate-result 标志
AsyncAPI 发送操作（观察）	async-observe 模式 — 添加 trigger 钩子
AsyncAPI 接收操作（注入）	async-inject 模式 — 添加 probe 钩子或使用 probe-topic
AsyncAPI 请求/响应	async-request-reply 模式 — 无需 trigger
解析复杂的 AsyncAPI 消息架构	asyncapi-parser 技能
搭建 AsyncAPI 测试存根	extract_channels.py --scaffold
检查 AsyncAPI 覆盖率	check_coverage.py（自动检测规范类型）