当前位置：首页 > ai >正文

【接口自动化】-5- 接口关联处理

ai 2025/8/11 5:55:38

一、用例执行顺序

增加一个冒烟用例的文件夹，把必须提取的 yaml 放到这个文件夹
asmoke 文件夹
pytest 默认的顺序是按 ASCII 进行排序的。ASCII 只能针对一个字符。
print(ord("a"))

如果默认不是按 ASCII 码执行，可以加入如下代码:

yaml_case_list = list(testcase_path.glob("**/*.yaml"))
yaml_case_list.sort()

二、接口关联封装设计

⭐ 接口关联的核心问题

→ “前一个接口的输出，作为后一个接口的输入”

比如：

登录接口返回 token → 下单接口需要携带该 token 才能调用
上传文件接口返回 file_id → 发布商品接口需要用该 file_id

要解决这类问题，需实现 “提取响应数据 → 存储为变量 → 传递给后续接口” 的完整流程。

⭐ 整体设计思路

用例标准化（CaseInfo 数据模型）：
通过 @dataclass 定义 CaseInfo，强制约束用例必须包含 feature/story/request/extract 等字段，让 YAML 用例格式统一。
响应数据提取（extract 字段设计）：
在 YAML 用例中通过 extract 声明 “需要从响应中提取哪些变量”，支持 正则、jsonpath 两种提取方式。
提取逻辑封装（ExtractUtil 工具类）：
编写 ExtractUtil.extract() 方法，统一处理 “根据提取规则（正则 /jsonpath ）从响应中取值” 的逻辑。
用例执行流程（create_testcase 动态生成测试）：
发送接口请求后，自动检查用例是否有 extract 配置 → 若有，调用 ExtractUtil 提取变量并存储（如写入 YAML ），供后续接口使用。

⭐ 核心代码模块拆解

1. 用例数据模型：`CaseInfo`

from dataclasses import dataclass@dataclass
class CaseInfo:# 用例基础信息feature: str  story: str    title: str    # 接口请求信息（method/url/params 等）request: dict # 断言信息（可扩展）validate: dict # 新增：提取规则配置（关键！实现接口关联）extract: dict = None

作用：用 dataclass 规范 YAML 用例的结构，确保所有用例包含 extract 字段（选填）

默认值： dict=None

2. 提取规则配置：YAML 中的 `extract` 设计

YAML 示例（get_token.yaml ）：

feature: 公众号模块
story: 获取鉴权码接口
title: 验证获取鉴权码接口成功返回
request:method: geturl: https://api.weixin.qq.com/cgi-bin/tokenparams:grant_type: client_credentialappid: wx8a9de038e93f77absecret: 8326fc915928dee3165720c910effb86
# 关键：定义提取规则
extract:  # csrf_token：变量名；[json, "$.access_token", 0]：提取规则（jsonpath 方式）csrf_token: [json, "$.access_token", 0]  
validate: null

规则解析：

[json, "$.access_token", 0] 表示：
- 用 jsonpath 方式提取
- 提取响应中 access_token 的值（$.access_token 是 jsonpath 表达式）
- 取列表第 0 个值（jsonpath 返回列表，通常取第 0 个元素）

3. 提取工具类：`ExtractUtil`

import re
import jsonpathclass ExtractUtil:def extract(self, res, var_name, attr_name, expr, index):"""解析并提取响应中的变量:param res: 请求响应对象（包含 text/json/cookies 等）:param var_name: 要提取的变量名（如 csrf_token ）:param attr_name: 提取方式（json/re ，对应 yaml 中的 json/re 关键字 ）:param expr: 提取表达式（如 jsonpath 表达式 `$.access_token` ，正则表达式 `pattern` ）:param index: 提取结果的下标（如取列表第 0 个元素 ）"""try:if attr_name == "json":# 用 jsonpath 提取values = jsonpath.jsonpath(res.json(), expr)  # 取指定下标值（通常是 0 ）result = values[index]  elif attr_name == "re":# 用正则提取values = re.findall(expr, res.text)  result = values[index] if values else Noneelse:raise Exception(f"不支持的提取方式：{attr_name}")# 提取后可存储变量（如写入 YAML ，供后续接口使用）# 示例：write_yaml({var_name: result})  return resultexcept Exception as e:raise Exception(f"提取变量 {var_name} 失败：{str(e)}")

