十四、【测试执行篇】让测试跑起来:API 接口测试执行器设计与实现 (后端执行逻辑)
@[TOC](【测试执行篇】让测试跑起来:API 接口测试执行器设计与实现 (后端执行逻辑))
前言
测试执行是测试平台的核心价值所在。一个好的测试执行器需要能够:
- 准确解析测试用例: 正确理解用例中定义的请求参数和断言条件。
- 可靠地发送请求: 模拟真实的客户端行为与被测 API 交互。
- 有效地执行断言: 根据预设规则验证 API 响应的正确性。
- 详细地记录结果: 保存每次执行的详细信息,包括请求、响应、断言结果、耗时等,以便后续分析和报告。
在本文中,我们将主要关注后端 API 接口测试执行器的设计与实现。我们将学习如何根据测试用例中定义的请求信息(URL、方法、头部、请求体等),使用 Python 的 requests
库实际发送 HTTP 请求,然后根据定义的断言规则来判断测试是否通过,并记录执行结果。
重要:更新 TestCase
模型以支持 API 测试细节
我们之前定义的 TestCase
模型中的 steps_text
字段对于描述手动测试步骤是足够的,但对于自动化 API 测试,我们需要更结构化的字段来存储请求细节和断言规则。
因此,在开始之前,我们需要对 TestCase
模型进行一次重要的升级。
准备工作
-
Django 项目已就绪: 后端代码结构完整。
-
requests
库: 这是 Python 中非常流行的 HTTP 请求库。如果你的虚拟环境中还没有安装,请安装它:# 在 Django 项目的虚拟环境中 pip install requests
-
之前定义的模型:
Project
,Module
,TestCase
,TestPlan
都已存在并迁移。
第一部分:升级 TestCase
模型和相关后端组件
a. 修改 api/models.py
中的 TestCase
模型:
# test-platform/api/models.py
from django.db import models
# ... Project, Module ...class TestCase(BaseModel):module = models.ForeignKey(Module, on_delete=models.CASCADE, verbose_name="所属模块", related_name="testcases")priority_choices = [('P0', 'P0-最高'), ('P1', 'P1-高'), ('P2', 'P2-中'), ('P3', 'P3-低')]priority = models.CharField(max_length=2, choices=priority_choices, default='P1', verbose_name="优先级")case_type_choices = [('functional', '功能测试'), ('api', '接口测试'), ('ui', 'UI测试')]case_type = models.CharField(max_length=20, choices=case_type_choices, default='api',verbose_name="用例类型") # 默认改为api# --- 新增 API 测试相关字段 ---request_method_choices = [('GET', 'GET'), ('POST', 'POST'), ('PUT', 'PUT'),('DELETE', 'DELETE'), ('PATCH', 'PATCH'), ('OPTIONS', 'OPTIONS'), ('HEAD', 'HEAD')]request_method = models.CharField(max_length=10, choices=request_method_choices, default='GET',verbose_name="请求方法")request_url = models.CharField(max_length=1024, default='http://example.com', verbose_name="请求URL")# TextField 用于存储 JSON 字符串,也可以使用 JSONField (如果数据库支持,如 PostgreSQL)request_headers = models.TextField(null=True, blank=True, verbose_name="请求头 (JSON格式)")request_body = models.TextField(null=True, blank=True, verbose_name="请求体 (JSON或其他格式)")assertions = models.TextField(null=True, blank=True, default='[]', verbose_name="断言规则 (JSON格式)")# --- 原有字段 ---precondition = models.TextField(null=True, blank=True, verbose_name="前置条件")# steps_text 可以保留用于备注,或者移除如果不再需要steps_text = models.TextField(null=True, blank=True, verbose_name="步骤/备注 (文本描述)")expected_result = models.TextField(null=True, blank=True, verbose_name="预期结果 (文本描述)") # 可用于备注maintainer = models.CharField(max_length=50, null=True, blank=True, verbose_name="维护人")class Meta:verbose_name = "测试用例"verbose_name_plural = "测试用例列表"unique_together = ('module', 'name')ordering = ['-create_time']def __str__(self):return f"{self.module.project.name} - {self.module.name} - {self.name}"
新增字段解释:
request_method
: HTTP 请求方法 (GET, POST 等)。request_url
: 请求的完整 URL 或路径 (如果配置了 Base URL)。request_headers
: JSON 格式的字符串,存储请求头,例如{"Content-Type": "application/json", "Authorization": "Bearer xyz"}
。request_body
: 请求体内容,对于 POST/PUT 等方法。可以是 JSON 字符串、表单数据字符串等。assertions
: JSON 格式的字符串,存储断言规则列表。每个规则是一个对象,例如:{"type": "status_code", "expected": 200}
{"type": "body_contains", "expected": "success message"}
{"type": "json_path_equals", "expression": "$.data.id", "expected": 100}
{"type": "header_equals", "header_name": "Content-Type", "expected": "application/json"}
b. 生成并应用数据库迁移:
python manage.py makemigrations api
python manage.py migrate
c. 更新 TestCaseSerializer
(api/serializers.py
):
确保新的字段被包含进来。
# test-platform/api/serializers.py
class TestCaseSerializer(serializers.ModelSerializer):module_name = serializers.CharField(source='module.name', read_only=True)project_name = serializers.CharField(source='module.project.name', read_only=True)project_id = serializers.IntegerField(source='module.project.id', read_only=True)priority_display = serializers.CharField(source='get_priority_display', read_only=True)case_type_display = serializers.CharField(source='get_case_type_display', read_only=True)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', # 新增字段'precondition', 'steps_text', 'expected_result', 'case_type', 'case_type_display', 'maintainer','create_time', 'update_time']extra_kwargs = {'create_time': {'read_only': True},'update_time': {'read_only': True},'module': {'help_text': "关联的模块ID"},}
d. 前端调整 (重要提示):
你需要更新前端的 TestCaseEditView.vue
表单,使其能够输入这些新的 API 测试相关字段 (request_method
, request_url
, request_headers
, request_body
, assertions
),并将 steps_text
用于更通用的备注。本篇文章将假设这些数据可以被正确地存储和获取,重点在于后端的执行逻辑。
第二部分:设计测试执行结果的数据模型
我们需要模型来存储每次测试计划执行的总体情况,以及计划中每个用例的单独执行结果。
a. 在 api/models.py
中添加 TestRun
和 TestCaseRun
模型:
# test-platform/api/models.py
import uuid # 用于生成唯一的执行ID# ... (TestPlan 模型之后) ...
class TestRun(BaseModel):"""一次测试计划的执行记录"""id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False, verbose_name="执行ID")test_plan = models.ForeignKey(TestPlan, on_delete=models.CASCADE, related_name="runs", verbose_name="关联测试计划")status_choices = [('PENDING', '待执行'), ('RUNNING', '执行中'), ('COMPLETED', '已完成'), ('ERROR', '执行出错')]status = models.CharField(max_length=10, choices=status_choices, default='PENDING', verbose_name="执行状态")total_cases = models.PositiveIntegerField(default=0, verbose_name="用例总数")passed_cases = models.PositiveIntegerField(default=0, verbose_name="通过数")failed_cases = models.PositiveIntegerField(default=0, verbose_name="失败数")error_cases = models.PositiveIntegerField(default=0, verbose_name="错误数") # 用例执行本身出错,非断言失败start_time = models.DateTimeField(null=True, blank=True, verbose_name="开始时间")end_time = models.DateTimeField(null=True, blank=True, verbose_name="结束时间")duration = models.FloatField(null=True, blank=True, verbose_name="持续时间 (秒)")# environment = models.ForeignKey(Environment, ...) # 如果有环境管理# name 和 description 可以从 BaseModel 继承,或者这里覆盖# name 字段可以设置为执行时的快照名称,例如 "计划X - 2023-10-27 10:00"# description 可以用于备注本次执行的目的等class Meta:verbose_name = "测试执行记录"verbose_name_plural = "测试执行记录列表"ordering = ['-create_time']def __str__(self):return f"Run {self.id} for {self.test_plan.name}"class TestCaseRun(models.Model):"""单个测试用例在某次 TestRun 中的执行结果"""id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False, verbose_name="单用例执行ID")test_run = models.ForeignKey(TestRun, on_delete=models.CASCADE, related_name="case_runs", verbose_name="所属测试执行")test_case = models.ForeignKey(TestCase, on_delete=models.SET_NULL, null=True, verbose_name="关联测试用例") # 用例可能被删除# 快照用例信息 (可选,但推荐,防止原用例修改导致报告不准)case_name_snapshot = models.CharField(max_length=255, verbose_name="用例名称快照")# ...可以快照更多用例字段...status_choices = [('PASS', '通过'), ('FAIL', '失败'), ('ERROR', '执行错误'), ('SKIP', '跳过') ]status = models.CharField(max_length=10, choices=status_choices, verbose_name="执行结果")# 存储请求和响应的详细信息 (可以是 JSON 字符串)request_data = models.TextField(null=True, blank=True, verbose_name="实际请求数据") # 含url,method,headers,bodyresponse_data = models.TextField(null=True, blank=True, verbose_name="实际响应数据") # 含status_code,headers,body# 存储断言结果 (可以是 JSON 字符串,记录每个断言的细节)# 例如: [{"type": "status_code", "expected": 200, "actual": 200, "passed": true}, ...]assertion_results = models.TextField(null=True, blank=True, verbose_name="断言结果详情")error_message = models.TextField(null=True, blank=True, verbose_name="错误信息") # 如果 status 是 ERRORduration = models.FloatField(null=True, blank=True, verbose_name="耗时 (秒)")run_time = models.DateTimeField(auto_now_add=True, verbose_name="执行时间")class Meta:verbose_name = "单用例执行结果"verbose_name_plural = "单用例执行结果列表"ordering = ['run_time']def __str__(self):return f"{self.case_name_snapshot} - {self.status}"
关键点:
TestRun.id
: 使用UUIDField
作为主键,更适合分布式或异步场景。TestRun
记录了整体执行情况和统计数据。TestCaseRun
记录了每个用例的详细执行情况,包括请求快照、响应快照、断言结果等。case_name_snapshot
: 在TestCaseRun
中存储执行时用例的名称,即使原用例后来被修改或删除,报告中的名称依然准确。
b. 生成并应用数据库迁移:
python manage.py makemigrations api
python manage.py migrate
c. 注册到 Django Admin (api/admin.py
):
from .models import TestRun, TestCaseRun # 导入
# ...
admin.site.register(TestRun)
admin.site.register(TestCaseRun)
第三部分:实现测试执行核心服务
我们将创建一个服务函数或类来处理单个 API 测试用例的执行。
a. 创建api/services/
目录和api/services/test_executor.py
文件
# test-platform/api/services/test_executor.py
import requests
import json
import time
from typing import Dict, List, Any, Tuple
from ..models import TestCase # 从父级 models 导入# --- 断言类型 ---
ASSERTION_TYPE_STATUS_CODE = "status_code"
ASSERTION_TYPE_BODY_CONTAINS = "body_contains"
ASSERTION_TYPE_JSON_PATH_EQUALS = "json_path_equals" # 需要 jsonpath_ng
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) -> Dict[str, Any]:"""执行单个 API 测试用例并返回结果字典。"""result = {"status": "ERROR", # 默认是错误状态"request_data": {},"response_data": {},"assertion_results": [],"error_message": None,"duration": 0.0}start_time = time.time()try:# 1. 解析请求参数method = test_case.request_method.upper()url = test_case.request_urlheaders = {}if test_case.request_headers:try:headers = json.loads(test_case.request_headers)except json.JSONDecodeError:result["error_message"] = "请求头 JSON 格式错误"result["duration"] = time.time() - start_timereturn resultbody = test_case.request_body # 请求体可能是 JSON 字符串,也可能是其他# 记录实际发出的请求 (用于报告)result["request_data"] = {"method": method,"url": url,"headers": headers,"body": body, # 注意:对于文件上传等,这里可能需要特殊处理或不记录完整内容}# 2. 发送 HTTP 请求response = Noneif method == 'GET':response = requests.get(url, headers=headers, timeout=10) # params 可以从 url 中解析或单独字段elif method == 'POST':# 假设 body 是 JSON 字符串,如果 Content-Type 是 application/jsonif headers.get('Content-Type', '').lower().startswith('application/json') and body:try:parsed_body = json.loads(body)response = requests.post(url, headers=headers, json=parsed_body, timeout=10)except json.JSONDecodeError:result["error_message"] = "请求体 JSON 格式错误 (当 Content-Type 为 JSON 时)"result["duration"] = time.time() - start_timereturn resultelse: # 其他 Content-Type 或无 bodyresponse = requests.post(url, headers=headers, data=body, timeout=10)elif method == 'PUT':if headers.get('Content-Type', '').lower().startswith('application/json') and body:try:parsed_body = json.loads(body)response = requests.put(url, headers=headers, json=parsed_body, timeout=10)except json.JSONDecodeError:result["error_message"] = "请求体 JSON 格式错误 (当 Content-Type 为 JSON 时)"result["duration"] = time.time() - start_timereturn resultelse:response = requests.put(url, headers=headers, data=body, timeout=10)elif method == 'DELETE':response = requests.delete(url, headers=headers, timeout=10)# ... (可以添加 PATCH 等其他方法) ...else:result["error_message"] = f"不支持的请求方法: {method}"result["duration"] = time.time() - start_timereturn resultresult["duration"] = time.time() - start_time # 请求完成后的总时长# 3. 记录响应response_body_text = ""try:# 尝试以 JSON 解析响应体,如果失败则作为文本response_json = response.json()response_body_text = json.dumps(response_json, indent=2, ensure_ascii=False)except json.JSONDecodeError:response_body_text = response.textresponse_json = None # 用于 JSONPath 断言result["response_data"] = {"status_code": response.status_code,"headers": dict(response.headers),"body": response_body_text,}# 4. 执行断言assertions_rules = []if test_case.assertions:try:assertions_rules = json.loads(test_case.assertions)except json.JSONDecodeError:result["error_message"] = (result["error_message"] or "") + "; 断言规则 JSON 格式错误"# 即使断言格式错误,也可能已经有一个 error_message,所以追加# 此时不直接 return,因为请求可能已成功,只是断言部分失败all_assertions_passed = Trueif not assertions_rules: # 如果没有断言规则# 如果没有断言规则,但HTTP请求成功 (2xx),则认为用例通过all_assertions_passed = 200 <= response.status_code < 300for rule in assertions_rules:assertion_result_item = {"type": rule.get("type"),"expression": rule.get("expression"), # for json_path, header_name"expected": rule.get("expected"),"actual": None,"passed": False,}if rule["type"] == ASSERTION_TYPE_STATUS_CODE:assertion_result_item["actual"] = response.status_codeassertion_result_item["passed"] = (response.status_code == rule["expected"])elif rule["type"] == ASSERTION_TYPE_BODY_CONTAINS:assertion_result_item["actual"] = response_body_text # 整个响应体作为实际值assertion_result_item["passed"] = (str(rule["expected"]) in response_body_text)elif rule["type"] == ASSERTION_TYPE_HEADER_EQUALS:header_name = rule.get("expression") # 用 expression 字段存 header_nameactual_header_value = response.headers.get(header_name)assertion_result_item["actual"] = actual_header_valueassertion_result_item["passed"] = (actual_header_value == rule["expected"])elif rule["type"] == ASSERTION_TYPE_JSON_PATH_EQUALS and jsonpath_parse:if response_json is None: # 如果响应体不是有效 JSONassertion_result_item["actual"] = "Response body is not valid JSON"assertion_result_item["passed"] = Falseelse:try:jsonpath_expr = jsonpath_parse(rule["expression"])matches = [match.value for match in jsonpath_expr.find(response_json)]if matches:assertion_result_item["actual"] = matches[0] # 取第一个匹配项assertion_result_item["passed"] = (matches[0] == rule["expected"])else:assertion_result_item["actual"] = "JSONPath did not match any element"assertion_result_item["passed"] = Falseexcept Exception as e:assertion_result_item["actual"] = f"JSONPath evaluation error: {str(e)}"assertion_result_item["passed"] = Falseelse:# 未知断言类型,标记为失败assertion_result_item["actual"] = "Unknown assertion type"assertion_result_item["passed"] = Falseresult["assertion_results"].append(assertion_result_item)if not assertion_result_item["passed"]:all_assertions_passed = Falseresult["status"] = "PASS" if all_assertions_passed else "FAIL"except requests.exceptions.RequestException as e:result["error_message"] = f"请求执行出错: {str(e)}"result["status"] = "ERROR"result["duration"] = time.time() - start_time # 更新出错时的总时长except Exception as e:result["error_message"] = f"执行过程中发生未知错误: {str(e)}"result["status"] = "ERROR"result["duration"] = time.time() - start_timereturn result
关键点与解释:
- 输入: 函数接收一个
TestCase
模型实例。 - 输出: 返回一个字典,包含执行状态、请求/响应数据、断言结果等,这个字典的结构将用于创建
TestCaseRun
对象。 - 解析请求参数: 从
test_case
对象中获取method
,url
,headers
(需要json.loads
),body
。 - 发送请求: 使用
requests
库的对应方法 (requests.get
,requests.post
等) 发送请求。- 对于
POST
/PUT
,如果Content-Type
是application/json
,则使用json=
参数传递解析后的 body;否则使用data=
参数。 - 设置了
timeout
。
- 对于
- 记录响应: 获取响应的状态码、头部、响应体。尝试将响应体按 JSON 解析,如果失败则作为纯文本。
- 执行断言:
- 从
test_case.assertions
(JSON 字符串) 解析断言规则列表。 - 遍历每条规则,根据
type
执行相应的断言逻辑:status_code
: 比较响应状态码。body_contains
: 检查响应体文本是否包含预期字符串。header_equals
: 比较指定响应头的值。json_path_equals
: (需要pip install jsonpath-ng
) 使用 JSONPath 表达式从 JSON 响应中提取值并进行比较。
- 记录每个断言的实际值和通过状态。
- 根据所有断言是否通过来设置最终的
status
(PASS
或FAIL
)。
- 从
- 错误处理: 使用
try...except
捕获requests.exceptions.RequestException
(网络请求错误) 和其他一般错误,并记录到error_message
,设置status
为ERROR
。 - 耗时: 记录了整个执行过程的耗时。
安装 jsonpath-ng
(如果需要 JSONPath 断言):
pip install jsonpath-ng
第四部分:创建测试执行的 API 端点
我们将为 TestPlanViewSet
添加一个自定义的 run
action,用于触发测试计划的执行。
a. 修改 api/views.py
中的 TestPlanViewSet
:
# test-platform/api/views.py
import json # 用于序列化 request/response/assertion data
from rest_framework import viewsets, filters
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework import status as http_status # 避免与模型 status 冲突
from django.utils import timezone
from .models import Project, Module, TestCase, TestPlan, TestRun, TestCaseRun # 导入 TestRun, TestCaseRun
from .services.test_executor import execute_api_test_case # 导入执行函数
from .serializers import ProjectSerializer, ModuleSerializer, TestCaseSerializer, TestPlanSerializer,TestRunSerializer # 稍后创建 TestRunSerializer# ... TestPlanViewSet ...
class TestPlanViewSet(viewsets.ModelViewSet):queryset = TestPlan.objects.all().order_by('-update_time')serializer_class = TestPlanSerializer # TestPlan 的序列化器# ... filter_backends, filterset_fields, etc. ...def get_queryset(self):return super().get_queryset().prefetch_related('test_cases', 'project')@action(detail=True, methods=['post'], url_path='run')def run_test_plan(self, request, pk=None):"""执行指定的测试计划POST /api/testplans/{pk}/run/"""try:test_plan = self.get_object() # 获取 TestPlan 实例 (pk 即 test_plan_id)except TestPlan.DoesNotExist:return Response({"detail": "测试计划未找到。"}, status=http_status.HTTP_404_NOT_FOUND)# 1. 创建 TestRun 记录run_start_time = timezone.now()test_run = TestRun.objects.create(test_plan=test_plan,name=f"{test_plan.name} - {run_start_time.strftime('%Y-%m-%d %H:%M')}", # 执行名称description=f"手动触发执行: {test_plan.name}",status='RUNNING',start_time=run_start_time,total_cases=test_plan.test_cases.count())passed_count = 0failed_count = 0error_count = 0# 2. 遍历计划中的用例并执行# 使用 select_related('module__project') 提前加载关联数据,减少后续查询test_cases_in_plan = test_plan.test_cases.select_related('module__project').all()for test_case_instance in test_cases_in_plan:execution_result = execute_api_test_case(test_case_instance)# 创建 TestCaseRun 记录TestCaseRun.objects.create(test_run=test_run,test_case=test_case_instance,case_name_snapshot=test_case_instance.name,status=execution_result["status"],request_data=json.dumps(execution_result["request_data"], ensure_ascii=False, indent=2),response_data=json.dumps(execution_result["response_data"], ensure_ascii=False, indent=2),assertion_results=json.dumps(execution_result["assertion_results"], ensure_ascii=False, indent=2),error_message=execution_result["error_message"],duration=execution_result["duration"])if execution_result["status"] == "PASS":passed_count += 1elif execution_result["status"] == "FAIL":failed_count += 1else: # ERRORerror_count += 1# 3. 更新 TestRun 状态和统计run_end_time = timezone.now()test_run.end_time = run_end_timetest_run.duration = (run_end_time - run_start_time).total_seconds()test_run.passed_cases = passed_counttest_run.failed_cases = failed_counttest_run.error_cases = error_counttest_run.status = 'COMPLETED' if error_count == 0 else 'ERROR' # 如果有执行错误,整体标记为ERRORtest_run.save()# 4. 返回 TestRun 的结果 (可以使用 TestRunSerializer)# serializer = TestRunSerializer(test_run) # 创建 TestRunSerializer 用于返回# return Response(serializer.data, status=http_status.HTTP_200_OK)return Response({"message": "测试计划执行完成","test_run_id": str(test_run.id), # 返回 UUID 字符串"status": test_run.status,"passed": passed_count,"failed": failed_count,"errors": error_count,"total": test_run.total_cases}, status=http_status.HTTP_200_OK)
关键点:
@action(detail=True, methods=['post'], url_path='run')
:- 在
TestPlanViewSet
上定义了一个名为run_test_plan
的自定义 action。 detail=True
表示这个 action 是针对单个TestPlan
实例的 (需要pk
)。methods=['post']
表示它响应 POST 请求。url_path='run'
定义了 URL 的最后一部分,所以完整的 URL 会是/api/testplans/{pk}/run/
。
- 在
- 流程:
- 获取要执行的
TestPlan
对象。 - 创建一个新的
TestRun
记录,状态为RUNNING
,记录开始时间。 - 遍历
TestPlan
中的所有test_cases
。 - 对每个
test_case_instance
调用execute_api_test_case()
函数。 - 将执行结果保存为一个新的
TestCaseRun
记录,并关联到当前的TestRun
。 - 统计通过、失败、错误的用例数量。
- 所有用例执行完毕后,更新
TestRun
的结束时间、耗时、统计数据和最终状态 (COMPLETED
或ERROR
)。 - 返回一个包含
test_run_id
和执行摘要的响应。
- 获取要执行的
json.dumps(...)
用于将请求/响应/断言的字典数据序列化为 JSON 字符串存储到TextField
。
b. 创建 TestRunSerializer
(可选,用于更规范地返回 TestRun
数据):
在 api/serializers.py
中:
# test-platform/api/serializers.py
from .models import TestRun # 导入class TestRunSerializer(serializers.ModelSerializer):test_plan_name = serializers.CharField(source='test_plan.name', read_only=True)# 可以添加 TestCaseRun 的嵌套序列化器,如果需要在获取 TestRun 详情时一并返回# case_runs = TestCaseRunSerializer(many=True, read_only=True) class Meta:model = TestRunfields = '__all__' # 或者明确指定字段read_only_fields = ('create_time', 'update_time', 'start_time', 'end_time', 'duration', 'total_cases', 'passed_cases', 'failed_cases', 'error_cases')
第五步:测试后端执行逻辑
现在,我们可以使用 Postman 或类似的 API 测试工具来测试我们的执行器了。
-
准备数据:
- 确保你至少有一个项目、一个模块。
- 在该模块下创建一个或多个接口类型的测试用例,并填写真实的、可访问的
request_url
、request_method
、request_headers
(如果需要,例如{"Content-Type": "application/json"}
),request_body
(如果方法是 POST/PUT 等),以及一些简单的assertions
。- 你可以找一些公开的免费 API 来测试,例如
https://jsonplaceholder.typicode.com/todos/1
(GET 请求)。
- 你可以找一些公开的免费 API 来测试,例如
- 创建一个测试计划,将这些测试用例关联进去。记下这个测试计划的 ID。
-
使用 Postman 发送请求:
- URL:
http://127.0.0.1:8000/api/testplans/7/run/
(假设测试计划 ID 为 7) - Method:
POST
- Body: 不需要请求体。
- URL:
-
观察响应:
-
检查数据库:
- 去 Django Admin (
http://127.0.0.1:8000/admin/
)。 - 查看 Test runs,应该有一条新的记录,对应于你返回的
test_run_id
。检查其状态、统计数据等。
- 查看 Test case runs,应该有对应于计划中每个用例的执行结果记录。点开查看
request_data
,response_data
,assertion_results
等字段,确认它们是否被正确记录。
通过这些步骤,你可以验证后端测试执行器的基本功能是否按预期工作。
总结
在这篇文章中,我们为测试平台的核心——API 接口测试执行器——构建了坚实的后端逻辑:
- ✅ 升级了
TestCase
模型,添加了request_method
,request_url
,request_headers
,request_body
,assertions
等结构化字段,以支持 API 自动化测试。并相应更新了TestCaseSerializer
。 - ✅ 设计并创建了
TestRun
和TestCaseRun
模型,用于存储测试计划的整体执行记录和单个用例的详细执行结果 (包括请求/响应快照、断言详情、耗时等)。 - ✅ 实现了核心的
execute_api_test_case
服务函数 (api/services/test_executor.py
):- 使用
requests
库发送 HTTP 请求。 - 能够解析用例中的请求参数。
- 能够根据预定义的断言规则 (状态码、响应体包含、JSONPath 等) 验证响应。
- 处理请求和断言过程中的错误。
- 返回结构化的执行结果。
- 使用
- ✅ 在
TestPlanViewSet
中添加了一个自定义的run
action (POST /api/testplans/{pk}/run/
) 作为触发测试计划执行的 API 端点。该 action 会:- 创建
TestRun
记录。 - 遍历计划中的用例,调用执行服务。
- 创建
TestCaseRun
记录。 - 更新
TestRun
的最终状态和统计数据。
- 创建
- ✅ 指导了如何使用 Postman 测试后端执行逻辑,并检查数据库中的结果。
现在,我们的测试平台后端已经具备了实际执行 API 测试并记录结果的能力!这为后续的前端触发界面和测试报告展示奠定了基础。
在下一篇文章中,我们将探讨如何使用 Celery 实现测试任务的后台异步执行,以避免长时间运行的测试计划阻塞 API 请求,并提升用户体验。