在上一篇文章《Pytest 测试用例自动生成：接口自动化进阶实践》中，我们已经解决了“如何高效编写和维护接口自动化用例”的问题。
然而，随着业务的发展和团队规模的扩大，很多公司会选择开发自己的测试平台，以实现更高效、更统一的管理。
企业接口自动化通常会经历如下过程：
- 初期：测试人员本地用 Pytest 写脚本（即脚本项目）
- 中期：接入 Jenkins，定时跑一跑
- 后期：公司开始建设统一测试平台
一. 为什么需要将脚本项目接入测试平台
1. 当前面临的挑战
- 用例复用难：已有的大量 Pytest 测试用例无法直接在平台上运行，重新编写成本高。
- 管理复杂：多个脚本项目分散管理，执行效率低、维护困难。
- 报告分散：测试报告散落在各个项目中，难以统一查看和分析。
因此，需要考虑在【不重写、不侵入、不破坏】现有 Pytest 脚本项目的前提下，让它具备“测试平台可接入能力”。
2. 两种可选方案
我们考虑以下两种方案将脚本项目接入测试平台：
- 方案一：将脚本项目直接集成到测试平台目录中。
- 方案二：使用 FastAPI 改造脚本项目，提供接口供平台调用。
为什么选择方案二？
- 脚本项目经常需要修改和更新，若集成到平台中，每次修改都需要重新发布平台代码，繁琐且易出错。
- FastAPI 轻量、高性能，适合快速构建RESTful接口，实现脚本与平台的解耦。
二. 使用 FastAPI 框架改造脚本项目
1. 核心思路
我们的目标是：不改动原有Pytest脚本逻辑，仅通过"封装+接口"的方式，让脚本项目具备平台接入能力。
实现原则：
- 不重写：保持原有
testcases/、utils/目录结构不变 - 不侵入：原有脚本文件无需修改任何代码
- 不解耦：仅新增API层，不改变原有执行逻辑
2. 目录改造
原有脚本项目核心目录保留不变，改造后新增「API 层、配置层」，确保原有脚本无需修改即可复用。
-
改造前目录（即脚本项目原始目录）示例如下：
SUPER-API-AUTO-TEST/ # 接口自动化测试项目根目录 ├── auth.py # 鉴权相关 ├── case_generator.py # 用例生成逻辑 ├── config.yaml # 项目配置 ├── conftest.py # Pytest夹具 ├── runner.py # 用例执行入口 ├── reports/ # 测试报告目录 ├── logs/ # 日志目录 ├── testcases/ # 测试用例脚本（Python） ├── testcases_data/ # 测试用例数据（YAML） └── utils/ # 通用工具类 -
Fastapi 改造后目录结构示例如下：
FASTAPI-SUPER-API-AUTO-TEST/ # 基于FastAPI的改造后目录 ├── auth.py ├── case_generator.py ├── conftest.py ├── main.py # FastAPI应用入口（核心新增） ├── pytest.ini ├── runner.py ├── api/ # API层（核心新增） │ ├── testcase_route.py # 接口路由定义（URL、请求参数） │ ├── testcase_service.py # 接口业务逻辑（与原有脚本交互） │ └── __init__.py ├── configs/ # 配置目录（拆分原有config.yaml） │ └── config.yaml ├── logs/ ├── reports/ ├── testcases/ # 完全复用原有目录 ├── testcases_data/ # 完全复用原有目录 └── utils/ # 完全复用原有目录
改造后，新增了 Fastapi 路由层 api/、配置层 configs/，以及 FastAPI 应用入口 main.py。
三. 核心接口设计与实现
1. 接口设计示例
我们设计了以下关键接口供测试平台调用：
① 获取项目与模块信息
GET /api_test/testcases/projects：获取所有项目列表GET /api_test/testcases/modules：获取指定项目下的模块列表
② 获取用例列表
GET /api_test/testcases/list：支持按项目/模块筛选测试用例
③ 生成测试用例
POST /api_test/testcases/generate：根据YAML用例文件生成Python测试脚本
④ 执行测试任务
POST /api_test/testcases/run：后台执行指定测试用例，支持环境选择、报告类型、回调通知等
⑤ 获取测试报告
GET /api_test/reports/get_by_task：根据任务ID获取报告访问地址
2. 代码示例
testcase_service.py
# @author: xiaoqqimport os, re
from datetime import datetime
from typing import List, Optional, Dict
from pathlib import PathTESTCASE_ROOT = "testcases"def get_abs_root_path(root_path: str) -> Path:"""使用当前文件相对路径构造 testcases/ 的绝对路径:param root_path: 目录名:return:"""base_dir = Path(__file__).resolve().parent # 当前文件所在目录abs_root_path = (base_dir.parent / root_path).resolve()return abs_root_pathdef get_all_testcases(project: Optional[str] = None,module: Optional[str] = None,root_path: str = TESTCASE_ROOT) -> List[Dict]:"""获取所有测试用例（支持通过 project/module 筛选）返回字段包括 filename（无后缀）、path（绝对路径字符串）、Allure 元信息等"""abs_root_path = get_abs_root_path(root_path)if not abs_root_path.exists():return []# 路径校验if module and not project:raise ValueError("传入 module 前必须先传入 project")# 构造起始目录路径search_path = abs_root_pathif project:search_path = search_path / projectif module:search_path = search_path / moduleif not search_path.exists():return []testcases = []for dirpath, _, filenames in os.walk(search_path):for file in filenames:if file.startswith("test_") and file.endswith(".py"):full_path = os.path.join(dirpath, file)rel_path = os.path.relpath(full_path, abs_root_path) # 相对路径，如 merchant/device/test_xxx.pypath_parts = Path(rel_path).parts # 使用 pathlib 安全拆解路径if len(path_parts ) < 2:continue # 至少要有 project/filename 结构_project = path_parts [0]_filename = path_parts [-1]_module = path_parts [1] if len(path_parts ) > 2 else None # module 可选# 按传参过滤if project and _project != project:continueif module and _module != module:continuefilename = os.path.splitext(_filename)[0] # 去掉 .py 后缀last_modified = datetime.fromtimestamp(os.path.getmtime(full_path)).isoformat()# 提取用例元信息try:case_name, epic, feature, story = extract_case_info(full_path)except Exception as e:case_name, epic, feature, story = None, None, None, None# 拼接最终 path 字段为 TESTCASE_ROOT/... 形式full_case_path = str(Path(root_path) / rel_path).replace("\\", "/")# 构造 external_id：project|module|filename|pathexternal_id = f"{_project}|{_module or 'nomodule'}|{filename}|{full_case_path}"testcases.append({"project": _project,"module": _module, # None 表示无 module 层级"file": _filename,"filename": filename,"path": full_case_path,"last_modified": last_modified,"case_name": case_name or filename,"allure_epic": epic,"allure_feature": feature,"allure_story": story,"external_id": external_id # 加入唯一标识})return testcasesdef extract_case_info(file_path):"""解析测试用例文件，获取相应信息:param file_path::return:"""with open(file_path, 'r', encoding='utf-8') as file:content = file.read()case_name_match = re.search(r'def setup_class.*?\(.*?\):.*?log\.info\(\'========== 开始执行测试用例：(.+?) ==========\'',content, re.DOTALL)case_name = case_name_match.group(1).strip() if case_name_match else \os.path.splitext(os.path.basename(file_path))[0]allure_epic_match = re.search(r'@allure\.epic\(\'(.+?)\'\)', content)allure_feature_match = re.search(r'@allure\.feature\(\'(.+?)\'\)', content)allure_story_match = re.search(r'@allure\.story\(\'(.+?)\'\)', content)allure_epic = allure_epic_match.group(1).strip() if allure_epic_match else Noneallure_feature = allure_feature_match.group(1).strip() if allure_feature_match else Noneallure_story = allure_story_match.group(1).strip() if allure_story_match else Nonereturn case_name, allure_epic, allure_feature, allure_storydef get_all_projects(root_path: str = TESTCASE_ROOT) -> List[Dict[str, str]]:"""获取 testcases/ 下所有项目名、相对路径及创建时间（倒序排序）"""abs_root_path = get_abs_root_path(root_path)if not abs_root_path.exists():return []projects = []for d in abs_root_path.iterdir():if d.is_dir():created_time = datetime.fromtimestamp(d.stat().st_ctime)projects.append({"name": d.name,"path": str(Path(root_path) / d.name).replace("\\", "/"),"created_time": created_time.isoformat()})# 按创建时间倒序return sorted(projects, key=lambda x: x["created_time"], reverse=True)def get_all_projects_and_modules(project: Optional[str] = None,root_path: str = TESTCASE_ROOT
) -> List[Dict]:"""获取所有项目和模块结构（支持指定项目）。包含路径、创建时间，按项目时间倒序。"""abs_root_path = get_abs_root_path(root_path)if not abs_root_path.exists():return []result = []for proj_dir in abs_root_path.iterdir():if not proj_dir.is_dir():continueproj_name = proj_dir.nameif project and proj_name != project:continueproj_created_time = datetime.fromtimestamp(proj_dir.stat().st_ctime)modules = []# 遍历模块目录时需要忽略的子目录EXCLUDE_DIRS = {"__pycache__", ".pytest_cache", ".git", ".idea"}for mod_dir in proj_dir.iterdir():if mod_dir.is_dir() and mod_dir.name not in EXCLUDE_DIRS:mod_created_time = datetime.fromtimestamp(mod_dir.stat().st_ctime)modules.append({"name": mod_dir.name,"path": str(Path(root_path) / proj_name / mod_dir.name).replace("\\", "/"),"created_time": mod_created_time.isoformat()})# 模块也可以排序（如有需求）modules.sort(key=lambda x: x["created_time"], reverse=True)result.append({"project": proj_name,"path": str(Path(root_path) / proj_name).replace("\\", "/"),"created_time": proj_created_time.isoformat(),"modules": modules})if project:break# 项目排序return sorted(result, key=lambda x: x["created_time"], reverse=True)def generate_testcase(case_yaml_list: list = None):"""生成测试用例:return:"""from case_generator import CaseGeneratorCG = CaseGenerator()CG.generate_testcases(project_yaml_list=case_yaml_list)if __name__ == '__main__':# print(get_all_testcases())# print(get_all_projects())print(get_all_projects_and_modules(project="merchant"))
testcase_route.py 示例如下：
# @author: xiaoqqfrom pathlib import Path
from fastapi import APIRouter, BackgroundTasks, Query, Body
from pydantic import BaseModel
from typing import List, Optional
from runner import run_tests
from api.testcase_service import (get_all_testcases,get_all_projects,get_all_projects_and_modules,generate_testcase,
)router = APIRouter()class TestExecutionRequest(BaseModel):testcases: Optional[List[str]] = ['testcases/'] # 默认运行所有目录env: Optional[str] = 'pre'report_type: Optional[str] = 'pytest-html'dingtalk_notify: Optional[bool] = Truetask_id: Optional[str]callback_url: Optional[str]auth_token: Optional[str] = None # 新增字段：从平台传入的 token# 执行测试用例
@router.post("/testcases/run")
def run_testcases(request: TestExecutionRequest, background_tasks: BackgroundTasks):try:background_tasks.add_task(run_tests,testcases=request.testcases,env=request.env,report_type=request.report_type,dingtalk_notify=request.dingtalk_notify,task_id=request.task_id,callback_url=request.callback_url,auth_token=request.auth_token, # 测试平台回调 auth_token)return {"code": 0,"msg": "测试任务已提交后台执行","task_id": request.task_id}except Exception as e:return {"code":1, "msg": f"测试任务失败：{str(e)}"}# 获取测试用例
@router.get("/testcases/list")
def list_testcases(project: str = Query(None), module: str = Query(None)):try:testcases = get_all_testcases(project, module)return {"code": 0,"msg": "success","testcases": testcases}except Exception as e:return {"code": 1, "msg": f"获取测试用例失败：{str(e)}"}# 获取 testcases/ 中的所有测试项目
@router.get("/testcases/projects")
def list_projects():try:projects = get_all_projects()return {"code": 0, "msg": "success", "projects": projects}except Exception as e:return {"code": 1, "msg": f"获取测试项目失败：{str(e)}"}# 获取 testcases/ 中的所有测试项目及模块
@router.get("/testcases/modules")
def list_modules(project: str = Query(None)):try:modules = get_all_projects_and_modules(project)return {"code": 0, "msg": "success", "modules": modules}except Exception as e:return {"code": 1, "msg": f"获取测试项目-模块失败：{str(e)}"}class GenerateCaseRequest(BaseModel):case_yaml_list: Optional[List[str]] = None# 根据 testcases_data/ 中的测试数据生成测试用例文件
@router.post("/testcases/generate")
def generate_testcase_route(req: GenerateCaseRequest):try:generate_testcase(req.case_yaml_list)return {"code": 0, "msg": "success"}except Exception as e:return {"code": 1, "msg": f"获取测试项目-模块失败：{str(e)}"}@router.get("/reports/get_by_task")
def get_report_by_task(task_id: str,report_type: str,created_at: str # 格式: "20250814"
):"""根据 task_id + 创建时间 + report_type 获取报告 URL"""if not created_at:return {"code": 1, "msg": "created_at 必填", "url": None}base_path = Path(__file__).resolve().parent.parent / "reports" / created_atif report_type == "pytest-html":report_file = base_path / f"report_{task_id}.html"elif report_type == "allure":report_file = base_path / f"report_{task_id}_allure/html/index.html"else:return {"code": 1, "msg": "未知 report_type", "url": None}if not report_file.exists():return {"code": 1, "msg": "报告文件不存在", "url": None}relative_url = str(report_file.relative_to(Path(__file__).resolve().parent.parent)).replace("\\", "/")return {"code": 0, "msg": "success", "url": f"/{relative_url}"}
mian.py
from fastapi import FastAPI
from api import testcase_route
from pathlib import Path
from fastapi.staticfiles import StaticFilesapp = FastAPI(title="接口自动化测试服务")# 挂载测试用例路由
app.include_router(testcase_route.router, prefix="/api_test", tags=["测试任务"])# 挂载 reports 目录为静态文件目录
reports_dir = Path(__file__).parent / "reports"
reports_dir.mkdir(exist_ok=True) # 确保目录存在
app.mount("/reports", StaticFiles(directory=reports_dir), name="reports")if __name__ == "__main__":from utils.log_manager import LogManagerLogManager.setup_logging() # 启动时显式初始化日志import uvicornuvicorn.run("main:app",host="0.0.0.0",port=8000,# reload=True,reload_excludes=["testcases/*", "logs/*", "reports/*"] # 排除这些目录的文件变更)
四. 测试平台调用
执行mian.py，启动 Fastapi 项目后，便可在测试平台通过调用相关接口来管理该脚本测试项目（平台调用代码不具体提供）。
1. 调用示意图
测试平台││ HTTP 调用▼
FastAPI 测试服务││ pytest 执行▼
测试报告生成││ 回调结果▼
测试平台展示
这样，职责边界非常清晰：
- 测试平台：调度、记录、展示，
- 改造后的测试服务：执行、产出报告
2. 测试平台界面
平台测试用例列表：
测试报告列表：
五. 总结
方案优势总结如下：
-
解耦与复用：脚本项目独立维护，平台通过接口调用，互不影响
-
灵活执行：支持按项目、模块、用例筛选执行，适应不同测试场景。
-
异步处理：长时间任务后台执行，平台可实时获取状态与报告。
-
报告统一管理：所有报告集中存储，支持在线统一查看。
当然，示例代码还可以进行优化扩展，如加入用户认证机制来保障接口安全等。
当接口自动化发展到一定规模，单机脚本 或 Jenkins Job 都会成为瓶颈，而“脚本服务化 + 平台调度”，几乎是所有成熟团队最终都会走到的一步。
如果你：
- 正在做接口自动化
- 或正在参与测试平台建设
- 或正在被“脚本怎么接平台”折磨
那么，希望这篇文章能少让你走一点弯路。