根据 attr_name（json 或 re ）选择提取方式：
- json → 用 jsonpath.jsonpath 解析响应 JSON
- re → 用 re.findall 解析响应文本
提取结果通过 index（如下标 0 ）取值，确保拿到目标数据。

4. 测试用例生成：`create_testcase`

def create_testcase(yaml_path):# 参数化装饰器：让每个 YAML 用例数据驱动一次测试@pytest.mark.parametrize("caseinfo", read_testcase(yaml_path))def func(self, caseinfo):# 1. 校验用例是否符合 CaseInfo 规范new_caseinfo = verify_yaml(caseinfo)  # 2. 发送接口请求res = RequestUtil().send_all_request(**new_caseinfo.request)  # 3. 关键：检查是否需要提取变量（接口关联核心逻辑）if new_caseinfo.extract:  for var_name, extract_rule in new_caseinfo.extract.items():# extract_rule 格式：[attr_name, expr, index]# 如：[json, "$.access_token", 0]attr_name, expr, index = extract_rule  # 调用 ExtractUtil 提取变量ExtractUtil().extract(res, var_name, attr_name, expr, index)  return func

细节：提取关联接口的变量在发送请求后→因为这个变量是响应中提取的必须先发送请求得到响应。

流程拆解：

发送请求后，若用例定义了 extract → 遍历 extract 的每个变量配置。
对每个变量（如 csrf_token ），调用 ExtractUtil.extract 提取数据并存储（如写入 YAML ）。

5. YAML 用例示例

# 示例：get_token.yaml
feature: 公众号模块
story: 获取鉴权码接口
title: 验证获取鉴权码接口成功返回
request:method: geturl: https://api.weixin.qq.com/cgi-bin/tokenparams:grant_type: client_credentialappid: wx8a9de038e93f77absecret: 8326fc915928dee3165720c910effb86
# 提取规则：用 jsonpath 从响应中提取 access_token ，存为 csrf_token
extract:  csrf_token: [json, "$.access_token", 0]  
validate: null

执行流程：

发送 get_token 请求 → 响应包含 access_token。
触发 extract 逻辑 → 调用 ExtractUtil.extract → 用 jsonpath 提取 $.access_token → 结果存入 csrf_token（如写入 YAML ）。
后续接口（如下单接口）可读取 csrf_token 作为入参，实现接口关联。

⭐ 接口关联完整流程示例

步骤 1：获取 token（`get_token.yaml` ）

发送请求 → 响应 JSON 包含 access_token。
extract 配置触发提取 → csrf_token: [json, "$.access_token", 0] → 提取 access_token 并存为 csrf_token。

步骤 2：下单接口（`order.yaml` ）

feature: 电商模块
story: 下单接口
title: 验证下单接口需携带 token
request:method: posturl: /api/orderdata:# 读取之前提取的 csrf_token（如从 YAML 读取）token: ${csrf_token}  product_id: 123
extract: null
validate:status_code: 200

执行时，token 会自动替换为 get_token 接口提取的 csrf_token → 实现前接口的输出作为后接口的输入。

三、优化代码 `ExtractUtil`

⭐ 先复习：深拷贝 vs 浅拷贝（理解代码里的 `copy.deepcopy`）

浅拷贝：复制对象的 “引用”，新对象和原对象共享同一块内存。修改新对象会影响原对象（如 list.copy()、字典赋值）。
深拷贝：复制对象的 “值”，新对象和原对象内存地址完全独立。修改新对象不影响原对象（需用 copy.deepcopy 实现）。

