三十一、【高级特性篇】接口用例参数化与关联：实现上下文数据传递

- 前言
- - 准备工作
  - 第一部分：后端数据模型调整
  - - 1. 升级 `TestCase` 模型
    - 2. 生成并应用数据库迁移
    - 3. 更新 `TestCaseSerializer`
  - 第二部分：后端测试执行器强化
  - - 1. 修改 `execute_api_test_case` 函数
    - 2. 修改 Celery 任务 (`api/tasks.py`)
    - 1. 更新 `api/testcase.ts` 中的类型定义
    - 2. 大改 `TestCaseEditView.vue`
  - 第四部分：全面测试与验证
- 总结

前言

在接口测试中，用例之间往往存在依赖关系：一个接口的响应数据（例如，用户 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 = TestCasefields = ['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: ignoreprint("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_urlfinal_url = original_url # 最终发送请求的 URL# 如果提供了环境，应用其 base_url 和 config_dataenvironment_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_urlfinal_url = f"{environment.base_url.rstrip('/')}/{original_url.lstrip('/')}"elif not original_url: # 如果用例没有填写 URL，并且有环境，则使用环境的 base_urlfinal_url = environment.base_url# 解析环境的 config_dataif environment.config_data:try:environment_config_data = environment.config_data# 例如，从 config_data 中获取 headersenvironment_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_timereturn resultheaders = {}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_timereturn resultexcept Exception as e:result["error_message"] = f"请求头参数注入失败: {str(e)}"result["duration"] = time.time() - start_timereturn 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_timereturn resultresult["request_data"] = {"method": method,"url": final_url,"headers": final_headers,"body": body,}# 3. 发送 HTTP 请求response = Noneif method == 'GET':response = requests.get(final_url, headers=final_headers, timeout=10)elif method == 'POST':if final_headers.get