#!/usr/bin/env python3 """ 测试运行器 - 统一的测试入口点 使用方法: python run_tests.py --help # 显示帮助 python run_tests.py api # 运行 API 测试 python run_tests.py quick # 运行快速测试 python run_tests.py concurrent # 运行并发测试 python run_tests.py performance # 运行性能监控 python run_tests.py all # 运行所有测试 """ import argparse import asyncio import sys import os from pathlib import Path # 添加项目根目录到 Python 路径 project_root = Path(__file__).parent sys.path.insert(0, str(project_root)) from tests.utils import wait_for_server, TestReporter from tests.config import BASE_URL def run_api_test(): """运行基础 API 测试""" print("🔧 运行基础 API 测试...") import subprocess result = subprocess.run([sys.executable, "tests/test_api.py"], cwd=project_root, capture_output=True, text=True) print(result.stdout) if result.stderr: print("错误输出:", result.stderr) return result.returncode == 0 async def run_quick_test(): """运行快速测试""" print("⚡ 运行快速测试...") # 检查服务器 if not await wait_for_server(BASE_URL, timeout=10): return False from tests.quick_test import quick_test, mini_concurrent_test try: success1 = await quick_test() success2 = await mini_concurrent_test() if success1 else False return success1 and success2 except Exception as e: print(f"❌ 快速测试失败: {e}") return False async def run_concurrent_test(): """运行并发测试""" print("🚀 运行并发测试...") # 检查服务器 if not await wait_for_server(BASE_URL, timeout=10): return False from tests.test_concurrent import run_comprehensive_concurrent_test try: await run_comprehensive_concurrent_test() return True except Exception as e: print(f"❌ 并发测试失败: {e}") return False async def run_performance_test(): """运行性能监控测试""" print("📊 运行性能监控...") # 检查服务器 if not await wait_for_server(BASE_URL, timeout=10): return False from tests.performance_monitor import run_load_test_with_monitoring try: await run_load_test_with_monitoring() return True except Exception as e: print(f"❌ 性能测试失败: {e}") return False async def run_all_tests(): """运行所有测试""" print("🎯 运行完整测试套件") print("=" * 60) results = {} # 1. API 测试 print("\n1️⃣ 基础 API 测试") results["api"] = run_api_test() # 2. 快速测试 print("\n2️⃣ 快速功能测试") results["quick"] = await run_quick_test() # 3. 并发测试 print("\n3️⃣ 并发性能测试") results["concurrent"] = await run_concurrent_test() # 4. 性能监控(可选) print("\n4️⃣ 性能监控测试") results["performance"] = await run_performance_test() # 生成总结报告 print("\n" + "=" * 60) print("📋 测试总结:") total_tests = len(results) passed_tests = sum(1 for success in results.values() if success) for test_name, success in results.items(): status = "✅ 通过" if success else "❌ 失败" print(f" {test_name.upper()}: {status}") print(f"\n🎯 总体结果: {passed_tests}/{total_tests} 测试通过") if passed_tests == total_tests: print("🎉 所有测试都通过了!") return True else: print("⚠️ 部分测试失败,请检查日志。") return False def main(): """主函数""" parser = argparse.ArgumentParser(description="RAG 系统测试运行器") parser.add_argument( "test_type", choices=["api", "quick", "concurrent", "performance", "all"], help="要运行的测试类型" ) parser.add_argument( "--timeout", type=int, default=30, help="服务器启动超时时间(秒)" ) parser.add_argument( "--no-server-check", action="store_true", help="跳过服务器检查" ) args = parser.parse_args() # 根据参数运行相应的测试 try: if args.test_type == "api": success = run_api_test() elif args.test_type == "quick": success = asyncio.run(run_quick_test()) elif args.test_type == "concurrent": success = asyncio.run(run_concurrent_test()) elif args.test_type == "performance": success = asyncio.run(run_performance_test()) elif args.test_type == "all": success = asyncio.run(run_all_tests()) else: print(f"❌ 未知的测试类型: {args.test_type}") return 1 return 0 if success else 1 except KeyboardInterrupt: print("\n⏹️ 测试被用户中断") return 1 except Exception as e: print(f"❌ 测试运行失败: {e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": sys.exit(main())