root
4 months ago
3 changed files with 172 additions and 11 deletions
@ -0,0 +1,148 @@ |
|||
#!/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.log import get_logger |
|||
|
|||
logger = get_logger(__name__) |
|||
|
|||
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) |
|||
Loading…
Reference in new issue