十四、【测试执行篇】让测试跑起来：API 接口测试执行器设计与实现 (后端执行逻辑)

@[TOC](【测试执行篇】让测试跑起来：API 接口测试执行器设计与实现 (后端执行逻辑))

前言

测试执行是测试平台的核心价值所在。一个好的测试执行器需要能够：

1. 准确解析测试用例： 正确理解用例中定义的请求参数和断言条件。
2. 可靠地发送请求： 模拟真实的客户端行为与被测 API 交互。
3. 有效地执行断言： 根据预设规则验证 API 响应的正确性。
4. 详细地记录结果： 保存每次执行的详细信息，包括请求、响应、断言结果、耗时等，以便后续分析和报告。

在本文中，我们将主要关注后端 API 接口测试执行器的设计与实现。我们将学习如何根据测试用例中定义的请求信息（URL、方法、头部、请求体等），使用 Python 的 requests 库实际发送 HTTP 请求，然后根据定义的断言规则来判断测试是否通过，并记录执行结果。

重要：更新 TestCase 模型以支持 API 测试细节

我们之前定义的 TestCase 模型中的 steps_text 字段对于描述手动测试步骤是足够的，但对于自动化 API 测试，我们需要更结构化的字段来存储请求细节和断言规则。

因此，在开始之前，我们需要对 TestCase 模型进行一次重要的升级。

准备工作

1. Django 项目已就绪： 后端代码结构完整。

2. requests 库： 这是 Python 中非常流行的 HTTP 请求库。如果你的虚拟环境中还没有安装，请安装它：

   # 在 Django 项目的虚拟环境中
   pip install requests
   

   

3. 之前定义的模型： 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 = TestCase
        fields = [
            '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,body
    response_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 是 ERROR
    duration = 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: ignore
    print("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_url
        
        headers = {}
        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_time
                return result
        
        body = test_case.request_body # 请求体可能是 JSON 字符串，也可能是其他
        
        # 记录实际发出的请求 (用于报告)
        result["request_data"] = {
            "method": method,
            "url": url,
            "headers": headers,
            "body": body, # 注意：对于文件上传等，这里可能需要特殊处理或不记录完整内容
        }

        # 2. 发送 HTTP 请求
        response = None
        if method == 'GET':
            response = requests.get(url, headers=headers, timeout=10) # params 可以从 url 中解析或单独字段
        elif method == 'POST':
            # 假设 body 是 JSON 字符串，如果 Content-Type 是 application/json
            if 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_time
                    return result
            else: # 其他 Content-Type 或无 body
                response = 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_time
                    return result
            else:
                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_time
            return result

        result["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.text
            response_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 = True
        if not assertions_rules: # 如果没有断言规则
            # 如果没有断言规则，但HTTP请求成功 (2xx)，则认为用例通过
             all_assertions_passed = 200 <= response.status_code < 300
        
        for 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_code
                assertion_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_name
                actual_header_value = response.headers.get(header_name)
                assertion_result_item["actual"] = actual_header_value
                assertion_result_item["passed"] = (actual_header_value == rule["expected"])

            elif rule["type"] == ASSERTION_TYPE_JSON_PATH_EQUALS and jsonpath_parse:
                if response_json is None: # 如果响应体不是有效 JSON
                    assertion_result_item["actual"] = "Response body is not valid JSON"
                    assertion_result_item["passed"] = False
                else:
                    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"] = False
                    except Exception as e:
                        assertion_result_item["actual"] = f"JSONPath evaluation error: {str(e)}"
                        assertion_result_item["passed"] = False
            
            else:
                # 未知断言类型，标记为失败
                assertion_result_item["actual"] = "Unknown assertion type"
                assertion_result_item["passed"] = False

            result["assertion_results"].append(assertion_result_item)
            if not assertion_result_item["passed"]:
                all_assertions_passed = False
        
        result["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_time

    return 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 = 0
        failed_count = 0
        error_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 += 1
            elif execution_result["status"] == "FAIL":
                failed_count += 1
            else: # ERROR
                error_count += 1
        
        # 3. 更新 TestRun 状态和统计
        run_end_time = timezone.now()
        test_run.end_time = run_end_time
        test_run.duration = (run_end_time - run_start_time).total_seconds()
        test_run.passed_cases = passed_count
        test_run.failed_cases = failed_count
        test_run.error_cases = error_count
        test_run.status = 'COMPLETED' if error_count == 0 else 'ERROR' # 如果有执行错误，整体标记为ERROR
        test_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/。
• 流程：
  1. 获取要执行的 TestPlan 对象。
  2. 创建一个新的 TestRun 记录，状态为 RUNNING，记录开始时间。
  3. 遍历 TestPlan 中的所有 test_cases。
  4. 对每个 test_case_instance 调用 execute_api_test_case() 函数。
  5. 将执行结果保存为一个新的 TestCaseRun 记录，并关联到当前的 TestRun。
  6. 统计通过、失败、错误的用例数量。
  7. 所有用例执行完毕后，更新 TestRun 的结束时间、耗时、统计数据和最终状态 (COMPLETED 或 ERROR)。
  8. 返回一个包含 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 = TestRun
        fields = '__all__' # 或者明确指定字段
        read_only_fields = ('create_time', 'update_time', 'start_time', 'end_time', 'duration', 
                            'total_cases', 'passed_cases', 'failed_cases', 'error_cases')

第五步：测试后端执行逻辑

现在，我们可以使用 Postman 或类似的 API 测试工具来测试我们的执行器了。

1. 准备数据：

   • 确保你至少有一个项目、一个模块。
   • 在该模块下创建一个或多个接口类型的测试用例，并填写真实的、可访问的 request_url、request_method、request_headers (如果需要，例如 {"Content-Type": "application/json"}), request_body (如果方法是 POST/PUT 等)，以及一些简单的 assertions。
     • 你可以找一些公开的免费 API 来测试，例如 https://jsonplaceholder.typicode.com/todos/1 (GET 请求)。
   • 创建一个测试计划，将这些测试用例关联进去。记下这个测试计划的 ID。
     
     

2. 使用 Postman 发送请求：

   • URL: http://127.0.0.1:8000/api/testplans/7/run/ (假设测试计划 ID 为 7)
   • Method: POST
   • Body: 不需要请求体。

3. 观察响应：
   

4. 检查数据库：

• 去 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 请求，并提升用户体验。