BrunoGaoandClaude
aa41afaf00
- 新增科技风格的抽屉组件样式和主题系统，支持健康监测、设备管理、用户管理、告警配置等多个业务主题 - 优化TenantInfo组件的事件处理，修复Vue 3兼容性问题 - 重构路由器创建逻辑，提升代码可维护性和扩展性 - 统一所有抽屉弹窗的UI风格，增加操作横幅、分组布局、图标提示等现代化设计元素 - 修复customer租户管理权限控制逻辑，改进按钮状态管理和权限验证流程 - 新增批量数据上传测试工具，支持curl和Python接口测试 - 优化设备信息编辑界面，增加分组显示、帮助文本、状态指示等功能 - 创建完整的界面增强指南和自动化脚本，便于后续维护和扩展 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <[email protected]>
349 lines
14 KiB
Python
349 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
ljwx-boot 批量上传接口测试脚本
|
|
|
|
测试三个核心接口：
|
|
1. upload_health_data - 健康数据批量上传
|
|
2. upload_device_info - 设备信息批量上传
|
|
3. upload_common_event - 通用事件上传
|
|
|
|
兼容原 Python ljwx-bigscreen 接口格式
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
import random
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Any
|
|
import argparse
|
|
|
|
|
|
class BatchUploadTester:
|
|
"""批量上传接口测试器"""
|
|
|
|
def __init__(self, base_url: str = "http://localhost:8080"):
|
|
self.base_url = base_url
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'ljwx-batch-upload-tester/1.0'
|
|
})
|
|
|
|
def generate_health_data(self, count: int = 10) -> List[Dict[str, Any]]:
|
|
"""生成测试健康数据"""
|
|
health_data = []
|
|
|
|
for i in range(count):
|
|
# 模拟不同设备和用户
|
|
device_id = f"DEVICE_{i % 5:03d}"
|
|
user_id = str(100 + (i % 20))
|
|
org_id = str(1 + (i % 3))
|
|
|
|
data = {
|
|
"device_id": device_id,
|
|
"user_id": user_id,
|
|
"org_id": org_id,
|
|
"customer_id": "8",
|
|
|
|
# 健康指标数据
|
|
"heart_rate": random.randint(60, 100),
|
|
"blood_oxygen": random.randint(95, 100),
|
|
"temperature": round(36.0 + random.uniform(0, 2.0), 1),
|
|
"pressure_high": random.randint(110, 140),
|
|
"pressure_low": random.randint(70, 90),
|
|
"stress": random.randint(1, 10),
|
|
"step": random.randint(1000, 15000),
|
|
"distance": round(random.uniform(0.5, 10.0), 2),
|
|
"calorie": round(random.uniform(200, 800), 1),
|
|
|
|
# 位置信息
|
|
"latitude": round(39.9 + random.uniform(-0.1, 0.1), 6),
|
|
"longitude": round(116.4 + random.uniform(-0.1, 0.1), 6),
|
|
"altitude": round(random.uniform(40, 60), 1),
|
|
|
|
# 时间戳
|
|
"create_time": (datetime.now() - timedelta(minutes=random.randint(1, 60))).strftime("%Y-%m-%d %H:%M:%S")
|
|
}
|
|
health_data.append(data)
|
|
|
|
return health_data
|
|
|
|
def generate_device_data(self, count: int = 5) -> List[Dict[str, Any]]:
|
|
"""生成测试设备数据"""
|
|
device_data = []
|
|
|
|
for i in range(count):
|
|
data = {
|
|
"device_id": f"DEVICE_{i:03d}",
|
|
"device_name": f"智能手环_{i}",
|
|
"device_type": "wearable",
|
|
"firmware_version": f"v2.{random.randint(1, 9)}.{random.randint(0, 9)}",
|
|
"battery_level": random.randint(10, 100),
|
|
"signal_strength": random.randint(-80, -30),
|
|
"customer_id": "8",
|
|
"status": random.choice(["online", "offline", "charging"]),
|
|
"last_sync": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
}
|
|
device_data.append(data)
|
|
|
|
return device_data
|
|
|
|
def generate_common_event_data(self) -> Dict[str, Any]:
|
|
"""生成通用事件测试数据"""
|
|
return {
|
|
"health_data": self.generate_health_data(5),
|
|
"device_info": self.generate_device_data(2),
|
|
"alert_data": [
|
|
{
|
|
"alert_type": "heart_rate_abnormal",
|
|
"device_id": "DEVICE_001",
|
|
"user_id": "101",
|
|
"severity": "high",
|
|
"message": "心率异常：检测到心率持续过高",
|
|
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
}
|
|
]
|
|
}
|
|
|
|
def test_upload_health_data(self, data_count: int = 10) -> Dict[str, Any]:
|
|
"""测试健康数据上传接口"""
|
|
print(f"\n🩺 测试健康数据上传接口 (数据量: {data_count})")
|
|
|
|
# 生成测试数据
|
|
health_data = self.generate_health_data(data_count)
|
|
|
|
# 测试新格式接口
|
|
url = f"{self.base_url}/batch/upload-health-data"
|
|
result1 = self._make_request("POST", url, health_data, "upload-health-data")
|
|
|
|
# 测试Python兼容接口
|
|
url_compat = f"{self.base_url}/batch/upload_health_data"
|
|
result2 = self._make_request("POST", url_compat, health_data, "upload_health_data (兼容)")
|
|
|
|
return {"modern": result1, "compatible": result2}
|
|
|
|
def test_upload_device_info(self, device_count: int = 5) -> Dict[str, Any]:
|
|
"""测试设备信息上传接口"""
|
|
print(f"\n📱 测试设备信息上传接口 (设备数: {device_count})")
|
|
|
|
# 生成测试数据
|
|
device_data = self.generate_device_data(device_count)
|
|
|
|
# 测试新格式接口
|
|
url = f"{self.base_url}/batch/upload-device-info"
|
|
result1 = self._make_request("POST", url, device_data, "upload-device-info")
|
|
|
|
# 测试Python兼容接口
|
|
url_compat = f"{self.base_url}/batch/upload_device_info"
|
|
result2 = self._make_request("POST", url_compat, device_data, "upload_device_info (兼容)")
|
|
|
|
return {"modern": result1, "compatible": result2}
|
|
|
|
def test_upload_common_event(self) -> Dict[str, Any]:
|
|
"""测试通用事件上传接口"""
|
|
print(f"\n🔄 测试通用事件上传接口")
|
|
|
|
# 生成测试数据
|
|
event_data = self.generate_common_event_data()
|
|
|
|
# 测试新格式接口
|
|
url = f"{self.base_url}/batch/upload-common-event"
|
|
result1 = self._make_request("POST", url, event_data, "upload-common-event")
|
|
|
|
# 测试Python兼容接口
|
|
url_compat = f"{self.base_url}/batch/upload_common_event"
|
|
result2 = self._make_request("POST", url_compat, event_data, "upload_common_event (兼容)")
|
|
|
|
return {"modern": result1, "compatible": result2}
|
|
|
|
def test_get_stats(self) -> Dict[str, Any]:
|
|
"""测试统计信息接口"""
|
|
print(f"\n📊 测试统计信息接口")
|
|
|
|
url = f"{self.base_url}/batch/stats"
|
|
return self._make_request("GET", url, None, "stats")
|
|
|
|
def test_health_check(self) -> Dict[str, Any]:
|
|
"""测试健康检查接口"""
|
|
print(f"\n💚 测试健康检查接口")
|
|
|
|
url = f"{self.base_url}/batch/health"
|
|
return self._make_request("GET", url, None, "health")
|
|
|
|
def test_performance(self, data_size: int = 1000) -> Dict[str, Any]:
|
|
"""测试性能测试接口"""
|
|
print(f"\n⚡ 测试性能测试接口 (数据规模: {data_size})")
|
|
|
|
url = f"{self.base_url}/batch/performance-test?dataSize={data_size}"
|
|
return self._make_request("POST", url, None, "performance-test")
|
|
|
|
def _make_request(self, method: str, url: str, data: Any, operation: str) -> Dict[str, Any]:
|
|
"""发送HTTP请求"""
|
|
start_time = time.time()
|
|
|
|
try:
|
|
if method.upper() == "GET":
|
|
response = self.session.get(url, timeout=30)
|
|
else:
|
|
response = self.session.post(url, json=data, timeout=30)
|
|
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|
|
|
result = {
|
|
"operation": operation,
|
|
"method": method,
|
|
"url": url,
|
|
"status_code": response.status_code,
|
|
"elapsed_ms": round(elapsed_time, 2),
|
|
"success": response.status_code == 200
|
|
}
|
|
|
|
if response.status_code == 200:
|
|
try:
|
|
response_data = response.json()
|
|
result["response"] = response_data
|
|
print(f" ✅ {operation}: {response.status_code} ({elapsed_time:.1f}ms)")
|
|
|
|
# 显示关键统计信息
|
|
if "result" in response_data and isinstance(response_data["result"], dict):
|
|
stats = response_data["result"]
|
|
if "processed" in stats:
|
|
print(f" 📈 处理: {stats.get('processed', 0)} 条")
|
|
if "duplicates" in stats:
|
|
print(f" 🔄 重复: {stats.get('duplicates', 0)} 条")
|
|
if "processing_time_ms" in stats:
|
|
print(f" ⏱️ 耗时: {stats.get('processing_time_ms', 0)} ms")
|
|
|
|
except json.JSONDecodeError:
|
|
result["response"] = response.text
|
|
print(f" ⚠️ {operation}: {response.status_code} (响应格式异常)")
|
|
|
|
else:
|
|
result["error"] = response.text
|
|
print(f" ❌ {operation}: {response.status_code} - {response.text[:100]}")
|
|
|
|
except Exception as e:
|
|
elapsed_time = (time.time() - start_time) * 1000
|
|
result = {
|
|
"operation": operation,
|
|
"method": method,
|
|
"url": url,
|
|
"error": str(e),
|
|
"elapsed_ms": round(elapsed_time, 2),
|
|
"success": False
|
|
}
|
|
print(f" 💥 {operation}: 请求异常 - {str(e)}")
|
|
|
|
return result
|
|
|
|
def run_all_tests(self, health_count: int = 10, device_count: int = 5, perf_size: int = 1000):
|
|
"""运行所有测试"""
|
|
print("🚀 开始 ljwx-boot 批量上传接口测试")
|
|
print(f" 目标服务: {self.base_url}")
|
|
print(f" 测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
|
results = {}
|
|
|
|
# 1. 健康检查
|
|
results["health_check"] = self.test_health_check()
|
|
|
|
# 2. 健康数据上传测试
|
|
results["health_data"] = self.test_upload_health_data(health_count)
|
|
|
|
# 3. 设备信息上传测试
|
|
results["device_info"] = self.test_upload_device_info(device_count)
|
|
|
|
# 4. 通用事件上传测试
|
|
results["common_event"] = self.test_upload_common_event()
|
|
|
|
# 5. 统计信息查询
|
|
results["stats"] = self.test_get_stats()
|
|
|
|
# 6. 性能测试
|
|
results["performance"] = self.test_performance(perf_size)
|
|
|
|
print(f"\n📋 测试完成总结:")
|
|
self._print_test_summary(results)
|
|
|
|
return results
|
|
|
|
def _print_test_summary(self, results: Dict[str, Any]):
|
|
"""打印测试总结"""
|
|
total_tests = 0
|
|
passed_tests = 0
|
|
|
|
for test_name, test_result in results.items():
|
|
if isinstance(test_result, dict):
|
|
if "success" in test_result:
|
|
# 单个测试结果
|
|
total_tests += 1
|
|
if test_result["success"]:
|
|
passed_tests += 1
|
|
print(f" ✅ {test_name}: 通过 ({test_result.get('elapsed_ms', 0):.1f}ms)")
|
|
else:
|
|
print(f" ❌ {test_name}: 失败")
|
|
else:
|
|
# 多个测试结果 (如health_data的modern和compatible)
|
|
for sub_test, sub_result in test_result.items():
|
|
if isinstance(sub_result, dict) and "success" in sub_result:
|
|
total_tests += 1
|
|
if sub_result["success"]:
|
|
passed_tests += 1
|
|
print(f" ✅ {test_name}.{sub_test}: 通过 ({sub_result.get('elapsed_ms', 0):.1f}ms)")
|
|
else:
|
|
print(f" ❌ {test_name}.{sub_test}: 失败")
|
|
|
|
success_rate = (passed_tests / total_tests * 100) if total_tests > 0 else 0
|
|
print(f"\n🎯 测试通过率: {passed_tests}/{total_tests} ({success_rate:.1f}%)")
|
|
|
|
if success_rate == 100:
|
|
print("🎉 所有测试通过！接口运行正常")
|
|
elif success_rate >= 80:
|
|
print("⚠️ 大部分测试通过，部分功能可能存在问题")
|
|
else:
|
|
print("🚨 多个测试失败，建议检查服务状态")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
parser = argparse.ArgumentParser(description="ljwx-boot 批量上传接口测试")
|
|
parser.add_argument("--url", default="http://localhost:8080", help="服务基础URL")
|
|
parser.add_argument("--health-count", type=int, default=10, help="健康数据测试条数")
|
|
parser.add_argument("--device-count", type=int, default=5, help="设备信息测试条数")
|
|
parser.add_argument("--perf-size", type=int, default=1000, help="性能测试数据规模")
|
|
|
|
# 添加单独测试选项
|
|
parser.add_argument("--test-health", action="store_true", help="仅测试健康数据接口")
|
|
parser.add_argument("--test-device", action="store_true", help="仅测试设备信息接口")
|
|
parser.add_argument("--test-event", action="store_true", help="仅测试通用事件接口")
|
|
parser.add_argument("--test-stats", action="store_true", help="仅测试统计信息接口")
|
|
parser.add_argument("--test-check", action="store_true", help="仅测试健康检查接口")
|
|
parser.add_argument("--test-perf", action="store_true", help="仅测试性能接口")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# 初始化测试器
|
|
tester = BatchUploadTester(args.url)
|
|
|
|
# 运行指定的测试
|
|
if args.test_health:
|
|
tester.test_upload_health_data(args.health_count)
|
|
elif args.test_device:
|
|
tester.test_upload_device_info(args.device_count)
|
|
elif args.test_event:
|
|
tester.test_upload_common_event()
|
|
elif args.test_stats:
|
|
tester.test_get_stats()
|
|
elif args.test_check:
|
|
tester.test_health_check()
|
|
elif args.test_perf:
|
|
tester.test_performance(args.perf_size)
|
|
else:
|
|
# 运行所有测试
|
|
tester.run_all_tests(args.health_count, args.device_count, args.perf_size)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |