260 lines
7.9 KiB
Python
260 lines
7.9 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"
|
|
|
|
# 测试结果存储
|
|
test_results = []
|
|
api_docs = []
|
|
|
|
def test_root_endpoint():
|
|
"""测试根路径接口"""
|
|
print("\n" + "="*80)
|
|
print("[1/3] 测试根路径接口...")
|
|
|
|
url = f"{API_BASE_URL}/"
|
|
|
|
# 测试多次以获取平均性能
|
|
test_count = 5
|
|
print(f" 执行{test_count}次测试以获取平均性能...")
|
|
|
|
times = []
|
|
for i in range(test_count):
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.get(url, 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': '/',
|
|
'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': '/',
|
|
'method': 'GET',
|
|
'description': '根路径接口（服务状态查询）',
|
|
'auth': '无需认证',
|
|
'params': '无',
|
|
'response': '返回服务版本和运行状态信息'
|
|
})
|
|
|
|
def test_health_check():
|
|
"""测试健康检查接口"""
|
|
print("\n" + "="*80)
|
|
print("[2/3] 测试健康检查接口...")
|
|
|
|
url = f"{API_BASE_URL}/health"
|
|
|
|
# 测试多次以获取平均性能
|
|
test_count = 5
|
|
print(f" 执行{test_count}次测试以获取平均性能...")
|
|
|
|
times = []
|
|
for i in range(test_count):
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.get(url, 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': '/health',
|
|
'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': '/health',
|
|
'method': 'GET',
|
|
'description': '健康检查接口（包含数据库连接测试）',
|
|
'auth': '无需认证',
|
|
'params': '无',
|
|
'response': '返回服务健康状态、数据库连接状态、时间戳和版本号'
|
|
})
|
|
|
|
def test_404_endpoint():
|
|
"""测试404错误处理"""
|
|
print("\n" + "="*80)
|
|
print("[3/3] 测试404错误处理...")
|
|
|
|
url = f"{API_BASE_URL}/nonexistent-endpoint-test"
|
|
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.get(url, timeout=30)
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|
|
|
success = response.status_code == 404
|
|
error = '' if success else response.text
|
|
|
|
print(f" 状态码: {response.status_code} - 耗时: {elapsed_time:.2f}ms")
|
|
|
|
test_results.append({
|
|
'endpoint': '/nonexistent-endpoint (404测试)',
|
|
'method': 'GET',
|
|
'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': '/nonexistent-endpoint (404测试)',
|
|
'method': 'GET',
|
|
'status_code': 0,
|
|
'elapsed_time_ms': '0',
|
|
'success': False,
|
|
'error': str(e)
|
|
})
|
|
|
|
api_docs.append({
|
|
'endpoint': '404错误处理',
|
|
'method': 'ALL',
|
|
'description': '处理所有不存在的接口请求',
|
|
'auth': '无',
|
|
'params': '任意',
|
|
'response': '返回404状态码和错误信息'
|
|
})
|
|
|
|
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)
|
|
|
|
# 执行所有测试
|
|
test_root_endpoint()
|
|
test_health_check()
|
|
test_404_endpoint()
|
|
|
|
# 打印摘要
|
|
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()
|