# perf_api_test

**Repository Path**: hankaizhou/perf_api_test

## Basic Information

- **Project Name**: perf_api_test
- **Description**: 一个专注于性能测试的API工具库，提供高效、易用的接口测试方案，支持多种测试场景和性能分析，助力开发者快速定位和优化系统性能问题。
- **Primary Language**: Unknown
- **License**: AGPL-3.0
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-09
- **Last Updated**: 2026-04-03

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 性能测试框架

## 📋 项目概述

性能测试框架是一个功能完整、架构清晰的测试工具，用于对API接口进行性能测试。框架采用模块化设计，具有高度的可扩展性，能够支持不同类型的接口进行性能测试。

### ✨ 主要特性

- **模块化设计**：核心功能、接口适配、配置管理等模块分离
- **扩展性架构**：统一的接口抽象层和插件机制
- **灵活的配置系统**：支持通过配置文件或API动态调整测试参数
- **可复用组件**：提供测试工具类、数据生成器和结果分析组件
- **兼容主流性能测试指标**：支持响应时间、成功率、吞吐量等指标采集
- **详细的报告生成**：支持Markdown和Word格式的测试报告
- **Grafana集成**：支持从Grafana获取内存监控数据
- **AI分析集成**：支持使用DeepSeek AI分析测试结果

## 📁 目录结构

```
e:\api_test\
├── adapters/          # 接口适配模块
├── config/            # 配置管理模块
├── core/              # 核心功能模块
├── docs/              # 文档目录
├── examples/          # 示例脚本
│   ├── example_usage.py    # 使用示例脚本
│   ├── grafana_monitor.py  # Grafana监控示例
│   └── test_oom.py         # OOM测试示例（遗留脚本）
├── logs/              # 日志目录
├── tests/             # 测试目录
├── utils/             # 工具模块
├── .env.example       # 环境变量示例文件
├── .gitignore         # Git忽略文件
├── cli.py             # 命令行工具
├── main.py            # 主入口文件
├── README.md          # 项目说明文件
├── requirements.txt   # 依赖文件
├── setup.py           # 安装脚本
└── test_framework.py  # 框架测试脚本
```

## 🚀 快速开始

### 1. 环境准备

