三十一、【高级特性篇】接口用例参数化与关联：实现上下文数据传递
前言
在接口测试中，用例之间往往存在依赖关系：一个接口的响应数据（例如，用户 ID、认证 Token、订单号）需要作为下一个接口的请求参数。目前，测试平台还没有实现这种上下游数据传递，导致测试用例的复用性低、维护成本高。
本文的目标是：
在测试平台中实现参数提取 (Extraction) 和参数注入 (Injection) 机制，使测试用例之间能够自动传递数据，从而实现真正的端到端自动化测试。
准备工作
- 前端项目就绪：
test-platform/frontend项目可以正常运行 (npm run dev)。 - 后端 API 运行中： Django 后端服务运行。
TestCase模型已存在。requests库和jsonpath-ng已安装在后端虚拟环境。- Celery 和 Redis 已配置并运行。
- Element Plus 集成完毕。
第一部分：后端数据模型调整
修改 TestCase 模型，增加字段来存储提取和注入的规则。
1. 升级 TestCase 模型
打开 test-platform/api/models.py。
在 TestCase 模型中，新增 extract_params 字段。
# test-platform/api/models.py
# ... (其他导入和模型定义) ...
class TestCase(BaseModel):
# ... (原有字段，如 module, priority, case_type, request_method, request_url 等) ...
# --- 新增：变量提取和参数注入规则 ---
extract_params = models.TextField(null=True, blank=True, default='[]', verbose_name="参数提取规则 (JSON格式)")
# ...
2. 生成并应用数据库迁移
# 在 test-platform 目录下
python manage.py makemigrations api
python manage.py migrate api
3. 更新 TestCaseSerializer
打开 test-platform/api/serializers.py，将新的 extract_params 字段添加到 TestCaseSerializer 的 fields 列表中。
# test-platform/api/serializers.py
# ... (其他导入和 Serializer) ...
class TestCaseSerializer(serializers.ModelSerializer):
# ... (原有字段定义) ...
class Meta:
model = TestCase
fields = [
'id', 'name', 'description', 'module', 'module_name', 'project_id', 'project_name',
'priority', 'priority_display',
'request_method', 'request_url', 'request_headers', 'request_body', 'assertions', 'extract_params', # 新增字段
'precondition', 'steps_text', 'expected_result',
'case_type', 'case_type_display', 'maintainer',
'create_time', 'update_time'
]
# ... (extra_kwargs 保持不变) ...
第二部分：后端测试执行器强化
改造测试执行器，使其支持参数提取和注入。
1. 修改 execute_api_test_case 函数
打开 test-platform/api/services/test_executor.py进行修改。
# test-platform/api/services/test_executor.py
import requests
import json
import time
import urllib.parse
import re
from typing import Dict, List, Any, Tuple, Optional
from ..models import TestCase, Environment
# --- 断言类型 ---
ASSERTION_TYPE_STATUS_CODE = "status_code"
ASSERTION_TYPE_BODY_CONTAINS = "body_contains"
ASSERTION_TYPE_JSON_PATH_EQUALS = "json_path_equals"
ASSERTION_TYPE_HEADER_EQUALS = "header_equals"
try:
from jsonpath_ng import jsonpath, parse as jsonpath_parse
except ImportError:
jsonpath_parse = None # type: ignore
print("WARNING: jsonpath_ng not installed. JSONPath assertions will not work.")
def execute_api_test_case(
test_case: TestCase,
environment: Optional[Environment] = None,
context_variables: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
执行单个 API 测试用例并返回结果字典。
可以传入 environment 对象，用于动态替换 base_url 和添加公共请求头。
context_variables: 上下文变量字典，用于参数注入和提取
"""
# 初始化上下文变量字典（如果未提供）
if context_variables is None:
context_variables = {
}
result = {
"status": "ERROR", # 默认是错误状态
"request_data": {
},
"response_data": {
},
"assertion_results": [],
"error_message": None,
"duration": 0.0,
"context_variables": context_variables # 返回更新后的上下文变量
}
start_time = time.time()
try:
# 1. 解析请求参数并应用环境配置
method = test_case.request_method.upper()
original_url = test_case.request_url
final_url = original_url # 最终发送请求的 URL
# 如果提供了环境，应用其 base_url 和 config_data
environment_headers = {
}
environment_config_data = {
}
if environment:
if environment.base_url and original_url and not original_url.startswith(('http://', 'https://')):
# 如果 test_case.request_url 是相对路径，则拼接 base_url
final_url = f"{
environment.base_url.rstrip('/')}/{
original_url.lstrip('/')}"
elif not original_url: # 如果用例没有填写 URL，并且有环境，则使用环境的 base_url
final_url = environment.base_url
# 解析环境的 config_data
if environment.config_data:
try:
environment_config_data = environment.config_data
# 例如，从 config_data 中获取 headers
environment_headers = environment_config_data.get('headers', {
})
except Exception as e:
result["error_message"] = f"环境配置数据解析失败: {
e}"
return result
# 验证URL格式
parsed_url = urllib.parse.urlparse(final_url)
if not parsed_url.scheme:
final_url = f"https://{
final_url}" # 默认使用https
# 2. 参数注入 - 替换URL中的变量占位符
try:
final_url = replace_variables(final_url, context_variables)
except Exception as e:
result["error_message"] = f"URL参数注入失败: {
str(e)}"
result["duration"] = time.time() - start_time
return result
headers = {
}
if test_case.request_headers:
try:
headers = json.loads(test_case.request_headers)
# 参数注入 - 替换请求头中的变量占位符
headers = replace_variables_in_dict(headers, context_variables)
except json.JSONDecodeError:
result["error_message"] = "请求头 JSON 格式错误"
result["duration"] = time.time() - start_time
return result
except Exception as e:
result["error_message"] = f"请求头参数注入失败: {
str(e)}"
result["duration"] = time.time() - start_time
return result
# 合并环境的通用请求头和用例自定义的请求头
# 用例自定义的头可以覆盖环境的头
final_headers = {
**environment_headers, **headers}
body = test_case.request_body
# 参数注入 - 替换请求体中的变量占位符
if body:
try:
# 检查是否是JSON格式的请求体
if final_headers.get('Content-Type', '').lower().startswith('application/json'):
try:
body_dict = json.loads(body)
body_dict = replace_variables_in_dict(body_dict, context_variables)
body = json.dumps(body_dict)
except json.JSONDecodeError:
# 如果不是有效的JSON，则作为字符串处理
body = replace_variables(body, context_variables)
else:
# 非JSON格式，直接替换字符串
body = replace_variables(body, context_variables)
except Exception as e:
result["error_message"] = f"请求体参数注入失败: {
str(e)}"
result["duration"] = time.time() - start_time
return result
result["request_data"] = {
"method": method,
"url": final_url,
"headers": final_headers,
"body": body,
}
# 3. 发送 HTTP 请求
response = None
if method == 'GET':
response = requests.get(final_url, headers=final_headers, timeout=10)
elif method == 'POST':
if final_headers