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.
148 lines
4.8 KiB
148 lines
4.8 KiB
#!/usr/bin/env python3
|
|
"""
|
|
测试WebSocket SSL配置冲突修复
|
|
验证修复后的SSL配置逻辑是否正确
|
|
"""
|
|
import asyncio
|
|
import ssl
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from app.core.config.settings import config
|
|
from app.core.websocket.client import WebSocketClient
|
|
from app.utils.structured_log import get_structured_logger, LogLevel
|
|
|
|
logger = get_structured_logger(__name__, LogLevel.DEBUG)
|
|
|
|
async def test_ssl_configuration():
|
|
"""测试SSL配置逻辑"""
|
|
print("=== 测试WebSocket SSL配置 ===")
|
|
|
|
# 测试配置
|
|
test_urls = [
|
|
"ws://localhost:8080", # 非SSL连接
|
|
"wss://localhost:8080", # SSL连接
|
|
]
|
|
|
|
for url in test_urls:
|
|
print(f"\n测试URL: {url}")
|
|
|
|
# 创建WebSocket客户端
|
|
client = WebSocketClient(url, f"test_{url.split('://')[1]}")
|
|
|
|
# 模拟连接过程（不实际连接）
|
|
try:
|
|
# 检查SSL配置逻辑
|
|
if url.startswith('wss://'):
|
|
print(" SSL连接检测到，检查配置...")
|
|
print(f" 当前配置: ssl_verify_certificate={config.websocket.ssl_verify_certificate}")
|
|
print(f" 当前配置: ssl_verify_hostname={config.websocket.ssl_verify_hostname}")
|
|
|
|
# 模拟SSL上下文创建
|
|
ssl_context = ssl.create_default_context()
|
|
|
|
# 先设置check_hostname，再设置verify_mode
|
|
if not config.websocket.ssl_verify_hostname:
|
|
ssl_context.check_hostname = False
|
|
print(" ✅ 设置: check_hostname=False")
|
|
|
|
if not config.websocket.ssl_verify_certificate:
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
print(" ✅ 设置: verify_mode=CERT_NONE")
|
|
else:
|
|
print(" ✅ 使用默认SSL验证")
|
|
|
|
print(" ✅ SSL配置逻辑正确，无冲突")
|
|
else:
|
|
print(" ✅ 非SSL连接，无需SSL配置")
|
|
|
|
except Exception as e:
|
|
print(f" ❌ SSL配置错误: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
async def test_websocket_client_creation():
|
|
"""测试WebSocket客户端创建"""
|
|
print("\n=== 测试WebSocket客户端创建 ===")
|
|
|
|
try:
|
|
# 测试创建客户端（不实际连接）
|
|
client = WebSocketClient("wss://localhost:8080", "test_client")
|
|
print(f"✅ 客户端创建成功: {client.name}")
|
|
print(f" 状态: {client.state.value}")
|
|
print(f" 连接状态: {client.is_connected}")
|
|
|
|
# 测试获取统计信息
|
|
stats = client.get_stats()
|
|
print(f" 统计信息: {stats}")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 客户端创建失败: {e}")
|
|
return False
|
|
|
|
async def test_configuration_values():
|
|
"""测试配置值"""
|
|
print("\n=== 测试配置值 ===")
|
|
|
|
print(f"WebSocket SSL配置:")
|
|
print(f" ssl_verify_certificate: {config.websocket.ssl_verify_certificate}")
|
|
print(f" ssl_verify_hostname: {config.websocket.ssl_verify_hostname}")
|
|
print(f" connection_timeout: {config.websocket.connection_timeout}")
|
|
print(f" reconnect_attempts: {config.websocket.reconnect_attempts}")
|
|
print(f" heartbeat_interval: {config.websocket.heartbeat_interval}")
|
|
|
|
# 验证配置逻辑
|
|
if not config.websocket.ssl_verify_certificate:
|
|
print("✅ 开发环境配置: 跳过SSL证书验证")
|
|
else:
|
|
print("✅ 生产环境配置: 启用SSL证书验证")
|
|
|
|
return True
|
|
|
|
async def main():
|
|
"""主测试函数"""
|
|
print("开始测试WebSocket SSL配置冲突修复...")
|
|
|
|
tests = [
|
|
test_configuration_values,
|
|
test_websocket_client_creation,
|
|
test_ssl_configuration,
|
|
]
|
|
|
|
results = []
|
|
for test in tests:
|
|
try:
|
|
result = await test()
|
|
results.append(result)
|
|
except Exception as e:
|
|
print(f"❌ 测试异常: {e}")
|
|
results.append(False)
|
|
|
|
# 输出测试结果
|
|
print("\n" + "="*50)
|
|
print("测试结果汇总:")
|
|
for i, result in enumerate(results):
|
|
status = "✅ 通过" if result else "❌ 失败"
|
|
print(f" 测试 {i+1}: {status}")
|
|
|
|
all_passed = all(results)
|
|
print(f"\n总体结果: {'✅ 所有测试通过' if all_passed else '❌ 部分测试失败'}")
|
|
|
|
return all_passed
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
result = asyncio.run(main())
|
|
sys.exit(0 if result else 1)
|
|
except KeyboardInterrupt:
|
|
print("\n测试被用户中断")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"测试异常: {e}")
|
|
sys.exit(1)
|