138 lines
4.3 KiB
Python
138 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
快速测试运行脚本
|
||
用于快速验证系统功能
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
import time
|
||
import os
|
||
|
||
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
|
||
from tests.test_concurrent import (
|
||
ConcurrentRAGTester,
|
||
test_concurrent_health_check,
|
||
test_concurrent_upload,
|
||
test_concurrent_chat,
|
||
)
|
||
|
||
|
||
async def quick_test():
|
||
"""快速测试所有主要功能"""
|
||
print("🚀 快速功能验证测试")
|
||
print("=" * 40)
|
||
|
||
try:
|
||
async with ConcurrentRAGTester() as tester:
|
||
# 1. 健康检查
|
||
print("1️⃣ 健康检查...")
|
||
health = await tester.health_check()
|
||
if health["status_code"] != 200:
|
||
print(f"❌ 服务器不可用: {health}")
|
||
return False
|
||
print(f"✅ 服务器正常 (响应时间: {health['response_time']:.3f}s)")
|
||
|
||
# 2. 单个文档上传
|
||
print("\n2️⃣ 文档上传测试...")
|
||
upload_result = await tester.upload_document(
|
||
"这是一个快速测试文档。包含关于人工智能和机器学习的基础知识。",
|
||
"quick_test.txt",
|
||
)
|
||
if upload_result["status_code"] != 200:
|
||
print(f"❌ 上传失败: {upload_result}")
|
||
return False
|
||
print(f"✅ 上传成功 (文档ID: {upload_result.get('document_id', 'N/A')})")
|
||
|
||
# 3. 等待处理
|
||
await asyncio.sleep(1)
|
||
|
||
# 4. 聊天测试
|
||
print("\n3️⃣ 聊天功能测试...")
|
||
chat_result = await tester.chat_query("什么是人工智能?")
|
||
if chat_result["status_code"] != 200:
|
||
print(f"❌ 聊天失败: {chat_result}")
|
||
return False
|
||
print(
|
||
f"✅ 聊天成功 (处理时间: {chat_result.get('processing_time', 0):.2f}s)"
|
||
)
|
||
print(f" 回答长度: {len(chat_result.get('answer', ''))} 字符")
|
||
print(f" 来源数量: {chat_result.get('sources_count', 0)}")
|
||
|
||
# 5. 文档列表
|
||
print("\n4️⃣ 文档列表测试...")
|
||
docs_result = await tester.get_documents()
|
||
if docs_result["status_code"] != 200:
|
||
print(f"❌ 获取文档列表失败: {docs_result}")
|
||
return False
|
||
doc_count = len(docs_result["data"])
|
||
print(f"✅ 文档列表获取成功 (文档数量: {doc_count})")
|
||
|
||
print("\n" + "=" * 40)
|
||
print("🎉 所有基础功能测试通过!")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试过程中发生错误: {e}")
|
||
return False
|
||
|
||
|
||
async def mini_concurrent_test():
|
||
"""迷你并发测试"""
|
||
print("\n🔥 迷你并发测试")
|
||
print("=" * 40)
|
||
|
||
try:
|
||
# 小规模并发测试
|
||
await test_concurrent_health_check(3)
|
||
await test_concurrent_upload(2)
|
||
await asyncio.sleep(1)
|
||
await test_concurrent_chat(3)
|
||
|
||
print("🎯 迷你并发测试完成!")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 并发测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
if len(sys.argv) > 1:
|
||
test_type = sys.argv[1].lower()
|
||
if test_type == "quick":
|
||
success = asyncio.run(quick_test())
|
||
elif test_type == "concurrent":
|
||
success = asyncio.run(mini_concurrent_test())
|
||
elif test_type == "both":
|
||
success1 = asyncio.run(quick_test())
|
||
success2 = asyncio.run(mini_concurrent_test()) if success1 else False
|
||
success = success1 and success2
|
||
else:
|
||
print("❌ 未知的测试类型")
|
||
print("用法: python quick_test.py [quick|concurrent|both]")
|
||
return
|
||
else:
|
||
# 默认运行所有测试
|
||
success1 = asyncio.run(quick_test())
|
||
success2 = asyncio.run(mini_concurrent_test()) if success1 else False
|
||
success = success1 and success2
|
||
|
||
if success:
|
||
print("\n✅ 所有测试通过!")
|
||
sys.exit(0)
|
||
else:
|
||
print("\n❌ 测试失败!")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\n⏹️ 测试被中断")
|
||
except Exception as e:
|
||
print(f"❌ 运行失败: {e}")
|
||
sys.exit(1)
|