test_cherry_sse.py2.57 kB
#!/usr/bin/env python3
"""
测试CherryStudio兼容的SSE服务器
"""
import requests
import json
import time
def test_cherry_sse():
"""测试CherryStudio兼容性"""
base_url = "http://localhost:8001"
print("🚀 测试CherryStudio兼容的SSE服务器...")
print("=" * 60)
# 1. 测试SSE信息端点
try:
response = requests.get(f"{base_url}/sse")
print("✅ SSE信息端点 - 成功")
data = response.json()
print(f"服务名称: {data['name']}")
print(f"协议: {data['protocol']}")
print(f"状态: {data['status']}")
print(f"内容类型: {data['content_type']}")
print(f"可用工具: {len(data['tools'])}个")
except Exception as e:
print(f"❌ SSE信息端点 - 失败: {e}")
return
print("\n" + "=" * 60)
# 2. 测试SSE流的HTTP头
try:
response = requests.head(f"{base_url}/sse/stream/realtime?symbol=000001")
print("✅ SSE流HTTP头检查 - 成功")
print(f"Content-Type: {response.headers.get('Content-Type')}")
print(f"Cache-Control: {response.headers.get('Cache-Control')}")
print(f"Connection: {response.headers.get('Connection')}")
# 检查是否是正确的content-type
content_type = response.headers.get('Content-Type', '')
if 'text/event-stream' in content_type:
print("✅ 内容类型正确: text/event-stream")
else:
print(f"❌ 内容类型错误: {content_type}")
except Exception as e:
print(f"❌ SSE流HTTP头检查 - 失败: {e}")
print("\n" + "=" * 60)
# 3. 测试SSE数据流（短时间）
try:
print("🔄 测试SSE数据流...")
response = requests.get(f"{base_url}/sse/stream/realtime?symbol=000001",
stream=True, timeout=8)
lines_count = 0
for line in response.iter_lines(decode_unicode=True):
if line:
print(f"SSE: {line}")
lines_count += 1
if lines_count >= 8: # 只读取前8行
break
print(f"✅ 成功读取 {lines_count} 行SSE数据")
except requests.exceptions.Timeout:
print("✅ SSE流测试完成（超时正常）")
except Exception as e:
print(f"❌ SSE数据流测试 - 失败: {e}")
print("\n" + "=" * 60)
print("🎉 CherryStudio兼容性测试完成！")
if __name__ == "__main__":
test_cherry_sse()