new_res = copy.deepcopy(res)

作用是 “完全复制一份响应对象（res ）”，后续对 new_res 的修改（如 new_res.json = ... ）不会影响原 res，避免污染原始响应数据。

`⭐ ExtractUtil.extract` 完整流程

import copy
import jsonpath
import re
from commons.yaml_util import write_yaml  # 假设封装了 YAML 写入class ExtractUtil:def extract(self, res, var_name, attr_name, expr: str, index):# 1. 深拷贝响应对象，避免修改原数据new_res = copy.deepcopy(res)  # 2. 处理响应的 json 属性（兼容非 JSON 响应）try:# 尝试把响应转成 JSON，赋值给 new_res.jsonnew_res.json = new_res.json()  except Exception:# 若响应不是 JSON，给 new_res.json 赋默认值new_res.json = {"msg": "response not json data"}  # 3. 通过反射获取响应属性（attr_name 是提取来源，如 'json'/'text' ）# 例如：attr_name='json' → 获取 new_res.json（处理后的 JSON 数据）data = getattr(new_res, attr_name)  print(f"data: %s" % data)  # 调试用：打印提取到的原始数据# 4. 根据表达式（expr）判断提取方式（jsonpath 或正则）if expr.startswith("$"):# 4.1 jsonpath 提取（expr 是 jsonpath 表达式，如 '$.access_token' ）# 注意：老师代码里的 dict(data) 可能有问题！如果 data 本身是字典，无需转换lis = jsonpath.jsonpath(data, expr)  else:# 4.2 正则提取（expr 是正则表达式，如 'pattern' ）lis = re.findall(expr, data)  # 5. 提取结果处理：取指定下标值，写入 YAMLif lis:# 取列表的第 index 个元素（通常是 0 ）value = lis[index]  # 写入 extract.yaml，变量名是 var_name（如 'csrf_token' ）write_yaml({var_name: value})

流程拆解：

深拷贝响应：new_res = copy.deepcopy(res) → 安全操作响应数据，不影响原响应。
处理 JSON 响应：new_res.json = new_res.json() → 尝试转 JSON，失败则赋默认值，避免后续提取报错。只有json()是方法，需要单独处理成属性
反射获取属性：getattr(new_res, attr_name) → 动态获取响应的某个属性（如 res.json/res.text/res.cookies ），作为提取的 “数据源”。
判断提取方式：
- expr.startswith("$") → 用 jsonpath 提取（适用于 JSON 响应）。
- 否则 → 用正则（re.findall ）提取（适用于文本响应）。
结果写入 YAML：提取到值（lis 非空）→ 取第 index 个元素 → 写入 extract.yaml，供后续接口使用。

⭐ 为什么单独处理json()这个唯一的方法：“数据是否需要动态计算 / 解析”

类型	示例（`res` 是 `requests.Response` 对象）	为什么是方法 / 属性？
属性	`res.text`、`res.status_code`	数据是直接存储在响应对象里的原始值，访问时无需额外计算（如状态码、响应文本）。
方法	`res.json()`	数据需要动态解析 / 计算（把响应文本转 JSON 字典），每次调用可能有不同结果（或抛出异常）。

⭐ 代码细节 & 潜在问题分析

1. 关键细节：`attr_name` 的作用

attr_name 是 “提取数据源”，决定从响应的哪个部分提取数据：

若 attr_name='json' → 从响应的 JSON 数据提取（需先 new_res.json = ... 处理）。
若 attr_name='text' → 从响应文本（res.text ）提取（适合正则）。
若 attr_name='cookies' → 从响应 cookies（res.cookies ）提取（需结合正则或自定义逻辑）。

extract:csrf_token: [json, "$.access_token", 0]

2. 潜在问题：`dict(data)` 的冗余

lis = jsonpath.jsonpath(dict(data), expr)

如果 data 本身已经是字典（如 new_res.json 处理后是字典），dict(data) 是多余的，直接用 data 即可。
修正后：

