You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
136 lines
4.0 KiB
136 lines
4.0 KiB
#!/usr/bin/env python3
|
|
"""
|
|
WebSocket测试运行脚本
|
|
同时启动测试服务器和运行功能测试
|
|
"""
|
|
import asyncio
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from app.utils.structured_log import get_structured_logger, LogLevel
|
|
|
|
logger = get_structured_logger(__name__, LogLevel.DEBUG)
|
|
|
|
|
|
async def run_websocket_test():
|
|
"""运行WebSocket完整测试"""
|
|
print("🚀 启动WebSocket完整功能测试")
|
|
print("="*60)
|
|
|
|
# 启动测试服务器
|
|
print("📡 启动WebSocket测试服务器...")
|
|
server_process = subprocess.Popen([
|
|
sys.executable, "test_websocket_server_simple.py"
|
|
], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
# 等待服务器启动
|
|
print("⏳ 等待服务器启动...")
|
|
await asyncio.sleep(3)
|
|
|
|
try:
|
|
# 运行功能测试
|
|
print("🧪 运行WebSocket功能测试...")
|
|
test_process = subprocess.run([
|
|
sys.executable, "test_websocket_complete.py"
|
|
], capture_output=True, text=True)
|
|
|
|
# 输出测试结果
|
|
print("\n" + "="*60)
|
|
print("测试输出:")
|
|
print("="*60)
|
|
|
|
if test_process.stdout:
|
|
print("标准输出:")
|
|
print(test_process.stdout)
|
|
|
|
if test_process.stderr:
|
|
print("错误输出:")
|
|
print(test_process.stderr)
|
|
|
|
print("="*60)
|
|
print(f"测试退出码: {test_process.returncode}")
|
|
|
|
if test_process.returncode == 0:
|
|
print("✅ 测试完成")
|
|
else:
|
|
print("❌ 测试失败")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 运行测试时发生异常: {e}")
|
|
|
|
finally:
|
|
# 停止服务器
|
|
print("\n🛑 停止WebSocket测试服务器...")
|
|
server_process.terminate()
|
|
try:
|
|
server_process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
server_process.kill()
|
|
|
|
print("✅ 测试服务器已停止")
|
|
|
|
|
|
def run_simple_test():
|
|
"""运行简单测试（不依赖外部服务器）"""
|
|
print("🧪 运行简单功能测试（不依赖外部服务器）")
|
|
print("="*60)
|
|
|
|
try:
|
|
# 导入测试模块
|
|
from test_websocket_complete import WebSocketCompleteTest
|
|
|
|
async def run_test():
|
|
test = WebSocketCompleteTest()
|
|
# 修改测试URL为无效地址，只测试内部功能
|
|
test.test_url = "ws://invalid-server:9999/ws"
|
|
|
|
# 只运行不依赖连接的测试
|
|
await test.test_create_client_and_channels()
|
|
await test.test_send_messages()
|
|
await test.test_receive_message_processing()
|
|
await test.test_priority_control()
|
|
await test.test_cleanup()
|
|
test.print_test_results()
|
|
|
|
asyncio.run(run_test())
|
|
|
|
except Exception as e:
|
|
print(f"❌ 简单测试失败: {e}")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("WebSocket测试运行器")
|
|
print("="*60)
|
|
print("选择测试模式:")
|
|
print("1. 完整测试（需要启动测试服务器）")
|
|
print("2. 简单测试（仅测试内部功能）")
|
|
print("3. 退出")
|
|
|
|
while True:
|
|
try:
|
|
choice = input("\n请选择 (1-3): ").strip()
|
|
|
|
if choice == "1":
|
|
print("\n选择完整测试模式...")
|
|
asyncio.run(run_websocket_test())
|
|
break
|
|
elif choice == "2":
|
|
print("\n选择简单测试模式...")
|
|
run_simple_test()
|
|
break
|
|
elif choice == "3":
|
|
print("退出测试")
|
|
break
|
|
else:
|
|
print("无效选择，请输入 1、2 或 3")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n用户中断，退出测试")
|
|
break
|
|
except Exception as e:
|
|
print(f"发生错误: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|