feat(admin): 添加用户管理相关文件

添加用户管理视图、API和状态管理文件
This commit is contained in:
张翔
2026-03-28 14:37:29 +08:00
commit 08ea5fbe98
1643 changed files with 255646 additions and 0 deletions
@@ -0,0 +1,204 @@
# test-tools - API测试工具
## 概述
test-tools是一个基于Python的API测试工具模块,提供完整的API测试功能,包括认证、测试用例执行、报告生成等功能。
## 目录结构
```
test-tools/
├── config/ # 配置文件目录
│ ├── test_config.yaml # 测试配置文件
│ └── settings.py # 配置管理模块
├── core/ # 核心功能模块
├── data/ # 测试数据目录
├── utils/ # 工具函数目录
├── test_cases/ # 测试用例目录
├── interfaces/ # 接口定义目录
├── main.py # 主程序入口
├── main_enhanced.py # 增强版主程序
├── requirements.txt # Python依赖
├── pyproject.toml # Poetry配置文件
└── README.md # 本文档
```
## 功能特性
- API测试自动化
- 支持多种认证方式
- 测试用例管理
- 测试报告生成(JSON、HTML格式)
- 日志记录
- 并行测试支持
- 测试重试机制
## 快速开始
### 安装依赖
```bash
cd test-tools
poetry install
```
或者使用pip
```bash
pip install -r requirements.txt
```
### 配置环境变量
在项目根目录的`.env`文件中配置test-tools相关环境变量:
```bash
# test-tools API测试配置
TEST_TOOLS_API_BASE_URL=http://127.0.0.1:8080/api
TEST_TOOLS_API_TIMEOUT=30
TEST_TOOLS_API_MAX_RETRIES=3
TEST_TOOLS_AUTH_LOGIN_ENDPOINT=/sys/auth/login
TEST_TOOLS_AUTH_USERNAME=admin
TEST_TOOLS_AUTH_PASSWORD=admin123
TEST_TOOLS_AUTH_TOKEN_STORAGE=memory
TEST_TOOLS_REPORT_OUTPUT_DIR=test-results/test-tools/reports
TEST_TOOLS_LOGGING_LEVEL=INFO
TEST_TOOLS_LOGGING_FILE=test-results/test-tools/logs/test.log
TEST_TOOLS_LOGGING_CONSOLE=true
```
### 运行测试
使用npm脚本:
```bash
npm run test:test-tools
```
或者直接运行Python脚本:
```bash
cd test-tools
python main.py
```
运行增强版主程序:
```bash
cd test-tools
python main_enhanced.py
```
## 配置说明
### test_config.yaml
配置文件位于`config/test_config.yaml`,支持环境变量覆盖:
```yaml
api:
base_url: ${TEST_TOOLS_API_BASE_URL:http://127.0.0.1:8080/api}
timeout: ${TEST_TOOLS_API_TIMEOUT:30}
max_retries: ${TEST_TOOLS_API_MAX_RETRIES:3}
auth:
login_endpoint: ${TEST_TOOLS_AUTH_LOGIN_ENDPOINT:/sys/auth/login}
username: ${TEST_TOOLS_AUTH_USERNAME:admin}
password: ${TEST_TOOLS_AUTH_PASSWORD:admin123}
token_storage: ${TEST_TOOLS_AUTH_TOKEN_STORAGE:memory}
report:
output_dir: ${TEST_TOOLS_REPORT_OUTPUT_DIR:../test-results/test-tools/reports}
formats:
- json
- html
logging:
level: ${TEST_TOOLS_LOGGING_LEVEL:INFO}
file: ${TEST_TOOLS_LOGGING_FILE:../test-results/test-tools/logs/test.log}
console: ${TEST_TOOLS_LOGGING_CONSOLE:true}
```
## 测试报告
测试报告将生成在`test-results/test-tools/reports/`目录下,支持以下格式:
- JSON格式:`test_results.json`
- HTML格式:`test_results.html`
日志文件将生成在`test-results/test-tools/logs/`目录下。
## 集成到主测试项目
test-tools已集成到主测试项目中,可以通过以下命令统一执行:
```bash
# 执行所有测试(包括test-tools)
npm test
# 只执行test-tools测试
npm run test:test-tools
# 查看test-tools测试报告
npm run test:test-tools:report
```
## 开发指南
### 添加新的测试用例
1.`test_cases/`目录下创建新的测试用例文件
2. 按照现有测试用例的格式编写测试代码
3. 使用`main.py``main_enhanced.py`运行测试
### 修改配置
1. 修改`config/test_config.yaml`文件
2. 或者在`.env`文件中设置环境变量覆盖配置
3. 环境变量优先级高于配置文件
## 依赖项
主要依赖项:
- Python >= 3.10
- requests >= 2.31.0
- httpx >= 0.25.0
- pyyaml >= 6.0
- bcrypt >= 4.0.0
- psycopg2-binary >= 2.9.0
- python-dotenv >= 1.0.0
开发依赖项:
- pytest >= 7.4.0
- pytest-asyncio >= 0.21.0
- pytest-cov >= 4.1.0
- black >= 23.0.0
- flake8 >= 6.0.0
- mypy >= 1.5.0
## 故障排除
### 常见问题
1. **依赖安装失败**
- 确保Python版本 >= 3.10
- 尝试使用`pip install --upgrade pip`升级pip
2. **测试执行失败**
- 检查API服务是否正常运行
- 检查环境变量配置是否正确
- 查看日志文件获取详细错误信息
3. **报告生成失败**
- 检查报告输出目录是否有写权限
- 确保磁盘空间充足
## 贡献指南
欢迎提交问题和改进建议。
## 许可证
MIT License
@@ -0,0 +1 @@
"""配置模块"""
@@ -0,0 +1,248 @@
"""
测试工具配置管理模块
"""
import os
import re
from pathlib import Path
from typing import Dict, Any
import yaml
class TestConfig:
"""测试配置类"""
def __init__(self, config_path: str = None):
"""
初始化配置
Args:
config_path: 配置文件路径,默认为config/test_config.yaml
"""
if config_path is None:
config_path = Path(__file__).parent.parent / "config" / "test_config.yaml"
self.config_path = Path(config_path)
self.config = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""
加载配置文件
Returns:
配置字典
"""
if not self.config_path.exists():
return self._get_default_config()
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
return self._resolve_env_vars(config)
except Exception as e:
print(f"加载配置文件失败: {e}")
return self._get_default_config()
def _resolve_env_vars(self, config: Any) -> Any:
"""
解析环境变量
Args:
config: 配置对象
Returns:
解析后的配置对象
"""
if isinstance(config, dict):
return {k: self._convert_type(self._resolve_env_vars(v)) for k, v in config.items()}
elif isinstance(config, list):
return [self._convert_type(self._resolve_env_vars(item)) for item in config]
elif isinstance(config, str):
return self._convert_type(self._parse_env_var(config))
else:
return self._convert_type(config)
def _parse_env_var(self, value: str) -> str:
"""
解析环境变量
Args:
value: 配置值
Returns:
解析后的值
"""
pattern = r'\$\{([^:}]+)(?::([^}]*))?\}'
match = re.search(pattern, value)
if match:
env_var = match.group(1)
default_value = match.group(2) if match.group(2) else ""
return os.getenv(env_var, default_value)
return value
def _convert_type(self, value: Any) -> Any:
"""
类型转换
Args:
value: 配置值
Returns:
转换后的值
"""
if isinstance(value, str):
if value.isdigit():
return int(value)
try:
return float(value)
except ValueError:
pass
if value.lower() in ('true', 'false'):
return value.lower() == 'true'
return value
def _get_default_config(self) -> Dict[str, Any]:
"""
获取默认配置
Returns:
默认配置字典
"""
return {
"api": {
"base_url": os.getenv("TEST_TOOLS_API_BASE_URL", "http://127.0.0.1:8080/api"),
"timeout": int(os.getenv("TEST_TOOLS_API_TIMEOUT", "30")),
"max_retries": int(os.getenv("TEST_TOOLS_API_MAX_RETRIES", "3"))
},
"auth": {
"login_endpoint": "/sys/auth/login",
"username": os.getenv("TEST_TOOLS_AUTH_USERNAME", "admin"),
"password": os.getenv("TEST_TOOLS_AUTH_PASSWORD", "admin123"),
"token_storage": "memory"
},
"test": {
"data_dir": "data",
"test_cases_dir": "test_cases",
"parallel": True,
"retry_count": 2
},
"report": {
"output_dir": "../test-results/test-tools/reports",
"formats": ["json", "html"],
"include_details": True
},
"logging": {
"level": "INFO",
"file": "../test-results/test-tools/logs/test.log",
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"console": True
}
}
def get(self, key: str, default: Any = None) -> Any:
"""
获取配置值
Args:
key: 配置键,支持点号分隔的嵌套键(如"api.base_url"
default: 默认值
Returns:
配置值
"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
def set(self, key: str, value: Any) -> None:
"""
设置配置值
Args:
key: 配置键,支持点号分隔的嵌套键
value: 配置值
"""
keys = key.split('.')
config = self.config
for k in keys[:-1]:
if k not in config:
config[k] = {}
config = config[k]
config[keys[-1]] = value
def save(self) -> None:
"""保存配置到文件"""
self.config_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True)
@property
def api_base_url(self) -> str:
"""获取API基础URL"""
return self.get("api.base_url")
@property
def api_timeout(self) -> int:
"""获取API超时时间"""
return self.get("api.timeout")
@property
def api_max_retries(self) -> int:
"""获取API最大重试次数"""
return self.get("api.max_retries")
@property
def auth_login_endpoint(self) -> str:
"""获取登录端点"""
return self.get("auth.login_endpoint")
@property
def auth_username(self) -> str:
"""获取测试用户名"""
return self.get("auth.username")
@property
def auth_password(self) -> str:
"""获取测试密码"""
return self.get("auth.password")
@property
def report_output_dir(self) -> str:
"""获取报告输出目录"""
return self.get("report.output_dir")
@property
def report_formats(self) -> list:
"""获取报告格式"""
return self.get("report.formats")
@property
def logging_level(self) -> str:
"""获取日志级别"""
return self.get("logging.level")
@property
def logging_file(self) -> str:
"""获取日志文件路径"""
return self.get("logging.file")
@property
def logging_console(self) -> bool:
"""获取是否输出到控制台"""
return self.get("logging.console", True)
# 全局配置实例
config = TestConfig()
@@ -0,0 +1,36 @@
# 测试工具配置文件
# API配置
api:
base_url: ${TEST_TOOLS_API_BASE_URL:http://127.0.0.1:8080/api}
timeout: ${TEST_TOOLS_API_TIMEOUT:30}
max_retries: ${TEST_TOOLS_API_MAX_RETRIES:3}
# 认证配置
auth:
login_endpoint: ${TEST_TOOLS_AUTH_LOGIN_ENDPOINT:/sys/auth/login}
username: ${TEST_TOOLS_AUTH_USERNAME:admin}
password: ${TEST_TOOLS_AUTH_PASSWORD:admin123}
token_storage: ${TEST_TOOLS_AUTH_TOKEN_STORAGE:memory}
# 测试配置
test:
data_dir: data
test_cases_dir: test_cases
parallel: true
retry_count: 2
# 报告配置
report:
output_dir: ${TEST_TOOLS_REPORT_OUTPUT_DIR:../test-results/test-tools/reports}
formats:
- json
- html
include_details: true
# 日志配置
logging:
level: ${TEST_TOOLS_LOGGING_LEVEL:INFO}
file: ${TEST_TOOLS_LOGGING_FILE:../test-results/test-tools/logs/test.log}
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
console: ${TEST_TOOLS_LOGGING_CONSOLE:true}
@@ -0,0 +1 @@
"""核心模块"""
@@ -0,0 +1,284 @@
"""
API测试器核心模块
"""
import requests
import time
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field
from .validation import ValidationEngine, ValidationRule
from .auth_manager import AuthManager
from config.settings import config
from utils.logger import TestLogger
from utils.reporter import TestResult as ReportTestResult, TestSummary
@dataclass
class TestResult:
"""测试结果"""
passed: bool
test_name: str
error_message: str = ""
response_time: float = 0.0
status_code: int = 0
request_data: Dict[str, Any] = field(default_factory=dict)
response_data: Dict[str, Any] = field(default_factory=dict)
class APITester:
"""API测试器"""
def __init__(self, logger: TestLogger = None, auto_auth: bool = True):
"""
初始化API测试器
Args:
logger: 日志记录器
auto_auth: 是否自动认证
"""
self.logger = logger or TestLogger("api_tester", config.logging_file, config.logging_level)
self.session = requests.Session()
self.auth_manager = AuthManager(self.logger)
self.validation_engine = ValidationEngine()
self.auto_auth = auto_auth
# 配置会话
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
# 如果启用自动认证,尝试登录
if auto_auth:
self._ensure_authenticated()
def _ensure_authenticated(self) -> bool:
"""
确保已认证
Returns:
是否认证成功
"""
return self.auth_manager.ensure_authenticated()
def _update_auth_header(self) -> None:
"""更新认证请求头"""
auth_header = self.auth_manager.get_auth_header(auto_refresh=True)
self.session.headers.update(auth_header)
def set_token(self, token: str) -> None:
"""
设置认证令牌(已弃用,建议使用AuthManager
Args:
token: JWT令牌
"""
self.logger.warning("set_token方法已弃用,建议使用AuthManager")
self.session.headers.update({
"Authorization": f"Bearer {token}"
})
self.logger.info(f"已设置认证令牌")
def clear_token(self) -> None:
"""清除认证令牌(已弃用,建议使用AuthManager"""
self.logger.warning("clear_token方法已弃用,建议使用AuthManager")
self.session.headers.pop("Authorization", None)
self.logger.info(f"已清除认证令牌")
def login(self, username: str = None, password: str = None) -> bool:
"""
用户登录
Args:
username: 用户名
password: 密码
Returns:
是否登录成功
"""
return self.auth_manager.login(username, password)
def request(
self,
method: str,
endpoint: str,
data: Dict[str, Any] = None,
params: Dict[str, Any] = None,
headers: Dict[str, str] = None,
expected_status: int = 200,
test_name: str = None,
require_auth: bool = True
) -> TestResult:
"""
发送HTTP请求
Args:
method: HTTP方法(GET, POST, PUT, DELETE
endpoint: API端点
data: 请求体数据
params: URL参数
headers: 请求头
expected_status: 期望的状态码
test_name: 测试名称
require_auth: 是否需要认证
Returns:
测试结果
"""
if test_name is None:
test_name = f"{method} {endpoint}"
url = f"{config.api_base_url}{endpoint}"
# 如果需要认证,确保已认证并更新认证头
if require_auth and self.auto_auth:
if not self._ensure_authenticated():
return TestResult(
passed=False,
test_name=test_name,
error_message="认证失败"
)
self._update_auth_header()
self.logger.log_test_start(test_name)
self.logger.log_request(method, url, data, headers)
try:
start_time = time.time()
if method.upper() == "GET":
response = self.session.get(
url,
params=params,
headers=headers,
timeout=config.api_timeout
)
elif method.upper() == "POST":
response = self.session.post(
url,
json=data,
params=params,
headers=headers,
timeout=config.api_timeout
)
elif method.upper() == "PUT":
response = self.session.put(
url,
json=data,
params=params,
headers=headers,
timeout=config.api_timeout
)
elif method.upper() == "DELETE":
response = self.session.delete(
url,
params=params,
headers=headers,
timeout=config.api_timeout
)
else:
raise ValueError(f"不支持的HTTP方法: {method}")
response_time = (time.time() - start_time) * 1000
try:
response_data = response.json()
except:
response_data = {"raw": response.text}
self.logger.log_response(response.status_code, response_time, response_data)
# 验证状态码
passed, error = self.validation_engine.validate_status_code(expected_status, response.status_code)
if passed:
self.logger.log_test_end(test_name, True, response_time)
return TestResult(
passed=True,
test_name=test_name,
response_time=response_time,
status_code=response.status_code,
request_data=data or params or {},
response_data=response_data
)
else:
self.logger.log_test_end(test_name, False, response_time)
return TestResult(
passed=False,
test_name=test_name,
error_message=error,
response_time=response_time,
status_code=response.status_code,
request_data=data or params or {},
response_data=response_data
)
except requests.exceptions.Timeout:
error_msg = "请求超时"
self.logger.error(f"{test_name} - {error_msg}")
return TestResult(
passed=False,
test_name=test_name,
error_message=error_msg,
response_time=config.api_timeout * 1000
)
except requests.exceptions.ConnectionError:
error_msg = "连接错误"
self.logger.error(f"{test_name} - {error_msg}")
return TestResult(
passed=False,
test_name=test_name,
error_message=error_msg
)
except Exception as e:
self.logger.log_error(e)
return TestResult(
passed=False,
test_name=test_name,
error_message=f"未知错误: {str(e)}"
)
def validate(
self,
test_result: TestResult,
validation_rules: List[ValidationRule]
) -> TestResult:
"""
验证测试结果
Args:
test_result: 测试结果
validation_rules: 验证规则列表
Returns:
验证后的测试结果
"""
if not test_result.passed:
return test_result
for rule in validation_rules:
if rule.rule_type == "status_code":
passed, error = rule.validate(test_result.status_code)
elif rule.rule_type == "response_time":
passed, error = rule.validate(test_result.response_time)
elif rule.rule_type in ["contains", "equals", "json_path", "regex", "header", "schema"]:
passed, error = rule.validate(actual_data=test_result.response_data)
else:
passed, error = False, f"未知的验证规则: {rule.rule_type}"
self.logger.log_validation(rule.rule_type, passed, error)
if not passed:
test_result.passed = False
test_result.error_message = error
break
return test_result
def close(self) -> None:
"""关闭会话"""
self.session.close()
self.logger.info("已关闭测试会话")
@@ -0,0 +1,338 @@
"""
认证管理器模块
提供自动令牌获取、存储、验证和刷新功能
"""
import time
import json
import hashlib
from typing import Optional, Dict, Any
from pathlib import Path
from dataclasses import dataclass
from datetime import datetime, timedelta
from utils.logger import TestLogger
from config.settings import config
@dataclass
class TokenInfo:
"""令牌信息"""
token: str
username: str
issued_at: float
expires_at: float
refresh_token: Optional[str] = None
def is_expired(self, buffer_seconds: int = 60) -> bool:
"""
检查令牌是否过期
Args:
buffer_seconds: 缓冲时间(秒),提前多少秒认为过期
Returns:
是否过期
"""
return time.time() > (self.expires_at - buffer_seconds)
def time_until_expiry(self) -> float:
"""
获取距离过期的时间
Returns:
距离过期的秒数
"""
return max(0, self.expires_at - time.time())
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
"token": self.token,
"username": self.username,
"issued_at": self.issued_at,
"expires_at": self.expires_at,
"refresh_token": self.refresh_token
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'TokenInfo':
"""从字典创建"""
return cls(
token=data["token"],
username=data["username"],
issued_at=data["issued_at"],
expires_at=data["expires_at"],
refresh_token=data.get("refresh_token")
)
class AuthManager:
"""认证管理器"""
def __init__(self, logger: TestLogger = None):
"""
初始化认证管理器
Args:
logger: 日志记录器
"""
self.logger = logger or TestLogger("auth_manager", config.logging_file, config.logging_level)
self.token_info: Optional[TokenInfo] = None
self.token_cache_file = Path(config.report_output_dir) / "token_cache.json"
# 令牌刷新缓冲时间(秒)
self.refresh_buffer = 60
# 加载缓存的令牌
self._load_cached_token()
def _load_cached_token(self) -> None:
"""从缓存加载令牌"""
if not self.token_cache_file.exists():
return
try:
with open(self.token_cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
self.token_info = TokenInfo.from_dict(data)
if self.token_info.is_expired():
self.logger.info("缓存的令牌已过期,将重新获取")
self.token_info = None
else:
self.logger.info(f"从缓存加载令牌,剩余有效期: {self.token_info.time_until_expiry():.0f}")
except Exception as e:
self.logger.warning(f"加载缓存令牌失败: {e}")
self.token_info = None
def _save_cached_token(self) -> None:
"""保存令牌到缓存"""
if self.token_info is None:
return
try:
self.token_cache_file.parent.mkdir(parents=True, exist_ok=True)
with open(self.token_cache_file, 'w', encoding='utf-8') as f:
json.dump(self.token_info.to_dict(), f, indent=2)
self.logger.info("令牌已缓存")
except Exception as e:
self.logger.warning(f"保存缓存令牌失败: {e}")
def _clear_cached_token(self) -> None:
"""清除缓存的令牌"""
try:
if self.token_cache_file.exists():
self.token_cache_file.unlink()
self.logger.info("缓存的令牌已清除")
except Exception as e:
self.logger.warning(f"清除缓存令牌失败: {e}")
def login(
self,
username: str = None,
password: str = None,
force_refresh: bool = False
) -> bool:
"""
用户登录
Args:
username: 用户名
password: 密码
force_refresh: 是否强制刷新令牌
Returns:
是否登录成功
"""
username = username or config.auth_username
password = password or config.auth_password
# 检查是否需要重新登录
if not force_refresh and self.token_info and not self.token_info.is_expired(self.refresh_buffer):
self.logger.info(f"使用现有令牌,剩余有效期: {self.token_info.time_until_expiry():.0f}")
return True
# 执行登录
self.logger.info(f"用户登录: {username}")
try:
import requests
login_url = f"{config.api_base_url}{config.auth_login_endpoint}"
response = requests.post(
login_url,
json={"username": username, "password": password},
timeout=config.api_timeout
)
if response.status_code == 200:
data = response.json()
# 兼容两种响应格式:
# 格式1: {"code": 200, "data": {"token": "...", "user": {...}}}
# 格式2: {"token": "...", "user": {...}}
token = None
if "data" in data and isinstance(data["data"], dict):
token = data["data"].get("token")
else:
token = data.get("token")
if token:
# 解析JWT令牌获取过期时间
expires_at = self._parse_token_expiry(token)
# 创建令牌信息
self.token_info = TokenInfo(
token=token,
username=username,
issued_at=time.time(),
expires_at=expires_at,
refresh_token=data.get("refreshToken") if "data" in data else None
)
# 缓存令牌
self._save_cached_token()
self.logger.info(f"✅ 登录成功,令牌有效期: {(expires_at - time.time()):.0f}")
return True
self.logger.error(f"❌ 登录失败: {response.text}")
return False
except Exception as e:
self.logger.error(f"❌ 登录异常: {str(e)}")
return False
def _parse_token_expiry(self, token: str) -> float:
"""
解析JWT令牌的过期时间
Args:
token: JWT令牌
Returns:
过期时间戳
"""
try:
# JWT格式: header.payload.signature
parts = token.split('.')
if len(parts) != 3:
raise ValueError("无效的JWT令牌格式")
# 解码payloadBase64URL编码)
import base64
payload = parts[1]
# 添加必要的填充
padding = 4 - len(payload) % 4
if padding != 4:
payload += '=' * padding
decoded = base64.urlsafe_b64decode(payload)
payload_data = json.loads(decoded)
# 获取过期时间(exp字段是Unix时间戳,秒)
exp = payload_data.get('exp')
if exp:
return float(exp)
# 如果没有exp字段,默认24小时后过期
return time.time() + 24 * 3600
except Exception as e:
self.logger.warning(f"解析令牌过期时间失败: {e},使用默认过期时间")
return time.time() + 24 * 3600
def get_token(self, auto_refresh: bool = True) -> Optional[str]:
"""
获取当前令牌
Args:
auto_refresh: 是否自动刷新过期令牌
Returns:
令牌字符串,如果未登录则返回None
"""
if self.token_info is None:
return None
# 检查令牌是否过期
if self.token_info.is_expired(self.refresh_buffer):
if auto_refresh:
self.logger.info("令牌即将过期,尝试自动刷新")
if self.login(force_refresh=True):
return self.token_info.token
else:
self.logger.error("自动刷新令牌失败")
return None
else:
self.logger.warning("令牌已过期")
return None
return self.token_info.token
def get_auth_header(self, auto_refresh: bool = True) -> Dict[str, str]:
"""
获取认证请求头
Args:
auto_refresh: 是否自动刷新过期令牌
Returns:
认证请求头字典
"""
token = self.get_token(auto_refresh)
if token:
return {"Authorization": f"Bearer {token}"}
else:
return {}
def logout(self) -> None:
"""用户登出"""
self.token_info = None
self._clear_cached_token()
self.logger.info("用户已登出")
def is_authenticated(self) -> bool:
"""
检查是否已认证
Returns:
是否已认证
"""
return self.token_info is not None and not self.token_info.is_expired()
def get_token_info(self) -> Optional[TokenInfo]:
"""
获取令牌信息
Returns:
令牌信息对象
"""
return self.token_info
def ensure_authenticated(self, username: str = None, password: str = None) -> bool:
"""
确保已认证,如果未认证则自动登录
Args:
username: 用户名
password: 密码
Returns:
是否认证成功
"""
if self.is_authenticated():
return True
return self.login(username, password)
@@ -0,0 +1,583 @@
"""
错误诊断模块
提供详细的错误分析、分类、归因和恢复建议
"""
import traceback
import re
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import json
class ErrorCategory(Enum):
"""错误分类"""
NETWORK_ERROR = "network_error"
AUTH_ERROR = "auth_error"
VALIDATION_ERROR = "validation_error"
SERVER_ERROR = "server_error"
TIMEOUT_ERROR = "timeout_error"
DATA_ERROR = "data_error"
CONFIG_ERROR = "config_error"
UNKNOWN_ERROR = "unknown_error"
class ErrorSeverity(Enum):
"""错误严重程度"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class ErrorAnalysis:
"""错误分析结果"""
error_type: str
error_category: ErrorCategory
error_severity: ErrorSeverity
error_message: str
stack_trace: str
context: Dict[str, Any]
possible_causes: List[str]
suggested_solutions: List[str]
recovery_actions: List[str]
related_tests: List[str]
class ErrorPattern:
"""错误模式"""
def __init__(self, pattern: str, category: ErrorCategory, severity: ErrorSeverity,
causes: List[str], solutions: List[str], recovery: List[str]):
"""
初始化错误模式
Args:
pattern: 错误模式(正则表达式)
category: 错误分类
severity: 错误严重程度
causes: 可能的原因
solutions: 建议的解决方案
recovery: 恢复操作
"""
self.pattern = re.compile(pattern, re.IGNORECASE)
self.category = category
self.severity = severity
self.causes = causes
self.solutions = solutions
self.recovery = recovery
def matches(self, error_message: str) -> bool:
"""
检查错误消息是否匹配此模式
Args:
error_message: 错误消息
Returns:
是否匹配
"""
return bool(self.pattern.search(error_message))
class ErrorDiagnoser:
"""错误诊断器"""
def __init__(self):
"""初始化错误诊断器"""
self.error_patterns = self._initialize_error_patterns()
self.error_history: List[ErrorAnalysis] = []
def _initialize_error_patterns(self) -> List[ErrorPattern]:
"""
初始化错误模式
Returns:
错误模式列表
"""
patterns = [
# 网络错误
ErrorPattern(
r"connection\s+(refused|timeout|reset|closed)",
ErrorCategory.NETWORK_ERROR,
ErrorSeverity.HIGH,
[
"网络连接被拒绝",
"服务器未启动或不可达",
"防火墙阻止连接",
"网络配置错误"
],
[
"检查服务器是否正在运行",
"验证网络连接",
"检查防火墙设置",
"确认服务器地址和端口正确"
],
[
"重启服务器",
"检查网络配置",
"重试连接"
]
),
ErrorPattern(
r"host\s+(unreachable|not\s+found)",
ErrorCategory.NETWORK_ERROR,
ErrorSeverity.HIGH,
[
"主机地址不存在",
"DNS解析失败",
"网络不可达"
],
[
"验证主机地址正确",
"检查DNS配置",
"确认网络连接"
],
[
"修正主机地址",
"重试连接"
]
),
# 认证错误
ErrorPattern(
r"(unauthorized|authentication\s+failed|invalid\s+(token|credentials))",
ErrorCategory.AUTH_ERROR,
ErrorSeverity.HIGH,
[
"认证令牌无效或过期",
"用户名或密码错误",
"权限不足"
],
[
"检查认证令牌",
"验证用户凭据",
"确认用户权限"
],
[
"重新登录",
"刷新认证令牌",
"联系管理员"
]
),
ErrorPattern(
r"(forbidden|access\s+denied)",
ErrorCategory.AUTH_ERROR,
ErrorSeverity.HIGH,
[
"访问被拒绝",
"权限不足",
"资源不存在"
],
[
"检查用户权限",
"验证资源是否存在",
"确认访问控制配置"
],
[
"联系管理员",
"申请相应权限"
]
),
# 验证错误
ErrorPattern(
r"(validation\s+failed|invalid\s+(parameter|input|data))",
ErrorCategory.VALIDATION_ERROR,
ErrorSeverity.MEDIUM,
[
"输入数据格式错误",
"参数验证失败",
"数据不符合要求"
],
[
"检查输入数据格式",
"验证参数类型和范围",
"参考API文档"
],
[
"修正输入数据",
"调整参数值"
]
),
ErrorPattern(
r"(required\s+field\s+missing|missing\s+required\s+parameter)",
ErrorCategory.VALIDATION_ERROR,
ErrorSeverity.MEDIUM,
[
"缺少必填字段",
"参数不完整"
],
[
"检查请求参数",
"确认必填字段",
"参考API文档"
],
[
"补充必填字段",
"修正请求参数"
]
),
# 服务器错误
ErrorPattern(
r"(internal\s+server\s+error|server\s+error)",
ErrorCategory.SERVER_ERROR,
ErrorSeverity.CRITICAL,
[
"服务器内部错误",
"服务器处理异常",
"服务器配置问题"
],
[
"检查服务器日志",
"验证服务器配置",
"检查数据库连接"
],
[
"联系技术支持",
"检查服务器状态",
"重启服务器"
]
),
ErrorPattern(
r"(database\s+(error|connection\s+failed|timeout))",
ErrorCategory.SERVER_ERROR,
ErrorSeverity.CRITICAL,
[
"数据库连接失败",
"数据库查询错误",
"数据库超时"
],
[
"检查数据库服务状态",
"验证数据库连接配置",
"检查SQL语句"
],
[
"重启数据库服务",
"修正连接配置",
"优化SQL查询"
]
),
# 超时错误
ErrorPattern(
r"(request\s+timeout|operation\s+timed\s+out)",
ErrorCategory.TIMEOUT_ERROR,
ErrorSeverity.MEDIUM,
[
"请求超时",
"服务器响应慢",
"网络延迟高"
],
[
"检查网络连接",
"增加超时时间",
"优化请求"
],
[
"重试请求",
"增加超时配置",
"优化网络环境"
]
),
# 数据错误
ErrorPattern(
r"(data\s+(not\s+found|does\s+not\s+exist)|record\s+not\s+found)",
ErrorCategory.DATA_ERROR,
ErrorSeverity.MEDIUM,
[
"数据不存在",
"记录未找到",
"数据已被删除"
],
[
"验证数据ID",
"检查数据是否存在",
"确认数据状态"
],
[
"使用正确的数据ID",
"重新创建数据"
]
),
ErrorPattern(
r"(duplicate\s+(key|entry|record)|constraint\s+violation)",
ErrorCategory.DATA_ERROR,
ErrorSeverity.MEDIUM,
[
"数据重复",
"唯一约束冲突",
"数据已存在"
],
[
"检查数据是否已存在",
"验证唯一字段",
"使用不同的值"
],
[
"删除重复数据",
"使用不同的值",
"更新现有数据"
]
),
# 配置错误
ErrorPattern(
r"(configuration\s+error|invalid\s+configuration|config\s+not\s+found)",
ErrorCategory.CONFIG_ERROR,
ErrorSeverity.HIGH,
[
"配置错误",
"配置文件缺失",
"配置参数无效"
],
[
"检查配置文件",
"验证配置参数",
"参考配置文档"
],
[
"修正配置文件",
"重置配置",
"重新加载配置"
]
)
]
return patterns
def diagnose_error(self, exception: Exception, context: Dict[str, Any] = None) -> ErrorAnalysis:
"""
诊断错误
Args:
exception: 异常对象
context: 上下文信息
Returns:
错误分析结果
"""
error_message = str(exception)
stack_trace = traceback.format_exc()
# 查找匹配的错误模式
matched_pattern = None
for pattern in self.error_patterns:
if pattern.matches(error_message):
matched_pattern = pattern
break
# 如果没有匹配的模式,使用默认分类
if matched_pattern is None:
matched_pattern = ErrorPattern(
r".*",
ErrorCategory.UNKNOWN_ERROR,
ErrorSeverity.MEDIUM,
["未知错误"],
["检查日志", "联系技术支持"],
["重试操作", "联系支持"]
)
# 分析上下文
context = context or {}
context.update({
"error_type": type(exception).__name__,
"error_message": error_message,
"timestamp": context.get("timestamp"),
"test_name": context.get("test_name"),
"url": context.get("url"),
"method": context.get("method")
})
# 创建错误分析结果
analysis = ErrorAnalysis(
error_type=type(exception).__name__,
error_category=matched_pattern.category,
error_severity=matched_pattern.severity,
error_message=error_message,
stack_trace=stack_trace,
context=context,
possible_causes=matched_pattern.causes,
suggested_solutions=matched_pattern.solutions,
recovery_actions=matched_pattern.recovery,
related_tests=self._find_related_tests(context)
)
# 记录错误历史
self.error_history.append(analysis)
return analysis
def _find_related_tests(self, context: Dict[str, Any]) -> List[str]:
"""
查找相关测试
Args:
context: 上下文信息
Returns:
相关测试列表
"""
related_tests = []
url = context.get("url", "")
method = context.get("method", "")
# 根据URL和方法查找相关测试
if "/auth/login" in url:
related_tests.extend(["用户登录", "认证测试", "权限测试"])
elif "/user/" in url:
related_tests.extend(["用户管理测试", "用户列表测试", "用户创建测试"])
elif "/role/" in url:
related_tests.extend(["角色管理测试", "角色列表测试", "角色创建测试"])
elif "/menu/" in url:
related_tests.extend(["菜单管理测试", "菜单列表测试"])
return related_tests
def get_error_statistics(self) -> Dict[str, Any]:
"""
获取错误统计信息
Returns:
错误统计信息
"""
if not self.error_history:
return {}
# 按分类统计
category_stats = {}
for analysis in self.error_history:
category = analysis.error_category.value
if category not in category_stats:
category_stats[category] = 0
category_stats[category] += 1
# 按严重程度统计
severity_stats = {}
for analysis in self.error_history:
severity = analysis.error_severity.value
if severity not in severity_stats:
severity_stats[severity] = 0
severity_stats[severity] += 1
# 按错误类型统计
type_stats = {}
for analysis in self.error_history:
error_type = analysis.error_type
if error_type not in type_stats:
type_stats[error_type] = 0
type_stats[error_type] += 1
return {
"total_errors": len(self.error_history),
"by_category": category_stats,
"by_severity": severity_stats,
"by_type": type_stats,
"most_common_errors": self._get_most_common_errors(5)
}
def _get_most_common_errors(self, limit: int = 5) -> List[Dict[str, Any]]:
"""
获取最常见的错误
Args:
limit: 返回数量限制
Returns:
最常见错误列表
"""
error_counts = {}
for analysis in self.error_history:
error_key = f"{analysis.error_type}: {analysis.error_message[:50]}"
if error_key not in error_counts:
error_counts[error_key] = 0
error_counts[error_key] += 1
sorted_errors = sorted(error_counts.items(), key=lambda x: x[1], reverse=True)
return [
{
"error": error[0],
"count": error[1]
}
for error in sorted_errors[:limit]
]
def generate_error_report(self, analysis: ErrorAnalysis) -> str:
"""
生成错误报告
Args:
analysis: 错误分析结果
Returns:
错误报告(Markdown格式)
"""
report = f"""# 错误诊断报告
## 错误信息
- **错误类型**: {analysis.error_type}
- **错误分类**: {analysis.error_category.value}
- **严重程度**: {analysis.error_severity.value}
- **错误消息**: {analysis.error_message}
## 错误堆栈
```
{analysis.stack_trace}
```
## 上下文信息
{self._format_context(analysis.context)}
## 可能的原因
{self._format_list(analysis.possible_causes)}
## 建议的解决方案
{self._format_list(analysis.suggested_solutions)}
## 恢复操作
{self._format_list(analysis.recovery_actions)}
## 相关测试
{self._format_list(analysis.related_tests)}
---
*报告生成时间: {self._get_current_timestamp()}*
"""
return report
def _format_context(self, context: Dict[str, Any]) -> str:
"""格式化上下文信息"""
lines = []
for key, value in context.items():
if value is not None:
lines.append(f"- **{key}**: {value}")
return "\n".join(lines)
def _format_list(self, items: List[str]) -> str:
"""格式化列表"""
return "\n".join([f"- {item}" for item in items])
def _get_current_timestamp(self) -> str:
"""获取当前时间戳"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def clear_history(self) -> None:
"""清空错误历史"""
self.error_history.clear()
@@ -0,0 +1,380 @@
"""
测试执行性能优化模块
支持并行执行、结果缓存、连接池等功能
"""
import asyncio
import time
from typing import List, Dict, Any, Optional, Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from functools import lru_cache
from dataclasses import dataclass
import hashlib
import json
from .api_tester import APITester, TestResult
from ..utils.logger import TestLogger
@dataclass
class CachedResult:
"""缓存结果"""
key: str
result: TestResult
timestamp: float
ttl: int = 3600 # 缓存有效期(秒)
def is_expired(self) -> bool:
"""检查缓存是否过期"""
return time.time() - self.timestamp > self.ttl
class ResultCache:
"""测试结果缓存"""
def __init__(self, default_ttl: int = 3600):
"""
初始化缓存
Args:
default_ttl: 默认缓存有效期(秒)
"""
self.cache: Dict[str, CachedResult] = {}
self.default_ttl = default_ttl
def _generate_key(self, method: str, endpoint: str, data: Dict[str, Any] = None,
params: Dict[str, Any] = None) -> str:
"""
生成缓存键
Args:
method: HTTP方法
endpoint: API端点
data: 请求体数据
params: URL参数
Returns:
缓存键
"""
key_data = {
"method": method,
"endpoint": endpoint,
"data": data or {},
"params": params or {}
}
key_str = json.dumps(key_data, sort_keys=True)
return hashlib.md5(key_str.encode()).hexdigest()
def get(self, method: str, endpoint: str, data: Dict[str, Any] = None,
params: Dict[str, Any] = None) -> Optional[TestResult]:
"""
获取缓存结果
Args:
method: HTTP方法
endpoint: API端点
data: 请求体数据
params: URL参数
Returns:
缓存的测试结果
"""
key = self._generate_key(method, endpoint, data, params)
if key in self.cache:
cached = self.cache[key]
if not cached.is_expired():
return cached.result
else:
del self.cache[key]
return None
def set(self, method: str, endpoint: str, result: TestResult,
data: Dict[str, Any] = None, params: Dict[str, Any] = None,
ttl: int = None) -> None:
"""
设置缓存结果
Args:
method: HTTP方法
endpoint: API端点
result: 测试结果
data: 请求体数据
params: URL参数
ttl: 缓存有效期(秒)
"""
key = self._generate_key(method, endpoint, data, params)
cached = CachedResult(
key=key,
result=result,
timestamp=time.time(),
ttl=ttl or self.default_ttl
)
self.cache[key] = cached
def clear(self) -> None:
"""清空缓存"""
self.cache.clear()
def cleanup_expired(self) -> None:
"""清理过期缓存"""
expired_keys = [key for key, cached in self.cache.items() if cached.is_expired()]
for key in expired_keys:
del self.cache[key]
class ParallelTestExecutor:
"""并行测试执行器"""
def __init__(self, max_workers: int = 4, logger: TestLogger = None):
"""
初始化并行执行器
Args:
max_workers: 最大工作线程数
logger: 日志记录器
"""
self.max_workers = max_workers
self.logger = logger
def execute_tests(self, test_functions: List[Callable],
use_cache: bool = True, cache: ResultCache = None) -> List[TestResult]:
"""
并行执行测试
Args:
test_functions: 测试函数列表
use_cache: 是否使用缓存
cache: 缓存实例
Returns:
测试结果列表
"""
results = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# 提交所有测试任务
future_to_test = {
executor.submit(self._execute_single_test, func, use_cache, cache): func
for func in test_functions
}
# 收集结果
for future in as_completed(future_to_test):
test_func = future_to_test[future]
try:
result = future.result()
results.append(result)
except Exception as e:
if self.logger:
self.logger.error(f"测试执行失败: {str(e)}")
results.append(TestResult(
passed=False,
test_name=test_func.__name__,
error_message=str(e)
))
return results
def _execute_single_test(self, test_func: Callable, use_cache: bool,
cache: ResultCache) -> TestResult:
"""
执行单个测试
Args:
test_func: 测试函数
use_cache: 是否使用缓存
cache: 缓存实例
Returns:
测试结果
"""
try:
# 尝试从缓存获取结果
if use_cache and cache:
# 这里简化实现,实际应该根据测试函数的参数生成缓存键
pass
# 执行测试
return test_func()
except Exception as e:
raise e
def execute_tests_async(self, test_functions: List[Callable]) -> List[TestResult]:
"""
异步并行执行测试
Args:
test_functions: 测试函数列表
Returns:
测试结果列表
"""
async def run_all():
tasks = [self._execute_single_test_async(func) for func in test_functions]
return await asyncio.gather(*tasks, return_exceptions=True)
results = asyncio.run(run_all())
# 处理异常结果
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append(TestResult(
passed=False,
test_name=test_functions[i].__name__,
error_message=str(result)
))
else:
processed_results.append(result)
return processed_results
async def _execute_single_test_async(self, test_func: Callable) -> TestResult:
"""
异步执行单个测试
Args:
test_func: 测试函数
Returns:
测试结果
"""
# 这里简化实现,实际应该使用异步HTTP客户端
return test_func()
class PerformanceOptimizer:
"""性能优化器"""
def __init__(self, logger: TestLogger = None):
"""
初始化性能优化器
Args:
logger: 日志记录器
"""
self.logger = logger
self.cache = ResultCache()
self.executor = ParallelTestExecutor(logger=logger)
def optimize_test_execution(self, test_functions: List[Callable],
parallel: bool = True, use_cache: bool = True) -> List[TestResult]:
"""
优化测试执行
Args:
test_functions: 测试函数列表
parallel: 是否并行执行
use_cache: 是否使用缓存
Returns:
测试结果列表
"""
start_time = time.time()
if self.logger:
self.logger.info(f"开始优化测试执行: {len(test_functions)}个测试用例")
self.logger.info(f"并行执行: {parallel}, 使用缓存: {use_cache}")
# 清理过期缓存
self.cache.cleanup_expired()
# 执行测试
if parallel:
results = self.executor.execute_tests(test_functions, use_cache, self.cache)
else:
results = []
for func in test_functions:
try:
result = func()
results.append(result)
except Exception as e:
if self.logger:
self.logger.error(f"测试执行失败: {str(e)}")
results.append(TestResult(
passed=False,
test_name=func.__name__,
error_message=str(e)
))
execution_time = time.time() - start_time
if self.logger:
self.logger.info(f"测试执行完成: {execution_time:.2f}")
self.logger.info(f"平均每个测试: {execution_time/len(test_functions):.2f}")
return results
def get_cache_stats(self) -> Dict[str, Any]:
"""
获取缓存统计信息
Returns:
缓存统计信息
"""
return {
"total_entries": len(self.cache.cache),
"expired_entries": sum(1 for cached in self.cache.cache.values() if cached.is_expired()),
"valid_entries": sum(1 for cached in self.cache.cache.values() if not cached.is_expired())
}
def clear_cache(self) -> None:
"""清空缓存"""
self.cache.clear()
if self.logger:
self.logger.info("缓存已清空")
class ConnectionPool:
"""连接池"""
def __init__(self, max_connections: int = 10):
"""
初始化连接池
Args:
max_connections: 最大连接数
"""
self.max_connections = max_connections
self.connections = []
def get_connection(self) -> APITester:
"""
获取连接
Returns:
API测试器实例
"""
if self.connections:
return self.connections.pop()
return APITester()
def return_connection(self, connection: APITester) -> None:
"""
归还连接
Args:
connection: API测试器实例
"""
if len(self.connections) < self.max_connections:
self.connections.append(connection)
else:
connection.close()
def close_all(self) -> None:
"""关闭所有连接"""
for connection in self.connections:
connection.close()
self.connections.clear()
@@ -0,0 +1,300 @@
"""
测试验证引擎模块
"""
from typing import Dict, Any, Tuple, Optional
import re
import json
from datetime import datetime
class ValidationEngine:
"""测试验证引擎"""
@staticmethod
def validate_status_code(expected: int, actual: int) -> Tuple[bool, str]:
"""
验证HTTP状态码
Args:
expected: 期望的状态码
actual: 实际的状态码
Returns:
(是否通过, 错误消息)
"""
if expected == actual:
return True, ""
return False, f"期望状态码{expected},实际{actual}"
@staticmethod
def validate_response_body(expected: Dict[str, Any], actual: Dict[str, Any]) -> Tuple[bool, str]:
"""
验证响应体
Args:
expected: 期望的响应体
actual: 实际的响应体
Returns:
(是否通过, 错误消息)
"""
if expected == actual:
return True, ""
# 找出差异
differences = ValidationEngine._find_differences(expected, actual)
return False, f"响应体不匹配: {differences}"
@staticmethod
def validate_contains(expected_value: Any, actual_value: Any) -> Tuple[bool, str]:
"""
验证包含关系
Args:
expected_value: 期望包含的值
actual_value: 实际值
Returns:
(是否通过, 错误消息)
"""
if isinstance(actual_value, (str, list, dict)):
if expected_value in actual_value:
return True, ""
return False, f"期望值'{expected_value}'不在实际值中"
if expected_value == actual_value:
return True, ""
return False, f"期望包含'{expected_value}',实际为'{actual_value}'"
@staticmethod
def validate_equals(expected_value: Any, actual_value: Any) -> Tuple[bool, str]:
"""
验证相等关系
Args:
expected_value: 期望的值
actual_value: 实际的值
Returns:
(是否通过, 错误消息)
"""
if expected_value == actual_value:
return True, ""
return False, f"期望值'{expected_value}',实际值'{actual_value}'"
@staticmethod
def validate_json_path(path: str, expected_value: Any, actual_data: Dict[str, Any]) -> Tuple[bool, str]:
"""
验证JSON路径
Args:
path: JSON路径(如"data.user.id"
expected_value: 期望的值
actual_data: 实际的数据
Returns:
(是否通过, 错误消息)
"""
try:
keys = path.split('.')
value = actual_data
for key in keys:
if isinstance(value, dict):
value = value.get(key)
elif isinstance(value, list) and key.isdigit():
value = value[int(key)]
else:
return False, f"路径'{path}'不存在"
if value == expected_value:
return True, ""
return False, f"路径'{path}'期望值'{expected_value}',实际值'{value}'"
except Exception as e:
return False, f"验证JSON路径失败: {str(e)}"
@staticmethod
def validate_regex(pattern: str, actual_value: str) -> Tuple[bool, str]:
"""
验证正则表达式
Args:
pattern: 正则表达式模式
actual_value: 实际的值
Returns:
(是否通过, 错误消息)
"""
try:
if re.match(pattern, str(actual_value)):
return True, ""
return False, f"'{actual_value}'不匹配正则表达式'{pattern}'"
except Exception as e:
return False, f"正则表达式验证失败: {str(e)}"
@staticmethod
def validate_header(expected_header: Dict[str, str], actual_headers: Dict[str, str]) -> Tuple[bool, str]:
"""
验证响应头
Args:
expected_header: 期望的响应头
actual_headers: 实际的响应头
Returns:
(是否通过, 错误消息)
"""
for key, expected_value in expected_header.items():
actual_value = actual_headers.get(key)
if actual_value is None:
return False, f"缺少响应头: {key}"
if actual_value != expected_value:
return False, f"响应头'{key}'期望值'{expected_value}',实际值'{actual_value}'"
return True, ""
@staticmethod
def validate_response_time(expected_max_time: float, actual_time: float) -> Tuple[bool, str]:
"""
验证响应时间
Args:
expected_max_time: 期望的最大响应时间(毫秒)
actual_time: 实际的响应时间(毫秒)
Returns:
(是否通过, 错误消息)
"""
if actual_time <= expected_max_time:
return True, ""
return False, f"响应时间{actual_time}ms超过期望最大值{expected_max_time}ms"
@staticmethod
def validate_schema(expected_schema: Dict[str, Any], actual_data: Dict[str, Any]) -> Tuple[bool, str]:
"""
验证数据结构
Args:
expected_schema: 期望的结构(简化版)
actual_data: 实际的数据
Returns:
(是否通过, 错误消息)
"""
for key, expected_type in expected_schema.items():
if key not in actual_data:
return False, f"缺少字段: {key}"
actual_value = actual_data[key]
if expected_type == "string" and not isinstance(actual_value, str):
return False, f"字段'{key}'期望类型string,实际类型{type(actual_value).__name__}"
elif expected_type == "number" and not isinstance(actual_value, (int, float)):
return False, f"字段'{key}'期望类型number,实际类型{type(actual_value).__name__}"
elif expected_type == "boolean" and not isinstance(actual_value, bool):
return False, f"字段'{key}'期望类型boolean,实际类型{type(actual_value).__name__}"
elif expected_type == "array" and not isinstance(actual_value, list):
return False, f"字段'{key}'期望类型array,实际类型{type(actual_value).__name__}"
elif expected_type == "object" and not isinstance(actual_value, dict):
return False, f"字段'{key}'期望类型object,实际类型{type(actual_value).__name__}"
return True, ""
@staticmethod
def _find_differences(expected: Any, actual: Any, path: str = "") -> str:
"""
找出两个值之间的差异
Args:
expected: 期望的值
actual: 实际的值
path: 当前路径
Returns:
差异描述
"""
if expected == actual:
return ""
if isinstance(expected, dict) and isinstance(actual, dict):
differences = []
all_keys = set(expected.keys()) | set(actual.keys())
for key in all_keys:
new_path = f"{path}.{key}" if path else key
if key not in expected:
differences.append(f"{new_path}: 实际存在但期望不存在")
elif key not in actual:
differences.append(f"{new_path}: 期望存在但实际不存在")
else:
diff = ValidationEngine._find_differences(expected[key], actual[key], new_path)
if diff:
differences.append(diff)
return "; ".join(differences)
elif isinstance(expected, list) and isinstance(actual, list):
if len(expected) != len(actual):
return f"{path}: 长度不匹配(期望{len(expected)},实际{len(actual)}"
differences = []
for i, (exp_item, act_item) in enumerate(zip(expected, actual)):
new_path = f"{path}[{i}]"
diff = ValidationEngine._find_differences(exp_item, act_item, new_path)
if diff:
differences.append(diff)
return "; ".join(differences)
else:
return f"{path}: 期望'{expected}',实际'{actual}'"
class ValidationRule:
"""验证规则"""
def __init__(self, rule_type: str, expected_value: Any = None, path: str = None):
"""
初始化验证规则
Args:
rule_type: 规则类型(status_code, contains, equals, json_path, regex, header, response_time, schema
expected_value: 期望值
path: JSON路径(仅用于json_path规则)
"""
self.rule_type = rule_type
self.expected_value = expected_value
self.path = path
def validate(self, actual_value: Any = None, actual_data: Dict[str, Any] = None) -> Tuple[bool, str]:
"""
执行验证
Args:
actual_value: 实际值
actual_data: 实际数据(用于JSON路径验证)
Returns:
(是否通过, 错误消息)
"""
if self.rule_type == "status_code":
return ValidationEngine.validate_status_code(self.expected_value, actual_value)
elif self.rule_type == "contains":
return ValidationEngine.validate_contains(self.expected_value, actual_value)
elif self.rule_type == "equals":
return ValidationEngine.validate_equals(self.expected_value, actual_value)
elif self.rule_type == "json_path":
return ValidationEngine.validate_json_path(self.path, self.expected_value, actual_data)
elif self.rule_type == "regex":
return ValidationEngine.validate_regex(self.expected_value, actual_value)
elif self.rule_type == "header":
return ValidationEngine.validate_header(self.expected_value, actual_value)
elif self.rule_type == "response_time":
return ValidationEngine.validate_response_time(self.expected_value, actual_value)
elif self.rule_type == "schema":
return ValidationEngine.validate_schema(self.expected_value, actual_data)
else:
return False, f"未知的验证规则类型: {self.rule_type}"
@@ -0,0 +1 @@
"""数据模块"""
@@ -0,0 +1,387 @@
"""
测试数据生成器模块
"""
import random
import string
from datetime import datetime, timedelta
from typing import Dict, Any, List
import hashlib
class TestDataGenerator:
"""测试数据生成器"""
@staticmethod
def random_string(length: int = 10) -> str:
"""
生成随机字符串
Args:
length: 字符串长度
Returns:
随机字符串
"""
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
@staticmethod
def random_email() -> str:
"""
生成随机邮箱
Returns:
随机邮箱
"""
username = TestDataGenerator.random_string(8)
domains = ['gmail.com', 'yahoo.com', 'hotmail.com', 'outlook.com', 'example.com']
domain = random.choice(domains)
return f"{username}@{domain}"
@staticmethod
def random_phone() -> str:
"""
生成随机手机号
Returns:
随机手机号
"""
return f"1{random.choice([3, 5, 7, 8, 9])}{random.randint(100000000, 999999999)}"
@staticmethod
def random_password(length: int = 12) -> str:
"""
生成随机密码
Args:
length: 密码长度
Returns:
随机密码
"""
chars = string.ascii_letters + string.digits + "!@#$%^&*"
return ''.join(random.choices(chars, k=length))
@staticmethod
def random_int(min_val: int = 0, max_val: int = 100) -> int:
"""
生成随机整数
Args:
min_val: 最小值
max_val: 最大值
Returns:
随机整数
"""
return random.randint(min_val, max_val)
@staticmethod
def random_float(min_val: float = 0.0, max_val: float = 100.0, decimals: int = 2) -> float:
"""
生成随机浮点数
Args:
min_val: 最小值
max_val: 最大值
decimals: 小数位数
Returns:
随机浮点数
"""
return round(random.uniform(min_val, max_val), decimals)
@staticmethod
def random_date(start_year: int = 2020, end_year: int = 2025) -> datetime:
"""
生成随机日期
Args:
start_year: 起始年份
end_year: 结束年份
Returns:
随机日期
"""
start_date = datetime(start_year, 1, 1)
end_date = datetime(end_year, 12, 31)
delta = end_date - start_date
random_days = random.randint(0, delta.days)
return start_date + timedelta(days=random_days)
@staticmethod
def random_boolean() -> bool:
"""
生成随机布尔值
Returns:
随机布尔值
"""
return random.choice([True, False])
@staticmethod
def random_choice(choices: List[Any]) -> Any:
"""
从列表中随机选择一个值
Args:
choices: 选项列表
Returns:
随机选择的值
"""
return random.choice(choices)
@staticmethod
def generate_user_data(overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成用户数据
Args:
overrides: 覆盖的默认值
Returns:
用户数据
"""
username = TestDataGenerator.random_string(8)
data = {
"username": username,
"password": "Test@123",
"email": f"{username}@example.com",
"phone": TestDataGenerator.random_phone(),
"nickname": f"用户{username}",
"status": 1,
"roleIds": []
}
if overrides:
data.update(overrides)
return data
@staticmethod
def generate_admin_data(overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成管理员数据
Args:
overrides: 覆盖的默认值
Returns:
管理员数据
"""
data = TestDataGenerator.generate_user_data({
"username": "admin",
"password": "admin123",
"email": "admin@example.com",
"nickname": "管理员",
"roleIds": [1]
})
if overrides:
data.update(overrides)
return data
@staticmethod
def generate_role_data(overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成角色数据
Args:
overrides: 覆盖的默认值
Returns:
角色数据
"""
role_code = f"role_{TestDataGenerator.random_string(6).lower()}"
data = {
"roleName": f"测试角色{role_code}",
"roleCode": role_code,
"description": f"这是一个测试角色{role_code}",
"status": 1,
"sort": TestDataGenerator.random_int(1, 100)
}
if overrides:
data.update(overrides)
return data
@staticmethod
def generate_menu_data(overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成菜单数据
Args:
overrides: 覆盖的默认值
Returns:
菜单数据
"""
menu_code = f"menu_{TestDataGenerator.random_string(6).lower()}"
data = {
"menuName": f"测试菜单{menu_code}",
"menuCode": menu_code,
"menuType": 1,
"path": f"/{menu_code}",
"component": f"{menu_code}/index",
"icon": "SettingOutlined",
"parentId": 0,
"sort": TestDataGenerator.random_int(1, 100),
"status": 0,
"visible": 1
}
if overrides:
data.update(overrides)
return data
@staticmethod
def generate_permission_data(overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成权限数据
Args:
overrides: 覆盖的默认值
Returns:
权限数据
"""
permission_code = f"perm:{TestDataGenerator.random_string(6).lower()}"
data = {
"permissionName": f"测试权限{permission_code}",
"permissionCode": permission_code,
"permissionType": "button",
"description": f"这是一个测试权限{permission_code}",
"status": 1
}
if overrides:
data.update(overrides)
return data
@staticmethod
def generate_user_list(count: int, overrides: Dict[str, Any] = None) -> List[Dict[str, Any]]:
"""
批量生成用户数据
Args:
count: 生成数量
overrides: 覆盖的默认值
Returns:
用户数据列表
"""
return [TestDataGenerator.generate_user_data(overrides) for _ in range(count)]
@staticmethod
def generate_role_list(count: int, overrides: Dict[str, Any] = None) -> List[Dict[str, Any]]:
"""
批量生成角色数据
Args:
count: 生成数量
overrides: 覆盖的默认值
Returns:
角色数据列表
"""
return [TestDataGenerator.generate_role_data(overrides) for _ in range(count)]
@staticmethod
def generate_menu_list(count: int, overrides: Dict[str, Any] = None) -> List[Dict[str, Any]]:
"""
批量生成菜单数据
Args:
count: 生成数量
overrides: 覆盖的默认值
Returns:
菜单数据列表
"""
return [TestDataGenerator.generate_menu_data(overrides) for _ in range(count)]
@staticmethod
def generate_pagination_data(data: List[Any], page: int = 1, page_size: int = 10) -> Dict[str, Any]:
"""
生成分页数据
Args:
data: 原始数据列表
page: 当前页码
page_size: 每页数量
Returns:
分页数据
"""
total = len(data)
total_pages = (total + page_size - 1) // page_size
start_index = (page - 1) * page_size
end_index = start_index + page_size
page_data = data[start_index:end_index]
return {
"records": page_data,
"total": total,
"page": page,
"pageSize": page_size,
"totalPages": total_pages
}
@staticmethod
def generate_search_query(keyword: str = None) -> Dict[str, Any]:
"""
生成搜索查询
Args:
keyword: 搜索关键词
Returns:
搜索查询参数
"""
query = {
"keyword": keyword or TestDataGenerator.random_string(5),
"page": TestDataGenerator.random_int(1, 10),
"pageSize": random.choice([10, 20, 50, 100]),
"sortField": "createTime",
"sortOrder": random.choice(["asc", "desc"])
}
return query
@staticmethod
def generate_test_credentials() -> Dict[str, Dict[str, str]]:
"""
生成测试凭据
Returns:
测试凭据字典
"""
return {
"admin": {
"username": "admin",
"password": "admin123"
},
"user": {
"username": "testuser",
"password": "test123"
}
}
@staticmethod
def generate_hash(data: str) -> str:
"""
生成数据的哈希值
Args:
data: 原始数据
Returns:
哈希值
"""
return hashlib.md5(data.encode()).hexdigest()
@@ -0,0 +1,313 @@
{
"version": "1.0",
"description": "测试数据标准格式",
"test_suites": [
{
"name": "用户管理测试",
"description": "用户管理模块的测试用例",
"test_cases": [
{
"id": "user_login_001",
"name": "用户登录",
"description": "验证用户登录功能",
"priority": "high",
"enabled": true,
"endpoint": "/auth/login",
"method": "POST",
"data": {
"username": "admin",
"password": "admin123"
},
"expected": {
"status_code": 200,
"response": {
"code": 200,
"data": {
"token": "${token}",
"user": {
"username": "admin"
}
}
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "data.token",
"expected_value": "${token}"
}
]
},
{
"id": "user_get_list_001",
"name": "获取用户列表",
"description": "验证获取用户列表功能",
"priority": "high",
"enabled": true,
"endpoint": "/user/list",
"method": "GET",
"params": {
"page": 1,
"pageSize": 10
},
"expected": {
"status_code": 200,
"response": {
"code": 200,
"data": {
"records": [],
"total": 0
}
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
}
],
"dependencies": ["user_login_001"]
},
{
"id": "user_create_001",
"name": "创建用户",
"description": "验证创建用户功能",
"priority": "high",
"enabled": true,
"endpoint": "/user/create",
"method": "POST",
"data": {
"username": "testuser001",
"password": "Test@123",
"email": "testuser001@example.com",
"phone": "13800138000",
"nickname": "测试用户001",
"status": 1,
"roleIds": []
},
"expected": {
"status_code": 200,
"response": {
"code": 200,
"data": {
"id": "${user_id}",
"username": "testuser001"
}
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "data.username",
"expected_value": "testuser001"
}
],
"dependencies": ["user_login_001"]
},
{
"id": "user_update_001",
"name": "更新用户",
"description": "验证更新用户功能",
"priority": "medium",
"enabled": true,
"endpoint": "/user/update",
"method": "PUT",
"data": {
"id": "${user_id}",
"username": "testuser001",
"nickname": "测试用户001-更新",
"status": 1
},
"expected": {
"status_code": 200,
"response": {
"code": 200
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
}
],
"dependencies": ["user_create_001"]
},
{
"id": "user_delete_001",
"name": "删除用户",
"description": "验证删除用户功能",
"priority": "medium",
"enabled": true,
"endpoint": "/user/delete/${user_id}",
"method": "DELETE",
"expected": {
"status_code": 200,
"response": {
"code": 200
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
}
],
"dependencies": ["user_create_001"]
}
]
},
{
"name": "角色管理测试",
"description": "角色管理模块的测试用例",
"test_cases": [
{
"id": "role_get_list_001",
"name": "获取角色列表",
"description": "验证获取角色列表功能",
"priority": "high",
"enabled": true,
"endpoint": "/role/list",
"method": "GET",
"params": {
"page": 1,
"pageSize": 10
},
"expected": {
"status_code": 200,
"response": {
"code": 200,
"data": {
"records": [],
"total": 0
}
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
}
],
"dependencies": ["user_login_001"]
},
{
"id": "role_create_001",
"name": "创建角色",
"description": "验证创建角色功能",
"priority": "high",
"enabled": true,
"endpoint": "/role/create",
"method": "POST",
"data": {
"roleName": "测试角色001",
"roleCode": "test_role_001",
"description": "这是一个测试角色",
"status": 1,
"sort": 1
},
"expected": {
"status_code": 200,
"response": {
"code": 200,
"data": {
"id": "${role_id}",
"roleName": "测试角色001"
}
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "data.roleName",
"expected_value": "测试角色001"
}
],
"dependencies": ["user_login_001"]
}
]
},
{
"name": "菜单管理测试",
"description": "菜单管理模块的测试用例",
"test_cases": [
{
"id": "menu_get_list_001",
"name": "获取菜单列表",
"description": "验证获取菜单列表功能",
"priority": "high",
"enabled": true,
"endpoint": "/menu/list",
"method": "GET",
"expected": {
"status_code": 200,
"response": {
"code": 200,
"data": {
"list": []
}
}
},
"validations": [
{
"rule_type": "status_code",
"expected_value": 200
},
{
"rule_type": "json_path",
"path": "code",
"expected_value": 200
}
],
"dependencies": ["user_login_001"]
}
]
}
]
}
@@ -0,0 +1,655 @@
"""
统一测试工具接口标准
定义所有测试工具应遵循的统一接口规范
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
class TestStatus(Enum):
"""测试状态枚举"""
PASSED = "passed"
FAILED = "failed"
SKIPPED = "skipped"
RUNNING = "running"
class LogLevel(Enum):
"""日志级别枚举"""
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
@dataclass
class TestConfig:
"""测试配置数据类"""
name: str
version: str
description: str
# API配置
api_base_url: str
api_timeout: int
api_max_retries: int
# 认证配置
auth_enabled: bool
auth_login_endpoint: str
auth_username: str
auth_password: str
# 测试配置
test_parallel: bool
test_retry_count: int
# 报告配置
report_output_dir: str
report_formats: List[str]
report_include_details: bool
# 日志配置
logging_level: str
logging_file: str
logging_console: bool
@dataclass
class TestResult:
"""测试结果数据类"""
test_name: str
test_type: str
status: TestStatus
url: str
method: str
status_code: int
response_time: float
error_message: str
request_data: Dict[str, Any]
response_data: Dict[str, Any]
timestamp: str
stack_trace: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
"test_name": self.test_name,
"test_type": self.test_type,
"status": self.status.value,
"url": self.url,
"method": self.method,
"status_code": self.status_code,
"response_time": self.response_time,
"error_message": self.error_message,
"request_data": self.request_data,
"response_data": self.response_data,
"timestamp": self.timestamp,
"stack_trace": self.stack_trace
}
@dataclass
class TestSummary:
"""测试摘要数据类"""
total: int
passed: int
failed: int
skipped: int
pass_rate: float
total_time: float
start_time: str
end_time: str
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return {
"total": self.total,
"passed": self.passed,
"failed": self.failed,
"skipped": self.skipped,
"pass_rate": self.pass_rate,
"total_time": self.total_time,
"start_time": self.start_time,
"end_time": self.end_time
}
@dataclass
class ValidationRule:
"""验证规则数据类"""
rule_type: str
expected_value: Any
path: Optional[str] = None
description: Optional[str] = None
class ITestTool(ABC):
"""测试工具接口"""
@abstractmethod
def get_name(self) -> str:
"""
获取测试工具名称
Returns:
工具名称
"""
pass
@abstractmethod
def get_version(self) -> str:
"""
获取测试工具版本
Returns:
工具版本
"""
pass
@abstractmethod
def load_config(self, config_path: str = None) -> None:
"""
加载配置
Args:
config_path: 配置文件路径
"""
pass
@abstractmethod
def get_config(self) -> TestConfig:
"""
获取配置
Returns:
配置对象
"""
pass
@abstractmethod
def run_tests(self, test_cases: List[Any] = None) -> Tuple[List[TestResult], TestSummary]:
"""
运行测试
Args:
test_cases: 测试用例列表
Returns:
(测试结果列表, 测试摘要)
"""
pass
@abstractmethod
def generate_report(self, results: List[TestResult], summary: TestSummary, format: str = "json") -> str:
"""
生成测试报告
Args:
results: 测试结果列表
summary: 测试摘要
format: 报告格式(json, html, pdf
Returns:
报告文件路径
"""
pass
@abstractmethod
def set_log_level(self, level: LogLevel) -> None:
"""
设置日志级别
Args:
level: 日志级别
"""
pass
@abstractmethod
def get_logs(self) -> List[str]:
"""
获取日志
Returns:
日志列表
"""
pass
@abstractmethod
def cleanup(self) -> None:
"""清理资源"""
pass
class IAPITester(ITestTool):
"""API测试工具接口"""
@abstractmethod
def login(self, username: str = None, password: str = None) -> bool:
"""
用户登录
Args:
username: 用户名
password: 密码
Returns:
是否登录成功
"""
pass
@abstractmethod
def request(
self,
method: str,
endpoint: str,
data: Dict[str, Any] = None,
params: Dict[str, Any] = None,
headers: Dict[str, str] = None,
expected_status: int = 200
) -> TestResult:
"""
发送HTTP请求
Args:
method: HTTP方法
endpoint: API端点
data: 请求体数据
params: URL参数
headers: 请求头
expected_status: 期望的状态码
Returns:
测试结果
"""
pass
@abstractmethod
def validate(self, result: TestResult, rules: List[ValidationRule]) -> TestResult:
"""
验证测试结果
Args:
result: 测试结果
rules: 验证规则列表
Returns:
验证后的测试结果
"""
pass
class IE2ETester(ITestTool):
"""E2E测试工具接口"""
@abstractmethod
def navigate(self, url: str) -> None:
"""
导航到指定URL
Args:
url: 目标URL
"""
pass
@abstractmethod
def click(self, selector: str) -> None:
"""
点击元素
Args:
selector: 元素选择器
"""
pass
@abstractmethod
def fill(self, selector: str, value: str) -> None:
"""
填写表单字段
Args:
selector: 元素选择器
value: 填充值
"""
pass
@abstractmethod
def wait_for_element(self, selector: str, timeout: int = 5000) -> None:
"""
等待元素出现
Args:
selector: 元素选择器
timeout: 超时时间(毫秒)
"""
pass
@abstractmethod
def take_screenshot(self, name: str) -> str:
"""
截取屏幕截图
Args:
name: 截图名称
Returns:
截图文件路径
"""
pass
@abstractmethod
def expect_visible(self, selector: str, timeout: int = 5000) -> None:
"""
期望元素可见
Args:
selector: 元素选择器
timeout: 超时时间(毫秒)
"""
pass
@abstractmethod
def expect_text(self, selector: str, expected_text: str, timeout: int = 5000) -> None:
"""
期望元素包含指定文本
Args:
selector: 元素选择器
expected_text: 期望的文本
timeout: 超时时间(毫秒)
"""
pass
class IValidationEngine(ABC):
"""验证引擎接口"""
@abstractmethod
def validate_status_code(self, expected: int, actual: int) -> Tuple[bool, str]:
"""
验证HTTP状态码
Args:
expected: 期望的状态码
actual: 实际的状态码
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_response_body(self, expected: Dict[str, Any], actual: Dict[str, Any]) -> Tuple[bool, str]:
"""
验证响应体
Args:
expected: 期望的响应体
actual: 实际的响应体
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_contains(self, expected_value: Any, actual_value: Any) -> Tuple[bool, str]:
"""
验证包含关系
Args:
expected_value: 期望包含的值
actual_value: 实际的值
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_equals(self, expected_value: Any, actual_value: Any) -> Tuple[bool, str]:
"""
验证相等关系
Args:
expected_value: 期望的值
actual_value: 实际的值
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_json_path(self, path: str, expected_value: Any, actual_data: Dict[str, Any]) -> Tuple[bool, str]:
"""
验证JSON路径
Args:
path: JSON路径
expected_value: 期望的值
actual_data: 实际的数据
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_regex(self, pattern: str, actual_value: str) -> Tuple[bool, str]:
"""
验证正则表达式
Args:
pattern: 正则表达式模式
actual_value: 实际的值
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_header(self, expected_header: Dict[str, str], actual_headers: Dict[str, str]) -> Tuple[bool, str]:
"""
验证响应头
Args:
expected_header: 期望的响应头
actual_headers: 实际的响应头
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_response_time(self, expected_max_time: float, actual_time: float) -> Tuple[bool, str]:
"""
验证响应时间
Args:
expected_max_time: 期望的最大响应时间(毫秒)
actual_time: 实际的响应时间(毫秒)
Returns:
(是否通过, 错误消息)
"""
pass
@abstractmethod
def validate_schema(self, expected_schema: Dict[str, Any], actual_data: Dict[str, Any]) -> Tuple[bool, str]:
"""
验证数据结构
Args:
expected_schema: 期望的结构
actual_data: 实际的数据
Returns:
(是否通过, 错误消息)
"""
pass
class IReportGenerator(ABC):
"""报告生成器接口"""
@abstractmethod
def generate_json_report(self, results: List[TestResult], summary: TestSummary, filename: str = None) -> str:
"""
生成JSON格式报告
Args:
results: 测试结果列表
summary: 测试摘要
filename: 报告文件名
Returns:
报告文件路径
"""
pass
@abstractmethod
def generate_html_report(self, results: List[TestResult], summary: TestSummary, filename: str = None) -> str:
"""
生成HTML格式报告
Args:
results: 测试结果列表
summary: 测试摘要
filename: 报告文件名
Returns:
报告文件路径
"""
pass
@abstractmethod
def generate_pdf_report(self, results: List[TestResult], summary: TestSummary, filename: str = None) -> str:
"""
生成PDF格式报告
Args:
results: 测试结果列表
summary: 测试摘要
filename: 报告文件名
Returns:
报告文件路径
"""
pass
class ILogger(ABC):
"""日志记录器接口"""
@abstractmethod
def debug(self, message: str) -> None:
"""记录调试信息"""
pass
@abstractmethod
def info(self, message: str) -> None:
"""记录信息"""
pass
@abstractmethod
def warning(self, message: str) -> None:
"""记录警告"""
pass
@abstractmethod
def error(self, message: str) -> None:
"""记录错误"""
pass
@abstractmethod
def critical(self, message: str) -> None:
"""记录严重错误"""
pass
@abstractmethod
def exception(self, message: str) -> None:
"""记录异常信息"""
pass
@abstractmethod
def set_level(self, level: LogLevel) -> None:
"""
设置日志级别
Args:
level: 日志级别
"""
pass
@abstractmethod
def get_logs(self) -> List[str]:
"""
获取日志
Returns:
日志列表
"""
pass
class ITestDataGenerator(ABC):
"""测试数据生成器接口"""
@abstractmethod
def generate_user_data(self, overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成用户数据
Args:
overrides: 覆盖的默认值
Returns:
用户数据
"""
pass
@abstractmethod
def generate_role_data(self, overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成角色数据
Args:
overrides: 覆盖的默认值
Returns:
角色数据
"""
pass
@abstractmethod
def generate_menu_data(self, overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成菜单数据
Args:
overrides: 覆盖的默认值
Returns:
菜单数据
"""
pass
@abstractmethod
def generate_permission_data(self, overrides: Dict[str, Any] = None) -> Dict[str, Any]:
"""
生成权限数据
Args:
overrides: 覆盖的默认值
Returns:
权限数据
"""
pass
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""
前后端API交互自动化测试工具(重构版)
模块化架构,支持配置文件化、数据生成、多种验证规则和多种报告格式
"""
import sys
import time
from pathlib import Path
from typing import List, Dict, Any
# 添加项目根目录到Python路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from config.settings import config
from core.api_tester import APITester
from utils.logger import TestLogger, LoggerFactory
from utils.reporter import ReportGenerator, TestResult as ReportTestResult, TestSummary
from test_cases.api_tests import APITestCases
class TestRunner:
"""测试运行器"""
def __init__(self):
"""初始化测试运行器"""
self.logger = LoggerFactory.get_logger(
"test_runner",
config.logging_file,
config.logging_level,
config.logging_console
)
self.tester = APITester(self.logger)
self.test_cases = APITestCases(self.tester)
self.report_generator = ReportGenerator(config.report_output_dir)
self.test_results: List[ReportTestResult] = []
self.start_time: float = 0
def run_test(self, test_name: str, test_func) -> bool:
"""
运行单个测试
Args:
test_name: 测试名称
test_func: 测试函数
Returns:
是否通过
"""
self.logger.log_test_start(test_name)
try:
start_time = time.time()
passed = test_func()
duration = (time.time() - start_time) * 1000
self.logger.log_test_end(test_name, passed, duration)
# 记录测试结果
test_result = ReportTestResult(
test_name=test_name,
test_type="API",
url=config.api_base_url,
method="N/A",
status_code=200 if passed else 0,
response_time=duration,
success=passed,
error_message="" if passed else f"测试失败",
request_data={},
response_data={},
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)
self.test_results.append(test_result)
return passed
except Exception as e:
duration = (time.time() - start_time) * 1000
self.logger.log_error(e)
test_result = ReportTestResult(
test_name=test_name,
test_type="API",
url=config.api_base_url,
method="N/A",
status_code=0,
response_time=duration,
success=False,
error_message=str(e),
request_data={},
response_data={},
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)
self.test_results.append(test_result)
return False
def run_all_tests(self) -> Dict[str, Any]:
"""
运行所有测试
Returns:
测试结果摘要
"""
self.start_time = time.time()
self.logger.info("="*60)
self.logger.info("开始运行所有测试用例")
self.logger.info("="*60)
# 1. 健康检查
self.run_test("健康检查", self.test_cases.test_health_check)
# 2. 用户登录
self.run_test("用户登录", self.test_cases.test_login)
# 3. 获取列表
self.run_test("获取用户列表", self.test_cases.test_get_user_list)
self.run_test("获取角色列表", self.test_cases.test_get_role_list)
self.run_test("获取菜单列表", self.test_cases.test_get_menu_list)
# 4. 创建数据
self.run_test("创建用户", self.test_cases.test_create_user)
self.run_test("创建角色", self.test_cases.test_create_role)
# 5. 搜索数据
self.run_test("搜索用户", self.test_cases.test_search_users)
# 计算测试摘要
total_time = (time.time() - self.start_time) * 1000
total = len(self.test_results)
passed = sum(1 for r in self.test_results if r.success)
failed = total - passed
skipped = 0
pass_rate = (passed / total * 100) if total > 0 else 0
summary = TestSummary(
total=total,
passed=passed,
failed=failed,
skipped=skipped,
pass_rate=pass_rate,
total_time=total_time,
start_time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.start_time)),
end_time=time.strftime("%Y-%m-%d %H:%M:%S")
)
# 记录摘要
self.logger.log_summary(total, passed, failed, skipped, total_time)
# 生成报告
self._generate_reports(summary)
return {
"total": total,
"passed": passed,
"failed": failed,
"skipped": skipped,
"pass_rate": pass_rate,
"total_time": total_time
}
def _generate_reports(self, summary: TestSummary) -> None:
"""
生成测试报告
Args:
summary: 测试摘要
"""
self.logger.info("生成测试报告...")
try:
reports = self.report_generator.generate_all_reports(self.test_results, summary)
for format_type, path in reports.items():
self.logger.info(f"{format_type.upper()}报告已生成: {path}")
except Exception as e:
self.logger.error(f"生成报告失败: {str(e)}")
def cleanup(self) -> None:
"""清理资源"""
self.tester.close()
self.logger.info("测试完成,资源已清理")
def main():
"""主函数"""
print("="*60)
print("前后端API交互自动化测试工具")
print("="*60)
print()
# 显示配置信息
print("配置信息:")
print(f" API基础URL: {config.api_base_url}")
print(f" 超时时间: {config.api_timeout}")
print(f" 最大重试次数: {config.api_max_retries}")
print(f" 报告输出目录: {config.report_output_dir}")
print(f" 报告格式: {', '.join(config.report_formats)}")
print()
# 创建测试运行器
runner = TestRunner()
try:
# 运行所有测试
results = runner.run_all_tests()
# 显示最终结果
print()
print("="*60)
print("测试完成!")
print("="*60)
print(f"总计: {results['total']}")
print(f"通过: {results['passed']}")
print(f"失败: {results['failed']}")
print(f"通过率: {results['pass_rate']:.2f}%")
print(f"总耗时: {results['total_time']:.2f}ms")
print("="*60)
# 返回退出码
sys.exit(0 if results['failed'] == 0 else 1)
except KeyboardInterrupt:
print("\n\n测试被用户中断")
sys.exit(130)
except Exception as e:
print(f"\n\n测试执行失败: {str(e)}")
sys.exit(1)
finally:
runner.cleanup()
if __name__ == "__main__":
main()
@@ -0,0 +1,371 @@
#!/usr/bin/env python3
"""
前后端API交互自动化测试工具(增强版)
支持参数化测试、边界条件测试、异常场景测试和自动认证
"""
import sys
import time
from pathlib import Path
from typing import List, Dict, Any
# 添加项目根目录到Python路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
from config.settings import config
from core.api_tester import APITester
from utils.logger import TestLogger, LoggerFactory
from utils.reporter import ReportGenerator, TestResult as ReportTestResult, TestSummary
from test_cases.api_tests_enhanced import (
APITestCases,
ParameterizedTestCases,
BoundaryTestCases,
ExceptionTestCases
)
class EnhancedTestRunner:
"""增强测试运行器"""
def __init__(self):
"""初始化测试运行器"""
self.logger = LoggerFactory.get_logger(
"enhanced_test_runner",
config.logging_file,
config.logging_level,
config.logging_console
)
self.tester = APITester(self.logger, auto_auth=True)
self.api_test_cases = APITestCases(self.tester)
self.parameterized_test_cases = ParameterizedTestCases(self.tester)
self.boundary_test_cases = BoundaryTestCases(self.tester)
self.exception_test_cases = ExceptionTestCases(self.tester)
self.report_generator = ReportGenerator(config.report_output_dir)
self.test_results: List[ReportTestResult] = []
self.start_time: float = 0
def run_test(self, test_name: str, test_func) -> bool:
"""
运行单个测试
Args:
test_name: 测试名称
test_func: 测试函数
Returns:
是否通过
"""
self.logger.log_test_start(test_name)
try:
start_time = time.time()
passed = test_func()
duration = (time.time() - start_time) * 1000
self.logger.log_test_end(test_name, passed, duration)
# 记录测试结果
test_result = ReportTestResult(
test_name=test_name,
test_type="API",
url=config.api_base_url,
method="N/A",
status_code=200 if passed else 0,
response_time=duration,
success=passed,
error_message="" if passed else f"测试失败",
request_data={},
response_data={},
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)
self.test_results.append(test_result)
return passed
except Exception as e:
duration = (time.time() - start_time) * 1000
self.logger.log_error(e)
test_result = ReportTestResult(
test_name=test_name,
test_type="API",
url=config.api_base_url,
method="N/A",
status_code=0,
response_time=duration,
success=False,
error_message=str(e),
request_data={},
response_data={},
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)
self.test_results.append(test_result)
return False
def run_basic_tests(self) -> Dict[str, bool]:
"""
运行基础测试用例
Returns:
测试结果字典
"""
self.logger.info("="*60)
self.logger.info("运行基础测试用例")
self.logger.info("="*60)
return self.api_test_cases.run_all_tests()
def run_parameterized_tests(self) -> Dict[str, Dict[str, bool]]:
"""
运行参数化测试用例
Returns:
测试结果字典
"""
self.logger.info("="*60)
self.logger.info("运行参数化测试用例")
self.logger.info("="*60)
results = {}
# 测试不同凭据登录
credentials = [
{"username": "admin", "password": "admin123"},
{"username": "test", "password": "test123"}
]
results["不同凭据登录"] = self.parameterized_test_cases.test_login_with_different_credentials(credentials)
# 测试不同页面大小
page_sizes = [10, 20, 50, 100]
results["分页功能"] = self.parameterized_test_cases.test_pagination(page_sizes)
return results
def run_boundary_tests(self) -> Dict[str, Dict[str, bool]]:
"""
运行边界条件测试用例
Returns:
测试结果字典
"""
self.logger.info("="*60)
self.logger.info("运行边界条件测试用例")
self.logger.info("="*60)
results = {}
# 测试用户名长度边界
results["用户名长度边界"] = self.boundary_test_cases.test_user_name_length()
# 测试页面大小边界
results["页面大小边界"] = self.boundary_test_cases.test_page_size_boundaries()
return results
def run_exception_tests(self) -> Dict[str, Dict[str, bool]]:
"""
运行异常场景测试用例
Returns:
测试结果字典
"""
self.logger.info("="*60)
self.logger.info("运行异常场景测试用例")
self.logger.info("="*60)
results = {}
# 测试无效凭据
results["无效凭据"] = self.exception_test_cases.test_invalid_credentials()
# 测试缺少必填字段
results["缺少必填字段"] = self.exception_test_cases.test_missing_required_fields()
# 测试未授权访问
results["未授权访问"] = self.exception_test_cases.test_unauthorized_access()
# 测试无效HTTP方法
results["无效HTTP方法"] = self.exception_test_cases.test_invalid_http_methods()
return results
def run_all_tests(self) -> Dict[str, Any]:
"""
运行所有测试用例
Returns:
测试结果摘要
"""
self.start_time = time.time()
self.logger.info("="*60)
self.logger.info("开始运行所有测试用例(增强版)")
self.logger.info("="*60)
# 1. 基础测试
basic_results = self.run_basic_tests()
for test_name, passed in basic_results.items():
full_test_name = f"基础测试 - {test_name}"
self._record_test_result(full_test_name, passed)
# 2. 参数化测试
param_results = self.run_parameterized_tests()
for category, results in param_results.items():
for test_name, passed in results.items():
full_test_name = f"参数化测试 - {category} - {test_name}"
self._record_test_result(full_test_name, passed)
# 3. 边界条件测试
boundary_results = self.run_boundary_tests()
for category, results in boundary_results.items():
for test_name, passed in results.items():
full_test_name = f"边界条件测试 - {category} - {test_name}"
self._record_test_result(full_test_name, passed)
# 4. 异常场景测试
exception_results = self.run_exception_tests()
for category, results in exception_results.items():
for test_name, passed in results.items():
full_test_name = f"异常场景测试 - {category} - {test_name}"
self._record_test_result(full_test_name, passed)
# 计算测试摘要
total_time = (time.time() - self.start_time) * 1000
total = len(self.test_results)
passed = sum(1 for r in self.test_results if r.success)
failed = total - passed
skipped = 0
pass_rate = (passed / total * 100) if total > 0 else 0
summary = TestSummary(
total=total,
passed=passed,
failed=failed,
skipped=skipped,
pass_rate=pass_rate,
total_time=total_time,
start_time=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.start_time)),
end_time=time.strftime("%Y-%m-%d %H:%M:%S")
)
# 记录摘要
self.logger.log_summary(total, passed, failed, skipped, total_time)
# 生成报告
self._generate_reports(summary)
return {
"total": total,
"passed": passed,
"failed": failed,
"skipped": skipped,
"pass_rate": pass_rate,
"total_time": total_time
}
def _record_test_result(self, test_name: str, passed: bool, response_time: float = 0) -> None:
"""
记录测试结果
Args:
test_name: 测试名称
passed: 是否通过
response_time: 响应时间
"""
test_result = ReportTestResult(
test_name=test_name,
test_type="API",
url=config.api_base_url,
method="N/A",
status_code=200 if passed else 0,
response_time=response_time,
success=passed,
error_message="" if passed else f"测试失败",
request_data={},
response_data={},
timestamp=time.strftime("%Y-%m-%d %H:%M:%S")
)
self.test_results.append(test_result)
def _generate_reports(self, summary: TestSummary) -> None:
"""
生成测试报告
Args:
summary: 测试摘要
"""
self.logger.info("生成测试报告...")
try:
reports = self.report_generator.generate_all_reports(self.test_results, summary)
for format_type, path in reports.items():
self.logger.info(f"{format_type.upper()}报告已生成: {path}")
except Exception as e:
self.logger.error(f"生成报告失败: {str(e)}")
def cleanup(self) -> None:
"""清理资源"""
self.tester.close()
self.logger.info("测试完成,资源已清理")
def main():
"""主函数"""
print("="*60)
print("前后端API交互自动化测试工具(增强版)")
print("="*60)
print()
# 显示配置信息
print("配置信息:")
print(f" API基础URL: {config.api_base_url}")
print(f" 超时时间: {config.api_timeout}")
print(f" 最大重试次数: {config.api_max_retries}")
print(f" 报告输出目录: {config.report_output_dir}")
print(f" 报告格式: {', '.join(config.report_formats)}")
print()
# 创建测试运行器
runner = EnhancedTestRunner()
try:
# 运行所有测试
results = runner.run_all_tests()
# 显示最终结果
print()
print("="*60)
print("测试完成!")
print("="*60)
print(f"总计: {results['total']}")
print(f"通过: {results['passed']}")
print(f"失败: {results['failed']}")
print(f"通过率: {results['pass_rate']:.2f}%")
print(f"总耗时: {results['total_time']:.2f}ms")
print("="*60)
# 返回退出码
sys.exit(0 if results['failed'] == 0 else 1)
except KeyboardInterrupt:
print("\n\n测试被用户中断")
sys.exit(130)
except Exception as e:
print(f"\n\n测试执行失败: {str(e)}")
sys.exit(1)
finally:
runner.cleanup()
if __name__ == "__main__":
main()
@@ -0,0 +1,41 @@
[tool.poetry]
name = "test-tools"
version = "1.0.0"
description = "test-tools - API测试工具"
authors = ["Test Team <test@example.com>"]
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.10"
requests = "^2.31.0"
httpx = "^0.25.0"
pyyaml = "^6.0"
bcrypt = "^4.0.0"
psycopg2-binary = "^2.9.0"
python-dotenv = "^1.0.0"
[tool.poetry.group.dev.dependencies]
pytest = "^7.4.0"
pytest-asyncio = "^0.21.0"
pytest-cov = "^4.1.0"
black = "^23.0.0"
flake8 = "^6.0.0"
mypy = "^1.5.0"
[tool.poetry.scripts]
test-tools = "main:main"
test-tools-enhanced = "main_enhanced:main"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.black]
line-length = 100
target-version = ['py310']
[tool.mypy]
python_version = "3.10"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = false
@@ -0,0 +1,27 @@
# Python测试工具依赖包
# HTTP客户端
requests>=2.31.0
httpx>=0.25.0
# 配置管理
pyyaml>=6.0
# 密码加密
bcrypt>=4.0.0
# 数据库驱动
psycopg2-binary>=2.9.0
# 测试框架
pytest>=7.4.0
pytest-asyncio>=0.21.0
pytest-cov>=4.1.0
# 代码质量
black>=23.0.0
flake8>=6.0.0
mypy>=1.5.0
# 工具库
python-dotenv>=1.0.0
@@ -0,0 +1 @@
"""测试用例模块"""
@@ -0,0 +1,338 @@
"""
API测试用例模块
"""
from typing import List, Dict, Any
from core.api_tester import APITester
from core.validation import ValidationRule
from data.test_data import TestDataGenerator
class APITestCases:
"""API测试用例集合"""
def __init__(self, tester: APITester):
"""
初始化测试用例
Args:
tester: API测试器实例
"""
self.tester = tester
self.test_data_generator = TestDataGenerator()
def test_health_check(self) -> bool:
"""
测试健康检查接口
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/actuator/health",
test_name="健康检查",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", "UP", path="status")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_login(self, username: str = None, password: str = None) -> bool:
"""
测试用户登录
Args:
username: 用户名
password: 密码
Returns:
是否通过
"""
username = username or "admin"
password = password or "admin123"
result = self.tester.request(
"POST",
"/auth/login",
data={"username": username, "password": password},
test_name="用户登录",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "token", path="data.token")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_get_user_list(self) -> bool:
"""
测试获取用户列表
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/user/list",
params={"page": 1, "pageSize": 10},
test_name="获取用户列表",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_get_role_list(self) -> bool:
"""
测试获取角色列表
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/role/list",
params={"page": 1, "pageSize": 10},
test_name="获取角色列表",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_get_menu_list(self) -> bool:
"""
测试获取菜单列表
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/menu/list",
test_name="获取菜单列表",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "list", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_create_user(self) -> bool:
"""
测试创建用户
Returns:
是否通过
"""
user_data = self.test_data_generator.generate_user_data()
result = self.tester.request(
"POST",
"/user/create",
data=user_data,
test_name="创建用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", user_data["username"], path="data.username")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_create_role(self) -> bool:
"""
测试创建角色
Returns:
是否通过
"""
role_data = self.test_data_generator.generate_role_data()
result = self.tester.request(
"POST",
"/role/create",
data=role_data,
test_name="创建角色",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", role_data["roleName"], path="data.roleName")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_update_user(self, user_id: int = None) -> bool:
"""
测试更新用户
Args:
user_id: 用户ID
Returns:
是否通过
"""
user_data = self.test_data_generator.generate_user_data()
if user_id:
user_data["id"] = user_id
result = self.tester.request(
"PUT",
"/user/update",
data=user_data,
test_name="更新用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_delete_user(self, user_id: int) -> bool:
"""
测试删除用户
Args:
user_id: 用户ID
Returns:
是否通过
"""
result = self.tester.request(
"DELETE",
f"/user/delete/{user_id}",
test_name="删除用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_get_user_info(self, user_id: int) -> bool:
"""
测试获取用户信息
Args:
user_id: 用户ID
Returns:
是否通过
"""
result = self.tester.request(
"GET",
f"/user/info/{user_id}",
test_name="获取用户信息",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", user_id, path="data.id")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_search_users(self, keyword: str = None) -> bool:
"""
测试搜索用户
Args:
keyword: 搜索关键词
Returns:
是否通过
"""
keyword = keyword or "admin"
result = self.tester.request(
"GET",
"/user/search",
params={"keyword": keyword, "page": 1, "pageSize": 10},
test_name="搜索用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def run_all_tests(self) -> Dict[str, bool]:
"""
运行所有测试用例
Returns:
测试结果字典
"""
results = {}
# 健康检查
results["health_check"] = self.test_health_check()
# 登录
results["login"] = self.test_login()
# 获取列表
results["get_user_list"] = self.test_get_user_list()
results["get_role_list"] = self.test_get_role_list()
results["get_menu_list"] = self.test_get_menu_list()
# 创建
results["create_user"] = self.test_create_user()
results["create_role"] = self.test_create_role()
# 搜索
results["search_users"] = self.test_search_users()
return results
@@ -0,0 +1,710 @@
"""
API测试用例模块(增强版)
支持参数化测试、边界条件测试和异常场景测试
"""
from typing import List, Dict, Any, Callable
from functools import wraps
import pytest
from core.api_tester import APITester
from core.validation import ValidationRule
from data.test_data import TestDataGenerator
def ensure_auth(test_func: Callable) -> Callable:
"""
装饰器:确保测试前已认证
Args:
test_func: 测试函数
Returns:
包装后的测试函数
"""
@wraps(test_func)
def wrapper(self, *args, **kwargs):
# 确保已认证
if not self.tester._ensure_authenticated():
return False
return test_func(self, *args, **kwargs)
return wrapper
class APITestCases:
"""API测试用例集合(增强版)"""
def __init__(self, tester: APITester):
"""
初始化测试用例
Args:
tester: API测试器实例
"""
self.tester = tester
self.test_data_generator = TestDataGenerator()
self.created_user_id = None
self.created_role_id = None
@ensure_auth
def test_health_check(self) -> bool:
"""
测试健康检查接口
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/actuator/health",
test_name="健康检查",
expected_status=200,
require_auth=False
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", "UP", path="status")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def test_login(self, username: str = None, password: str = None) -> bool:
"""
测试用户登录
Args:
username: 用户名
password: 密码
Returns:
是否通过
"""
username = username or "admin"
password = password or "admin123"
result = self.tester.request(
"POST",
"/auth/login",
data={"username": username, "password": password},
test_name="用户登录",
expected_status=200,
require_auth=False
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "token", path="data.token")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_get_user_list(self) -> bool:
"""
测试获取用户列表
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/user/list",
params={"page": 1, "pageSize": 10},
test_name="获取用户列表",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_get_role_list(self) -> bool:
"""
测试获取角色列表
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/role/list",
params={"page": 1, "pageSize": 10},
test_name="获取角色列表",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_get_menu_list(self) -> bool:
"""
测试获取菜单列表
Returns:
是否通过
"""
result = self.tester.request(
"GET",
"/menu/list",
test_name="获取菜单列表",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "list", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_create_user(self) -> bool:
"""
测试创建用户
Returns:
是否通过
"""
user_data = self.test_data_generator.generate_user_data()
result = self.tester.request(
"POST",
"/user/create",
data=user_data,
test_name="创建用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", user_data["username"], path="data.username")
]
result = self.tester.validate(result, validation_rules)
# 保存创建的用户ID
if result.passed and result.response_data.get("data"):
self.created_user_id = result.response_data["data"].get("id")
return result.passed
@ensure_auth
def test_create_role(self) -> bool:
"""
测试创建角色
Returns:
是否通过
"""
role_data = self.test_data_generator.generate_role_data()
result = self.tester.request(
"POST",
"/role/create",
data=role_data,
test_name="创建角色",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", role_data["roleName"], path="data.roleName")
]
result = self.tester.validate(result, validation_rules)
# 保存创建的角色ID
if result.passed and result.response_data.get("data"):
self.created_role_id = result.response_data["data"].get("id")
return result.passed
@ensure_auth
def test_update_user(self, user_id: int = None) -> bool:
"""
测试更新用户
Args:
user_id: 用户ID
Returns:
是否通过
"""
user_data = self.test_data_generator.generate_user_data()
if user_id:
user_data["id"] = user_id
elif self.created_user_id:
user_data["id"] = self.created_user_id
result = self.tester.request(
"PUT",
"/user/update",
data=user_data,
test_name="更新用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_delete_user(self, user_id: int = None) -> bool:
"""
测试删除用户
Args:
user_id: 用户ID
Returns:
是否通过
"""
if user_id is None:
user_id = self.created_user_id
if user_id is None:
return False
result = self.tester.request(
"DELETE",
f"/user/delete/{user_id}",
test_name="删除用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_get_user_info(self, user_id: int = None) -> bool:
"""
测试获取用户信息
Args:
user_id: 用户ID
Returns:
是否通过
"""
if user_id is None:
user_id = self.created_user_id
if user_id is None:
return False
result = self.tester.request(
"GET",
f"/user/info/{user_id}",
test_name="获取用户信息",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", user_id, path="data.id")
]
result = self.tester.validate(result, validation_rules)
return result.passed
@ensure_auth
def test_search_users(self, keyword: str = None) -> bool:
"""
测试搜索用户
Args:
keyword: 搜索关键词
Returns:
是否通过
"""
keyword = keyword or "admin"
result = self.tester.request(
"GET",
"/user/search",
params={"keyword": keyword, "page": 1, "pageSize": 10},
test_name="搜索用户",
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
return result.passed
def run_all_tests(self) -> Dict[str, bool]:
"""
运行所有基础测试用例
Returns:
测试结果字典
"""
results = {}
# 健康检查
results["health_check"] = self.test_health_check()
# 登录
results["login"] = self.test_login()
# 获取列表
results["get_user_list"] = self.test_get_user_list()
results["get_role_list"] = self.test_get_role_list()
results["get_menu_list"] = self.test_get_menu_list()
# 创建
results["create_user"] = self.test_create_user()
results["create_role"] = self.test_create_role()
# 搜索
results["search_users"] = self.test_search_users()
return results
class ParameterizedTestCases:
"""参数化测试用例"""
def __init__(self, tester: APITester):
"""
初始化参数化测试用例
Args:
tester: API测试器实例
"""
self.tester = tester
self.test_data_generator = TestDataGenerator()
@ensure_auth
def test_login_with_different_credentials(self, credentials: List[Dict[str, str]]) -> Dict[str, bool]:
"""
使用不同凭据测试登录
Args:
credentials: 凭据列表,每个元素包含username和password
Returns:
测试结果字典
"""
results = {}
for cred in credentials:
username = cred.get("username")
password = cred.get("password")
test_name = f"登录测试 - {username}"
result = self.tester.request(
"POST",
"/auth/login",
data={"username": username, "password": password},
test_name=test_name,
expected_status=200,
require_auth=False
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code")
]
result = self.tester.validate(result, validation_rules)
results[test_name] = result.passed
return results
@ensure_auth
def test_pagination(self, page_sizes: List[int]) -> Dict[str, bool]:
"""
测试分页功能
Args:
page_sizes: 页面大小列表
Returns:
测试结果字典
"""
results = {}
for page_size in page_sizes:
test_name = f"分页测试 - 页面大小: {page_size}"
result = self.tester.request(
"GET",
"/user/list",
params={"page": 1, "pageSize": page_size},
test_name=test_name,
expected_status=200
)
validation_rules = [
ValidationRule("status_code", 200),
ValidationRule("json_path", 200, path="code"),
ValidationRule("json_path", "records", path="data")
]
result = self.tester.validate(result, validation_rules)
results[test_name] = result.passed
return results
class BoundaryTestCases:
"""边界条件测试用例"""
def __init__(self, tester: APITester):
"""
初始化边界条件测试用例
Args:
tester: API测试器实例
"""
self.tester = tester
self.test_data_generator = TestDataGenerator()
@ensure_auth
def test_user_name_length(self) -> Dict[str, bool]:
"""
测试用户名长度边界
Returns:
测试结果字典
"""
results = {}
# 测试空用户名
result = self.tester.request(
"POST",
"/auth/login",
data={"username": "", "password": "admin123"},
test_name="登录测试 - 空用户名",
expected_status=400,
require_auth=False
)
results["空用户名"] = result.status_code == 400
# 测试超长用户名
long_username = "a" * 100
result = self.tester.request(
"POST",
"/auth/login",
data={"username": long_username, "password": "admin123"},
test_name="登录测试 - 超长用户名",
expected_status=400,
require_auth=False
)
results["超长用户名"] = result.status_code == 400
return results
@ensure_auth
def test_page_size_boundaries(self) -> Dict[str, bool]:
"""
测试页面大小边界
Returns:
测试结果字典
"""
results = {}
# 测试页面大小为0
result = self.tester.request(
"GET",
"/user/list",
params={"page": 1, "pageSize": 0},
test_name="分页测试 - 页面大小为0",
expected_status=400
)
results["页面大小为0"] = result.status_code == 400
# 测试页面大小为负数
result = self.tester.request(
"GET",
"/user/list",
params={"page": 1, "pageSize": -1},
test_name="分页测试 - 页面大小为负数",
expected_status=400
)
results["页面大小为负数"] = result.status_code == 400
# 测试超大页面大小
result = self.tester.request(
"GET",
"/user/list",
params={"page": 1, "pageSize": 10000},
test_name="分页测试 - 超大页面大小",
expected_status=200
)
results["超大页面大小"] = result.status_code == 200
return results
class ExceptionTestCases:
"""异常场景测试用例"""
def __init__(self, tester: APITester):
"""
初始化异常场景测试用例
Args:
tester: API测试器实例
"""
self.tester = tester
self.test_data_generator = TestDataGenerator()
@ensure_auth
def test_invalid_credentials(self) -> Dict[str, bool]:
"""
测试无效凭据
Returns:
测试结果字典
"""
results = {}
# 测试错误密码
result = self.tester.request(
"POST",
"/auth/login",
data={"username": "admin", "password": "wrongpassword"},
test_name="登录测试 - 错误密码",
expected_status=401,
require_auth=False
)
results["错误密码"] = result.status_code == 401
# 测试不存在的用户
result = self.tester.request(
"POST",
"/auth/login",
data={"username": "nonexistent", "password": "admin123"},
test_name="登录测试 - 不存在的用户",
expected_status=401,
require_auth=False
)
results["不存在的用户"] = result.status_code == 401
return results
@ensure_auth
def test_missing_required_fields(self) -> Dict[str, bool]:
"""
测试缺少必填字段
Returns:
测试结果字典
"""
results = {}
# 测试缺少用户名
result = self.tester.request(
"POST",
"/auth/login",
data={"password": "admin123"},
test_name="登录测试 - 缺少用户名",
expected_status=400,
require_auth=False
)
results["缺少用户名"] = result.status_code == 400
# 测试缺少密码
result = self.tester.request(
"POST",
"/auth/login",
data={"username": "admin"},
test_name="登录测试 - 缺少密码",
expected_status=400,
require_auth=False
)
results["缺少密码"] = result.status_code == 400
return results
@ensure_auth
def test_unauthorized_access(self) -> Dict[str, bool]:
"""
测试未授权访问
Returns:
测试结果字典
"""
results = {}
# 临时清除认证
self.tester.auth_manager.logout()
# 测试未授权访问用户列表
result = self.tester.request(
"GET",
"/user/list",
test_name="未授权访问 - 用户列表",
expected_status=401,
require_auth=False
)
results["未授权访问用户列表"] = result.status_code == 401
# 重新认证
self.tester._ensure_authenticated()
return results
@ensure_auth
def test_invalid_http_methods(self) -> Dict[str, bool]:
"""
测试无效的HTTP方法
Returns:
测试结果字典
"""
results = {}
# 测试不支持的HTTP方法
try:
result = self.tester.request(
"PATCH",
"/user/list",
test_name="无效HTTP方法 - PATCH",
expected_status=405
)
results["PATCH方法"] = result.status_code == 405
except:
results["PATCH方法"] = False
return results
@@ -0,0 +1 @@
"""工具模块"""
@@ -0,0 +1,507 @@
"""
增强版报告生成器模块
支持趋势分析、历史报告对比、性能基准对比等功能
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, asdict
import statistics
from .reporter import TestResult, TestSummary, ReportGenerator
@dataclass
class TrendData:
"""趋势数据"""
date: str
total: int
passed: int
failed: int
pass_rate: float
avg_response_time: float
total_time: float
@dataclass
class PerformanceBenchmark:
"""性能基准数据"""
endpoint: str
method: str
avg_response_time: float
min_response_time: float
max_response_time: float
p95_response_time: float
p99_response_time: float
class EnhancedReportGenerator(ReportGenerator):
"""增强版报告生成器"""
def __init__(self, output_dir: str = "reports", history_dir: str = "reports/history"):
"""
初始化增强版报告生成器
Args:
output_dir: 报告输出目录
history_dir: 历史报告目录
"""
super().__init__(output_dir)
self.history_dir = Path(history_dir)
self.history_dir.mkdir(parents=True, exist_ok=True)
def generate_trend_report(self, days: int = 7) -> str:
"""
生成趋势分析报告
Args:
days: 分析最近几天的数据
Returns:
报告文件路径
"""
trend_data = self._load_trend_data(days)
if not trend_data:
return ""
report_path = self.output_dir / f"trend_report_{datetime.now().strftime('%Y%m%d')}.html"
html_content = self._generate_trend_html(trend_data)
with open(report_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return str(report_path)
def generate_comparison_report(self, current_results: List[TestResult],
previous_results: List[TestResult] = None) -> str:
"""
生成对比报告
Args:
current_results: 当前测试结果
previous_results: 之前的测试结果
Returns:
报告文件路径
"""
if previous_results is None:
previous_results = self._load_latest_results()
if not previous_results:
return ""
comparison_data = self._compare_results(current_results, previous_results)
report_path = self.output_dir / f"comparison_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
html_content = self._generate_comparison_html(comparison_data)
with open(report_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return str(report_path)
def generate_performance_report(self, results: List[TestResult],
benchmarks: Dict[str, PerformanceBenchmark] = None) -> str:
"""
生成性能报告
Args:
results: 测试结果列表
benchmarks: 性能基准数据
Returns:
报告文件路径
"""
performance_data = self._analyze_performance(results)
if benchmarks:
performance_data = self._compare_with_benchmarks(performance_data, benchmarks)
report_path = self.output_dir / f"performance_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
html_content = self._generate_performance_html(performance_data)
with open(report_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return str(report_path)
def _load_trend_data(self, days: int) -> List[TrendData]:
"""
加载趋势数据
Args:
days: 加载最近几天的数据
Returns:
趋势数据列表
"""
trend_data = []
for i in range(days):
date = datetime.now() - timedelta(days=i)
date_str = date.strftime('%Y%m%d')
report_file = self.history_dir / f"test_report_{date_str}.json"
if report_file.exists():
with open(report_file, 'r', encoding='utf-8') as f:
report_data = json.load(f)
summary = report_data.get('summary', {})
results = report_data.get('results', [])
avg_response_time = statistics.mean([r['response_time'] for r in results]) if results else 0
trend_data.append(TrendData(
date=date.strftime('%Y-%m-%d'),
total=summary.get('total', 0),
passed=summary.get('passed', 0),
failed=summary.get('failed', 0),
pass_rate=summary.get('pass_rate', 0),
avg_response_time=avg_response_time,
total_time=summary.get('total_time', 0)
))
return sorted(trend_data, key=lambda x: x.date)
def _load_latest_results(self) -> List[TestResult]:
"""
加载最新的测试结果
Returns:
测试结果列表
"""
report_files = list(self.history_dir.glob("test_report_*.json"))
if not report_files:
return []
latest_file = max(report_files, key=lambda f: f.stat().st_mtime)
with open(latest_file, 'r', encoding='utf-8') as f:
report_data = json.load(f)
return [TestResult(**r) for r in report_data.get('results', [])]
def _compare_results(self, current: List[TestResult], previous: List[TestResult]) -> Dict[str, Any]:
"""
对比测试结果
Args:
current: 当前测试结果
previous: 之前的测试结果
Returns:
对比数据
"""
current_summary = self._calculate_summary(current)
previous_summary = self._calculate_summary(previous)
return {
"current": current_summary,
"previous": previous_summary,
"changes": {
"total": current_summary["total"] - previous_summary["total"],
"passed": current_summary["passed"] - previous_summary["passed"],
"failed": current_summary["failed"] - previous_summary["failed"],
"pass_rate": current_summary["pass_rate"] - previous_summary["pass_rate"],
"avg_response_time": current_summary["avg_response_time"] - previous_summary["avg_response_time"]
}
}
def _analyze_performance(self, results: List[TestResult]) -> Dict[str, Any]:
"""
分析性能数据
Args:
results: 测试结果列表
Returns:
性能分析数据
"""
response_times = [r.response_time for r in results if r.response_time > 0]
if not response_times:
return {}
return {
"avg_response_time": statistics.mean(response_times),
"min_response_time": min(response_times),
"max_response_time": max(response_times),
"median_response_time": statistics.median(response_times),
"std_response_time": statistics.stdev(response_times) if len(response_times) > 1 else 0,
"p95_response_time": self._calculate_percentile(response_times, 95),
"p99_response_time": self._calculate_percentile(response_times, 99),
"total_tests": len(results),
"failed_tests": sum(1 for r in results if not r.success)
}
def _compare_with_benchmarks(self, performance_data: Dict[str, Any],
benchmarks: Dict[str, PerformanceBenchmark]) -> Dict[str, Any]:
"""
与性能基准对比
Args:
performance_data: 性能数据
benchmarks: 性能基准
Returns:
对比后的性能数据
"""
comparisons = []
for endpoint, benchmark in benchmarks.items():
current_avg = performance_data.get("avg_response_time", 0)
comparison = {
"endpoint": endpoint,
"method": benchmark.method,
"benchmark_avg": benchmark.avg_response_time,
"current_avg": current_avg,
"diff": current_avg - benchmark.avg_response_time,
"diff_percent": ((current_avg - benchmark.avg_response_time) / benchmark.avg_response_time * 100) if benchmark.avg_response_time > 0 else 0,
"status": "good" if current_avg <= benchmark.avg_response_time * 1.2 else "warning" if current_avg <= benchmark.avg_response_time * 1.5 else "critical"
}
comparisons.append(comparison)
performance_data["benchmarks"] = comparisons
return performance_data
def _calculate_percentile(self, data: List[float], percentile: int) -> float:
"""
计算百分位数
Args:
data: 数据列表
percentile: 百分位数
Returns:
百分位数值
"""
sorted_data = sorted(data)
index = (percentile / 100) * (len(sorted_data) - 1)
if index.is_integer():
return sorted_data[int(index)]
else:
lower = sorted_data[int(index)]
upper = sorted_data[int(index) + 1]
return lower + (upper - lower) * (index - int(index))
def _calculate_summary(self, results: List[TestResult]) -> Dict[str, Any]:
"""
计算测试摘要
Args:
results: 测试结果列表
Returns:
测试摘要
"""
total = len(results)
passed = sum(1 for r in results if r.success)
failed = total - passed
response_times = [r.response_time for r in results if r.response_time > 0]
return {
"total": total,
"passed": passed,
"failed": failed,
"pass_rate": (passed / total * 100) if total > 0 else 0,
"avg_response_time": statistics.mean(response_times) if response_times else 0
}
def _generate_trend_html(self, trend_data: List[TrendData]) -> str:
"""生成趋势分析HTML"""
# 这里简化实现,实际应该使用图表库(如Chart.js)
return f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>趋势分析报告</title>
<style>
body {{ font-family: Arial, sans-serif; padding: 20px; }}
.trend-table {{ width: 100%; border-collapse: collapse; margin-top: 20px; }}
.trend-table th, .trend-table td {{ border: 1px solid #ddd; padding: 8px; text-align: center; }}
.trend-table th {{ background-color: #4CAF50; color: white; }}
.trend-table tr:nth-child(even) {{ background-color: #f2f2f2; }}
</style>
</head>
<body>
<h1>测试趋势分析报告</h1>
<p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<table class="trend-table">
<tr>
<th>日期</th>
<th>总数</th>
<th>通过</th>
<th>失败</th>
<th>通过率</th>
<th>平均响应时间</th>
</tr>
{"".join([f"""
<tr>
<td>{d.date}</td>
<td>{d.total}</td>
<td>{d.passed}</td>
<td>{d.failed}</td>
<td>{d.pass_rate:.2f}%</td>
<td>{d.avg_response_time:.2f}ms</td>
</tr>
""" for d in trend_data])}
</table>
</body>
</html>
"""
def _generate_comparison_html(self, comparison_data: Dict[str, Any]) -> str:
"""生成对比分析HTML"""
return f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>测试对比报告</title>
<style>
body {{ font-family: Arial, sans-serif; padding: 20px; }}
.comparison-table {{ width: 100%; border-collapse: collapse; margin-top: 20px; }}
.comparison-table th, .comparison-table td {{ border: 1px solid #ddd; padding: 8px; text-align: center; }}
.comparison-table th {{ background-color: #4CAF50; color: white; }}
.positive {{ color: green; }}
.negative {{ color: red; }}
</style>
</head>
<body>
<h1>测试对比报告</h1>
<p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<table class="trend-table">
<tr>
<th>指标</th>
<th>当前</th>
<th>之前</th>
<th>变化</th>
</tr>
<tr>
<td>总数</td>
<td>{comparison_data['current']['total']}</td>
<td>{comparison_data['previous']['total']}</td>
<td class="{'positive' if comparison_data['changes']['total'] >= 0 else 'negative'}">
{comparison_data['changes']['total']:+d}
</td>
</tr>
<tr>
<td>通过</td>
<td>{comparison_data['current']['passed']}</td>
<td>{comparison_data['previous']['passed']}</td>
<td class="{'positive' if comparison_data['changes']['passed'] >= 0 else 'negative'}">
{comparison_data['changes']['passed']:+d}
</td>
</tr>
<tr>
<td>失败</td>
<td>{comparison_data['current']['failed']}</td>
<td>{comparison_data['previous']['failed']}</td>
<td class="{'negative' if comparison_data['changes']['failed'] > 0 else 'positive'}">
{comparison_data['changes']['failed']:+d}
</td>
</tr>
<tr>
<td>通过率</td>
<td>{comparison_data['current']['pass_rate']:.2f}%</td>
<td>{comparison_data['previous']['pass_rate']:.2f}%</td>
<td class="{'positive' if comparison_data['changes']['pass_rate'] >= 0 else 'negative'}">
{comparison_data['changes']['pass_rate']:+.2f}%
</td>
</tr>
</table>
</body>
</html>
"""
def _generate_performance_html(self, performance_data: Dict[str, Any]) -> str:
"""生成性能分析HTML"""
return f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>性能分析报告</title>
<style>
body {{ font-family: Arial, sans-serif; padding: 20px; }}
.performance-table {{ width: 100%; border-collapse: collapse; margin-top: 20px; }}
.performance-table th, .performance-table td {{ border: 1px solid #ddd; padding: 8px; text-align: center; }}
.performance-table th {{ background-color: #4CAF50; color: white; }}
.good {{ color: green; }}
.warning {{ color: orange; }}
.critical {{ color: red; }}
</style>
</head>
<body>
<h1>性能分析报告</h1>
<p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<h2>性能指标</h2>
<table class="performance-table">
<tr>
<th>指标</th>
<th>值</th>
</tr>
<tr>
<td>平均响应时间</td>
<td>{performance_data.get('avg_response_time', 0):.2f}ms</td>
</tr>
<tr>
<td>最小响应时间</td>
<td>{performance_data.get('min_response_time', 0):.2f}ms</td>
</tr>
<tr>
<td>最大响应时间</td>
<td>{performance_data.get('max_response_time', 0):.2f}ms</td>
</tr>
<tr>
<td>中位数响应时间</td>
<td>{performance_data.get('median_response_time', 0):.2f}ms</td>
</tr>
<tr>
<td>P95响应时间</td>
<td>{performance_data.get('p95_response_time', 0):.2f}ms</td>
</tr>
<tr>
<td>P99响应时间</td>
<td>{performance_data.get('p99_response_time', 0):.2f}ms</td>
</tr>
</table>
</body>
</html>
"""
def save_to_history(self, results: List[TestResult], summary: TestSummary) -> None:
"""
保存测试结果到历史记录
Args:
results: 测试结果列表
summary: 测试摘要
"""
date_str = datetime.now().strftime('%Y%m%d')
history_file = self.history_dir / f"test_report_{date_str}.json"
report_data = {
"summary": summary.to_dict(),
"results": [r.to_dict() for r in results],
"generated_at": datetime.now().isoformat()
}
with open(history_file, 'w', encoding='utf-8') as f:
json.dump(report_data, f, ensure_ascii=False, indent=2)
@@ -0,0 +1,156 @@
"""
日志工具模块
"""
import logging
import sys
from pathlib import Path
from typing import Optional
from datetime import datetime
class TestLogger:
"""测试日志记录器"""
def __init__(self, name: str = "test", log_file: str = None, level: str = "INFO", console: bool = True):
"""
初始化日志记录器
Args:
name: 日志记录器名称
log_file: 日志文件路径
level: 日志级别
console: 是否输出到控制台
"""
self.logger = logging.getLogger(name)
self.logger.setLevel(getattr(logging, level.upper()))
# 清除现有的处理器
self.logger.handlers.clear()
# 创建格式化器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 添加文件处理器
if log_file:
log_path = Path(log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
# 添加控制台处理器
if console:
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
def debug(self, message: str) -> None:
"""记录调试信息"""
self.logger.debug(message)
def info(self, message: str) -> None:
"""记录信息"""
self.logger.info(message)
def warning(self, message: str) -> None:
"""记录警告"""
self.logger.warning(message)
def error(self, message: str) -> None:
"""记录错误"""
self.logger.error(message)
def critical(self, message: str) -> None:
"""记录严重错误"""
self.logger.critical(message)
def exception(self, message: str) -> None:
"""记录异常信息"""
self.logger.exception(message)
def log_test_start(self, test_name: str) -> None:
"""记录测试开始"""
self.info(f"{'='*60}")
self.info(f"开始测试: {test_name}")
self.info(f"{'='*60}")
def log_test_end(self, test_name: str, passed: bool, duration: float = None) -> None:
"""记录测试结束"""
status = "✅ 通过" if passed else "❌ 失败"
duration_str = f" (耗时: {duration:.2f}ms)" if duration else ""
self.info(f"测试结束: {test_name} - {status}{duration_str}")
def log_request(self, method: str, url: str, data: dict = None, headers: dict = None) -> None:
"""记录请求信息"""
self.info(f"发送请求: {method} {url}")
if data:
self.info(f"请求数据: {data}")
if headers:
self.info(f"请求头: {headers}")
def log_response(self, status_code: int, response_time: float, data: dict = None) -> None:
"""记录响应信息"""
self.info(f"响应状态码: {status_code}")
self.info(f"响应时间: {response_time:.2f}ms")
if data:
self.info(f"响应数据: {data}")
def log_validation(self, rule_type: str, passed: bool, message: str = "") -> None:
"""记录验证结果"""
status = "" if passed else ""
self.info(f"{status} 验证规则 [{rule_type}]: {message or '通过'}")
def log_error(self, error: Exception) -> None:
"""记录错误详情"""
self.error(f"错误类型: {type(error).__name__}")
self.error(f"错误信息: {str(error)}")
self.exception("错误堆栈:")
def log_summary(self, total: int, passed: int, failed: int, skipped: int = 0, duration: float = None) -> None:
"""记录测试摘要"""
self.info(f"{'='*60}")
self.info(f"测试摘要:")
self.info(f" 总数: {total}")
self.info(f" 通过: {passed}")
self.info(f" 失败: {failed}")
self.info(f" 跳过: {skipped}")
if duration:
self.info(f" 总耗时: {duration:.2f}ms")
self.info(f" 通过率: {(passed/total*100):.2f}%")
self.info(f"{'='*60}")
class LoggerFactory:
"""日志记录器工厂"""
_loggers = {}
@classmethod
def get_logger(cls, name: str = "test", log_file: str = None, level: str = "INFO", console: bool = True) -> TestLogger:
"""
获取日志记录器实例
Args:
name: 日志记录器名称
log_file: 日志文件路径
level: 日志级别
console: 是否输出到控制台
Returns:
日志记录器实例
"""
key = f"{name}_{log_file}_{level}_{console}"
if key not in cls._loggers:
cls._loggers[key] = TestLogger(name, log_file, level, console)
return cls._loggers[key]
@classmethod
def clear_all(cls) -> None:
"""清除所有日志记录器"""
cls._loggers.clear()
@@ -0,0 +1,516 @@
"""
报告生成器模块
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, List
from dataclasses import dataclass, asdict
@dataclass
class TestResult:
"""测试结果数据类"""
test_name: str
test_type: str
url: str
method: str
status_code: int
response_time: float
success: bool
error_message: str
request_data: Dict[str, Any]
response_data: Dict[str, Any]
timestamp: str
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return asdict(self)
@dataclass
class TestSummary:
"""测试摘要数据类"""
total: int
passed: int
failed: int
skipped: int
pass_rate: float
total_time: float
start_time: str
end_time: str
def to_dict(self) -> Dict[str, Any]:
"""转换为字典"""
return asdict(self)
class ReportGenerator:
"""测试报告生成器"""
def __init__(self, output_dir: str = "reports"):
"""
初始化报告生成器
Args:
output_dir: 报告输出目录
"""
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
def generate_json_report(self, results: List[TestResult], summary: TestSummary, filename: str = None) -> str:
"""
生成JSON格式报告
Args:
results: 测试结果列表
summary: 测试摘要
filename: 报告文件名
Returns:
报告文件路径
"""
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"test_report_{timestamp}.json"
report_path = self.output_dir / filename
report_data = {
"summary": summary.to_dict(),
"results": [result.to_dict() for result in results],
"generated_at": datetime.now().isoformat()
}
with open(report_path, 'w', encoding='utf-8') as f:
json.dump(report_data, f, ensure_ascii=False, indent=2)
return str(report_path)
def generate_html_report(self, results: List[TestResult], summary: TestSummary, filename: str = None) -> str:
"""
生成HTML格式报告
Args:
results: 测试结果列表
summary: 测试摘要
filename: 报告文件名
Returns:
报告文件路径
"""
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"test_report_{timestamp}.html"
report_path = self.output_dir / filename
html_content = self._generate_html_content(results, summary)
with open(report_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return str(report_path)
def _generate_html_content(self, results: List[TestResult], summary: TestSummary) -> str:
"""
生成HTML内容
Args:
results: 测试结果列表
summary: 测试摘要
Returns:
HTML内容
"""
passed_results = [r for r in results if r.success]
failed_results = [r for r in results if not r.success]
html = f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>测试报告 - {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</title>
<style>
* {{
margin: 0;
padding: 0;
box-sizing: border-box;
}}
body {{
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
padding: 20px;
color: #333;
}}
.container {{
max-width: 1200px;
margin: 0 auto;
background: white;
border-radius: 10px;
box-shadow: 0 10px 40px rgba(0, 0, 0, 0.1);
overflow: hidden;
}}
.header {{
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 30px;
text-align: center;
}}
.header h1 {{
font-size: 28px;
margin-bottom: 10px;
}}
.header p {{
font-size: 14px;
opacity: 0.9;
}}
.summary {{
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 20px;
padding: 30px;
background: #f8f9fa;
}}
.summary-card {{
background: white;
padding: 20px;
border-radius: 8px;
box-shadow: 0 2px 8px rgba(0, 0, 0, 0.1);
text-align: center;
}}
.summary-card h3 {{
font-size: 14px;
color: #666;
margin-bottom: 10px;
}}
.summary-card .value {{
font-size: 32px;
font-weight: bold;
color: #333;
}}
.summary-card.passed .value {{
color: #52c41a;
}}
.summary-card.failed .value {{
color: #ff4d4f;
}}
.progress-bar {{
grid-column: 1 / -1;
height: 30px;
background: #e8e8e8;
border-radius: 15px;
overflow: hidden;
margin-top: 20px;
}}
.progress-fill {{
height: 100%;
background: linear-gradient(90deg, #52c41a 0%, #73d13d 100%);
display: flex;
align-items: center;
justify-content: center;
color: white;
font-weight: bold;
transition: width 0.3s ease;
}}
.results {{
padding: 30px;
}}
.results h2 {{
font-size: 24px;
margin-bottom: 20px;
color: #333;
}}
.test-item {{
background: white;
border: 1px solid #e8e8e8;
border-radius: 8px;
margin-bottom: 15px;
overflow: hidden;
}}
.test-item.passed {{
border-left: 4px solid #52c41a;
}}
.test-item.failed {{
border-left: 4px solid #ff4d4f;
}}
.test-header {{
padding: 15px 20px;
background: #fafafa;
display: flex;
justify-content: space-between;
align-items: center;
cursor: pointer;
}}
.test-header:hover {{
background: #f0f0f0;
}}
.test-name {{
font-weight: bold;
color: #333;
}}
.test-status {{
padding: 4px 12px;
border-radius: 12px;
font-size: 12px;
font-weight: bold;
}}
.test-status.passed {{
background: #f6ffed;
color: #52c41a;
}}
.test-status.failed {{
background: #fff1f0;
color: #ff4d4f;
}}
.test-details {{
padding: 20px;
display: none;
border-top: 1px solid #e8e8e8;
}}
.test-item.expanded .test-details {{
display: block;
}}
.detail-row {{
margin-bottom: 10px;
}}
.detail-label {{
font-weight: bold;
color: #666;
margin-right: 10px;
}}
.detail-value {{
color: #333;
}}
.error-message {{
background: #fff1f0;
color: #ff4d4f;
padding: 10px;
border-radius: 4px;
margin-top: 10px;
}}
.code-block {{
background: #f5f5f5;
padding: 15px;
border-radius: 4px;
overflow-x: auto;
font-family: 'Courier New', monospace;
font-size: 13px;
margin-top: 10px;
}}
.footer {{
text-align: center;
padding: 20px;
background: #f8f9fa;
color: #666;
font-size: 14px;
}}
@media (max-width: 768px) {{
.summary {{
grid-template-columns: 1fr;
}}
}}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>API测试报告</h1>
<p>生成时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}</p>
</div>
<div class="summary">
<div class="summary-card">
<h3>测试总数</h3>
<div class="value">{summary.total}</div>
</div>
<div class="summary-card passed">
<h3>通过</h3>
<div class="value">{summary.passed}</div>
</div>
<div class="summary-card failed">
<h3>失败</h3>
<div class="value">{summary.failed}</div>
</div>
<div class="summary-card">
<h3>跳过</h3>
<div class="value">{summary.skipped}</div>
</div>
<div class="summary-card">
<h3>通过率</h3>
<div class="value">{summary.pass_rate:.1f}%</div>
</div>
<div class="summary-card">
<h3>总耗时</h3>
<div class="value">{summary.total_time:.0f}ms</div>
</div>
<div class="progress-bar">
<div class="progress-fill" style="width: {summary.pass_rate}%">
{summary.pass_rate:.1f}%
</div>
</div>
</div>
<div class="results">
<h2>测试结果</h2>
<div id="test-results">
"""
for result in results:
status_class = "passed" if result.success else "failed"
status_text = "通过" if result.success else "失败"
html += f"""
<div class="test-item {status_class}" onclick="toggleDetails(this)">
<div class="test-header">
<span class="test-name">{result.test_name}</span>
<span class="test-status {status_class}">{status_text}</span>
</div>
<div class="test-details">
<div class="detail-row">
<span class="detail-label">测试类型:</span>
<span class="detail-value">{result.test_type}</span>
</div>
<div class="detail-row">
<span class="detail-label">请求方法:</span>
<span class="detail-value">{result.method}</span>
</div>
<div class="detail-row">
<span class="detail-label">请求URL:</span>
<span class="detail-value">{result.url}</span>
</div>
<div class="detail-row">
<span class="detail-label">状态码:</span>
<span class="detail-value">{result.status_code}</span>
</div>
<div class="detail-row">
<span class="detail-label">响应时间:</span>
<span class="detail-value">{result.response_time:.2f}ms</span>
</div>
<div class="detail-row">
<span class="detail-label">时间戳:</span>
<span class="detail-value">{result.timestamp}</span>
</div>
{self._generate_request_details(result)}
{self._generate_response_details(result)}
{self._generate_error_details(result)}
</div>
</div>
"""
html += """
</div>
</div>
<div class="footer">
<p>Everything is Suitable - API测试工具</p>
</div>
</div>
<script>
function toggleDetails(element) {
element.classList.toggle('expanded');
}
</script>
</body>
</html>
"""
return html
def _generate_request_details(self, result: TestResult) -> str:
"""生成请求详情"""
if not result.request_data:
return ""
import json
return f"""
<div class="detail-row">
<span class="detail-label">请求数据:</span>
</div>
<div class="code-block">{json.dumps(result.request_data, ensure_ascii=False, indent=2)}</div>
"""
def _generate_response_details(self, result: TestResult) -> str:
"""生成响应详情"""
if not result.response_data:
return ""
import json
return f"""
<div class="detail-row">
<span class="detail-label">响应数据:</span>
</div>
<div class="code-block">{json.dumps(result.response_data, ensure_ascii=False, indent=2)}</div>
"""
def _generate_error_details(self, result: TestResult) -> str:
"""生成错误详情"""
if result.success or not result.error_message:
return ""
return f"""
<div class="error-message">
<strong>错误信息:</strong> {result.error_message}
</div>
"""
def generate_all_reports(self, results: List[TestResult], summary: TestSummary) -> Dict[str, str]:
"""
生成所有格式的报告
Args:
results: 测试结果列表
summary: 测试摘要
Returns:
报告文件路径字典
"""
reports = {}
# 生成JSON报告
json_path = self.generate_json_report(results, summary)
reports['json'] = json_path
# 生成HTML报告
html_path = self.generate_html_report(results, summary)
reports['html'] = html_path
return reports