lis = jsonpath.jsonpath(data, expr)  # 去掉 dict(data)

3. 调试技巧

打印 data：确认提取的原始数据是否正确（如 JSON 结构、文本内容）。
打印 lis：确认 jsonpath / 正则是否匹配到值（如 lis 是否为空，是否是预期列表）。

⭐ 代码优化建议（让逻辑更健壮）

1. 修复 `dict(data)` 问题

# 原代码（有问题）
lis = jsonpath.jsonpath(dict(data), expr)  # 优化后（直接用 data ，前提是 data 可被 jsonpath 解析）
lis = jsonpath.jsonpath(data, expr)

2. 增加异常处理（避免提取失败导致用例崩溃）

try:if expr.startswith("$"):lis = jsonpath.jsonpath(data, expr)else:lis = re.findall(expr, data)
except Exception as e:raise Exception(f"提取变量失败：{str(e)}，表达式：{expr}，数据：{data}")

3. 支持更多提取来源（`attr_name` 扩展）

除了 json/text，还可支持 cookies/headers：

# 例如：attr_name='cookies' → 获取响应的 cookies（字典格式）
data = getattr(new_res, attr_name)  
# 若 data 是字典，可直接用 jsonpath 提取（如 '$.session_id' ）

⭐ 总结：代码的设计思路

深拷贝响应：确保原始响应数据不被修改，避免影响其他逻辑。
动态提取来源：通过 attr_name（如 json/text ）灵活选择提取数据的来源（响应 JSON、响应文本等）。
多提取方式支持：同时兼容 jsonpath（适合 JSON 响应）和正则（适合文本响应）。
结果持久化：提取的变量写入 YAML，供后续接口通过 read_yaml 使用，实现接口关联。

四、重点解释：反射

⭐ 反射的核心工具：`getattr` 函数

Python 里的 getattr(object, name[, default]) 函数，作用是动态获取对象的属性或方法：

object：要操作的对象（这里是 new_res，即响应对象的深拷贝）
name：字符串，指定要获取的属性名（如 json/text/cookies ）
default（可选）：属性不存在时返回的默认值（代码里没用到，依赖 try-except 处理）

通俗理解：
你不用提前写死 new_res.json 或 new_res.text，而是用字符串（attr_name 的值）动态决定要取对象的哪个属性，实现 **“运行时动态决定访问哪个属性”**。

⭐ 结合代码看反射的作用

回顾关键代码：

def extract(self, res, var_name, attr_name, expr: str, index):# 深拷贝 & 处理 json 方法转属性（略）new_res = copy.deepcopy(res)  # ... 中间处理 json 方法转属性的逻辑 ... # 反射核心：用 getattr 动态获取 new_res 的 attr_name 属性data = getattr(new_res, attr_name)  print(f"data: %s" % data)  # 后续根据 expr 决定用 jsonpath 还是正则提取if expr.startswith("$"):lis = jsonpath.jsonpath(data, expr)else:lis = re.findall(expr, data)# ... 提取后写入 yaml ...

场景举例（假设 YAML 配置）：

YAML 里写：

extract:token: [json, "$.access_token", 0]

对应调用 extract 时：

attr_name = 'json'（第二个参数是 json ）
getattr(new_res, 'json') → 等价于直接写 new_res.json，拿到响应解析后的 JSON 数据

如果 YAML 配置是：

extract:session_id: [text, "session_id=(\w+)", 0]

attr_name = 'text'
getattr(new_res, 'text') → 等价于 new_res.text，拿到响应的文本内容，用正则提取 session_id

⭐ 反射的价值：让提取逻辑 “动态可配置”

如果不用反射（getattr ），代码会变成这样：

# 伪代码：不用反射，写死属性判断
if attr_name == 'json':data = new_res.json
elif attr_name == 'text':data = new_res.text
elif attr_name == 'cookies':data = new_res.cookies
# ... 每加一个属性，就要加一个判断 ...

