#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 综合测试脚本 - 验证现代化系统是否正确工作 """ import os import sys import urllib.request import json import time import threading from datetime import datetime # 设置UTF-8编码 os.environ['PYTHONIOENCODING'] = 'utf-8' os.environ['PYTHONLEGACYWINDOWSSTDIO'] = '0' if os.name == 'nt': try: os.system('chcp 65001 >nul 2>&1') if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8', errors='replace') sys.stderr.reconfigure(encoding='utf-8', errors='replace') except Exception: pass class ModernSystemTester: """现代化系统测试器""" def __init__(self): self.base_url = "http://127.0.0.1:5000" self.test_results = [] def log_test(self, test_name, status, message="", details=None): """记录测试结果""" result = { "test": test_name, "status": status, "message": message, "details": details, "timestamp": datetime.now().strftime('%H:%M:%S') } self.test_results.append(result) status_emoji = "✅" if status == "PASS" else "❌" if status == "FAIL" else "⏳" print(f"{status_emoji} [{result['timestamp']}] {test_name}: {message}") if details: print(f" 详情: {details}") def check_api_connectivity(self): """测试1: API连接性""" try: with urllib.request.urlopen(f"{self.base_url}/api/version", timeout=5) as response: if response.getcode() == 200: data = json.loads(response.read().decode('utf-8')) version = data.get('data', {}).get('version', 'unknown') features = data.get('data', {}).get('features', []) self.log_test("API连接性", "PASS", f"版本 {version}", f"特性: {len(features)}个") return True except Exception as e: self.log_test("API连接性", "FAIL", f"连接失败: {str(e)}") return False def test_modern_logging(self): """测试2: 现代化日志系统""" try: # 检查日志文件是否存在 log_files = [] today = datetime.now().strftime('%Y-%m-%d') expected_files = [ f"api_{today}.log", f"api_error_{today}.log" ] for file in expected_files: if os.path.exists(file): log_files.append(file) if log_files: self.log_test("日志文件生成", "PASS", f"发现 {len(log_files)} 个日志文件", log_files) return True else: self.log_test("日志文件生成", "FAIL", "未发现loguru日志文件") return False except Exception as e: self.log_test("日志文件生成", "FAIL", f"检查失败: {str(e)}") return False def test_training_process_manager(self): """测试3: 训练进程管理器""" try: # 提交训练任务 training_data = { "product_id": "P001", "model_type": "transformer", "epochs": 2, "training_mode": "product" } json_data = json.dumps(training_data).encode('utf-8') req = urllib.request.Request( f"{self.base_url}/api/training", data=json_data, headers={'Content-Type': 'application/json'} ) with urllib.request.urlopen(req, timeout=30) as response: if response.getcode() == 200: result = json.loads(response.read().decode('utf-8')) task_id = result.get('task_id') message = result.get('message', '') # 检查是否使用了现代化进程管理器 uses_modern_system = '独立进程' in message if uses_modern_system: self.log_test("进程管理器", "PASS", "使用独立进程管理器", f"任务ID: {task_id[:8]}") return task_id else: self.log_test("进程管理器", "FAIL", "未使用现代化进程管理器", message) return None except Exception as e: self.log_test("进程管理器", "FAIL", f"提交任务失败: {str(e)}") return None def test_task_status_tracking(self, task_id): """测试4: 任务状态跟踪""" if not task_id: self.log_test("任务状态跟踪", "SKIP", "无有效任务ID") return False try: max_checks = 10 for i in range(max_checks): with urllib.request.urlopen(f"{self.base_url}/api/training/{task_id}", timeout=5) as response: if response.getcode() == 200: result = json.loads(response.read().decode('utf-8')) if result.get('status') == 'success': data = result.get('data', {}) status = data.get('status', 'unknown') progress = data.get('progress', 0) process_id = data.get('process_id', 'N/A') if i == 0: # 第一次检查 self.log_test("任务状态跟踪", "PASS", f"状态={status}, 进度={progress}%, 进程ID={process_id}") if status in ['completed', 'failed']: return status == 'completed' time.sleep(2) self.log_test("任务状态跟踪", "FAIL", f"超时: {max_checks * 2}秒后任务仍未完成") return False except Exception as e: self.log_test("任务状态跟踪", "FAIL", f"状态查询失败: {str(e)}") return False def test_unicode_support(self): """测试5: 中文和emoji支持""" try: # 检查产品列表API是否正确返回中文 with urllib.request.urlopen(f"{self.base_url}/api/products", timeout=5) as response: if response.getcode() == 200: result = json.loads(response.read().decode('utf-8')) products = result.get('data', []) chinese_found = False for product in products: name = product.get('name', '') if any('\u4e00' <= char <= '\u9fff' for char in name): # 检查中文字符 chinese_found = True break if chinese_found: self.log_test("中文支持", "PASS", f"发现 {len(products)} 个产品，包含中文") # 测试emoji是否在控制台正常显示 test_emoji = "🚀📊🔧💬✅❌⏳🎯" print(f" Emoji测试: {test_emoji}") self.log_test("Emoji支持", "PASS", "控制台emoji显示正常") return True else: self.log_test("中文支持", "FAIL", "未发现中文产品名称") return False except Exception as e: self.log_test("中文支持", "FAIL", f"测试失败: {str(e)}") return False def test_concurrent_training(self): """测试6: 并发训练能力""" try: # 同时提交3个训练任务 models = ['mlstm', 'tcn', 'kan'] task_ids = [] for model in models: training_data = { "product_id": "P001", "model_type": model, "epochs": 1, "training_mode": "product" } json_data = json.dumps(training_data).encode('utf-8') req = urllib.request.Request( f"{self.base_url}/api/training", data=json_data, headers={'Content-Type': 'application/json'} ) with urllib.request.urlopen(req, timeout=30) as response: if response.getcode() == 200: result = json.loads(response.read().decode('utf-8')) task_ids.append(result.get('task_id')) time.sleep(0.5) # 稍微错开提交时间 if len(task_ids) == 3: self.log_test("并发训练", "PASS", f"成功提交 {len(task_ids)} 个并发任务") return True else: self.log_test("并发训练", "FAIL", f"只成功提交 {len(task_ids)}/3 个任务") return False except Exception as e: self.log_test("并发训练", "FAIL", f"并发测试失败: {str(e)}") return False def test_all_tasks_api(self): """测试7: 任务列表API""" try: with urllib.request.urlopen(f"{self.base_url}/api/training", timeout=5) as response: if response.getcode() == 200: result = json.loads(response.read().decode('utf-8')) if result.get('status') == 'success': tasks = result.get('data', []) self.log_test("任务列表API", "PASS", f"发现 {len(tasks)} 个历史任务") return True else: self.log_test("任务列表API", "FAIL", "API返回错误状态") return False except Exception as e: self.log_test("任务列表API", "FAIL", f"API调用失败: {str(e)}") return False def generate_report(self): """生成测试报告""" print("\n" + "=" * 80) print("🎯 现代化系统测试报告") print("=" * 80) total_tests = len(self.test_results) passed_tests = len([r for r in self.test_results if r['status'] == 'PASS']) failed_tests = len([r for r in self.test_results if r['status'] == 'FAIL']) skipped_tests = len([r for r in self.test_results if r['status'] == 'SKIP']) print(f"📊 测试统计:") print(f" 总计: {total_tests} 个测试") print(f" 通过: {passed_tests} 个 ✅") print(f" 失败: {failed_tests} 个 ❌") print(f" 跳过: {skipped_tests} 个 ⏭️") print(f" 成功率: {passed_tests/total_tests*100:.1f}%") print(f"\n📋 详细结果:") for result in self.test_results: status_emoji = "✅" if result['status'] == "PASS" else "❌" if result['status'] == "FAIL" else "⏭️" print(f" {status_emoji} {result['test']}: {result['message']}") print(f"\n🎯 总体评估:") if failed_tests == 0: print(" 🏆 所有测试通过！现代化系统工作完全正常") return True elif failed_tests <= 2: print(" ⚠️ 大部分测试通过，系统基本正常，有少量问题需要修复") return False else: print(" ❌ 多个测试失败，现代化系统存在问题需要进一步修复") return False def run_all_tests(self): """运行所有测试""" print("🚀 开始现代化系统综合测试") print("=" * 80) # 测试序列 if not self.check_api_connectivity(): print("❌ API服务器未运行，请先启动: uv run server/modern_api.py") return False self.test_modern_logging() task_id = self.test_training_process_manager() self.test_task_status_tracking(task_id) self.test_unicode_support() self.test_concurrent_training() self.test_all_tasks_api() return self.generate_report() def main(): """主函数""" tester = ModernSystemTester() success = tester.run_all_tests() if success: print("\n🎉 现代化系统修改完全正确！") exit(0) else: print("\n⚠️ 现代化系统存在问题，需要进一步修复") exit(1) if __name__ == "__main__": main()