聊一聊 C# 中的闭包陷阱：foreach 循环的坑你还记得吗？

一. 为什么需要将脚本项目接入测试平台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, refrom datetime import datetimefrom typing import List, Optional, Dictfrom pathlib import PathTESTCASE_ROOT testcasesdef 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_idproject|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, 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 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, 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():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(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 CaseGeneratorCG 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: xiaoqqfrom pathlib import Pathfrom fastapi import APIRouter, BackgroundTasks, Query, Bodyfrom pydantic import BaseModelfrom typing import List, Optionalfrom runner import run_testsfrom 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] prereport_type: Optional[str] pytest-htmldingtalk_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,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 获取报告 URLif 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 / freport_{task_id}.htmlelif report_type allure:report_file base_path / freport_{task_id}_allure/html/index.htmlelse: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 FastAPIfrom api import testcase_routefrom pathlib import Pathfrom fastapi.staticfiles import StaticFilesapp FastAPI(title接口自动化测试服务)# 挂载测试用例路由app.include_router(testcase_route.router, prefix/api_test, tags[测试任务])# 挂载 reports 目录为静态文件目录reports_dir Path(__file__).parent / reportsreports_dir.mkdir(exist_okTrue) # 确保目录存在app.mount(/reports, StaticFiles(directoryreports_dir), namereports)if __name__ __main__:from utils.log_manager import LogManagerLogManager.setup_logging() # 启动时显式初始化日志import uvicornuvicorn.run(main:app,host0.0.0.0,port8000,# reloadTrue,reload_excludes[testcases/*, logs/*, reports/*] # 排除这些目录的文件变更)四. 测试平台调用执行mian.py启动 Fastapi 项目后便可在测试平台通过调用相关接口来管理该脚本测试项目平台调用代码不具体提供。1. 调用示意图测试平台││ HTTP 调用▼FastAPI 测试服务││ pytest 执行▼测试报告生成││ 回调结果▼测试平台展示这样职责边界非常清晰测试平台调度、记录、展示改造后的测试服务执行、产出报告颗郧捍栽