⭐ 总结：反射在这里的作用

动态性：根据 YAML 里的 attr_name，动态决定从响应对象的哪个属性提取数据，无需写死判断。
扩展性：新增提取来源（如 headers/cookies ）时，代码无需修改，只需改 YAML 配置。
简洁性：用一行 getattr 替代大量 if-elif，让 extract 函数更简洁、易维护。

五、重点解释：.json()处理响应对象new_res

⭐ 代码中对非 JSON 响应的处理逻辑

try:# 尝试把响应转成 JSON，赋值给 new_res.jsonnew_res.json = new_res.json()  
except Exception:# 若响应不是 JSON，给 new_res.json 赋默认值new_res.json = {"msg": "response not json data"}

逻辑拆解：

尝试解析 JSON：
调用 new_res.json()（requests 响应对象的原生方法），如果响应是 JSON 格式（如 {"code":0, "data": "xxx"}），则正常解析为字典，赋值给 new_res.json。
非 JSON 响应的降级处理：
如果响应不是 JSON（如 HTML 页面 <html>...</html>、纯文本 success 等），new_res.json() 会抛出 JSONDecodeError 异常，此时进入 except 分支，给 new_res.json 赋值一个默认字典 {"msg": "response not json data"}。

`⭐ res.json()` 的核心作用

res.json() 是 requests 库中 Response 对象的内置方法，作用是：
将 HTTP 响应的原始字节流（JSON 格式字符串）解析为 Python 字典 / 列表对象。

简单说：

输入：响应体中的 JSON 格式字符串（如 '{"code":0, "data":"xxx"}'）。
输出：对应的 Python 字典（如 {"code":0, "data":"xxx"}）。

这样你就可以直接用 Python 语法操作数据（如 res.json()["code"] 获取状态码），无需手动写 json.loads() 解析。

⭐ 响应对象 `res` 的类型及数据流转

1. `res` 的类型

res 是 requests.Response 类型的对象（由 requests.get()/requests.post() 等方法返回），包含了 HTTP 响应的所有信息（状态码、响应体、 headers 等）。

2. 响应数据的流转过程（从字节流到 Python 对象）

当服务器返回一个 JSON 响应时，数据经历了 3 个阶段：

阶段	数据形态	说明
1. 服务器返回	字节流（`bytes` 类型）	如 `b'{"code":0, "data":"xxx"}'`，这是网络传输的原始格式。
2. `res.content`	字节流（`bytes` 类型）	`Response` 对象的 `content` 属性，直接存储原始字节流。
3. `res.text`	字符串（`str` 类型）	`Response` 对象自动将字节流解码为字符串（默认 UTF-8），如 `'{"code":0, "data":"xxx"}'`。
4. `res.json()`	Python 字典 / 列表（`dict`/`list`）	调用 `json()` 方法，将 `res.text` 解析为 Python 对象。

举例：JSON 响应的解析过程

import requests# 发送请求，获取响应对象 res（Response 类型）
res = requests.get("https://api.example.com/data")# 1. 原始字节流（bytes）
print(type(res.content))  # <class 'bytes'>
print(res.content)        # b'{"code":0, "data":"hello"}'# 2. 解码后的字符串（str）
print(type(res.text))     # <class 'str'>
print(res.text)           # '{"code":0, "data":"hello"}'# 3. 解析后的 Python 字典（dict）
print(type(res.json()))   # <class 'dict'>
print(res.json())         # {'code': 0, 'data': 'hello'}# 可以直接用字典语法操作
print(res.json()["data"])  # 'hello'

⭐ 调用 `new_res.json()` 的完整逻辑

try:new_res.json = new_res.json()  # 调用原生 json() 方法
except Exception:new_res.json = {"msg": "response not json data"}

当响应是 JSON 格式时：

