一. 为什么需要将脚本项目接入测试平台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: xiaoqq import os, re from datetime import datetime from typing import List, Optional, Dict from pathlib import Path TESTCASE_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_path def 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_path if project: search_path search_path / project if module: search_path search_path / module if 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.py path_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: continue if module and _module ! module: continue filename 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_idproject|module|filename|path external_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 testcases def extract_case_info(file_path): 解析测试用例文件获取相应信息 :param file_path: :return: with open(file_path, r, encodingutf-8) as file: content file.read() case_name_match re.search( rdef 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(rallure\.epic\(\(.?)\\), content) allure_feature_match re.search(rallure\.feature\(\(.?)\\), content) allure_story_match re.search(rallure\.story\(\(.?)\\), content) allure_epic allure_epic_match.group(1).strip() if allure_epic_match else None allure_feature allure_feature_match.group(1).strip() if allure_feature_match else None allure_story allure_story_match.group(1).strip() if allure_story_match else None return case_name, allure_epic, allure_feature, allure_story def 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, keylambda x: x[created_time], reverseTrue) 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(): continue proj_name proj_dir.name if project and proj_name ! project: continue proj_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(keylambda x: x[created_time], reverseTrue) 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, keylambda x: x[created_time], reverseTrue) def generate_testcase(case_yaml_list: list None): 生成测试用例 :return: from case_generator import CaseGenerator CG CaseGenerator() CG.generate_testcases(project_yaml_listcase_yaml_list) if __name__ __main__: # print(get_all_testcases()) # print(get_all_projects()) print(get_all_projects_and_modules(projectmerchant))testcase_route.py示例如下# author: xiaoqq from 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] True task_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, testcasesrequest.testcases, envrequest.env, report_typerequest.report_type, dingtalk_notifyrequest.dingtalk_notify, task_idrequest.task_id, callback_urlrequest.callback_url, auth_tokenrequest.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_at if report_type pytest-html: report_file base_path / freport_{task_id}.html elif report_type allure: report_file base_path / freport_{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.pyfrom fastapi import FastAPI from api import testcase_route from pathlib import Path from fastapi.staticfiles import StaticFiles app FastAPI(title接口自动化测试服务) # 挂载测试用例路由 app.include_router(testcase_route.router, prefix/api_test, tags[测试任务]) # 挂载 reports 目录为静态文件目录 reports_dir Path(__file__).parent / reports reports_dir.mkdir(exist_okTrue) # 确保目录存在 app.mount(/reports, StaticFiles(directoryreports_dir), namereports) if __name__ __main__: from utils.log_manager import LogManager LogManager.setup_logging() # 启动时显式初始化日志 import uvicorn uvicorn.run( main:app, host0.0.0.0, port8000, # reloadTrue, reload_excludes[testcases/*, logs/*, reports/*] # 排除这些目录的文件变更 )四. 测试平台调用执行mian.py启动 Fastapi 项目后便可在测试平台通过调用相关接口来管理该脚本测试项目平台调用代码不具体提供。1. 调用示意图测试平台 │ │ HTTP 调用 ▼ FastAPI 测试服务 │ │ pytest 执行 ▼ 测试报告生成 │ │ 回调结果 ▼ 测试平台展示这样职责边界非常清晰测试平台调度、记录、展示改造后的测试服务执行、产出报告2. 测试平台界面平台测试用例列表测试报告列表五. 总结方案优势总结如下解耦与复用脚本项目独立维护平台通过接口调用互不影响灵活执行支持按项目、模块、用例筛选执行适应不同测试场景。异步处理长时间任务后台执行平台可实时获取状态与报告。报告统一管理所有报告集中存储支持在线统一查看。