pytest-executor
pytest-executor 是一个测试环境调度器，核心功能是为测试任务分配、管理和释放运行环境，并支持在多种测试框架（pytest、unittest 等）上执行测试用例。
核心功能
环境调度
- 多环境提供商支持：支持 k8s 和 hidevlab 两种环境提供商，通过配置动态切换
- 环境生命周期管理：自动化环境申请、状态查询、使用和释放流程
- 并行环境分配：支持多线程并行分配多个测试环境
- 设备资源管理：支持 GPU、CPU 等多种设备类型的资源分配
测试执行
- 多框架支持：原生支持 pytest，可扩展支持 unittest 等其他测试框架
- 本地/远程执行：支持在本地或远程主机上执行测试
- 并行测试：支持多环境并行执行测试用例
结果管理
- 日志收集：自动收集测试日志并归档
- 结果上传：将测试结果上传到 OpenLibing 平台和 OBS 存储
项目架构
┌─────────────────────────────────────────────────────────────────┐
│ Scheduler Runner │
│ (调度入口，协调整个测试流程) │
└─────────────────────────────────────────────────────────────────┘
│
┌─────────────────────┼─────────────────────┐
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ EnvManager │ │ CaseCollector │ │ Executor │
│ (环境管理) │ │ (用例收集) │ │ (测试执行) │
└───────────────┘ └───────────────┘ └───────────────┘
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ k8s Provider │ │ pytest │ │ pytest │
│ hidevlab │ │ unittest │ │ unittest │
│ Provider │ │ (可扩展) │ │ (可扩展) │
└───────────────┘ └───────────────┘ └───────────────┘
环境调度流程
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 1. 申请环境 │ --> │ 2. 查询状态 │ --> │ 3. 执行测试 │ --> │ 4. 释放环境 │
│ allocate() │ │ check_status()│ │ execute() │ │ release() │
└──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘
│ │ │ │
▼ ▼ ▼ ▼
HTTP API HTTP API SSH/本地 HTTP API
(k8s/hidevlab) (k8s/hidevlab) 执行 (k8s/hidevlab)
项目文件结构
pytest-executor/
├── src/
│ ├── api/ # API 交互模块
│ │ ├── __init__.py
│ │ ├── api_handler.py # OpenLibing API 处理（环境申请、状态查询、释放）
│ │ └── obs_client.py # OBS 存储客户端
│ ├── case_collector/ # 测试用例收集模块
│ │ ├── __init__.py
│ │ ├── base_collector.py # 用例收集器基类
│ │ ├── collector_factory.py # 用例收集器工厂
│ │ └── pytest_collector.py # pytest 用例收集器
│ ├── executor/ # 测试执行模块
│ │ ├── __init__.py
│ │ ├── adapter.py # 执行器适配器
│ │ ├── base_executor.py # 执行器基类
│ │ ├── executor_factory.py # 执行器工厂
│ │ └── pytest_executor.py # pytest 执行器
│ ├── log/ # 日志模块
│ │ ├── __init__.py
│ │ └── logger_factory.py # 日志工厂（统一日志格式，包含线程ID）
│ ├── report/ # 报告处理模块
│ │ ├── __init__.py
│ │ ├── log_archiver.py # 日志归档
│ │ ├── report_uploader.py # 报告上传
│ │ └── xml_processor.py # XML 结果处理
│ ├── scheduler/ # 调度器核心模块
│ │ ├── __init__.py
│ │ ├── env_manager.py # 环境管理器（申请、查询、释放环境）
│ │ └── scheduler_config.py # 调度器配置
│ ├── utils/ # 工具模块
│ │ ├── __init__.py
│ │ ├── crypto_utils.py # 加密工具
│ │ ├── data_utils.py # 数据处理工具
│ │ └── path_utils.py # 路径处理工具
│ ├── __init__.py
│ └── run_scheduler.py # 调度器主入口
├── main.py # 命令行入口
├── requirements.txt # 依赖列表
├── setup.py # 包安装配置
├── start.sh # 启动脚本
└── README.md # 项目文档
快速开始
安装依赖
pip install -r requirements.txt
执行测试
python main.py \
--testcase_dir /path/to/testcases \
--log_dir /path/to/logs \
--obs \
--task_id test-task-001 \
--pipeline_id pipeline-001 \
--pipeline_run_id run-001 \
--job_run_id job-001 \
--scheduler_secret \
--max_workers 4
环境提供商配置
切换环境提供商
通过 Config.env_provider 配置：
# 使用 k8s（默认）
Config.env_provider = "k8s"
# 使用 hidevlab
Config.env_provider = "hidevlab"
失败用例重跑（only_fail 特性）
功能概述
only_fail 特性允许 pytest-executor 只运行上次测试失败的用例，从而节省测试时间和资源。该特性特别适用于以下场景：
- 增量测试：修复代码后只重新运行失败的测试用例
- 持续集成优化：减少 CI 流水线执行时间
- 快速验证：快速验证 bug 修复是否有效
使用方法
在测试用例目录下创建配置文件（例如 pytest.ini），添加 [only-fail] 配置段：
[only-fail]
run_only_fail_case = true
通过 --fail_case 参数指定失败测试结果 JSON 文件：
python main.py \
--testcase_dir /path/to/testcases \
--log_dir /path/to/logs \
--fail_case /path/to/failed_cases.json \
... 其他参数
其中 failed_cases.json 文件部分内容示例如下
{
"testCasesResult": [
{
"name": "test_login_failed",
"state": "failed"
},
{
"name": "test_data_error",
"state": "error"
},
{
"name": "test_success_case",
"state": "passed"
}
]
}
字段说明：
testCasesResult：测试用例结果数组name：测试用例名称（支持格式：test_function或TestClass::test_method）state：测试用例状态failed：测试失败error：测试错误passed：测试通过（会被过滤掉）
过滤逻辑：
- 只保留
state != "passed"的用例 - 支持多种失败状态：
failed、error等 - 自动匹配测试用例名称（支持
::分隔符格式）
注意事项
- JSON 文件路径：
--fail_case参数指定的 JSON 文件必须存在，否则会运行所有测试用例 - JSON 文件格式：必须包含
testCasesResult数组字段 - 用例名称匹配：支持
test_function和TestClass::test_method两种格式 - 无失败用例：如果 JSON 文件中没有失败用例，将执行所有测试用例
相关配置项
在 src/scheduler/scheduler_config.py 中定义了两个配置项：
only_fail = False # 是否只运行失败用例（从配置文件读取）
fail_case = "" # 失败测试结果 JSON 文件路径（从命令行参数读取）
环境提供商实现
系统通过 EnvironmentManager 统一管理环境，自动根据配置调用对应的提供商：
class EnvironmentManager:
def allocate_environments(self, env_case_info):
# 根据 Config.env_provider 选择执行逻辑
if Config.env_provider == "k8s":
return self._query_k8s_env_status(env_case_info)
else:
return self._query_hidevlab_env_status(env_case_info)
def release_env(self, task_id):
# 根据配置选择释放逻辑
if Config.env_provider == "k8s":
self._call_k8s_delete_api(task_id)
else:
self._call_hidevlab_delete_api(task_id)
测试框架扩展机制
框架接入概述
系统通过工厂模式支持多种测试框架的接入，主要涉及两个扩展点：
| 扩展点 | 职责 | 接口类 |
|---|---|---|
| CaseCollector | 收集测试用例 | BaseCollector |
| Executor | 执行测试用例 | BaseExecutor |
接入新框架（以 unittest 为例）
步骤 1：实现 Collector
创建 src/case_collector/unittest_collector.py：
from .base_collector import BaseCollector
class UnittestCollector(BaseCollector):
def collect(self):
"""收集 unittest 测试用例"""
test_cases = []
# 扫描测试目录，收集 unittest 测试用例
# ...
return test_cases
步骤 2：实现 Executor
创建 src/executor/unittest_executor.py：
from .base_executor import BaseExecutor
class UnittestExecutor(BaseExecutor):
def execute(self, test_cases, env_info):
"""执行 unittest 测试用例"""
# 执行 unittest 测试
# ...
return results
步骤 3：注册到工厂
修改 src/case_collector/collector_factory.py：
from .unittest_collector import UnittestCollector
class CollectorFactory:
@staticmethod
def create(framework_name):
if framework_name == "pytest":
return PytestCollector()
elif framework_name == "unittest":
return UnittestCollector() # 注册新框架
else:
raise ValueError(f"Unsupported framework: {framework_name}")
修改 src/executor/executor_factory.py：
from .unittest_executor import UnittestExecutor
class ExecutorFactory:
@staticmethod
def create(framework_name):
if framework_name == "pytest":
return PytestExecutor()
elif framework_name == "unittest":
return UnittestExecutor() # 注册新框架
else:
raise ValueError(f"Unsupported framework: {framework_name}")
步骤 4：配置使用
# 在 scheduler_config.py 中设置
Config.framework_name = "unittest"
扩展接口定义
BaseCollector：
class BaseCollector:
def collect(self) -> List[TestCase]:
"""收集测试用例，返回测试用例列表"""
raise NotImplementedError
BaseExecutor：
class BaseExecutor:
def execute(self, test_cases: List[TestCase], env_info: dict) -> ExecutionResult:
"""执行测试用例"""
raise NotImplementedError
核心模块说明
EnvManager（环境管理器）
负责环境的全生命周期管理：
allocate_environments()- 批量申请环境release_env()- 释放单个环境release_all()- 释放所有环境
SchedulerRunner（调度器）
协调整个测试流程：
- 收集测试用例
- 分配测试环境
- 执行测试用例
- 释放测试环境
技术栈
- Python 3.8+
- pytest（默认测试框架）
- paramiko（SSH 远程执行）
- requests（HTTP 请求）
- esdk-obs-python（OBS 存储）
许可证
MIT License