将分散的Pytest测试脚本统一接入测试平台：FastAPI改造方案详解

在上一篇文章《Pytest 测试用例自动生成：接口自动化进阶实践》中，我们已经解决了“如何高效编写和维护接口自动化用例”的问题。
然而，随着业务的发展和团队规模的扩大，很多公司会选择开发自己的测试平台，以实现更高效、更统一的管理。
企业接口自动化通常会经历如下过程：

• 初期：测试人员本地用 Pytest 写脚本（即脚本项目）
  
• 中期：接入 Jenkins，定时跑一跑
  
• 后期：公司开始建设统一测试平台
  

一. 为什么需要将脚本项目接入测试平台

1. 当前面临的挑战

• 用例复用难：已有的大量 Pytest 测试用例无法直接在平台上运行，重新编写成本高。
  
• 管理复杂：多个脚本项目分散管理，执行效率低、维护困难。
  
• 报告分散：测试报告散落在各个项目中，难以统一查看和分析。
  

因此，需要考虑在【不重写、不侵入、不破坏】现有 Pytest 脚本项目的前提下，让它具备“测试平台可接入能力”。
2. 两种可选方案

我们考虑以下两种方案将脚本项目接入测试平台：

• 方案一：将脚本项目直接集成到测试平台目录中。
  
• 方案二：使用 FastAPI 改造脚本项目，提供接口供平台调用。
  

为什么选择方案二？

• 脚本项目经常需要修改和更新，若集成到平台中，每次修改都需要重新发布平台代码，繁琐且易出错。
  
• FastAPI 轻量、高性能，适合快速构建RESTful接口，实现脚本与平台的解耦。
  

二. 使用 FastAPI 框架改造脚本项目

1. 核心思路

我们的目标是：不改动原有Pytest脚本逻辑，仅通过"封装+接口"的方式，让脚本项目具备平台接入能力。
实现原则：

• 不重写：保持原有 testcases/、utils/ 目录结构不变
  
• 不侵入：原有脚本文件无需修改任何代码
  
• 不解耦：仅新增API层，不改变原有执行逻辑
  

2. 目录改造

原有脚本项目核心目录保留不变，改造后新增「API 层、配置层」，确保原有脚本无需修改即可复用。

• 改造前目录（即脚本项目原始目录）示例如下：
  1. SUPER-API-AUTO-TEST/          # 接口自动化测试项目根目录
  2. ├── auth.py                    # 鉴权相关
  3. ├── case_generator.py          # 用例生成逻辑
  4. ├── config.yaml                # 项目配置
  5. ├── conftest.py                # Pytest夹具
  6. ├── runner.py                  # 用例执行入口
  7. ├── reports/                   # 测试报告目录
  8. ├── logs/                      # 日志目录
  9. ├── testcases/                 # 测试用例脚本（Python）
  10. ├── testcases_data/            # 测试用例数据（YAML）
  11. └── utils/                     # 通用工具类

  复制代码
• Fastapi 改造后目录结构示例如下：
  1. FASTAPI-SUPER-API-AUTO-TEST/  # 基于FastAPI的改造后目录
  2. ├── auth.py                    
  3. ├── case_generator.py         
  4. ├── conftest.py               
  5. ├── main.py                   # FastAPI应用入口（核心新增）
  6. ├── pytest.ini               
  7. ├── runner.py                  
  8. ├── api/                      # API层（核心新增）
  9. │   ├── testcase_route.py     # 接口路由定义（URL、请求参数）
  10. │   ├── testcase_service.py   # 接口业务逻辑（与原有脚本交互）
  11. │   └── __init__.py           
  12. ├── configs/                  # 配置目录（拆分原有config.yaml）
  13. │   └── config.yaml           
  14. ├── logs/                     
  15. ├── reports/                  
  16. ├── testcases/                # 完全复用原有目录
  17. ├── testcases_data/           # 完全复用原有目录
  18. └── 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

