659 lines
22 KiB
Python
659 lines
22 KiB
Python
|
#!/usr/bin/env python
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
测试企业接口性能并生成CSV报告和接口文档
|
|||
|
|
"""
|
|||
|
|
import requests
|
|||
|
|
import time
|
|||
|
|
import csv
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
# API配置
|
|||
|
|
API_BASE_URL = "http://127.0.0.1:8216"
|
|||
|
|
USERNAME = "13621242430"
|
|||
|
|
PASSWORD = "admin123"
|
|||
|
|
|
|||
|
|
# 测试结果存储
|
|||
|
|
test_results = []
|
|||
|
|
api_docs = []
|
|||
|
|
|
|||
|
|
def login():
|
|||
|
|
"""登录获取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:
|
|||
|
|
token = response.json()['data']['token']
|
|||
|
|
print(f"✓ 登录成功，耗时: {elapsed_time:.2f}ms")
|
|||
|
|
return token
|
|||
|
|
else:
|
|||
|
|
print(f"✗ 登录失败: {response.status_code}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def test_get_enterprises_list(token):
|
|||
|
|
"""测试获取企业列表"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[1/9] 测试获取企业列表...")
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/list"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
# 测试不同的查询参数组合
|
|||
|
|
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
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text
|
|||
|
|
|
|||
|
|
print(f" [{i}] {desc}: {response.status_code} - {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/list ({desc})',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
time.sleep(0.1)
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" [{i}] {desc}: 失败 - {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/list ({desc})',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/list',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'description': '获取企业列表（平台管理员）',
|
|||
|
|
'auth': '需要认证+角色(admin/enterprise)',
|
|||
|
|
'params': 'page, pageSize, keyword, status',
|
|||
|
|
'response': '返回企业列表（含企业ID、名称、状态、统计数据等）'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_get_enterprises_stats(token):
|
|||
|
|
"""测试获取企业统计"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[2/9] 测试获取企业统计...")
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/stats"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
# 测试多次以获取平均性能
|
|||
|
|
test_count = 5
|
|||
|
|
print(f" 执行{test_count}次测试以获取平均性能...")
|
|||
|
|
|
|||
|
|
times = []
|
|||
|
|
for i in range(test_count):
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.get(url, headers=headers, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
times.append(elapsed_time)
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
|
|||
|
|
print(f" [{i+1}] 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
time.sleep(0.1)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" [{i+1}] 失败: {str(e)}")
|
|||
|
|
times.append(0)
|
|||
|
|
|
|||
|
|
# 计算平均耗时
|
|||
|
|
valid_times = [t for t in times if t > 0]
|
|||
|
|
avg_time = sum(valid_times) / len(valid_times) if valid_times else 0
|
|||
|
|
max_time = max(valid_times) if valid_times else 0
|
|||
|
|
min_time = min(valid_times) if valid_times else 0
|
|||
|
|
|
|||
|
|
print(f" 平均耗时: {avg_time:.2f}ms, 最大: {max_time:.2f}ms, 最小: {min_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/stats',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'status_code': 200 if valid_times else 0,
|
|||
|
|
'elapsed_time_ms': f"{avg_time:.2f}",
|
|||
|
|
'success': len(valid_times) == test_count,
|
|||
|
|
'error': f'测试{test_count}次, 平均耗时{avg_time:.2f}ms'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/stats',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'description': '获取企业统计（总数、激活数、禁用数、总用户数等）',
|
|||
|
|
'auth': '需要认证+角色(admin/enterprise)',
|
|||
|
|
'params': '无',
|
|||
|
|
'response': '返回企业统计数据'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_get_enterprise_info(token):
|
|||
|
|
"""测试获取企业信息"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[3/9] 测试获取企业信息...")
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/info"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
# 测试多次以获取平均性能
|
|||
|
|
test_count = 5
|
|||
|
|
print(f" 执行{test_count}次测试以获取平均性能...")
|
|||
|
|
|
|||
|
|
times = []
|
|||
|
|
for i in range(test_count):
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.get(url, headers=headers, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
times.append(elapsed_time)
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
|
|||
|
|
print(f" [{i+1}] 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
time.sleep(0.1)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" [{i+1}] 失败: {str(e)}")
|
|||
|
|
times.append(0)
|
|||
|
|
|
|||
|
|
# 计算平均耗时
|
|||
|
|
valid_times = [t for t in times if t > 0]
|
|||
|
|
avg_time = sum(valid_times) / len(valid_times) if valid_times else 0
|
|||
|
|
max_time = max(valid_times) if valid_times else 0
|
|||
|
|
min_time = min(valid_times) if valid_times else 0
|
|||
|
|
|
|||
|
|
print(f" 平均耗时: {avg_time:.2f}ms, 最大: {max_time:.2f}ms, 最小: {min_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/info',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'status_code': 200 if valid_times else 0,
|
|||
|
|
'elapsed_time_ms': f"{avg_time:.2f}",
|
|||
|
|
'success': len(valid_times) == test_count,
|
|||
|
|
'error': f'测试{test_count}次, 平均耗时{avg_time:.2f}ms'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/info',
|
|||
|
|
'method': 'GET',
|
|||
|
|
'description': '获取企业信息（企业管理员）',
|
|||
|
|
'auth': '需要认证+角色(多种角色)',
|
|||
|
|
'params': '无',
|
|||
|
|
'response': '返回当前企业详细信息'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_create_enterprise(token):
|
|||
|
|
"""测试创建企业"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[4/9] 测试创建企业...")
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/create"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
# 生成唯一的测试数据
|
|||
|
|
timestamp = int(time.time())
|
|||
|
|
data = {
|
|||
|
|
"name": f"测试企业{timestamp}",
|
|||
|
|
"short_name": f"测试{timestamp}",
|
|||
|
|
"phone": f"1392{timestamp % 10000000:07d}",
|
|||
|
|
"password": "test123456",
|
|||
|
|
"email": f"test{timestamp}@example.com"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.post(url, headers=headers, json=data, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text
|
|||
|
|
|
|||
|
|
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/create',
|
|||
|
|
'method': 'POST',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
# 保存创建的企业ID用于后续测试
|
|||
|
|
created_enterprise_id = None
|
|||
|
|
if success:
|
|||
|
|
created_enterprise_id = response.json().get('data', {}).get('id')
|
|||
|
|
print(f" ✓ 创建成功，企业ID: {created_enterprise_id}")
|
|||
|
|
|
|||
|
|
return created_enterprise_id
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" 失败: {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/create',
|
|||
|
|
'method': 'POST',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
return None
|
|||
|
|
finally:
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/create',
|
|||
|
|
'method': 'POST',
|
|||
|
|
'description': '创建企业（平台管理员）',
|
|||
|
|
'auth': '需要认证+角色(admin/enterprise)',
|
|||
|
|
'params': 'name, short_name, phone, password, email',
|
|||
|
|
'response': '返回新创建的企业ID和企业ID字符串'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_update_enterprise(token, enterprise_id):
|
|||
|
|
"""测试更新企业信息"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[5/9] 测试更新企业信息...")
|
|||
|
|
|
|||
|
|
# 先添加文档
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/<enterprise_id>',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'description': '更新企业信息（平台管理员）',
|
|||
|
|
'auth': '需要认证+角色(admin/enterprise)',
|
|||
|
|
'params': 'name, short_name, email, password',
|
|||
|
|
'response': '返回更新成功消息'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
if not enterprise_id:
|
|||
|
|
print(" 跳过：没有可用的企业ID")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/<id> (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': '没有可用的企业ID'
|
|||
|
|
})
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/{enterprise_id}"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
data = {
|
|||
|
|
"short_name": f"更新测试{int(time.time())}",
|
|||
|
|
"email": "updated@example.com"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.put(url, headers=headers, json=data, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text
|
|||
|
|
|
|||
|
|
print(f" 企业ID: {enterprise_id}, 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/{enterprise_id} (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" 失败: {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/{enterprise_id} (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_update_enterprise_info(token):
|
|||
|
|
"""测试更新企业信息（企业管理员）"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[6/9] 测试更新企业信息（企业管理员）...")
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/info"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
data = {
|
|||
|
|
"short_name": f"管理员更新{int(time.time())}",
|
|||
|
|
"email": "admin_updated@example.com"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.put(url, headers=headers, json=data, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text
|
|||
|
|
|
|||
|
|
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/info (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" 失败: {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/info (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
finally:
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/info',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'description': '更新企业信息（企业管理员）',
|
|||
|
|
'auth': '需要认证+角色(enterprise/admin)',
|
|||
|
|
'params': 'short_name, email',
|
|||
|
|
'response': '返回更新成功消息'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_change_password(token):
|
|||
|
|
"""测试修改企业密码"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[7/9] 测试修改企业密码...")
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/change-password"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
data = {
|
|||
|
|
"old_password": "wrong_password", # 故意使用错误密码测试
|
|||
|
|
"new_password": "new_test_password"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.put(url, headers=headers, json=data, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text[:100] # 限制错误信息长度
|
|||
|
|
|
|||
|
|
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/change-password',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" 失败: {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/change-password',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
finally:
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/change-password',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'description': '修改企业密码（企业管理员）',
|
|||
|
|
'auth': '需要认证+角色(enterprise)',
|
|||
|
|
'params': 'old_password, new_password',
|
|||
|
|
'response': '返回密码修改成功消息'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_toggle_enterprise_status(token, enterprise_id):
|
|||
|
|
"""测试切换企业状态"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[8/9] 测试切换企业状态...")
|
|||
|
|
|
|||
|
|
# 先添加文档
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/<enterprise_id>/status',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'description': '切换企业状态（平台管理员）',
|
|||
|
|
'auth': '需要认证+角色(admin/enterprise)',
|
|||
|
|
'params': 'status (active/disabled)',
|
|||
|
|
'response': '返回操作成功消息'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
if not enterprise_id:
|
|||
|
|
print(" 跳过：没有可用的企业ID")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/<id>/status (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': '没有可用的企业ID'
|
|||
|
|
})
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/{enterprise_id}/status"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
data = {"status": "disabled"}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.put(url, headers=headers, json=data, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text
|
|||
|
|
|
|||
|
|
print(f" 企业ID: {enterprise_id}, 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/{enterprise_id}/status (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" 失败: {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/{enterprise_id}/status (PUT)',
|
|||
|
|
'method': 'PUT',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def test_delete_enterprise(token, enterprise_id):
|
|||
|
|
"""测试删除企业"""
|
|||
|
|
print("\n" + "-"*80)
|
|||
|
|
print("[9/9] 测试删除企业...")
|
|||
|
|
|
|||
|
|
# 先添加文档
|
|||
|
|
api_docs.append({
|
|||
|
|
'endpoint': '/api/enterprises/<enterprise_id>',
|
|||
|
|
'method': 'DELETE',
|
|||
|
|
'description': '删除企业（平台管理员，物理删除）',
|
|||
|
|
'auth': '需要认证+角色(admin/enterprise)',
|
|||
|
|
'params': 'enterprise_id (路径参数)',
|
|||
|
|
'response': '返回删除成功消息'
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
if not enterprise_id:
|
|||
|
|
print(" 跳过：没有可用的企业ID")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': '/api/enterprises/<id> (DELETE)',
|
|||
|
|
'method': 'DELETE',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': '没有可用的企业ID'
|
|||
|
|
})
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
url = f"{API_BASE_URL}/api/enterprises/{enterprise_id}"
|
|||
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
start_time = time.time()
|
|||
|
|
response = requests.delete(url, headers=headers, timeout=30)
|
|||
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|||
|
|
|
|||
|
|
success = response.status_code == 200
|
|||
|
|
error = '' if success else response.text
|
|||
|
|
|
|||
|
|
print(f" 企业ID: {enterprise_id}, 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|||
|
|
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/{enterprise_id} (DELETE)',
|
|||
|
|
'method': 'DELETE',
|
|||
|
|
'status_code': response.status_code,
|
|||
|
|
'elapsed_time_ms': f"{elapsed_time:.2f}",
|
|||
|
|
'success': success,
|
|||
|
|
'error': error
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f" 失败: {str(e)}")
|
|||
|
|
test_results.append({
|
|||
|
|
'endpoint': f'/api/enterprises/{enterprise_id} (DELETE)',
|
|||
|
|
'method': 'DELETE',
|
|||
|
|
'status_code': 0,
|
|||
|
|
'elapsed_time_ms': '0',
|
|||
|
|
'success': False,
|
|||
|
|
'error': str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
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=[
|
|||
|
|
'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=[
|
|||
|
|
'endpoint', 'method', 'description', 'auth', 'params', 'response'
|
|||
|
|
])
|
|||
|
|
writer.writeheader()
|
|||
|
|
writer.writerows(api_docs)
|
|||
|
|
|
|||
|
|
print(f"✓ 接口文档已保存到: {filename}")
|
|||
|
|
return filename
|
|||
|
|
|
|||
|
|
def print_summary():
|
|||
|
|
"""打印测试摘要"""
|
|||
|
|
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
|
|||
|
|
|
|||
|
|
print(f"总测试数: {total}")
|
|||
|
|
print(f"成功: {success} ({success/total*100:.1f}%)")
|
|||
|
|
print(f"失败: {failed} ({failed/total*100:.1f}%)")
|
|||
|
|
print(f"平均耗时: {avg_time:.2f}ms")
|
|||
|
|
print(f"最大耗时: {max_time:.2f}ms")
|
|||
|
|
print(f"最小耗时: {min_time:.2f}ms")
|
|||
|
|
print("="*80)
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
"""主函数"""
|
|||
|
|
print("="*80)
|
|||
|
|
print("企业接口性能测试工具")
|
|||
|
|
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|||
|
|
print(f"API地址: {API_BASE_URL}")
|
|||
|
|
print("="*80)
|
|||
|
|
|
|||
|
|
# 登录
|
|||
|
|
token = login()
|
|||
|
|
if not token:
|
|||
|
|
print("\n✗ 登录失败，无法继续测试")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 执行所有测试
|
|||
|
|
test_get_enterprises_list(token)
|
|||
|
|
test_get_enterprises_stats(token)
|
|||
|
|
test_get_enterprise_info(token)
|
|||
|
|
created_enterprise_id = test_create_enterprise(token)
|
|||
|
|
test_update_enterprise(token, created_enterprise_id)
|
|||
|
|
test_update_enterprise_info(token)
|
|||
|
|
test_change_password(token)
|
|||
|
|
test_toggle_enterprise_status(token, created_enterprise_id)
|
|||
|
|
test_delete_enterprise(token, created_enterprise_id)
|
|||
|
|
|
|||
|
|
# 打印摘要
|
|||
|
|
print_summary()
|
|||
|
|
|
|||
|
|
# 保存结果
|
|||
|
|
csv_file = save_to_csv()
|
|||
|
|
doc_file = save_api_docs()
|
|||
|
|
|
|||
|
|
print(f"\n✓ 测试完成！")
|
|||
|
|
print(f" - 性能数据: {csv_file}")
|
|||
|
|
print(f" - 接口文档: {doc_file}")
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
try:
|
|||
|
|
main()
|
|||
|
|
except KeyboardInterrupt:
|
|||
|
|
print("\n\n✗ 测试被用户中断")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"\n✗ 测试异常: {e}")
|
|||
|
|
import traceback
|
|||
|
|
traceback.print_exc()
|