new_res 是 Response 对象的深拷贝（仍为 Response 类型）。
调用 new_res.json() → 内部会先获取 new_res.text（JSON 字符串），再通过 json.loads() 解析为 Python 字典。
将解析后的字典赋值给 new_res.json（原本 json 是方法，现在变成属性，存储解析结果）。

后续通过 getattr(new_res, "json") 就能直接拿到这个字典，用 jsonpath 提取数据（如 $.data）。

⭐ 关键区别：`res.json()` 与 `json.loads(res.text)`

两者功能类似（都是解析 JSON 字符串），但 res.json() 是 requests 封装的便捷方法，额外做了 2 件事：

自动处理编码问题：确保 res.text 的编码正确（如 GBK、UTF-8）。
错误处理更友好：解析失败时抛出 JSONDecodeError，便于定位问题。

因此，在 requests 响应处理中，优先用 res.json() 而非手动 json.loads(res.text)。

⭐ 总结

res.json()：将 JSON 格式的响应字符串解析为 Python 字典 / 列表，方便直接操作。
res 的类型：requests.Response 对象，包含原始字节流（content）、解码字符串（text）等属性。
数据流转：字节流 → 字符串 → Python 对象，res.json() 完成最后一步解析。

六、`requests.Response` 对象

他是requests 库封装的 “HTTP 响应容器”，它像一个 “数据包”，包含了服务器返回的所有信息（状态码、响应体、头部等）。

`⭐ requests.Response` 对象的 “样子”（核心属性展示）

发送请求后得到的 res 对象，包含以下关键信息（可以理解为一个 “结构化的响应数据包”）：

import requests# 发送请求，获取 Response 对象
res = requests.get("https://api.example.com/data")# 1. 状态信息
print(res.status_code)  # 状态码（如 200/404）
print(res.headers)      # 响应头（字典格式，如 Content-Type: application/json）# 2. 响应体数据（核心）
print(res.content)      # 原始字节流（bytes 类型）
print(res.text)         # 解码后的字符串（str 类型）
print(res.json())       # 解析后的 Python 对象（仅当响应是 JSON 时可用）# 3. 其他信息
print(res.url)          # 请求的 URL
print(res.cookies)      # 响应的 Cookies

⭐ 为什么 `content`（原始字节流）必须存在？

content 存储的是服务器返回的 原始二进制数据（bytes 类型），比如：

文本响应：b'{"code":0}'（JSON 字符串的字节形式）
图片响应：b'\x89PNG\r\n\x1a\n\x00\x00...'（PNG 图片的二进制数据）
文件响应：b'PK\x03\x04\x14\x00\x00...'（ZIP 文件的二进制数据）

存在的必要性：

最原始的响应形态：
网络传输的数据本质上都是二进制（字节流），content 直接保存了这种原始形态，确保数据没有丢失或被篡改。
适配非文本响应：
对于图片、文件、视频等二进制内容，无法直接转成字符串（会乱码），必须用 content 处理（如保存图片到本地）：
```
# 保存图片（必须用 content）
with open("image.png", "wb") as f:f.write(res.content)  # 写入原始字节流
```

⭐ 为什么 `text`（解码字符串）必须存在？

text 是 requests 自动将 content（字节流）解码后的字符串（str 类型），比如：

字节流 b'{"code":0}' 解码后 → '{"code":0}'（JSON 字符串）
字节流 b'<html>hello</html>' 解码后 → '<html>hello</html>'（HTML 字符串）

存在的必要性：

方便处理文本响应：
对于接口返回的 JSON、HTML、纯文本等文本类响应，我们更习惯用字符串操作（如正则匹配、打印查看），text 省去了手动解码的步骤：
```
# 直接操作字符串（比字节流更直观）
if "success" in res.text:print("操作成功")
```
自动处理编码：
requests 会根据响应头的 Content-Type 或字节流中的编码标识，自动选择合适的编码（如 UTF-8、GBK）解码，避免手动处理 content.decode("utf-8") 的麻烦。