1. # @author:  xiaoqq
3. import os, re
4. from datetime import datetime
5. from typing import List, Optional, Dict
6. from pathlib import Path
8. TESTCASE_ROOT = "testcases"
10. def get_abs_root_path(root_path: str) -> Path:
11.     """
12.     使用当前文件相对路径构造 testcases/ 的绝对路径
13.     :param root_path: 目录名
14.     :return:
15.     """
16.     base_dir = Path(__file__).resolve().parent  # 当前文件所在目录
17.     abs_root_path = (base_dir.parent / root_path).resolve()
18.     return abs_root_path
20. def get_all_testcases(project: Optional[str] = None,
21.                   module: Optional[str] = None,
22.                   root_path: str = TESTCASE_ROOT) -> List[Dict]:
23.     """
24.     获取所有测试用例（支持通过 project/module 筛选）
25.     返回字段包括 filename（无后缀）、path（绝对路径字符串）、Allure 元信息等
26.     """
27.     abs_root_path = get_abs_root_path(root_path)
28.     if not abs_root_path.exists():
29.        return []
30.    
31.     # 路径校验
32.     if module and not project:
33.        raise ValueError("传入 module 前必须先传入 project")
34.    
35.     # 构造起始目录路径
36.     search_path = abs_root_path
37.     if project:
38.        search_path = search_path / project
39.     if module:
40.        search_path = search_path / module
41.    
42.     if not search_path.exists():
43.        return []
44.    
45.     testcases = []
46.    
47.     for dirpath, _, filenames in os.walk(search_path):
48.        for file in filenames:
49.           if file.startswith("test_") and file.endswith(".py"):
50.              full_path = os.path.join(dirpath, file)
51.              rel_path = os.path.relpath(full_path, abs_root_path)  # 相对路径，如 merchant/device/test_xxx.py
52.              path_parts = Path(rel_path).parts  # 使用 pathlib 安全拆解路径
53.             
54.              if len(path_parts ) < 2:
55.                 continue  # 至少要有 project/filename 结构
56.             
57.              _project = path_parts [0]
58.              _filename = path_parts [-1]
59.              _module = path_parts [1] if len(path_parts ) > 2 else None  # module 可选
60.             
61.              # 按传参过滤
62.              if project and _project != project:
63.                 continue
64.              if module and _module != module:
65.                 continue
66.             
67.              filename = os.path.splitext(_filename)[0]  # 去掉 .py 后缀
68.              last_modified = datetime.fromtimestamp(os.path.getmtime(full_path)).isoformat()
69.             
70.              # 提取用例元信息
71.              try:
72.                 case_name, epic, feature, story = extract_case_info(full_path)
73.              except Exception as e:
74.                 case_name, epic, feature, story = None, None, None, None
75.                
76.              # 拼接最终 path 字段为 TESTCASE_ROOT/... 形式
77.              full_case_path = str(Path(root_path) / rel_path).replace("\", "/")
78.             
79.              # 构造 external_id：project|module|filename|path
80.              external_id = f"{_project}|{_module or 'nomodule'}|{filename}|{full_case_path}"
81.             
82.              testcases.append({
83.                 "project": _project,
84.                 "module": _module,  # None 表示无 module 层级
85.                 "file": _filename,
86.                 "filename": filename,
87.                 "path": full_case_path,
88.                 "last_modified": last_modified,
89.                 "case_name": case_name or filename,
90.                 "allure_epic": epic,
91.                 "allure_feature": feature,
92.                 "allure_story": story,
93.                 "external_id": external_id # 加入唯一标识
94.              })
95.    
96.     return testcases
98. def extract_case_info(file_path):
99.     """
100.     解析测试用例文件，获取相应信息
101.     :param file_path:
102.     :return:
103.     """
104.     with open(file_path, 'r', encoding='utf-8') as file:
105.        content = file.read()
106.       
107.        case_name_match = re.search(
108.           r'def setup_class.*?\(.*?\):.*?log\.info\(\'========== 开始执行测试用例：(.+?) ==========\'',
109.           content, re.DOTALL
110.        )
111.        case_name = case_name_match.group(1).strip() if case_name_match else \
112.        os.path.splitext(os.path.basename(file_path))[0]
113.       
114.        allure_epic_match = re.search(r'@allure\.epic\(\'(.+?)\'\)', content)
115.        allure_feature_match = re.search(r'@allure\.feature\(\'(.+?)\'\)', content)
116.        allure_story_match = re.search(r'@allure\.story\(\'(.+?)\'\)', content)
117.       
118.        allure_epic = allure_epic_match.group(1).strip() if allure_epic_match else None
119.        allure_feature = allure_feature_match.group(1).strip() if allure_feature_match else None
120.        allure_story = allure_story_match.group(1).strip() if allure_story_match else None
121.       
122.        return case_name, allure_epic, allure_feature, allure_story
126. def get_all_projects(root_path: str = TESTCASE_ROOT) -> List[Dict[str, str]]:
127.     """
128.     获取 testcases/ 下所有项目名、相对路径及创建时间（倒序排序）
129.     """
130.     abs_root_path = get_abs_root_path(root_path)
131.     if not abs_root_path.exists():
132.        return []
133.    
134.     projects = []
135.     for d in abs_root_path.iterdir():
136.        if d.is_dir():
137.           created_time = datetime.fromtimestamp(d.stat().st_ctime)
138.           projects.append({
139.              "name": d.name,
140.              "path": str(Path(root_path) / d.name).replace("\", "/"),
141.              "created_time": created_time.isoformat()
142.           })
143.    
144.     # 按创建时间倒序
145.     return sorted(projects, key=lambda x: x["created_time"], reverse=True)
148. def get_all_projects_and_modules(
149.     project: Optional[str] = None,
150.     root_path: str = TESTCASE_ROOT
151. ) -> List[Dict]:
152.     """
153.     获取所有项目和模块结构（支持指定项目）。包含路径、创建时间，按项目时间倒序。
154.     """
155.     abs_root_path = get_abs_root_path(root_path)
156.     if not abs_root_path.exists():
157.        return []
158.    
159.     result = []
160.    
161.     for proj_dir in abs_root_path.iterdir():
162.        if not proj_dir.is_dir():
163.           continue
164.       
165.        proj_name = proj_dir.name
166.        if project and proj_name != project:
167.           continue
168.       
169.        proj_created_time = datetime.fromtimestamp(proj_dir.stat().st_ctime)
170.        modules = []
171.       
172.        # 遍历模块目录时需要忽略的子目录
173.        EXCLUDE_DIRS = {"__pycache__", ".pytest_cache", ".git", ".idea"}
174.        for mod_dir in proj_dir.iterdir():
175.           if mod_dir.is_dir() and mod_dir.name not in EXCLUDE_DIRS:
176.              mod_created_time = datetime.fromtimestamp(mod_dir.stat().st_ctime)
177.              modules.append({
178.                 "name": mod_dir.name,
179.                 "path": str(Path(root_path) / proj_name / mod_dir.name).replace("\", "/"),
180.                 "created_time": mod_created_time.isoformat()
181.              })
182.       
183.        # 模块也可以排序（如有需求）
184.        modules.sort(key=lambda x: x["created_time"], reverse=True)
185.       
186.        result.append({
187.           "project": proj_name,
188.           "path": str(Path(root_path) / proj_name).replace("\", "/"),
189.           "created_time": proj_created_time.isoformat(),
190.           "modules": modules
191.        })
192.       
193.        if project:
194.           break
195.    
196.     # 项目排序
197.     return sorted(result, key=lambda x: x["created_time"], reverse=True)
200. def generate_testcase(case_yaml_list: list = None):
201.     """
202.     生成测试用例
203.     :return:
204.     """
205.     from case_generator import CaseGenerator
206.     CG = CaseGenerator()
207.     CG.generate_testcases(project_yaml_list=case_yaml_list)
210. if __name__ == '__main__':
211.     # print(get_all_testcases())
212.     # print(get_all_projects())
213.     print(get_all_projects_and_modules(project="merchant"))

复制代码

testcase_route.py 示例如下：

1. # @author:  xiaoqq
3. from pathlib import Path
4. from fastapi import APIRouter, BackgroundTasks, Query, Body
5. from pydantic import BaseModel
6. from typing import List, Optional
7. from runner import run_tests
8. from api.testcase_service import (
9.         get_all_testcases,
10.         get_all_projects,
11.         get_all_projects_and_modules,
12.         generate_testcase,
13. )
15. router = APIRouter()
17. class TestExecutionRequest(BaseModel):
18.         testcases: Optional[List[str]] = ['testcases/']  # 默认运行所有目录
19.         env: Optional[str] = 'pre'
20.         report_type: Optional[str] = 'pytest-html'
21.         dingtalk_notify: Optional[bool] = True
22.         task_id: Optional[str]
23.         callback_url: Optional[str]
24.         auth_token: Optional[str] = None  # 新增字段：从平台传入的 token
26. # 执行测试用例
27. @router.post("/testcases/run")
28. def run_testcases(request: TestExecutionRequest, background_tasks: BackgroundTasks):
29.         try:
30.                 background_tasks.add_task(
31.                         run_tests,
32.                         testcases=request.testcases,
33.                         env=request.env,
34.                         report_type=request.report_type,
35.                         dingtalk_notify=request.dingtalk_notify,
36.                         task_id=request.task_id,
37.                         callback_url=request.callback_url,
38.                         auth_token=request.auth_token,  # 测试平台回调 auth_token
39.                 )
40.                 return {
41.                         "code": 0,
42.                         "msg": "测试任务已提交后台执行",
43.                         "task_id": request.task_id
44.                 }
45.         except Exception as e:
46.                 return {"code":1, "msg": f"测试任务失败：{str(e)}"}
48. # 获取测试用例
49. @router.get("/testcases/list")
50. def list_testcases(project: str = Query(None), module: str = Query(None)):
51.         try:
52.                 testcases = get_all_testcases(project, module)
53.                 return {
54.                         "code": 0,
55.                         "msg": "success",
56.                         "testcases": testcases
57.                 }
58.         except Exception as e:
59.                 return {"code": 1, "msg": f"获取测试用例失败：{str(e)}"}
62. # 获取 testcases/ 中的所有测试项目
63. @router.get("/testcases/projects")
64. def list_projects():
65.         try:
66.                 projects = get_all_projects()
67.                 return {"code": 0, "msg": "success", "projects": projects}
68.         except Exception as e:
69.                 return {"code": 1, "msg": f"获取测试项目失败：{str(e)}"}
70.        
72. # 获取 testcases/ 中的所有测试项目及模块
73. @router.get("/testcases/modules")
74. def list_modules(project: str = Query(None)):
75.         try:
76.                 modules = get_all_projects_and_modules(project)
77.                 return {"code": 0, "msg": "success", "modules": modules}
78.         except Exception as e:
79.                 return {"code": 1, "msg": f"获取测试项目-模块失败：{str(e)}"}
83. class GenerateCaseRequest(BaseModel):
84.         case_yaml_list: Optional[List[str]] = None
86. # 根据 testcases_data/ 中的测试数据生成测试用例文件
87. @router.post("/testcases/generate")
88. def generate_testcase_route(req: GenerateCaseRequest):
89.         try:
90.                 generate_testcase(req.case_yaml_list)
91.                 return {"code": 0, "msg": "success"}
92.         except Exception as e:
93.                 return {"code": 1, "msg": f"获取测试项目-模块失败：{str(e)}"}
96. @router.get("/reports/get_by_task")
97. def get_report_by_task(
98.                 task_id: str,
99.                 report_type: str,
100.                 created_at: str  # 格式: "20250814"
101. ):
102.         """
103.         根据 task_id + 创建时间 + report_type 获取报告 URL
104.         """
105.         if not created_at:
106.                 return {"code": 1, "msg": "created_at 必填", "url": None}
107.        
108.         base_path = Path(__file__).resolve().parent.parent / "reports" / created_at
109.        
110.         if report_type == "pytest-html":
111.                 report_file = base_path / f"report_{task_id}.html"
112.         elif report_type == "allure":
113.                 report_file = base_path / f"report_{task_id}_allure/html/index.html"
114.         else:
115.                 return {"code": 1, "msg": "未知 report_type", "url": None}
116.        
117.         if not report_file.exists():
118.                 return {"code": 1, "msg": "报告文件不存在", "url": None}
119.        
120.         relative_url = str(report_file.relative_to(Path(__file__).resolve().parent.parent)).replace("\", "/")
121.         return {"code": 0, "msg": "success", "url": f"/{relative_url}"}

复制代码

mian.py

1. from fastapi import FastAPI
2. from api import testcase_route
3. from pathlib import Path
4. from fastapi.staticfiles import StaticFiles
6. app = FastAPI(title="接口自动化测试服务")
8. # 挂载测试用例路由
9. app.include_router(testcase_route.router, prefix="/api_test", tags=["测试任务"])
11. # 挂载 reports 目录为静态文件目录
12. reports_dir = Path(__file__).parent / "reports"
13. reports_dir.mkdir(exist_ok=True)  # 确保目录存在
14. app.mount("/reports", StaticFiles(directory=reports_dir), name="reports")
16. if __name__ == "__main__":
17.     from utils.log_manager import LogManager
18.     LogManager.setup_logging()  # 启动时显式初始化日志
19.    
20.     import uvicorn
21.     uvicorn.run(
22.         "main:app",
23.         host="0.0.0.0",
24.         port=8000,
25.         # reload=True,
26.         reload_excludes=["testcases/*", "logs/*", "reports/*"]  # 排除这些目录的文件变更
27.     )

复制代码

四. 测试平台调用

执行mian.py，启动 Fastapi 项目后，便可在测试平台通过调用相关接口来管理该脚本测试项目（平台调用代码不具体提供）。
1. 调用示意图

1. 测试平台
2.    │
3.    │ HTTP 调用
4.    ▼
5. FastAPI 测试服务
6.    │
7.    │ pytest 执行
8.    ▼
9. 测试报告生成
10.    │
11.    │ 回调结果
12.    ▼
13. 测试平台展示

复制代码

这样，职责边界非常清晰：

• 测试平台：调度、记录、展示，
  
• 改造后的测试服务：执行、产出报告
  

2. 测试平台界面

平台测试用例列表：

<img alt="image-20260120134215814" loading="lazy">

测试报告列表：

<img alt="image-20260122143715247" loading="lazy">

五. 总结

方案优势总结如下：

• 解耦与复用：脚本项目独立维护，平台通过接口调用，互不影响
  
• 灵活执行：支持按项目、模块、用例筛选执行，适应不同测试场景。
  
• 异步处理：长时间任务后台执行，平台可实时获取状态与报告。
  
• 报告统一管理：所有报告集中存储，支持在线统一查看。
  

当然，示例代码还可以进行优化扩展，如加入用户认证机制来保障接口安全等。
当接口自动化发展到一定规模，单机脚本 或 Jenkins Job 都会成为瓶颈，而“脚本服务化 + 平台调度”，几乎是所有成熟团队最终都会走到的一步。
如果你：

• 正在做接口自动化
  
• 或正在参与测试平台建设
  
• 或正在被“脚本怎么接平台”折磨
  

那么，希望这篇文章能少让你走一点弯路。
               

        

        本文作者：给你一页白纸        版权申明：本博客所有文章除特殊声明外，均采用BY-NC-SA         许可协议。转载请注明出处！        声援博主：如果觉得这篇文章对您有帮助，请点一下右下角的                “推荐”                图标哦，您的                “推荐”                是我写作的最大动力。您也可以点击下方的        【关注我】        按钮，关注博主不迷路。
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

感谢发布原创作品，程序园因你更精彩