1. **安装Python 3.8+**
   - 从 [Python官网](https://www.python.org/downloads/) 下载并安装

2. **安装依赖**
   ```bash
   pip install -r requirements.txt
   ```

3. **配置环境变量**
   - 复制 `.env.example` 为 `.env`
   - 根据实际情况修改配置

### 2. 配置设置

1. **基础配置**
   - 打开 `config/default_config.py` 文件
   - 修改以下配置：
     - `http.base_url`：设置为你要测试的API基础地址
     - `http.headers`：添加必要的请求头（如认证Token）
     - `test.concurrency`：设置并发数
     - `test.duration`：设置测试持续时间（秒）

2. **AI分析配置（可选）**
   - 在系统环境变量中设置：
     - Windows：`set DEEPSEEK_API_KEY=你的DeepSeek API密钥`
     - macOS/Linux：`export DEEPSEEK_API_KEY=你的DeepSeek API密钥`

3. **Grafana监控配置（可选）**
   - 修改 `grafana` 部分配置：
     - `enabled`：设置为 `true`
     - `url`：设置为Grafana服务器地址
     - `api_key`：设置为Grafana API密钥
     - `datasource`：设置为数据源名称
     - `namespace`：设置为要监控的Kubernetes命名空间

## 🎯 测试执行

### 方式1：使用命令行工具（推荐）

| 测试类型 | 命令示例 | 详细说明 |
|----------|----------|----------|
| 快速测试 | `python main.py run quick --base-url https://api.example.com --endpoints /api/test` | **用途**：快速验证接口的基本性能和可用性<br>**特点**：执行时间短，并发数适中<br>**适用场景**：开发阶段快速验证、CI/CD集成测试、日常监控 |
| 基准测试 | `python main.py run baseline --base-url https://api.example.com --endpoints /api/test` | **用途**：建立性能基准，作为后续测试的参考点<br>**特点**：执行多次，获取稳定的性能数据<br>**适用场景**：系统上线前评估、性能优化前后对比 |
| 压力测试 | `python main.py run stress --base-url https://api.example.com --endpoints /api/test --concurrency-levels 10,20,30` | **用途**：测试系统在不同负载下的表现，找出系统极限<br>**特点**：逐步增加并发数，观察系统性能变化<br>**适用场景**：系统容量规划、稳定性测试 |
| 边界测试 | `python main.py run boundary --base-url https://api.example.com --endpoints /api/test --boundary-params '{"page":1,"size":0};{"page":99999,"size":10000}'` | **用途**：测试接口在边界值和异常情况下的表现<br>**特点**：使用边界参数和异常参数测试<br>**适用场景**：接口健壮性测试、异常处理验证 |

### 命令参数说明

**通用参数**：
- `--base-url`：API基础地址，如 `https://api.example.com`（可选，默认使用配置文件中的值）
- `--endpoints`：要测试的接口路径，多个接口用逗号分隔（可选，默认测试配置文件中的所有接口）
- `--concurrency`：并发数，如 `5`（适用于快速测试）
- `--duration`：测试持续时间（秒），如 `300`（适用于快速测试）

**压力测试特有参数**：
- `--concurrency-levels`：并发数级别，多个级别用逗号分隔，如 `10,20,30`

**边界测试特有参数**：
- `--boundary-params`：边界测试参数，多个参数用分号分隔，如 `{"page":1,"size":0};{"page":99999,"size":10000}`

### 使用示例

**示例1：运行快速测试（5线程，5分钟）**
```bash
# 测试所有配置的接口
# --concurrency 5：设置5个并发线程
# --duration 300：设置测试持续300秒（5分钟）
python main.py run quick --concurrency 5 --duration 300

# 测试指定接口
# --endpoints：指定要测试的接口路径，多个接口用逗号分隔
# --concurrency 5：设置5个并发线程
# --duration 300：设置测试持续300秒（5分钟）
python main.py run quick --endpoints /v2/account/media/listVo,/v2/account/business/listVo --concurrency 5 --duration 300
```

**示例2：运行基准测试**
```bash
python main.py run baseline
```

**示例3：运行压力测试**
```bash
python main.py run stress --concurrency-levels 5,10,15,20
```

**示例4：运行边界测试**
```bash
python main.py run boundary
```

**也可以使用cli.py执行相同的命令**：
```bash
python cli.py run quick --concurrency 5 --duration 300
```

### 方式2：使用测试框架

```bash
# 运行完整的框架测试
python test_framework.py
```

### 方式3：使用示例脚本

```bash
# 运行示例脚本
python examples/example_usage.py

# 运行Grafana监控示例
python examples/grafana_monitor.py

# 运行遗留的OOM测试脚本（仅作为示例）
python examples/test_oom.py
```

## 📊 结果分析

1. **查看测试报告**
   - 测试完成后，报告将生成在 `reports` 目录中
   - Markdown报告：`reports/test_report_<时间戳>.md`
   - Word报告：`reports/test_report_<时间戳>.docx`

2. **报告内容分析**
   - 📋 **测试概览**：包含测试开始/结束时间、总请求数、成功率等
   - 📈 **详细结果**：每个接口的响应时间、错误率等
   - 🧠 **内存监控数据**：系统和进程内存使用情况
   - 🤖 **AI分析结论**：如果启用了AI分析

3. **关键指标解读**
   - ✅ **成功率**：应达到99%以上
   - ⏱️ **响应时间**：根据业务需求评估，通常应在1秒以内
   - 🔄 **吞吐量**：单位时间内处理的请求数，反映系统处理能力
   - 📉 **TP95/TP99**：95%/99%请求的响应时间，反映系统的稳定性

## 🧪 核心模块使用指南

### 1. 测试执行引擎

测试执行引擎负责管理测试的执行过程，包括并发控制和任务调度。

**使用示例**：

```python
from core.test_executor import TestExecutor

# 初始化执行器，设置并发数为20
executor = TestExecutor(concurrency=20)

# 定义测试函数
def test_api():
    # 测试逻辑
    pass

# 运行并发测试，持续300秒
results = executor.run_concurrent_test(test_api, [], duration_seconds=300)

# 运行批量测试
test_cases = [
    {'target': test_api, 'args': []},
    # 更多测试用例
]
batch_results = executor.run_batch_test(test_cases)
```

### 2. 数据收集模块

数据收集模块负责收集和分析测试数据，包括响应时间、成功率等指标。

**使用示例**：

```python
from core.data_collector import DataCollector

collector = DataCollector()

# 添加测试结果
for result in test_results:
    collector.add_result(result)

# 分析结果
analysis = collector.analyze_results()
print(analysis)

# 导出结果
collector.export_results('results.json')
```

### 3. 报告生成模块

报告生成模块负责生成测试报告，支持Markdown和Word格式。

**使用示例**：

```python
from core.report_generator import ReportGenerator

# 初始化报告生成器
generator = ReportGenerator(report_dir='reports')

# 生成Markdown报告
markdown_report = generator.generate_markdown_report(test_results, ai_analysis=None)
print(f"Markdown报告已生成: {markdown_report}")

# 生成Word报告
word_report = generator.generate_word_report(test_results, ai_analysis=None)
if word_report:
    print(f"Word报告已生成: {word_report}")
```

### 4. 性能指标采集模块

性能指标采集模块负责采集和分析性能测试指标。

**使用示例**：

```python
from core.metrics import MetricsCollector

collector = MetricsCollector()

# 添加请求记录
collector.add_request(response_time=0.5, success=True, status_code=200)
collector.add_request(response_time=1.2, success=True, status_code=200)
collector.add_request(response_time=0.8, success=False, status_code=500)

# 停止采集
collector.stop()

# 获取性能指标
metrics = collector.get_metrics()
print(metrics)

# 导出指标
collector.export_metrics('metrics.json')
```

## 🔌 接口适配器开发指南

### 1. 适配器基类

所有接口适配器都必须继承自`BaseAdapter`类，并实现以下方法：

- `send_request()`: 发送请求并返回结果
- `get_name()`: 获取适配器名称
- `get_supported_methods()`: 获取支持的方法

### 2. HTTP适配器

HTTP适配器用于测试HTTP接口，支持GET、POST、PUT、DELETE等方法。

**使用示例**：

```python
from adapters.http import HTTPAdapter

# 初始化HTTP适配器
adapter = HTTPAdapter(
    base_url="https://api.example.com",
    headers={"Content-Type": "application/json", "Authorization": "Bearer token"}
)

# 发送POST请求
result = adapter.send_request(
    endpoint="/api/users",
    data={"name": "test", "email": "test@example.com"},
    method="POST"
)

# 发送GET请求
result = adapter.send_request(
    endpoint="/api/users",
    params={"page": 1, "size": 10},
    method="GET"
)
```

### 3. 自定义适配器

要创建自定义适配器，需要继承`BaseAdapter`类并实现必要的方法。

**示例**：

```python
from adapters.base import BaseAdapter

class WebSocketAdapter(BaseAdapter):
    def send_request(self, endpoint, data=None, **kwargs):
        # 实现WebSocket请求逻辑
        pass
    
    def get_name(self):
        return "WebSocket"
    
    def get_supported_methods(self):
        return ["CONNECT", "SEND", "CLOSE"]
```

## ⚙️ 配置系统使用指南

配置系统支持从文件加载配置，以及动态调整配置参数。

**使用示例**：

```python
from config import ConfigManager, DEFAULT_CONFIG

# 初始化配置管理器
config = ConfigManager("config.json")

# 获取配置值
concurrency = config.get("test.concurrency", 10)

# 设置配置值
config.set("test.duration", 600)

# 更新配置
new_config = {"test": {"timeout": 60}}
config.update_config(new_config)

# 保存配置
config.save_config("new_config.json")

# 重置为默认配置
config.reset_to_default()
```

## 🛠️ 工具类使用指南

### 1. 日志工具

日志工具用于配置和管理日志系统。

**使用示例**：

```python
from utils import setup_logger

# 设置日志
logger = setup_logger(
    name="performance_test",
    config={
        "level": "INFO",
        "log_dir": "logs",
        "max_bytes": 10485760,  # 10MB
        "backup_count": 5
    }
)

# 使用日志
logger.info("测试开始")
logger.error("测试失败")
```

### 2. 数据生成器

数据生成器用于生成测试数据，支持多种类型的数据生成。

**使用示例**：

```python
from utils import DataGenerator

# 生成随机字符串
random_str = DataGenerator.generate_string(length=20)

# 生成随机数字
random_num = DataGenerator.generate_number(min_value=1, max_value=100)

# 生成分页参数
pagination = DataGenerator.generate_pagination_params(page_num=1, page_size=20)

# 生成时间范围参数
time_range = DataGenerator.generate_time_range(days=30)
```

### 3. 结果分析器

结果分析器用于分析测试结果并生成统计数据。

**使用示例**：

```python
from utils import ResultAnalyzer

# 分析测试结果
analysis = ResultAnalyzer.analyze(test_results)

# 生成分析摘要
summary = ResultAnalyzer.generate_summary(analysis)
print(summary)

# 比较两个测试结果
comparison = ResultAnalyzer.compare_results(results1, results2)

# 导出分析结果
ResultAnalyzer.export_analysis(analysis, "analysis.json")
```

### 4. 内存监控工具

内存监控工具用于监控系统和进程内存使用情况。

**使用示例**：

```python
from utils import MemoryMonitor

# 初始化内存监控器
monitor = MemoryMonitor(interval=30)  # 每30秒采集一次

# 开始监控，持续3600秒
monitor.start_monitoring(duration=3600)

# 执行测试...

# 停止监控
monitor.stop_monitoring()

# 获取内存数据
memory_data = monitor.get_memory_data()

# 获取内存监控摘要
summary = monitor.get_summary()

# 导出内存数据
monitor.export_data("memory_data.json")
```

## 📈 性能测试指标说明

框架支持采集以下性能测试指标：

| 指标名称 | 说明 | 单位 |
|---------|------|------|
| 总请求数 | 测试过程中发送的总请求数量 | 个 |
| 成功数 | 成功的请求数量 | 个 |
| 错误数 | 失败的请求数量 | 个 |
| 成功率 | 成功请求占总请求的百分比 | % |
| 平均响应时间 | 所有请求的平均响应时间 | 秒 |
| 最大响应时间 | 所有请求中的最大响应时间 | 秒 |
| 最小响应时间 | 所有请求中的最小响应时间 | 秒 |
| TP50 | 50%请求的响应时间不超过此值 | 秒 |
| TP90 | 90%请求的响应时间不超过此值 | 秒 |
| TP95 | 95%请求的响应时间不超过此值 | 秒 |
| TP99 | 99%请求的响应时间不超过此值 | 秒 |
| 吞吐量 | 单位时间内处理的请求数 | req/s |
| 响应时间分布 | 不同响应时间区间的请求数分布 | 个 |
| 状态码分布 | 各状态码的请求数分布 | 个 |

## 🚀 扩展开发示例

### 1. 自定义适配器

**示例：创建gRPC适配器**

```python
from adapters.base import BaseAdapter

class GRPCAdapter(BaseAdapter):
    def __init__(self, channel=None):
        self.channel = channel
    
    def send_request(self, endpoint, data=None, **kwargs):
        # 实现gRPC请求逻辑
        # ...
        return {
            "success": True,
            "elapsed": 0.1,
            "status": 0,
            "request_info": {"endpoint": endpoint, "data": data},
            "response_info": {"data": "response data"}
        }
    
    def get_name(self):
        return "gRPC"
    
    def get_supported_methods(self):
        return ["UNARY", "STREAM"]
    
    def initialize(self, config):
        # 初始化gRPC通道
        pass
    
    def cleanup(self):
        # 清理gRPC通道
        if self.channel:
            self.channel.close()
```

### 2. 自定义测试场景

**示例：创建压力测试场景**

```python
from core.test_executor import TestExecutor
from core.data_collector import DataCollector
from adapters.http import HTTPAdapter

# 初始化组件
adapter = HTTPAdapter(base_url="https://api.example.com")
executor = TestExecutor(concurrency=50)
collector = DataCollector()

# 定义测试函数
def test_api():
    result = adapter.send_request(
        endpoint="/api/resource",
        data={"key": "value"},
        method="POST"
    )
    return result

# 运行压力测试
print("开始压力测试...")
results = executor.run_concurrent_test(test_api, [], duration_seconds=1800)  # 30分钟

# 收集和分析结果
collector.add_results(results)
analysis = collector.analyze_results()

# 打印分析结果
print("测试结果分析:")
print(f"总请求数: {analysis['total']}")
print(f"成功率: {analysis['success_rate']:.2f}%")
print(f"平均响应时间: {analysis['avg_response_time']:.3f}s")
print(f"吞吐量: {analysis['throughput']:.2f} req/s")
```

## 🌟 最佳实践

### 1. 测试设计

- **明确测试目标**：在开始测试前，明确测试的目标和预期结果
- **选择合适的并发数**：根据系统性能和测试目标选择合适的并发数
- **设置合理的测试时长**：确保测试时长足够长，能够反映系统的真实性能
- **准备多样化的测试数据**：使用数据生成器生成多样化的测试数据

### 2. 配置管理

- **使用配置文件**：将测试参数集中到配置文件中，便于管理和调整
- **环境分离**：为不同环境（开发、测试、生产）创建不同的配置文件
- **版本控制**：将配置文件纳入版本控制，确保测试的可重复性

### 3. 结果分析

- **关注关键指标**：重点关注成功率、响应时间和吞吐量等关键指标
- **对比分析**：与历史测试结果进行对比，评估系统性能的变化
- **深入分析错误**：分析错误类型和分布，找出系统的薄弱环节
- **生成详细报告**：使用报告生成器生成详细的测试报告，便于团队分析和决策

### 4. 扩展开发

- **遵循适配器接口**：自定义适配器时，严格遵循BaseAdapter接口
- **保持代码简洁**：确保扩展代码简洁、清晰，易于维护
- **添加文档**：为扩展功能添加详细的文档和使用示例
- **测试扩展**：为扩展功能编写测试，确保其可靠性

## ➕ 如何添加新接口

### 1. 接口配置

要添加新接口到性能测试框架，只需在 `config/api_config.py` 文件中添加接口配置：

```python
# 在 API_CONFIG 字典中添加新的接口配置
API_CONFIG = {
    "media_account": {
        "endpoint": "/v2/account/media/listVo",
        "method": "POST",
        "payload": {"pageNum": 1, "pageSize": 10},
        "boundary_params": [
            {"pageNum": 99999, "pageSize": 10},  # 大页码
            {"pageNum": 1, "pageSize": 0},  # 零页大小
            {"pageNum": 1, "pageSize": -1},  # 负数页大小
            {"pageNum": 1, "pageSize": 10000}  # 大页大小
        ],
        "description": "媒体账户列表接口"
    },
    "business_account": {
        "endpoint": "/v2/account/business/listVo",
        "method": "POST",
        "payload": {
            "pageNum": 1,
            "pageSize": 10,
            "params": {
                "beginEffectTime": "2026-01-27",
                "endEffectTime": "2026-02-25"
            }
        },
        "boundary_params": [
            {"pageNum": 99999, "pageSize": 10, "params": {"beginEffectTime": "2026-01-27", "endEffectTime": "2026-02-25"}},
            {"pageNum": 1, "pageSize": 0, "params": {"beginEffectTime": "2026-01-27", "endEffectTime": "2026-02-25"}},
            {"pageNum": 1, "pageSize": -1, "params": {"beginEffectTime": "2026-01-27", "endEffectTime": "2026-02-25"}},
            {"pageNum": 1, "pageSize": 10000, "params": {"beginEffectTime": "2026-01-27", "endEffectTime": "2026-02-25"}},
            {"beginEffectTime": "2026/02/25", "endEffectTime": "2026/02/25"},  # 格式错误
            {"endEffectTime": "2026-02-25"},  # 缺失开始时间
            {"beginEffectTime": "2026-01-27"}   # 缺失结束时间
        ],
        "description": "商机账户列表接口"
    },
    # 新接口配置示例
    "new_interface": {
        "endpoint": "/v2/path/to/interface",
        "method": "POST",
        "payload": {"key": "value"},
        "boundary_params": [
            {"key": "value1"},
            {"key": "value2"}
        ],
        "description": "新接口描述"
    }
}
```

### 2. 接口配置字段说明

- **endpoint**：接口路径，如 `/v2/account/media/listVo`
- **method**：请求方法，默认为 `POST`
- **payload**：默认请求参数
- **boundary_params**：边界测试参数，用于测试接口在边界值和异常情况下的表现
- **description**：接口描述，用于在报告中显示

### 3. 运行新接口测试

添加新接口后，框架会自动识别并将其包含在测试中：

```python
from tests.test_scenarios import TestScenarios

# 初始化测试场景
scenarios = TestScenarios(base_url="http://your-api-server.com", headers={"Authorization": "Bearer your-token"})

# 运行所有接口测试（包括新添加的接口）
results = scenarios.run_all_interfaces_test(test_type="baseline")

# 运行指定接口测试
results = scenarios.run_interface_test("new_interface", test_type="stress")

# 运行所有边界测试
results = scenarios.run_all_boundary_tests()
```

### 4. 报告生成

新接口的测试结果会自动包含在生成的测试报告中，包括接口描述、测试指标等信息。

## 🔧 常见问题处理

| 问题 | 可能原因 | 解决方案 |
|------|----------|----------|
| API认证失败 | 认证Token错误或过期 | 检查请求头中的认证Token是否正确，确认Token是否过期 |
| 测试结果异常 | 网络不稳定或系统负载高 | 检查网络连接，确认目标系统是否正常运行，调整测试参数 |
| 内存监控失败 | 缺少psutil库或权限不足 | 确保安装了psutil库，检查系统权限 |
| 报告生成失败 | 缺少python-docx库或权限不足 | 确保安装了python-docx库，检查报告目录权限 |
| Grafana连接失败 | URL或API密钥错误 | 检查Grafana URL和API密钥是否正确，确认网络是否能访问Grafana |

## 📄 版本历史

### v1.0.0
- 初始版本
- 实现核心功能模块
- 支持HTTP接口测试
- 提供基本的性能指标采集
- 生成Markdown和Word格式报告

## 🤝 贡献指南

欢迎贡献代码和提出改进建议！请遵循以下步骤：

1. Fork 仓库
2. 创建特性分支
3. 提交更改
4. 推送到分支
5. 创建Pull Request

## 📄 许可证

本框架采用MIT许可证，详见LICENSE文件。