shengyudong/ai_wht_B
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5a384b694e99fadfe53dc2d96edcc96224390e85
ai_wht_B/测试整体后端接口性能.py
“shengyudong” 5a384b694e 2026-1-6
2026-01-06 14:18:39 +08:00

2534 lines
99 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
整合后端接口性能测试工具
整合了所有10个模块的接口测试：
1. 基础接口 (flask_wht_server_api.py)
2. 认证接口 (auth_routes.py)
3. 工作台接口 (dashboard_routes.py)
4. 企业接口 (enterprise_routes.py)
5. 图片接口 (image_routes.py)
6. 日志接口 (log_routes.py)
7. 文章接口 (article_routes.py)
8. 员工接口 (employee_routes.py)
9. 作者接口 (author_routes.py)
10. 搜索服务接口 (flask_wht_server_search.py)
"""
import requests
import time
import csv
from datetime import datetime
import io
from PIL import Image
import sys
# API配置
API_BASE_URL = "http://127.0.0.1:8216" # 主服务
SEARCH_API_BASE_URL = "http://127.0.0.1:8321" # 搜索服务
USERNAME = "13621242430"
PASSWORD = "admin123"
# 测试结果存储
test_results = []
api_docs = []
# 全局token
global_token = None
# 测试模块选择（可配置）
TEST_MODULES = {
'basic': True, # 基础接口
'auth': True, # 认证接口
'dashboard': True, # 工作台
'enterprise': True, # 企业
'image': True, # 图片
'log': True, # 日志
'article': True, # 文章
'employee': True, # 员工
'author': True, # 作者
'search': False, # 搜索服务（默认关闭，需要单独服务）
}
def print_module_header(module_name, module_num, total_modules):
"""打印模块测试头部"""
print("\n" + "="*80)
print(f"[{module_num}/{total_modules}] {module_name}")
print("="*80)
def login():
"""登录获取token"""
global global_token
print("\n" + "="*80)
print("正在登录...")
url = f"{API_BASE_URL}/api/auth/login"
data = {"username": USERNAME, "password": PASSWORD}
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
if response.status_code == 200:
global_token = response.json()['data']['token']
print(f"✓ 登录成功，耗时: {elapsed_time:.2f}ms")
return global_token
else:
print(f"✗ 登录失败: {response.status_code}")
return None
# ============================================================================
# 模块1: 基础接口测试
# ============================================================================
def test_basic_apis():
"""测试基础接口"""
if not TEST_MODULES['basic']:
return
print_module_header("基础接口测试", 1, 10)
# 根路径
print("\n[1/3] 测试根路径...")
url = f"{API_BASE_URL}/"
for i in range(3):
try:
start_time = time.time()
response = requests.get(url, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '基础接口',
'endpoint': '/',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '基础接口',
'endpoint': '/',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 健康检查
print("[2/3] 测试健康检查...")
url = f"{API_BASE_URL}/health"
for i in range(3):
try:
start_time = time.time()
response = requests.get(url, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '基础接口',
'endpoint': '/health',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '基础接口',
'endpoint': '/health',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 404测试
print("[3/3] 测试404错误处理...")
url = f"{API_BASE_URL}/nonexistent-test"
try:
start_time = time.time()
response = requests.get(url, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '基础接口',
'endpoint': '/nonexistent-test',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 404,
'error': ''
})
except Exception as e:
test_results.append({
'module': '基础接口',
'endpoint': '/nonexistent-test',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 添加API文档
api_docs.extend([
{'module': '基础接口', 'endpoint': '/', 'method': 'GET', 'description': '根路径', 'auth': '无'},
{'module': '基础接口', 'endpoint': '/health', 'method': 'GET', 'description': '健康检查', 'auth': '无'},
])
# ============================================================================
# 模块2: 认证接口测试 (原脚本测试数: 11个)
# ============================================================================
def test_auth_apis(token):
"""测试认证接口 - 包含统一登录、员工登录、登出、错误场景等"""
if not TEST_MODULES['auth']:
return
print_module_header("认证接口测试", 2, 10)
# 1. 统一登录 - 用户名
print("\n[1/11] 测试统一登录(用户名)...")
url = f"{API_BASE_URL}/api/auth/login"
data = {"username": USERNAME, "password": PASSWORD}
try:
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (用户名)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (用户名)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 2. 统一登录 - 手机号
print("[2/11] 测试统一登录(手机号)...")
data = {"phone": USERNAME, "password": PASSWORD}
try:
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (手机号)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (手机号)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 3. 统一登录 - 错误密码
print("[3/11] 测试统一登录(错误密码)...")
data = {"username": USERNAME, "password": "wrong_password"}
try:
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (错误密码)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 401,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (错误密码)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 4. 统一登录 - 缺少参数
print("[4/11] 测试统一登录(缺少参数)...")
data = {"username": USERNAME}
try:
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (缺少参数)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 400,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/login (缺少参数)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 5. 员工登录 - 正常
print("[5/11] 测试员工登录...")
url = f"{API_BASE_URL}/api/auth/employee/login"
data = {"phone": USERNAME, "password": PASSWORD}
try:
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/employee/login',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/employee/login',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6. 员工登录 - 错误密码
print("[6/11] 测试员工登录(错误密码)...")
data = {"phone": USERNAME, "password": "wrong_password"}
try:
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/employee/login (错误密码)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 401,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/employee/login (错误密码)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 7. 登出 - 带token
print("[7/11] 测试用户登出(带token)...")
url = f"{API_BASE_URL}/api/auth/logout"
headers = {"Authorization": f"Bearer {token}"}
try:
start_time = time.time()
response = requests.post(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/logout (带token)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/logout (带token)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 8. 登出 - 不带token
print("[8/11] 测试用户登出(不带token)...")
try:
start_time = time.time()
response = requests.post(url, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/logout (不带token)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '认证接口',
'endpoint': '/api/auth/logout (不带token)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 9-11. 并发登录测试
print("[9-11/11] 测试并发登录(3次)...")
url = f"{API_BASE_URL}/api/auth/login"
data = {"username": USERNAME, "password": PASSWORD}
import concurrent.futures
def single_login():
start_time = time.time()
response = requests.post(url, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
return response.status_code, elapsed_time
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = [executor.submit(single_login) for _ in range(3)]
results = [f.result() for f in concurrent.futures.as_completed(futures)]
for i, (status, elapsed) in enumerate(results, 1):
test_results.append({
'module': '认证接口',
'endpoint': f'/api/auth/login (并发{i})',
'method': 'POST',
'status_code': status,
'elapsed_time_ms': f"{elapsed:.2f}",
'success': status == 200,
'error': ''
})
except Exception as e:
for i in range(3):
test_results.append({
'module': '认证接口',
'endpoint': f'/api/auth/login (并发{i+1})',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '认证接口', 'endpoint': '/api/auth/login', 'method': 'POST', 'description': '统一登录(支持用户名/手机号)', 'auth': '无'},
{'module': '认证接口', 'endpoint': '/api/auth/employee/login', 'method': 'POST', 'description': '员工登录', 'auth': '无'},
{'module': '认证接口', 'endpoint': '/api/auth/logout', 'method': 'POST', 'description': '用户登出', 'auth': '可选'},
])
# ============================================================================
# 模块3: 工作台接口测试 (原脚本测试数: 7个)
# ============================================================================
def test_dashboard_apis(token):
"""测试工作台接口 - 概览、最近发布、热门产品"""
if not TEST_MODULES['dashboard']:
return
print_module_header("工作台接口测试", 3, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-5. 工作台概览 (多次测试获取平均性能)
print("\n[1-5/9] 测试工作台概览(5次)...")
url = f"{API_BASE_URL}/api/dashboard/overview"
for i in range(5):
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '工作台',
'endpoint': f'/api/dashboard/overview (第{i+1}次)',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '工作台',
'endpoint': f'/api/dashboard/overview (第{i+1}次)',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6-8. 最近发布 (不同limit参数)
print("[6-8/9] 测试最近发布(不同参数)...")
url = f"{API_BASE_URL}/api/dashboard/recent-publishes"
test_cases = [
{"limit": 5, "desc": "5条记录"},
{"limit": 10, "desc": "10条记录"},
{"limit": 20, "desc": "20条记录"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '工作台',
'endpoint': f'/api/dashboard/recent-publishes ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '工作台',
'endpoint': f'/api/dashboard/recent-publishes ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 9. 热门产品
print("[9/9] 测试热门产品...")
url = f"{API_BASE_URL}/api/dashboard/hot-products"
test_cases = [
{"limit": 4, "desc": "4个产品"},
{"limit": 8, "desc": "8个产品"},
{"limit": 12, "desc": "12个产品"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '工作台',
'endpoint': f'/api/dashboard/hot-products ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '工作台',
'endpoint': f'/api/dashboard/hot-products ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '工作台', 'endpoint': '/api/dashboard/overview', 'method': 'GET', 'description': '工作台概览(统计数据)', 'auth': '需要'},
{'module': '工作台', 'endpoint': '/api/dashboard/recent-publishes', 'method': 'GET', 'description': '最近发布(可指定条数)', 'auth': '需要'},
{'module': '工作台', 'endpoint': '/api/dashboard/hot-products', 'method': 'GET', 'description': '热门产品(按发布次数)', 'auth': '需要'},
])
# ============================================================================
# 模块4: 企业接口测试 (原脚本测试数: 15个)
# ============================================================================
def test_enterprise_apis(token):
"""测试企业接口 - 列表、统计、信息、CRUD操作"""
if not TEST_MODULES['enterprise']:
return
print_module_header("企业接口测试", 4, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-5. 企业列表 (不同查询参数)
print("\n[1-5/17] 测试企业列表(不同参数)...")
url = f"{API_BASE_URL}/api/enterprises/list"
test_cases = [
{"page": 1, "pageSize": 20, "desc": "基本查询"},
{"page": 1, "pageSize": 10, "desc": "每页10条"},
{"page": 1, "pageSize": 20, "keyword": "企业", "desc": "关键词搜索"},
{"page": 1, "pageSize": 20, "status": "active", "desc": "状态筛选"},
{"page": 2, "pageSize": 20, "desc": "第2页"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '企业接口',
'endpoint': f'/api/enterprises/list ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '企业接口',
'endpoint': f'/api/enterprises/list ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6-10. 企业统计 (多次测试)
print("[6-10/17] 测试企业统计(5次)...")
url = f"{API_BASE_URL}/api/enterprises/stats"
for i in range(5):
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '企业接口',
'endpoint': f'/api/enterprises/stats (第{i+1}次)',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '企业接口',
'endpoint': f'/api/enterprises/stats (第{i+1}次)',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 11-15. 企业信息 (多次测试)
print("[11-15/17] 测试企业信息(5次)...")
url = f"{API_BASE_URL}/api/enterprises/info"
for i in range(5):
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '企业接口',
'endpoint': f'/api/enterprises/info (第{i+1}次)',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '企业接口',
'endpoint': f'/api/enterprises/info (第{i+1}次)',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 16. 更新企业信息
print("[16/17] 测试更新企业信息...")
url = f"{API_BASE_URL}/api/enterprises/info"
data = {"short_name": f"测试更新{int(time.time())}", "email": "test@example.com"}
try:
start_time = time.time()
response = requests.put(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '企业接口',
'endpoint': '/api/enterprises/info (PUT)',
'method': 'PUT',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '企业接口',
'endpoint': '/api/enterprises/info (PUT)',
'method': 'PUT',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 17. 修改密码
print("[17/17] 测试修改密码...")
url = f"{API_BASE_URL}/api/enterprises/change-password"
data = {"old_password": "wrong", "new_password": "test123"}
try:
start_time = time.time()
response = requests.put(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '企业接口',
'endpoint': '/api/enterprises/change-password',
'method': 'PUT',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '企业接口',
'endpoint': '/api/enterprises/change-password',
'method': 'PUT',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '企业接口', 'endpoint': '/api/enterprises/list', 'method': 'GET', 'description': '企业列表(支持分页、搜索、筛选)', 'auth': '需要'},
{'module': '企业接口', 'endpoint': '/api/enterprises/stats', 'method': 'GET', 'description': '企业统计(总数、激活数等)', 'auth': '需要'},
{'module': '企业接口', 'endpoint': '/api/enterprises/info', 'method': 'GET', 'description': '企业信息', 'auth': '需要'},
{'module': '企业接口', 'endpoint': '/api/enterprises/info', 'method': 'PUT', 'description': '更新企业信息', 'auth': '需要'},
{'module': '企业接口', 'endpoint': '/api/enterprises/change-password', 'method': 'PUT', 'description': '修改密码', 'auth': '需要'},
])
# ============================================================================
# 模块5: 图片接口测试 (原脚本测试数: 17个)
# ============================================================================
def test_image_apis(token):
"""测试图片接口 - 图片、标签、类型管理"""
if not TEST_MODULES['image']:
return
print_module_header("图片接口测试", 5, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-2. 图片列表
print("\n[1-2/17] 测试图片列表(不同参数)...")
url = f"{API_BASE_URL}/api/images/list"
test_cases = [
{"page": 1, "pageSize": 20, "desc": "第1页"},
{"page": 1, "pageSize": 10, "desc": "每页10条"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': f'/api/images/list ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': f'/api/images/list ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 3. 图片仪表盘
print("[3/17] 测试图片仪表盘...")
url = f"{API_BASE_URL}/api/images/list_dashboard"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/list_dashboard',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/list_dashboard',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 4. 上传图片
print("[4/17] 测试上传图片...")
url = f"{API_BASE_URL}/api/images/upload"
try:
img = Image.new('RGB', (100, 100))
img_bytes = io.BytesIO()
img.save(img_bytes, format='PNG')
img_bytes.seek(0)
files = {'image': ('test.png', img_bytes, 'image/png')}
data = {'image_name': '测试图片', 'image_type_name': '测试类型', 'tag_keywords': '测试,性能'}
start_time = time.time()
response = requests.post(url, headers=headers, files=files, data=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/upload',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/upload',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 5. 删除图片
print("[5/17] 测试删除图片...")
url = f"{API_BASE_URL}/api/images/999999"
try:
start_time = time.time()
response = requests.delete(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/<id>',
'method': 'DELETE',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 404],
'error': ''
})
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/<id>',
'method': 'DELETE',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6. 标签名称列表
print("[6/17] 测试标签名称列表...")
url = f"{API_BASE_URL}/api/images/tags/names/list"
try:
start_time = time.time()
response = requests.get(url, headers=headers, params={"page": 1, "pageSize": 20}, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/tags/names/list',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/tags/names/list',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 7. 创建标签名称
print("[7/17] 测试创建标签名称...")
url = f"{API_BASE_URL}/api/images/tags/names/create"
data = {"tag_name": f"测试标签{int(time.time())}", "description": "测试"}
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/tags/names/create',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': '/api/images/tags/names/create',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 8-11. 标签名称CRUD操作
print("[8-11/17] 测试标签名称CRUD...")
endpoints = [
('/api/images/tags/names/999999', 'PUT', '更新标签名称'),
('/api/images/tags/names/999999', 'DELETE', '删除标签名称'),
('/api/images/tags/relations/list', 'GET', '标签关系列表'),
('/api/images/types/list', 'GET', '图片类型列表'),
]
for endpoint, method, desc in endpoints:
url = f"{API_BASE_URL}{endpoint}"
try:
start_time = time.time()
if method == 'GET':
response = requests.get(url, headers=headers, params={"page": 1, "pageSize": 20}, timeout=30)
elif method == 'PUT':
response = requests.put(url, headers=headers, json={"description": "更新"}, timeout=30)
else:
response = requests.delete(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': f'{endpoint} ({desc})',
'method': method,
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 404],
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': f'{endpoint} ({desc})',
'method': method,
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 12-17. 图片类型和标签管理
print("[12-17/17] 测试类型和标签管理...")
test_cases = [
('/api/images/types/create', 'POST', {"type_name": f"测试类型{int(time.time())}"}, '创建类型'),
('/api/images/types/999999', 'PUT', {"type_name": "更新类型"}, '更新类型'),
('/api/images/types/999999', 'DELETE', None, '删除类型'),
('/api/images/tags/list', 'GET', None, '标签列表'),
('/api/images/tags/create', 'POST', {"image_id": 1, "tag_id": 1}, '创建标签'),
('/api/images/tags/999999', 'DELETE', None, '删除标签'),
]
for endpoint, method, data, desc in test_cases:
url = f"{API_BASE_URL}{endpoint}"
try:
start_time = time.time()
if method == 'GET':
response = requests.get(url, headers=headers, params={"page": 1, "pageSize": 20}, timeout=30)
elif method == 'POST':
response = requests.post(url, headers=headers, json=data, timeout=30)
elif method == 'PUT':
response = requests.put(url, headers=headers, json=data, timeout=30)
else:
response = requests.delete(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '图片接口',
'endpoint': f'{endpoint} ({desc})',
'method': method,
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400, 404, 409],
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '图片接口',
'endpoint': f'{endpoint} ({desc})',
'method': method,
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '图片接口', 'endpoint': '/api/images/list', 'method': 'GET', 'description': '图片列表', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/list_dashboard', 'method': 'GET', 'description': '图片仪表盘', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/upload', 'method': 'POST', 'description': '上传图片', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/<id>', 'method': 'DELETE', 'description': '删除图片', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/names/list', 'method': 'GET', 'description': '标签名称列表', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/names/create', 'method': 'POST', 'description': '创建标签名称', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/names/<id>', 'method': 'PUT', 'description': '更新标签名称', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/names/<id>', 'method': 'DELETE', 'description': '删除标签名称', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/relations/list', 'method': 'GET', 'description': '标签关系列表', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/types/list', 'method': 'GET', 'description': '图片类型列表', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/types/create', 'method': 'POST', 'description': '创建类型', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/types/<id>', 'method': 'PUT', 'description': '更新类型', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/types/<id>', 'method': 'DELETE', 'description': '删除类型', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/list', 'method': 'GET', 'description': '标签列表', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/create', 'method': 'POST', 'description': '创建标签', 'auth': '需要'},
{'module': '图片接口', 'endpoint': '/api/images/tags/<id>', 'method': 'DELETE', 'description': '删除标签', 'auth': '需要'},
])
# ============================================================================
# 模块6: 日志接口测试 (原脚本测试数: 7个)
# ============================================================================
def test_log_apis(token):
"""测试日志接口 - 操作日志、文件日志"""
if not TEST_MODULES['log']:
return
print_module_header("日志接口测试", 6, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-3. 获取操作日志 (不同参数)
print("\n[1-3/7] 测试获取操作日志...")
url = f"{API_BASE_URL}/api/log/logs"
test_cases = [
{"page": 1, "size": 10, "desc": "第1页10条"},
{"page": 1, "size": 20, "desc": "第1页20条"},
{"page": 2, "size": 10, "desc": "第2页10条"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '日志接口',
'endpoint': f'/api/log/logs ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 500],
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '日志接口',
'endpoint': f'/api/log/logs ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 4. 日志文件列表
print("[4/7] 测试日志文件列表...")
url = f"{API_BASE_URL}/api/log/logs/files"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '日志接口',
'endpoint': '/api/log/logs/files',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '日志接口',
'endpoint': '/api/log/logs/files',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 5-7. 文件日志内容 (不同类型和行数)
print("[5-7/7] 测试文件日志内容...")
url = f"{API_BASE_URL}/api/log/logs/file"
test_cases = [
{"type": "article", "lines": 50, "desc": "article日志50行"},
{"type": "error", "lines": 50, "desc": "error日志50行"},
{"type": "article", "lines": 100, "desc": "article日志100行"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '日志接口',
'endpoint': f'/api/log/logs/file ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 404],
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '日志接口',
'endpoint': f'/api/log/logs/file ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '日志接口', 'endpoint': '/api/log/logs', 'method': 'GET', 'description': '获取操作日志(分页)', 'auth': '需要'},
{'module': '日志接口', 'endpoint': '/api/log/logs/files', 'method': 'GET', 'description': '日志文件列表', 'auth': '需要'},
{'module': '日志接口', 'endpoint': '/api/log/logs/file', 'method': 'GET', 'description': '文件日志内容', 'auth': '需要'},
])
# ============================================================================
# 模块7: 文章接口测试 (原脚本测试数: 10个)
# ============================================================================
def test_article_apis(token):
"""测试文章接口 - 列表、详情、发布"""
if not TEST_MODULES['article']:
return
print_module_header("文章接口测试", 7, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-3. 文章列表 (不同参数)
print("\n[1-3/10] 测试文章列表...")
url = f"{API_BASE_URL}/api/articles/list"
test_cases = [
{"page": 1, "pageSize": 20, "desc": "基本查询"},
{"page": 1, "pageSize": 10, "keyword": "文章", "desc": "关键词搜索"},
{"page": 2, "pageSize": 20, "desc": "第2页"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '文章接口',
'endpoint': f'/api/articles/list ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '文章接口',
'endpoint': f'/api/articles/list ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 4. 文章仪表盘
print("[4/10] 测试文章仪表盘...")
url = f"{API_BASE_URL}/api/articles/list_dashboard"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '文章接口',
'endpoint': '/api/articles/list_dashboard',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '文章接口',
'endpoint': '/api/articles/list_dashboard',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 5. 生成文章
print("[5/10] 测试生成文章...")
url = f"{API_BASE_URL}/api/articles/generate"
data = {"product_id": 1, "prompt_workflow_id": 1, "topics": ["测试主题1", "测试主题2"], "count": 2}
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '文章接口',
'endpoint': '/api/articles/generate',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '文章接口',
'endpoint': '/api/articles/generate',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6-10. 文章详情和更新（如果有文章）
print("[6-10/10] 测试文章详情和操作...")
list_response = requests.get(f"{API_BASE_URL}/api/articles/list", headers=headers,
params={"page": 1, "pageSize": 1}, timeout=30)
if list_response.status_code == 200:
articles = list_response.json().get('data', {}).get('list', [])
if articles:
article_id = articles[0]['id']
# 文章详情
url = f"{API_BASE_URL}/api/articles/{article_id}"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '文章接口',
'endpoint': f'/api/articles/{article_id}',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '文章接口',
'endpoint': f'/api/articles/{article_id}',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 更新文章
data = {"title": f"测试更新标题_{int(time.time())}", "status": "draft"}
try:
start_time = time.time()
response = requests.put(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '文章接口',
'endpoint': f'/api/articles/{article_id} (PUT)',
'method': 'PUT',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '文章接口',
'endpoint': f'/api/articles/{article_id} (PUT)',
'method': 'PUT',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 批量发布测试
print(" 测试批量发布接口...")
endpoints = [
('/api/articles/batch-published-account-cycle', {"article_ids": [1, 2], "author_ids": [1, 2]}, '账号循环'),
('/api/articles/batch-published-review', {"article_ids": [1, 2], "author_id": 1}, '指定作者'),
]
for endpoint, data, desc in endpoints:
url = f"{API_BASE_URL}{endpoint}"
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '文章接口',
'endpoint': f'{endpoint} ({desc})',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '文章接口',
'endpoint': f'{endpoint} ({desc})',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '文章接口', 'endpoint': '/api/articles/list', 'method': 'GET', 'description': '文章列表', 'auth': '需要'},
{'module': '文章接口', 'endpoint': '/api/articles/list_dashboard', 'method': 'GET', 'description': '文章仪表盘', 'auth': '需要'},
{'module': '文章接口', 'endpoint': '/api/articles/generate', 'method': 'POST', 'description': '生成文章', 'auth': '需要'},
{'module': '文章接口', 'endpoint': '/api/articles/<id>', 'method': 'GET', 'description': '文章详情', 'auth': '需要'},
{'module': '文章接口', 'endpoint': '/api/articles/<id>', 'method': 'PUT', 'description': '更新文章', 'auth': '需要'},
{'module': '文章接口', 'endpoint': '/api/articles/batch-published-account-cycle', 'method': 'POST', 'description': '批量发布(账号循环)', 'auth': '需要'},
{'module': '文章接口', 'endpoint': '/api/articles/batch-published-review', 'method': 'POST', 'description': '批量发布(指定作者)', 'auth': '需要'},
])
# ============================================================================
# 模块8: 员工接口测试 (原脚本测试数: 8个)
# ============================================================================
def test_employee_apis(token):
"""测试员工接口 - 列表、统计、CRUD"""
if not TEST_MODULES['employee']:
return
print_module_header("员工接口测试", 8, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-5. 员工列表 (不同参数)
print("\n[1-5/8] 测试员工列表(不同参数)...")
url = f"{API_BASE_URL}/api/employees/list"
test_cases = [
{"page": 1, "pageSize": 20, "desc": "基本查询"},
{"page": 1, "pageSize": 10, "desc": "每页10条"},
{"page": 1, "pageSize": 20, "keyword": "员工", "desc": "关键词搜索"},
{"page": 1, "pageSize": 20, "status": "active", "desc": "状态筛选"},
{"page": 2, "pageSize": 20, "desc": "第2页"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '员工接口',
'endpoint': f'/api/employees/list ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '员工接口',
'endpoint': f'/api/employees/list ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6. 员工统计
print("[6/8] 测试员工统计...")
url = f"{API_BASE_URL}/api/employees/stats"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '员工接口',
'endpoint': '/api/employees/stats',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '员工接口',
'endpoint': '/api/employees/stats',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 7. 添加员工
print("[7/8] 测试添加员工...")
url = f"{API_BASE_URL}/api/employees/add"
timestamp = int(time.time())
data = {
"name": f"测试员工{timestamp}",
"phone": f"1382{timestamp % 10000000:07d}",
"password": "test123456",
"role": "employee",
"department": "测试部门"
}
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '员工接口',
'endpoint': '/api/employees/add',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '员工接口',
'endpoint': '/api/employees/add',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 8. 删除员工
print("[8/8] 测试删除员工...")
url = f"{API_BASE_URL}/api/employees/999999"
try:
start_time = time.time()
response = requests.delete(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '员工接口',
'endpoint': '/api/employees/<id>',
'method': 'DELETE',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 404],
'error': ''
})
except Exception as e:
test_results.append({
'module': '员工接口',
'endpoint': '/api/employees/<id>',
'method': 'DELETE',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '员工接口', 'endpoint': '/api/employees/list', 'method': 'GET', 'description': '员工列表(分页、搜索)', 'auth': '需要'},
{'module': '员工接口', 'endpoint': '/api/employees/stats', 'method': 'GET', 'description': '员工统计', 'auth': '需要'},
{'module': '员工接口', 'endpoint': '/api/employees/add', 'method': 'POST', 'description': '添加员工', 'auth': '需要'},
{'module': '员工接口', 'endpoint': '/api/employees/<id>', 'method': 'DELETE', 'description': '删除员工', 'auth': '需要'},
])
# ============================================================================
# 模块9: 作者接口测试 (原脚本测试数: 20个)
# ============================================================================
def test_author_apis(token):
"""测试作者接口 - 列表、搜索、CRUD"""
if not TEST_MODULES['author']:
return
print_module_header("作者接口测试", 9, 10)
headers = {"Authorization": f"Bearer {token}"}
# 1-3. 作者列表 (不同参数)
print("\n[1-3/12] 测试作者列表...")
url = f"{API_BASE_URL}/api/authors"
test_cases = [
{"page": 1, "size": 10, "desc": "第1页"},
{"page": 1, "size": 5, "desc": "每页5条"},
{"page": 1, "size": 10, "search": "医生", "desc": "搜索关键词"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': f'/api/authors ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': f'/api/authors ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 4. 作者列表(用户关联)
print("[4/12] 测试作者列表(用户关联)...")
url = f"{API_BASE_URL}/api/authors/list"
try:
start_time = time.time()
response = requests.get(url, headers=headers, params={"page": 1, "size": 10}, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': '/api/authors/list',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': '/api/authors/list',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 5-6. 作者详细列表
print("[5-6/12] 测试作者详细列表...")
url = f"{API_BASE_URL}/api/authors/detail_list"
test_cases = [
{"page": 1, "size": 10, "desc": "基本查询"},
{"page": 1, "size": 20, "search": "医生", "desc": "搜索查询"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': f'/api/authors/detail_list ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': f'/api/authors/detail_list ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 7. 科室列表
print("[7/12] 测试科室列表...")
url = f"{API_BASE_URL}/api/authors/departments"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': '/api/authors/departments',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': '/api/authors/departments',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 8-9. 搜索作者
print("[8-9/12] 测试搜索作者...")
url = f"{API_BASE_URL}/api/authors/search"
test_cases = [
{"keyword": "医生", "limit": 10, "desc": "搜索医生"},
{"keyword": "科室", "limit": 5, "desc": "搜索科室"},
]
for i, params in enumerate(test_cases, 1):
desc = params.pop('desc')
try:
start_time = time.time()
response = requests.get(url, headers=headers, params=params, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': f'/api/authors/search ({desc})',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': f'/api/authors/search ({desc})',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 10-12. 作者CRUD操作
print("[10-12/12] 测试作者CRUD...")
# 创建作者
url = f"{API_BASE_URL}/api/authors"
timestamp = int(time.time())
data = {
"author_name": f"测试医生{timestamp}",
"app_id": f"test_app_{timestamp}",
"app_token": f"test_token_{timestamp}",
"department_name": "测试科室",
"title": "主任医师",
"hospital": "测试医院",
"channel": 1
}
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': '/api/authors (POST)',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 201, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': '/api/authors (POST)',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 更新和删除
for endpoint, method, desc in [
('/api/authors/999999', 'PUT', '更新作者'),
('/api/authors/999999', 'DELETE', '删除作者'),
]:
url = f"{API_BASE_URL}{endpoint}"
try:
start_time = time.time()
if method == 'PUT':
response = requests.put(url, headers=headers, json={"author_name": "更新测试"}, timeout=30)
else:
response = requests.delete(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '作者接口',
'endpoint': f'{endpoint} ({desc})',
'method': method,
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 404],
'error': ''
})
time.sleep(0.1)
except Exception as e:
test_results.append({
'module': '作者接口',
'endpoint': f'{endpoint} ({desc})',
'method': method,
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '作者接口', 'endpoint': '/api/authors', 'method': 'GET', 'description': '作者列表(按科室分组)', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors', 'method': 'POST', 'description': '创建作者', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors/<id>', 'method': 'PUT', 'description': '更新作者', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors/<id>', 'method': 'DELETE', 'description': '删除作者', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors/list', 'method': 'GET', 'description': '作者列表(用户关联)', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors/detail_list', 'method': 'GET', 'description': '作者详细列表', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors/departments', 'method': 'GET', 'description': '科室列表', 'auth': '需要'},
{'module': '作者接口', 'endpoint': '/api/authors/search', 'method': 'GET', 'description': '搜索作者', 'auth': '需要'},
])
# ============================================================================
# 模块10: 搜索服务接口测试 (原脚本测试数: 6个)
# ============================================================================
def test_search_apis(token):
"""测试搜索服务接口 - 独立服务（8321端口）"""
if not TEST_MODULES['search']:
return
print_module_header("搜索服务接口测试", 10, 10)
print("注意: 搜索服务使用独立端口 8321")
headers = {"Authorization": f"Bearer {token}"}
# 1. 健康检查
print("\n[1/6] 测试健康检查...")
url = f"{SEARCH_API_BASE_URL}/health"
try:
start_time = time.time()
response = requests.get(url, timeout=5)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '搜索服务',
'endpoint': '/health',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '搜索服务',
'endpoint': '/health',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 2. 查询日志
print("[2/6] 测试查询日志...")
url = f"{SEARCH_API_BASE_URL}/api/search-service/logs"
try:
start_time = time.time()
response = requests.get(url, headers=headers, params={"page": 1, "size": 10}, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/logs',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 500],
'error': ''
})
except Exception as e:
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/logs',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 3. 日志文件列表
print("[3/6] 测试日志文件列表...")
url = f"{SEARCH_API_BASE_URL}/api/search-service/logs/files"
try:
start_time = time.time()
response = requests.get(url, headers=headers, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/logs/files',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code == 200,
'error': ''
})
except Exception as e:
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/logs/files',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 4. 文件日志内容
print("[4/6] 测试文件日志内容...")
url = f"{SEARCH_API_BASE_URL}/api/search-service/logs/file"
try:
start_time = time.time()
response = requests.get(url, headers=headers, params={"type": "article", "lines": 50}, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/logs/file',
'method': 'GET',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 404],
'error': ''
})
except Exception as e:
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/logs/file',
'method': 'GET',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 5. 生成文章
print("[5/6] 测试生成文章...")
url = f"{SEARCH_API_BASE_URL}/api/search-service/generate_article"
data = {"query": "测试查询", "max_results": 5}
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/generate_article',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/generate_article',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
# 6. 关键词搜索
print("[6/6] 测试关键词搜索...")
url = f"{SEARCH_API_BASE_URL}/api/search-service/search"
data = {"keyword": "测试", "limit": 10}
try:
start_time = time.time()
response = requests.post(url, headers=headers, json=data, timeout=30)
elapsed_time = (time.time() - start_time) * 1000
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/search',
'method': 'POST',
'status_code': response.status_code,
'elapsed_time_ms': f"{elapsed_time:.2f}",
'success': response.status_code in [200, 400],
'error': ''
})
except Exception as e:
test_results.append({
'module': '搜索服务',
'endpoint': '/api/search-service/search',
'method': 'POST',
'status_code': 0,
'elapsed_time_ms': '0',
'success': False,
'error': str(e)
})
api_docs.extend([
{'module': '搜索服务', 'endpoint': '/health', 'method': 'GET', 'description': '健康检查', 'auth': '不需要'},
{'module': '搜索服务', 'endpoint': '/api/search-service/logs', 'method': 'GET', 'description': '查询日志', 'auth': '需要'},
{'module': '搜索服务', 'endpoint': '/api/search-service/logs/files', 'method': 'GET', 'description': '日志文件列表', 'auth': '需要'},
{'module': '搜索服务', 'endpoint': '/api/search-service/logs/file', 'method': 'GET', 'description': '文件日志内容', 'auth': '需要'},
{'module': '搜索服务', 'endpoint': '/api/search-service/generate_article', 'method': 'POST', 'description': '生成文章', 'auth': '需要'},
{'module': '搜索服务', 'endpoint': '/api/search-service/search', 'method': 'POST', 'description': '关键词搜索', 'auth': '需要'},
])
# ============================================================================
# 结果保存和输出
# ============================================================================
def save_to_csv():
"""保存测试结果到CSV文件"""
timestamp = datetime.now().strftime('%Y%m%d')
filename = f'整体后端接口性能测试_{timestamp}.csv'
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=[
'module', 'endpoint', 'method', 'status_code', 'elapsed_time_ms', 'success', 'error'
])
writer.writeheader()
writer.writerows(test_results)
print(f"\n✓ 性能测试结果已保存到: {filename}")
return filename
def save_api_docs():
"""保存接口文档到CSV文件"""
timestamp = datetime.now().strftime('%Y%m%d')
filename = f'整体后端接口文档_{timestamp}.csv'
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=[
'module', 'endpoint', 'method', 'description', 'auth'
])
writer.writeheader()
writer.writerows(api_docs)
print(f"✓ 接口文档已保存到: {filename}")
return filename
def generate_markdown_report():
"""生成Markdown格式的综合数据报表"""
from collections import defaultdict
timestamp = datetime.now().strftime('%Y%m%d')
filename = f'整体后端接口测试报表_{timestamp}.md'
# 总体统计
total = len(test_results)
success = sum(1 for r in test_results if r['success'])
failed = total - success
times = [float(r['elapsed_time_ms']) for r in test_results if r['elapsed_time_ms'] != '0']
avg_time = sum(times) / len(times) if times else 0
max_time = max(times) if times else 0
min_time = min(times) if times else 0
# 找出最快和最慢的接口
max_item = max(test_results, key=lambda x: float(x['elapsed_time_ms']) if x['elapsed_time_ms'] != '0' else 0)
min_item = min([r for r in test_results if r['elapsed_time_ms'] != '0'],
key=lambda x: float(x['elapsed_time_ms'])) if times else None
# 按模块统计
modules = {}
for r in test_results:
module = r['module']
if module not in modules:
modules[module] = {
'total': 0,
'success': 0,
'times': [],
'endpoints': [],
'errors': []
}
modules[module]['total'] += 1
if r['success']:
modules[module]['success'] += 1
else:
modules[module]['errors'].append(r['endpoint'])
if r['elapsed_time_ms'] != '0':
modules[module]['times'].append(float(r['elapsed_time_ms']))
modules[module]['endpoints'].append((r['endpoint'], float(r['elapsed_time_ms'])))
# HTTP方法统计
methods = defaultdict(lambda: {'total': 0, 'success': 0})
for r in test_results:
method = r['method']
methods[method]['total'] += 1
if r['success']:
methods[method]['success'] += 1
# 性能分布
fast = sum(1 for t in times if t < 100)
normal = sum(1 for t in times if 100 <= t < 500)
slow = sum(1 for t in times if 500 <= t < 1000)
very_slow = sum(1 for t in times if t >= 1000)
# 状态码分布
status_codes = defaultdict(int)
for r in test_results:
status_codes[r['status_code']] += 1
# 接口文档统计
doc_modules = defaultdict(int)
doc_methods = defaultdict(int)
auth_required = sum(1 for d in api_docs if d['auth'] == '需要')
for doc in api_docs:
doc_modules[doc['module']] += 1
doc_methods[doc['method']] += 1
# 开始写入Markdown文件
with open(filename, 'w', encoding='utf-8') as f:
# 标题和概览
f.write("# 整体后端接口性能测试报表\n\n")
f.write(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
f.write("---\n\n")
# 总体统计
f.write("## 📈 总体测试统计\n\n")
f.write("| 指标 | 数值 |\n")
f.write("|------|------|\n")
f.write(f"| 测试总数 | {total} 个 |\n")
f.write(f"| 成功数量 | {success} 个 ✓ |\n")
f.write(f"| 失败数量 | {failed} 个 ✗ |\n")
f.write(f"| **成功率** | **{success/total*100:.2f}%** |\n")
f.write(f"| 平均耗时 | {avg_time:.2f}ms |\n")
f.write(f"| 最大耗时 | {max_time:.2f}ms |\n")
f.write(f"| 最小耗时 | {min_time:.2f}ms |\n\n")
if max_item:
f.write("**最慢接口**:\n")
f.write(f"- `{max_item['endpoint']}` ({max_item['module']}) - {max_item['elapsed_time_ms']}ms\n\n")
if min_item:
f.write("**最快接口**:\n")
f.write(f"- `{min_item['endpoint']}` ({min_item['module']}) - {min_item['elapsed_time_ms']}ms\n\n")
f.write("---\n\n")
# 各模块性能分析
f.write("## 📊 各模块性能分析\n\n")
sorted_modules = sorted(modules.items(),
key=lambda x: (x[1]['success']/x[1]['total'] if x[1]['total'] > 0 else 0),
reverse=True)
f.write("| 模块 | 测试数 | 成功 | 失败 | 成功率 | 平均耗时 | 最大耗时 | 最小耗时 | 状态 |\n")
f.write("|------|--------|------|------|--------|----------|----------|----------|------|\n")
for module, stats in sorted_modules:
success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0
avg_time = (sum(stats['times']) / len(stats['times'])) if stats['times'] else 0
max_time = max(stats['times']) if stats['times'] else 0
min_time = min(stats['times']) if stats['times'] else 0
if success_rate >= 90:
status = "🟢 优秀"
elif success_rate >= 70:
status = "🟡 良好"
elif success_rate >= 50:
status = "🟠 一般"
else:
status = "🔴 需优化"
f.write(f"| {module} | {stats['total']} | {stats['success']} | {stats['total']-stats['success']} | "
f"{success_rate:.1f}% | {avg_time:.2f}ms | {max_time:.2f}ms | {min_time:.2f}ms | {status} |\n")
f.write("\n")
# 失败接口详情
f.write("### ⚠️ 失败接口详情\n\n")
has_errors = False
for module, stats in sorted_modules:
if stats['errors']:
has_errors = True
f.write(f"**{module}** ({len(stats['errors'])}个失败): \n")
for err_endpoint in stats['errors']:
f.write(f"- `{err_endpoint}`\n")
f.write("\n")
if not has_errors:
f.write("✅ 所有接口测试通过！\n\n")
f.write("---\n\n")
# HTTP方法分布
f.write("## 🔧 HTTP方法分布\n\n")
f.write("| 方法 | 总数 | 成功 | 失败 | 成功率 |\n")
f.write("|------|------|------|------|--------|\n")
for method, stats in sorted(methods.items()):
success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0
f.write(f"| {method} | {stats['total']} | {stats['success']} | "
f"{stats['total']-stats['success']} | {success_rate:.1f}% |\n")
f.write("\n---\n\n")
# 性能分布
f.write("## ⚡ 性能分布分析\n\n")
f.write("| 性能等级 | 数量 | 占比 |\n")
f.write("|----------|------|------|\n")
f.write(f"| 🚀 极快 (<100ms) | {fast} | {fast/len(times)*100:.1f}% |\n")
f.write(f"| ✅ 正常 (100-500ms) | {normal} | {normal/len(times)*100:.1f}% |\n")
f.write(f"| ⚠️ 较慢 (500-1000ms) | {slow} | {slow/len(times)*100:.1f}% |\n")
f.write(f"| 🐌 很慢 (>1000ms) | {very_slow} | {very_slow/len(times)*100:.1f}% |\n")
f.write("\n")
# 性能分布图（文本版）
f.write("### 性能分布可视化\n\n")
f.write("```\n")
max_bar_len = 50
if times:
fast_bar = '█' * int((fast/len(times)) * max_bar_len)
normal_bar = '█' * int((normal/len(times)) * max_bar_len)
slow_bar = '█' * int((slow/len(times)) * max_bar_len)
very_slow_bar = '█' * int((very_slow/len(times)) * max_bar_len)
f.write(f"极快 (<100ms): {fast_bar} {fast}个\n")
f.write(f"正常 (100-500ms): {normal_bar} {normal}个\n")
f.write(f"较慢 (500-1000ms):{slow_bar} {slow}个\n")
f.write(f"很慢 (>1000ms): {very_slow_bar} {very_slow}个\n")
f.write("```\n\n")
f.write("---\n\n")
# HTTP状态码分布
f.write("## 📋 HTTP状态码分布\n\n")
f.write("| 状态码 | 数量 | 占比 | 说明 |\n")
f.write("|--------|------|------|------|\n")
for code, count in sorted(status_codes.items()):
percentage = (count / total * 100)
if code == 200:
desc = "✅ 成功"
elif code >= 500:
desc = "❌ 服务器错误"
elif code >= 400:
desc = "⚠️ 客户端错误"
elif code == 0:
desc = "❌ 请求失败"
else:
desc = "ℹ️ 其他"
f.write(f"| {code} | {count} | {percentage:.1f}% | {desc} |\n")
f.write("\n---\n\n")
# 接口文档统计
f.write("## 📚 接口文档统计\n\n")
f.write(f"**已记录接口总数**: {len(api_docs)} 个 \n")
f.write(f"**需要认证**: {auth_required} 个 ({auth_required/len(api_docs)*100:.1f}%)\n\n")
f.write("### 按模块分类\n\n")
f.write("| 模块 | 接口数 |\n")
f.write("|------|--------|\n")
for module, count in sorted(doc_modules.items(), key=lambda x: x[1], reverse=True):
f.write(f"| {module} | {count} |\n")
f.write("\n")
f.write("### 按方法分类\n\n")
f.write("| 方法 | 接口数 |\n")
f.write("|------|--------|\n")
for method, count in sorted(doc_methods.items()):
f.write(f"| {method} | {count} |\n")
f.write("\n")
f.write("---\n\n")
# 优化建议
f.write("## 💡 优化建议\n\n")
suggestions = []
# 检查失败率高的模块
for module, stats in modules.items():
success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0
if success_rate < 70:
suggestions.append(f"🔴 **{module}** 模块成功率仅 {success_rate:.1f}%，建议优先排查认证、参数和数据库连接问题")
# 检查慢接口
slow_endpoints = [(r['module'], r['endpoint'], float(r['elapsed_time_ms']))
for r in test_results if r['elapsed_time_ms'] != '0' and float(r['elapsed_time_ms']) > 500]
if slow_endpoints:
slow_endpoints.sort(key=lambda x: x[2], reverse=True)
suggestions.append(f"⚠️ 发现 {len(slow_endpoints)} 个耗时超过500ms的接口，建议添加缓存或优化查询")
suggestions.append(" 慢接口TOP 3:")
for module, endpoint, time in slow_endpoints[:3]:
suggestions.append(f" - `{endpoint}` ({module}) - {time:.2f}ms")
# 检查错误率
if failed > 0:
error_rate = failed / total * 100
if error_rate > 20:
suggestions.append(f"❌ 总体错误率达到 {error_rate:.1f}%，建议进行全面的接口测试和修复")
if suggestions:
for suggestion in suggestions:
f.write(f"{suggestion}\n")
else:
f.write("✅ 所有接口性能良好，无需特别优化\n")
f.write("\n---\n\n")
# 测试配置信息
f.write("## ⚙️ 测试配置\n\n")
f.write(f"- **主服务地址**: `{API_BASE_URL}`\n")
f.write(f"- **搜索服务地址**: `{SEARCH_API_BASE_URL}`\n")
f.write(f"- **测试账号**: `{USERNAME}`\n")
enabled_modules = [name for name, enabled in TEST_MODULES.items() if enabled]
disabled_modules = [name for name, enabled in TEST_MODULES.items() if not enabled]
f.write(f"- **启用模块** ({len(enabled_modules)}个): {', '.join(enabled_modules)}\n")
if disabled_modules:
f.write(f"- **禁用模块** ({len(disabled_modules)}个): {', '.join(disabled_modules)}\n")
f.write("\n---\n\n")
f.write("*报表由整体后端接口性能测试工具自动生成*\n")
print(f"\n✅ Markdown报表已生成: {filename}")
return filename
def generate_data_report():
"""生成综合数据报表"""
from collections import defaultdict
print("\n" + "█"*80)
print("📊 综合数据报表")
print("█"*80)
# 总体统计
total = len(test_results)
success = sum(1 for r in test_results if r['success'])
failed = total - success
times = [float(r['elapsed_time_ms']) for r in test_results if r['elapsed_time_ms'] != '0']
avg_time = sum(times) / len(times) if times else 0
max_time = max(times) if times else 0
min_time = min(times) if times else 0
# 找出最快和最慢的接口
max_item = min(test_results, key=lambda x: float(x['elapsed_time_ms']) if x['elapsed_time_ms'] != '0' else float('inf'))
min_item = max(test_results, key=lambda x: float(x['elapsed_time_ms']) if x['elapsed_time_ms'] != '0' else 0)
print("\n" + "="*80)
print("📈 总体测试统计")
print("="*80)
print(f" 测试总数: {total} 个")
print(f" 成功数量: {success} 个 ✓")
print(f" 失败数量: {failed} 个 ✗")
print(f" 成功率: {success/total*100:.2f}%")
print(f" 平均耗时: {avg_time:.2f}ms")
print(f" 最大耗时: {max_time:.2f}ms")
print(f" └─ 接口: {max_item['endpoint']} ({max_item['module']})")
print(f" 最小耗时: {min_time:.2f}ms")
print(f" └─ 接口: {min_item['endpoint']} ({min_item['module']})")
# 按模块统计
print("\n" + "="*80)
print("📊 各模块性能分析")
print("="*80)
modules = {}
for r in test_results:
module = r['module']
if module not in modules:
modules[module] = {
'total': 0,
'success': 0,
'times': [],
'endpoints': [],
'errors': []
}
modules[module]['total'] += 1
if r['success']:
modules[module]['success'] += 1
else:
modules[module]['errors'].append(r['endpoint'])
if r['elapsed_time_ms'] != '0':
modules[module]['times'].append(float(r['elapsed_time_ms']))
modules[module]['endpoints'].append((r['endpoint'], float(r['elapsed_time_ms'])))
# 按成功率排序
sorted_modules = sorted(modules.items(),
key=lambda x: (x[1]['success']/x[1]['total'] if x[1]['total'] > 0 else 0),
reverse=True)
for module, stats in sorted_modules:
success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0
avg_time = (sum(stats['times']) / len(stats['times'])) if stats['times'] else 0
max_time = max(stats['times']) if stats['times'] else 0
min_time = min(stats['times']) if stats['times'] else 0
# 确定性能等级
if success_rate >= 90:
status = "🟢 优秀"
elif success_rate >= 70:
status = "🟡 良好"
elif success_rate >= 50:
status = "🟠 一般"
else:
status = "🔴 需优化"
print(f"\n{status} {module}")
print(f" 测试数: {stats['total']} | 成功: {stats['success']} | 失败: {stats['total']-stats['success']}")
print(f" 成功率: {success_rate:.2f}%")
print(f" 平均耗时: {avg_time:.2f}ms | 最大: {max_time:.2f}ms | 最小: {min_time:.2f}ms")
# 显示失败的接口
if stats['errors']:
print(f" ⚠️ 失败接口 ({len(stats['errors'])}个):")
for err_endpoint in stats['errors'][:3]: # 最多显示3个
print(f" • {err_endpoint}")
if len(stats['errors']) > 3:
print(f" • ... 还有 {len(stats['errors'])-3} 个")
# HTTP方法统计
print("\n" + "="*80)
print("🔧 HTTP方法分布")
print("="*80)
methods = defaultdict(lambda: {'total': 0, 'success': 0})
for r in test_results:
method = r['method']
methods[method]['total'] += 1
if r['success']:
methods[method]['success'] += 1
for method, stats in sorted(methods.items()):
success_rate = (stats['success'] / stats['total'] * 100) if stats['total'] > 0 else 0
print(f" {method:6} | 总数: {stats['total']:3} | 成功: {stats['success']:3} | 成功率: {success_rate:5.1f}%")
# 性能分布
print("\n" + "="*80)
print("⚡ 性能分布分析")
print("="*80)
fast = sum(1 for t in times if t < 100)
normal = sum(1 for t in times if 100 <= t < 500)
slow = sum(1 for t in times if 500 <= t < 1000)
very_slow = sum(1 for t in times if t >= 1000)
print(f" 🚀 极快 (<100ms): {fast:3} 个 ({fast/len(times)*100:.1f}%)")
print(f" ✅ 正常 (100-500ms): {normal:3} 个 ({normal/len(times)*100:.1f}%)")
print(f" ⚠️ 较慢 (500-1000ms): {slow:3} 个 ({slow/len(times)*100:.1f}%)")
print(f" 🐌 很慢 (>1000ms): {very_slow:3} 个 ({very_slow/len(times)*100:.1f}%)")
# 状态码分布
print("\n" + "="*80)
print("📋 HTTP状态码分布")
print("="*80)
status_codes = defaultdict(int)
for r in test_results:
status_codes[r['status_code']] += 1
for code, count in sorted(status_codes.items()):
percentage = (count / total * 100)
if code == 200:
print(f" ✅ {code}: {count:3} 个 ({percentage:.1f}%)")
elif code >= 400:
print(f" ❌ {code}: {count:3} 个 ({percentage:.1f}%)")
else:
print(f" ℹ️ {code}: {count:3} 个 ({percentage:.1f}%)")
# 接口文档统计
print("\n" + "="*80)
print("📚 接口文档统计")
print("="*80)
print(f" 已记录接口: {len(api_docs)} 个")
doc_modules = defaultdict(int)
doc_methods = defaultdict(int)
auth_required = sum(1 for d in api_docs if d['auth'] == '需要')
for doc in api_docs:
doc_modules[doc['module']] += 1
doc_methods[doc['method']] += 1
print(f" 需要认证: {auth_required} 个")
print(f"\n 按模块分类:")
for module, count in sorted(doc_modules.items(), key=lambda x: x[1], reverse=True):
print(f" • {module}: {count} 个")
print(f"\n 按方法分类:")
for method, count in sorted(doc_methods.items()):
print(f" • {method}: {count} 个")
print("\n" + "█"*80)
def print_summary():
"""打印测试摘要（兼容旧版）"""
generate_data_report()
def print_config():
"""打印测试配置"""
print("测试配置:")
print("-"*80)
enabled_modules = [name for name, enabled in TEST_MODULES.items() if enabled]
disabled_modules = [name for name, enabled in TEST_MODULES.items() if not enabled]
print(f"启用的模块 ({len(enabled_modules)}个): {', '.join(enabled_modules)}")
if disabled_modules:
print(f"禁用的模块 ({len(disabled_modules)}个): {', '.join(disabled_modules)}")
print("-"*80)
def main():
"""主函数"""
print("="*80)
print("整合后端接口性能测试工具")
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"主服务地址: {API_BASE_URL}")
print(f"搜索服务地址: {SEARCH_API_BASE_URL}")
print("="*80)
# 打印配置
print_config()
# 登录
token = login()
if not token:
print("\n✗ 登录失败，无法继续测试")
return 1
# 执行所有测试
try:
test_basic_apis()
test_auth_apis(token)
test_dashboard_apis(token)
test_enterprise_apis(token)
test_image_apis(token)
test_log_apis(token)
test_article_apis(token)
test_employee_apis(token)
test_author_apis(token)
test_search_apis(token)
except Exception as e:
print(f"\n✗ 测试过程中出现异常: {e}")
import traceback
traceback.print_exc()
# 生成综合数据报表（控制台输出）
generate_data_report()
# 保存结果
csv_file = save_to_csv()
doc_file = save_api_docs()
md_report_file = generate_markdown_report() # 生成Markdown报表
print("\n" + "="*80)
print("✅ 测试完成！")
print("="*80)
print(f"📁 输出文件:")
print(f" • 性能数据: {csv_file}")
print(f" • 接口文档: {doc_file}")
print(f" • 测试报表: {md_report_file}")
print(f"\n💡 提示:")
print(f" • 查看Markdown报表: {md_report_file}")
print(f" • 使用 'python 查看综合测试报表.py' 查看控制台版详细报表")
print(f" • 成功率低于70%的模块需要优化")
print(f" • 耗时超过500ms的接口建议添加缓存或优化查询")
print("="*80)
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
print("\n\n✗ 测试被用户中断")
sys.exit(1)
except Exception as e:
print(f"\n✗ 测试异常: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
Reference in New Issue View Git Blame Copy Permalink