easy-rag/run_tests.py

191 lines
5.3 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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())