Files
ai_wht_B/测试企业接口性能.py
“shengyudong” 5a384b694e 2026-1-6
2026-01-06 14:18:39 +08:00

659 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

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 -*-
"""
测试企业接口性能并生成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()