张翔
dc53a233b9
重构项目结构，将分散在各模块的领域模型统一移动到manage-common模块 更新相关依赖和引用路径 调整docker-compose配置和测试标记 添加新的Playwright测试配置 优化Dockerfile构建过程
200 lines
7.7 KiB
Python
200 lines
7.7 KiB
Python
"""
|
|
性能测试基础框架
|
|
"""
|
|
|
|
import pytest
|
|
import time
|
|
import asyncio
|
|
import statistics
|
|
from typing import List, Dict, Any
|
|
from httpx import AsyncClient
|
|
from loguru import logger
|
|
|
|
|
|
@pytest.mark.performance
|
|
@pytest.mark.slow
|
|
class PerformanceTest:
|
|
"""性能测试基类"""
|
|
|
|
@pytest.fixture
|
|
async def perf_client(self, authenticated_client: AsyncClient) -> AsyncClient:
|
|
"""性能测试客户端"""
|
|
return authenticated_client
|
|
|
|
@pytest.fixture
|
|
def performance_thresholds(self):
|
|
"""性能阈值配置"""
|
|
return {
|
|
"response_time_p95": 2000, # 95%的请求响应时间应小于2秒
|
|
"response_time_p99": 5000, # 99%的请求响应时间应小于5秒
|
|
"error_rate": 0.05, # 错误率应小于5%
|
|
"throughput_min": 10, # 最小吞吐量（请求/秒）
|
|
}
|
|
|
|
async def measure_request_time(self, client: AsyncClient, method: str,
|
|
url: str, **kwargs) -> float:
|
|
"""测量单个请求时间"""
|
|
start_time = time.time()
|
|
|
|
if method.upper() == "GET":
|
|
response = await client.get(url, **kwargs)
|
|
elif method.upper() == "POST":
|
|
response = await client.post(url, **kwargs)
|
|
elif method.upper() == "PUT":
|
|
response = await client.put(url, **kwargs)
|
|
elif method.upper() == "DELETE":
|
|
response = await client.delete(url, **kwargs)
|
|
else:
|
|
raise ValueError(f"Unsupported method: {method}")
|
|
|
|
end_time = time.time()
|
|
response_time = (end_time - start_time) * 1000 # 转换为毫秒
|
|
|
|
return response_time
|
|
|
|
async def measure_concurrent_requests(self, client: AsyncClient, method: str,
|
|
url: str, concurrency: int = 10,
|
|
**kwargs) -> Dict[str, Any]:
|
|
"""测量并发请求性能"""
|
|
async def make_request():
|
|
return await self.measure_request_time(client, method, url, **kwargs)
|
|
|
|
start_time = time.time()
|
|
results = await asyncio.gather(*[make_request() for _ in range(concurrency)])
|
|
end_time = time.time()
|
|
|
|
total_time = (end_time - start_time) * 1000 # 毫秒
|
|
response_times = results
|
|
|
|
return {
|
|
"concurrency": concurrency,
|
|
"total_time_ms": total_time,
|
|
"response_times_ms": response_times,
|
|
"min_time_ms": min(response_times),
|
|
"max_time_ms": max(response_times),
|
|
"avg_time_ms": statistics.mean(response_times),
|
|
"median_time_ms": statistics.median(response_times),
|
|
"p95_time_ms": self._percentile(response_times, 95),
|
|
"p99_time_ms": self._percentile(response_times, 99),
|
|
"throughput_rps": concurrency / (total_time / 1000),
|
|
"success_count": len(response_times),
|
|
}
|
|
|
|
def _percentile(self, data: List[float], percentile: float) -> float:
|
|
"""计算百分位数"""
|
|
sorted_data = sorted(data)
|
|
index = int(len(sorted_data) * percentile / 100)
|
|
return sorted_data[min(index, len(sorted_data) - 1)]
|
|
|
|
def assert_performance(self, results: Dict[str, Any], thresholds: Dict[str, Any]):
|
|
"""断言性能指标"""
|
|
p95_time = results["p95_time_ms"]
|
|
p99_time = results["p99_time_ms"]
|
|
throughput = results["throughput_rps"]
|
|
|
|
if p95_time > thresholds["response_time_p95"]:
|
|
pytest.fail(f"P95响应时间 {p95_time:.2f}ms 超过阈值 {thresholds['response_time_p95']}ms")
|
|
|
|
if p99_time > thresholds["response_time_p99"]:
|
|
pytest.fail(f"P99响应时间 {p99_time:.2f}ms 超过阈值 {thresholds['response_time_p99']}ms")
|
|
|
|
if throughput < thresholds["throughput_min"]:
|
|
pytest.fail(f"吞吐量 {throughput:.2f} rps 低于最小值 {thresholds['throughput_min']} rps")
|
|
|
|
logger.info(f"性能测试通过: P95={p95_time:.2f}ms, P99={p99_time:.2f}ms, 吞吐量={throughput:.2f} rps")
|
|
|
|
|
|
@pytest.mark.performance
|
|
@pytest.mark.slow
|
|
class TestAPIPerformance(PerformanceTest):
|
|
"""API性能测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_list_performance(self, perf_client: AsyncClient, performance_thresholds):
|
|
"""测试用户列表API性能"""
|
|
results = await self.measure_concurrent_requests(
|
|
perf_client, "GET", "/api/users", concurrency=20
|
|
)
|
|
|
|
self.assert_performance(results, performance_thresholds)
|
|
logger.info(f"用户列表API性能: {results}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_role_list_performance(self, perf_client: AsyncClient, performance_thresholds):
|
|
"""测试角色列表API性能"""
|
|
results = await self.measure_concurrent_requests(
|
|
perf_client, "GET", "/api/roles", concurrency=20
|
|
)
|
|
|
|
self.assert_performance(results, performance_thresholds)
|
|
logger.info(f"角色列表API性能: {results}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notice_list_performance(self, perf_client: AsyncClient, performance_thresholds):
|
|
"""测试通知列表API性能"""
|
|
results = await self.measure_concurrent_requests(
|
|
perf_client, "GET", "/api/notices", concurrency=20
|
|
)
|
|
|
|
self.assert_performance(results, performance_thresholds)
|
|
logger.info(f"通知列表API性能: {results}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_performance(self, perf_client: AsyncClient, performance_thresholds):
|
|
"""测试搜索API性能"""
|
|
results = await self.measure_concurrent_requests(
|
|
perf_client, "GET", "/api/users/page?keyword=test", concurrency=15
|
|
)
|
|
|
|
self.assert_performance(results, performance_thresholds)
|
|
logger.info(f"搜索API性能: {results}")
|
|
|
|
|
|
@pytest.mark.performance
|
|
@pytest.mark.slow
|
|
class TestLoadTesting(PerformanceTest):
|
|
"""负载测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sustained_load(self, perf_client: AsyncClient):
|
|
"""测试持续负载"""
|
|
duration_seconds = 30
|
|
requests_per_second = 5
|
|
total_requests = duration_seconds * requests_per_second
|
|
|
|
response_times = []
|
|
start_time = time.time()
|
|
|
|
for i in range(total_requests):
|
|
response_time = await self.measure_request_time(
|
|
perf_client, "GET", "/api/users"
|
|
)
|
|
response_times.append(response_time)
|
|
|
|
elapsed = time.time() - start_time
|
|
if elapsed < duration_seconds:
|
|
sleep_time = max(0, (i + 1) / requests_per_second - elapsed)
|
|
await asyncio.sleep(max(0, sleep_time))
|
|
|
|
avg_time = statistics.mean(response_times)
|
|
p95_time = self._percentile(response_times, 95)
|
|
|
|
logger.info(f"持续负载测试 - 平均响应时间: {avg_time:.2f}ms, P95: {p95_time:.2f}ms")
|
|
|
|
assert avg_time < 3000, f"平均响应时间 {avg_time:.2f}ms 超过阈值 3000ms"
|
|
assert p95_time < 5000, f"P95响应时间 {p95_time:.2f}ms 超过阈值 5000ms"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spike_load(self, perf_client: AsyncClient):
|
|
"""测试突发负载"""
|
|
spike_sizes = [10, 50, 100, 50, 10]
|
|
|
|
for spike_size in spike_sizes:
|
|
results = await self.measure_concurrent_requests(
|
|
perf_client, "GET", "/api/users", concurrency=spike_size
|
|
)
|
|
|
|
logger.info(f"突发负载测试 (并发={spike_size}): P95={results['p95_time_ms']:.2f}ms")
|
|
|
|
assert results["p95_time_ms"] < 10000, \
|
|
f"突发负载 {spike_size} 并发时 P95响应时间超时" |