191 lines
5.3 KiB
Python
191 lines
5.3 KiB
Python
#!/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())
|