refactor(test): 重构测试套件结构并优化测试配置
feat(test-suite): 新增测试套件模块，包含API测试客户端和测试配置 fix(api): 修复数据库实体和仓库的删除操作返回值 style(api): 统一数据库表名和字段命名 perf(api): 添加缓存注解提升配置查询性能 test(api): 添加H2测试数据库配置支持 chore: 清理旧的测试文件和脚本
This commit is contained in:
张翔
@@ -0,0 +1,225 @@
|
||||
# API 集成测试 - 基础API客户端
|
||||
import pytest
|
||||
import requests
|
||||
import time
|
||||
import os
|
||||
from typing import Optional, Dict, Any, Union
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
from dotenv import load_dotenv
|
||||
import httpx
|
||||
|
||||
# 加载环境变量
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class BaseAPIClient:
|
||||
"""基础 API 客户端，提供通用的 HTTP 请求方法"""
|
||||
|
||||
def __init__(self, base_url: Optional[str] = None, timeout: int = 30):
|
||||
self.base_url = base_url or os.getenv('BASE_URL', 'http://localhost:8084')
|
||||
self.timeout = timeout
|
||||
self.session = requests.Session()
|
||||
self.token: Optional[str] = None
|
||||
self.user_id: Optional[int] = None
|
||||
|
||||
# 配置重试策略
|
||||
retries = Retry(
|
||||
total=3,
|
||||
backoff_factor=0.1,
|
||||
status_forcelist=[500, 502, 503, 504]
|
||||
)
|
||||
self.session.mount('http', HTTPAdapter(max_retries=retries))
|
||||
self.session.mount('https', HTTPAdapter(max_retries=retries))
|
||||
|
||||
def login(self, username: str, password: str) -> bool:
|
||||
"""登录并获取 Token"""
|
||||
response = self.post(
|
||||
'/api/auth/login',
|
||||
json={'username': username, 'password': password},
|
||||
include_auth=False
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
self.token = data.get('token')
|
||||
self.user_id = data.get('userId')
|
||||
print(f"✅ 登录成功: {username} (User ID: {self.user_id})")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ 登录失败: {response.status_code}")
|
||||
return False
|
||||
|
||||
def _build_url(self, path: str) -> str:
|
||||
"""构建完整 URL"""
|
||||
if path.startswith('http'):
|
||||
return path
|
||||
return f"{self.base_url}{path}"
|
||||
|
||||
def _get_headers(self, include_auth: bool = True) -> Dict[str, str]:
|
||||
"""获取请求头"""
|
||||
headers = {
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
|
||||
if include_auth and self.token:
|
||||
headers['Authorization'] = f"Bearer {self.token}"
|
||||
|
||||
return headers
|
||||
|
||||
def get(self, path: str, params: Optional[Dict] = None, include_auth: bool = True) -> requests.Response:
|
||||
"""GET 请求"""
|
||||
url = self._build_url(path)
|
||||
headers = self._get_headers(include_auth)
|
||||
|
||||
response = self.session.get(
|
||||
url,
|
||||
headers=headers,
|
||||
params=params,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
def post(self, path: str, data: Optional[Dict] = None, json: Optional[Dict] = None,
|
||||
include_auth: bool = True) -> requests.Response:
|
||||
"""POST 请求"""
|
||||
url = self._build_url(path)
|
||||
headers = self._get_headers(include_auth)
|
||||
|
||||
response = self.session.post(
|
||||
url,
|
||||
headers=headers,
|
||||
data=data,
|
||||
json=json,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
def put(self, path: str, data: Optional[Dict] = None, json: Optional[Dict] = None,
|
||||
include_auth: bool = True) -> requests.Response:
|
||||
"""PUT 请求"""
|
||||
url = self._build_url(path)
|
||||
headers = self._get_headers(include_auth)
|
||||
|
||||
response = self.session.put(
|
||||
url,
|
||||
headers=headers,
|
||||
data=data,
|
||||
json=json,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
def delete(self, path: str, include_auth: bool = True) -> requests.Response:
|
||||
"""DELETE 请求"""
|
||||
url = self._build_url(path)
|
||||
headers = self._get_headers(include_auth)
|
||||
|
||||
response = self.session.delete(
|
||||
url,
|
||||
headers=headers,
|
||||
timeout=self.timeout
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
def cleanup_resource(self, resource_type: str, resource_id: int) -> bool:
|
||||
"""清理测试资源"""
|
||||
try:
|
||||
response = self.delete(f'/api/{resource_type}/{resource_id}')
|
||||
return response.status_code == 200
|
||||
except Exception as e:
|
||||
print(f"清理资源失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
class APIFixture:
|
||||
"""API 测试固定装置，提供测试数据管理"""
|
||||
|
||||
def __init__(self, api_client: BaseAPIClient):
|
||||
self.api_client = api_client
|
||||
self.created_users = []
|
||||
self.created_roles = []
|
||||
self.created_menus = []
|
||||
self.created_configs = []
|
||||
self.created_dicts = []
|
||||
|
||||
def cleanup(self):
|
||||
"""清理所有创建的测试数据"""
|
||||
print("\n🧹 清理测试数据...")
|
||||
|
||||
# 清理用户
|
||||
for user_id in self.created_users:
|
||||
self.api_client.cleanup_resource('users', user_id)
|
||||
self.created_users.clear()
|
||||
|
||||
# 清理角色
|
||||
for role_id in self.created_roles:
|
||||
self.api_client.cleanup_resource('roles', role_id)
|
||||
self.created_roles.clear()
|
||||
|
||||
# 清理菜单
|
||||
for menu_id in self.created_menus:
|
||||
self.api_client.cleanup_resource('menus', menu_id)
|
||||
self.created_menus.clear()
|
||||
|
||||
# 清理配置
|
||||
for config_id in self.created_configs:
|
||||
self.api_client.cleanup_resource('config', config_id)
|
||||
self.created_configs.clear()
|
||||
|
||||
# 清理字典
|
||||
for dict_id in self.created_dicts:
|
||||
self.api_client.cleanup_resource('dict', dict_id)
|
||||
self.created_dicts.clear()
|
||||
|
||||
print("✅ 测试数据清理完成")
|
||||
|
||||
|
||||
class AsyncAPIClient:
|
||||
"""异步 API 客户端，使用 httpx"""
|
||||
|
||||
def __init__(self, client: httpx.AsyncClient):
|
||||
self.client = client
|
||||
self.token: Optional[str] = None
|
||||
self.user_id: Optional[int] = None
|
||||
|
||||
def set_auth(self, token: str, user_id: int = None):
|
||||
"""设置认证信息"""
|
||||
self.token = token
|
||||
self.user_id = user_id
|
||||
self.client.headers.update({'Authorization': f'Bearer {token}'})
|
||||
|
||||
async def login(self, username: str, password: str) -> httpx.Response:
|
||||
"""登录并获取 Token"""
|
||||
response = await self.client.post(
|
||||
'/api/auth/login',
|
||||
json={'username': username, 'password': password}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
self.token = data.get('token')
|
||||
self.user_id = data.get('userId')
|
||||
print(f"✅ 登录成功: {username} (User ID: {self.user_id})")
|
||||
|
||||
return response
|
||||
|
||||
async def get(self, path: str, params: Optional[Dict] = None) -> httpx.Response:
|
||||
"""GET 请求"""
|
||||
return await self.client.get(path, params=params)
|
||||
|
||||
async def post(self, path: str, json: Optional[Dict] = None) -> httpx.Response:
|
||||
"""POST 请求"""
|
||||
return await self.client.post(path, json=json)
|
||||
|
||||
async def put(self, path: str, json: Optional[Dict] = None) -> httpx.Response:
|
||||
"""PUT 请求"""
|
||||
return await self.client.put(path, json=json)
|
||||
|
||||
async def delete(self, path: str) -> httpx.Response:
|
||||
"""DELETE 请求"""
|
||||
return await self.client.delete(path)
|
||||
Reference in New Issue
Block a user