提交 e7c07c60 authored 作者: 陈泽健's avatar 陈泽健

feat(api-test): 接口测试模块完整实现——用例管理/执行引擎/断言/报告集成

- 新增 ApiTestExecutor(HttpClient 复用 + 模板变量 + 4类型断言引擎)
- 新增 ApiTestService(用例查询/执行创建/异步执行/结果保存)
- 新增 /api/api-test 路由(7个端点:用例 CRUD + 执行管理 + 结果查询)
- 新增 ApiCaseList.vue(支持 HTTP 配置 + Headers/Body/断言规则编辑器)
- 新增 apiTest.ts 前端 API 封装
- 复用通用 Execution.vue(api 类型分支)和 Reports.vue(api 类型标签)
- 复用 test_cases 表(case_type='api'),API 配置存 steps 字段
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 3f368601
此差异已折叠。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:api_test.py
模块描述:接口测试 API 路由 — 用例管理、执行、结果查询
作者:czj
创建日期:2026-08-17
"""
import asyncio
import logging
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.api_test import (
ApiTestExecuteRequest,
ApiTestExecuteResponse,
ApiTestResultItem,
ApiTestExecutionSummary,
)
from app.services.api_test_service import ApiTestService
logger = logging.getLogger(__name__)
router = APIRouter()
# ==================== 用例管理 ====================
@router.get("/cases", summary="获取接口测试用例列表")
async def list_api_cases(
skip: int = Query(0, ge=0, description="跳过记录数"),
limit: int = Query(20, ge=1, le=100, description="返回记录数"),
module_id: Optional[str] = Query(None, description="模块ID筛选"),
keyword: Optional[str] = Query(None, description="关键字搜索"),
db: AsyncSession = Depends(get_db),
):
"""
获取接口测试用例列表
Returns:
dict: 用例列表和总数
"""
service = ApiTestService(db)
cases, total = await service.get_api_cases(
skip=skip, limit=limit, module_id=module_id, keyword=keyword
)
return {
"items": [c.to_dict() for c in cases],
"total": total,
}
@router.get("/cases/{case_id}", summary="获取接口测试用例详情")
async def get_api_case(
case_id: str,
db: AsyncSession = Depends(get_db),
):
"""
获取接口测试用例详情
Args:
case_id: 用例ID
Returns:
dict: 用例详情
"""
service = ApiTestService(db)
case = await service.get_api_case(case_id)
if not case:
raise HTTPException(status_code=404, detail=f"接口测试用例不存在: {case_id}")
return case.to_dict()
# ==================== 执行管理 ====================
@router.post("/executions", summary="创建接口测试执行")
async def create_execution(
request: ApiTestExecuteRequest,
db: AsyncSession = Depends(get_db),
):
"""
创建接口测试执行记录
Args:
request: 执行请求(含用例ID列表、目标URL、配置ID)
Returns:
ApiTestExecuteResponse: 执行记录
"""
service = ApiTestService(db)
execution = await service.create_execution(
case_ids=request.case_ids,
target_url=request.target_url,
config_id=request.config_id,
)
return ApiTestExecuteResponse(
execution_id=execution.id,
status=execution.status,
message=f"已创建执行记录,共 {execution.total_cases} 个用例",
)
@router.post("/executions/{execution_id}/run", summary="执行接口测试")
async def run_execution(
execution_id: str,
db: AsyncSession = Depends(get_db),
):
"""
触发接口测试执行(异步)
Args:
execution_id: 执行ID
Returns:
dict: 执行状态
"""
service = ApiTestService(db)
# 异步执行(不等待完成)
asyncio.create_task(service.run_execution(execution_id))
return {
"execution_id": execution_id,
"status": "running",
"message": "接口测试已开始执行",
}
@router.get("/executions", summary="获取接口测试执行记录列表")
async def list_executions(
skip: int = Query(0, ge=0, description="跳过记录数"),
limit: int = Query(20, ge=1, le=100, description="返回记录数"),
db: AsyncSession = Depends(get_db),
):
"""
获取接口测试执行记录列表
Returns:
dict: 执行记录列表和总数
"""
service = ApiTestService(db)
executions, total = await service.get_executions(skip=skip, limit=limit)
return {
"items": [e.to_dict() for e in executions],
"total": total,
}
@router.get("/executions/{execution_id}/results", summary="获取接口测试执行结果")
async def get_execution_results(
execution_id: str,
skip: int = Query(0, ge=0, description="跳过记录数"),
limit: int = Query(50, ge=1, le=200, description="返回记录数"),
db: AsyncSession = Depends(get_db),
):
"""
获取接口测试执行结果详情
Args:
execution_id: 执行ID
Returns:
ApiTestExecutionSummary: 执行结果摘要
"""
service = ApiTestService(db)
# 获取执行记录
from app.models.execution import Execution
from sqlalchemy import select
result = await db.execute(
select(Execution).where(Execution.id == execution_id)
)
execution = result.scalar_one_or_none()
if not execution:
raise HTTPException(status_code=404, detail=f"执行记录不存在: {execution_id}")
# 获取结果列表
case_results, total = await service.get_execution_results(
execution_id=execution_id, skip=skip, limit=limit
)
# 转换结果
items = []
for cr in case_results:
items.append(ApiTestResultItem(
case_id=cr.case_id or "",
case_name=cr.case_name,
status=cr.status,
duration=cr.duration,
error=cr.error_message or "",
))
return ApiTestExecutionSummary(
execution_id=execution.id,
name=execution.name,
status=execution.status,
total=execution.total_cases,
passed=execution.passed,
failed=execution.failed,
start_time=execution.start_time.isoformat() if execution.start_time else None,
end_time=execution.end_time.isoformat() if execution.end_time else None,
duration=execution.duration,
results=items,
)
@router.get("/executions/{execution_id}/result/{case_result_id}", summary="获取单个用例执行详情")
async def get_case_result_detail(
execution_id: str,
case_result_id: str,
db: AsyncSession = Depends(get_db),
):
"""
获取单个用例的执行详情(含请求/响应/断言结果)
Args:
execution_id: 执行ID
case_result_id: 用例结果ID
Returns:
dict: 用例执行详情
"""
from app.models.case_result import CaseResult
from sqlalchemy import select
result = await db.execute(
select(CaseResult).where(
CaseResult.id == case_result_id,
CaseResult.execution_id == execution_id,
)
)
case_result = result.scalar_one_or_none()
if not case_result:
raise HTTPException(status_code=404, detail="用例结果不存在")
return case_result.to_dict()
\ No newline at end of file
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:api_test.py
模块描述:接口测试 Pydantic 模式定义 — 断言规则、请求配置、执行结果
作者:czj
创建日期:2026-08-17
"""
from datetime import datetime
from typing import Optional, List, Dict, Any
from pydantic import BaseModel, Field
# ==================== 断言规则 ====================
class AssertionRule(BaseModel):
"""
断言规则
支持三种断言类型:
- status_code: HTTP 状态码比较
- body_field: 响应体 JSONPath 字段检查
- response_time: 响应时间比较(ms)
- header: 响应头检查
Attributes:
type (str): 断言类型
operator (str): 运算符 (equals/not_equals/contains/regex/greater_than/less_than)
value (Any): 期望值
path (Optional[str]): JSONPath(body_field 类型必填)
"""
type: str = Field(..., description="断言类型: status_code/body_field/response_time/header")
operator: str = Field("equals", description="运算符: equals/not_equals/contains/regex/greater_than/less_than")
value: Any = Field(..., description="期望值")
path: Optional[str] = Field(None, description="JSONPath(body_field 类型必填,如 $.code)")
# ==================== 请求配置 ====================
class ApiRequestConfig(BaseModel):
"""
接口测试用例的请求配置(存储在 test_cases.steps 字段中)
Attributes:
method (str): HTTP 方法
url (str): 接口路径(相对路径,如 /meetingV3/api/message/book)
headers (dict): 自定义请求头
body (Any): 请求体(支持 dict/list/str)
auth_required (bool): 是否需要登录 Token
sign_request (bool): 是否需要签名
account_key (str): 账号 key
assertions (list): 断言规则列表
description (str): 用例描述/预期说明
"""
method: str = Field("GET", description="HTTP 方法")
url: str = Field(..., description="接口路径(相对路径)")
headers: Dict[str, str] = Field(default_factory=dict, description="自定义请求头")
body: Any = Field(None, description="请求体")
auth_required: bool = Field(True, description="是否需要登录 Token")
sign_request: bool = Field(False, description="是否需要签名")
account_key: str = Field("superadmin", description="账号 key")
assertions: List[AssertionRule] = Field(default_factory=list, description="断言规则列表")
description: str = Field("", description="用例描述/预期说明")
# ==================== 执行相关 ====================
class ApiAssertionResult(BaseModel):
"""
单条断言执行结果
Attributes:
type (str): 断言类型
operator (str): 运算符
expected (Any): 期望值
actual (Any): 实际值
passed (bool): 是否通过
message (str): 描述信息
"""
type: str = Field("", description="断言类型")
operator: str = Field("", description="运算符")
expected: Any = Field(None, description="期望值")
actual: Any = Field(None, description="实际值")
passed: bool = Field(False, description="是否通过")
message: str = Field("", description="描述信息")
class ApiStepResult(BaseModel):
"""
单次 API 请求执行结果(存储在 case_results.steps_result 中)
Attributes:
order (int): 步骤序号
name (str): 步骤名称
method (str): 请求方法
url (str): 请求完整 URL
status_code (int): HTTP 状态码
response_time (float): 响应时间(ms)
request_headers (dict): 请求头
request_body (Any): 请求体
response_headers (dict): 响应头
response_body (str): 响应体文本
assertions (list): 断言执行结果
passed (bool): 是否通过
error (str): 错误信息
"""
order: int = Field(0, description="步骤序号")
name: str = Field("", description="步骤名称")
method: str = Field("", description="请求方法")
url: str = Field("", description="请求完整 URL")
status_code: int = Field(0, description="HTTP 状态码")
response_time: float = Field(0.0, description="响应时间(ms)")
request_headers: Dict[str, str] = Field(default_factory=dict, description="请求头")
request_body: Any = Field(None, description="请求体")
response_headers: Dict[str, str] = Field(default_factory=dict, description="响应头")
response_body: str = Field("", description="响应体文本")
assertions: List[ApiAssertionResult] = Field(default_factory=list, description="断言执行结果")
passed: bool = Field(False, description="是否通过")
error: str = Field("", description="错误信息")
# ==================== API 请求/响应 ====================
class ApiTestExecuteRequest(BaseModel):
"""
执行接口测试请求
Attributes:
case_ids (list): 用例 ID 列表
config_id (Optional[str]): 安全测试配置 ID(复用登录配置)
target_url (Optional[str]): 目标服务器 base URL
name (Optional[str]): 执行名称(可选,留空自动生成)
"""
case_ids: List[str] = Field(..., description="用例 ID 列表", min_length=1)
config_id: Optional[str] = Field(None, description="安全测试配置 ID,复用登录配置")
target_url: Optional[str] = Field(None, description="目标服务器 base URL,如 https://192.168.5.44")
name: Optional[str] = Field(None, description="执行名称,留空自动生成")
class ApiTestExecuteResponse(BaseModel):
"""
执行接口测试响应
Attributes:
execution_id (str): 执行记录 ID
status (str): 状态
message (str): 消息
"""
execution_id: str = Field("", description="执行记录 ID")
status: str = Field("", description="状态")
message: str = Field("", description="消息")
class ApiTestResultItem(BaseModel):
"""
接口测试结果项(单条用例)
Attributes:
case_id (str): 用例 ID
case_name (str): 用例名称
status (str): 状态 (passed/failed/skipped)
duration (float): 耗时(秒)
step_result (ApiStepResult): 执行详情
error (str): 错误信息
"""
case_id: str = Field("", description="用例 ID")
case_name: str = Field("", description="用例名称")
status: str = Field("", description="状态: passed/failed/skipped")
duration: float = Field(0.0, description="耗时(秒)")
step_result: Optional[ApiStepResult] = Field(None, description="执行详情")
error: str = Field("", description="错误信息")
class ApiTestExecutionSummary(BaseModel):
"""
接口测试执行摘要
Attributes:
execution_id (str): 执行记录 ID
name (str): 执行名称
status (str): 状态
total (int): 总用例数
passed (int): 通过数
failed (int): 失败数
start_time (str): 开始时间
end_time (str): 结束时间
duration (float): 总耗时
results (list): 详细结果列表
"""
execution_id: str = Field("", description="执行记录 ID")
name: str = Field("", description="执行名称")
status: str = Field("", description="状态")
total: int = Field(0, description="总用例数")
passed: int = Field(0, description="通过数")
failed: int = Field(0, description="失败数")
start_time: Optional[str] = Field(None, description="开始时间")
end_time: Optional[str] = Field(None, description="结束时间")
duration: float = Field(0.0, description="总耗时(秒)")
results: List[ApiTestResultItem] = Field(default_factory=list, description="详细结果列表")
\ No newline at end of file
此差异已折叠。
/**
* 接口测试 API 封装
*
* @author czj
* @date 2026-08-17
*/
import request from '@/utils/request'
export interface AssertionRule {
type: 'status_code' | 'body_field' | 'response_time' | 'header'
operator: 'equals' | 'not_equals' | 'contains' | 'regex' | 'greater_than' | 'less_than'
value: any
path?: string
}
export interface ApiRequestConfig {
method: string
url: string
headers: Record<string, string>
body?: any
auth_required: boolean
sign_request: boolean
account_key: string
assertions: AssertionRule[]
description: string
}
export interface ApiTestExecuteRequest {
case_ids: string[]
config_id?: string
target_url?: string
name?: string
}
export const apiTestApi = {
/** 获取接口测试用例列表 */
listCases(params?: {
skip?: number
limit?: number
module_id?: string
keyword?: string
}) {
return request.get('/api/api-test/cases', { params }) as any
},
/** 获取用例详情 */
getCase(caseId: string) {
return request.get(`/api/api-test/cases/${caseId}`) as any
},
/** 创建执行记录 */
createExecution(data: ApiTestExecuteRequest) {
return request.post('/api/api-test/executions', data) as any
},
/** 执行接口测试 */
runExecution(executionId: string) {
return request.post(`/api/api-test/executions/${executionId}/run`) as any
},
/** 获取执行记录列表 */
listExecutions(params?: { skip?: number; limit?: number }) {
return request.get('/api/api-test/executions', { params }) as any
},
/** 获取执行结果 */
getExecutionResults(executionId: string, params?: { skip?: number; limit?: number }) {
return request.get(`/api/api-test/executions/${executionId}/results`, { params }) as any
},
/** 获取单个用例执行详情 */
getCaseResultDetail(executionId: string, caseResultId: string) {
return request.get(`/api/api-test/executions/${executionId}/result/${caseResultId}`) as any
},
}
\ No newline at end of file
此差异已折叠。
<!--
页面名称:api-test/index.vue
页面描述:接口测试模块父页面(路由容器)
@author czj
@date 2026-08-17
-->
<template>
<router-view />
</template>
<script setup lang="ts">
/**
* 接口测试模块路由容器
*
* 子页面:用例管理 / 执行中心 / 报告中心
*/
</script>
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论