test_dashboard.py•8.43 kB
#!/usr/bin/env python3
"""
MemOS Dashboard 测试脚本
测试Dashboard的API接口和功能
"""
import os
import sys
import time
import requests
import threading
from pathlib import Path
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
def test_api_endpoints():
"""测试API端点"""
print("🧪 测试API端点")
print("-" * 30)
base_url = "http://localhost:8000"
# 测试端点列表
endpoints = [
("/health", "健康检查"),
("/status", "系统状态"),
("/metrics", "性能指标"),
("/memories/stats", "记忆统计"),
("/capacity/report", "容量报告"),
("/dashboard/topic-drift", "主题漂移统计"),
("/mcp/tools", "MCP工具列表")
]
results = []
for endpoint, description in endpoints:
try:
print(f"测试 {endpoint} ({description})...")
response = requests.get(f"{base_url}{endpoint}", timeout=5)
if response.status_code == 200:
data = response.json()
if data.get('success', False):
print(f" ✅ {description} - 成功")
results.append((endpoint, True, "成功"))
else:
print(f" ⚠️ {description} - 响应格式异常")
results.append((endpoint, False, "响应格式异常"))
else:
print(f" ❌ {description} - HTTP {response.status_code}")
results.append((endpoint, False, f"HTTP {response.status_code}"))
except requests.exceptions.ConnectionError:
print(f" ❌ {description} - 连接失败")
results.append((endpoint, False, "连接失败"))
except requests.exceptions.Timeout:
print(f" ❌ {description} - 请求超时")
results.append((endpoint, False, "请求超时"))
except Exception as e:
print(f" ❌ {description} - 错误: {e}")
results.append((endpoint, False, str(e)))
return results
def test_dashboard_files():
"""测试Dashboard文件"""
print("\n📁 测试Dashboard文件")
print("-" * 30)
dashboard_dir = Path(__file__).parent / "dashboard"
required_files = [
("index.html", "主页面"),
("dashboard.js", "JavaScript脚本")
]
results = []
for filename, description in required_files:
file_path = dashboard_dir / filename
if file_path.exists():
file_size = file_path.stat().st_size
print(f"✅ {description} ({filename}) - {file_size} 字节")
results.append((filename, True, f"{file_size} 字节"))
else:
print(f"❌ {description} ({filename}) - 文件不存在")
results.append((filename, False, "文件不存在"))
return results
def test_dashboard_access():
"""测试Dashboard访问"""
print("\n🌐 测试Dashboard访问")
print("-" * 30)
dashboard_url = "http://localhost:3000"
try:
print(f"测试访问 {dashboard_url}...")
response = requests.get(dashboard_url, timeout=5)
if response.status_code == 200:
content = response.text
# 检查关键内容
checks = [
("MemOS 迷你Dashboard", "页面标题"),
("系统状态", "系统状态卡片"),
("记忆统计", "记忆统计卡片"),
("性能指标", "性能指标卡片"),
("容量管理", "容量管理卡片"),
("主题漂移检测", "主题漂移检测卡片"),
("dashboard.js", "JavaScript引用")
]
for check_text, description in checks:
if check_text in content:
print(f" ✅ {description} - 找到")
else:
print(f" ❌ {description} - 未找到")
print(f"✅ Dashboard页面访问成功 ({len(content)} 字符)")
return True
else:
print(f"❌ Dashboard访问失败 - HTTP {response.status_code}")
return False
except requests.exceptions.ConnectionError:
print("❌ Dashboard访问失败 - 连接失败")
return False
except Exception as e:
print(f"❌ Dashboard访问失败 - 错误: {e}")
return False
def start_test_servers():
"""启动测试服务器"""
print("🚀 启动测试服务器...")
try:
# 启动API服务器
from official_api_server import create_official_api_app, setup_exception_handlers
import uvicorn
app, manager = create_official_api_app()
setup_exception_handlers(app)
# 在后台线程启动API服务器
def run_api_server():
uvicorn.run(app, host="localhost", port=8000, log_level="error")
api_thread = threading.Thread(target=run_api_server, daemon=True)
api_thread.start()
print("✅ API服务器启动中...")
# 启动Dashboard服务器
from http.server import HTTPServer, SimpleHTTPRequestHandler
class DashboardHandler(SimpleHTTPRequestHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, directory=str(Path(__file__).parent / "dashboard"), **kwargs)
def end_headers(self):
self.send_header('Access-Control-Allow-Origin', '*')
super().end_headers()
def run_dashboard_server():
server = HTTPServer(('localhost', 3000), DashboardHandler)
server.serve_forever()
dashboard_thread = threading.Thread(target=run_dashboard_server, daemon=True)
dashboard_thread.start()
print("✅ Dashboard服务器启动中...")
# 等待服务器启动
time.sleep(3)
return True
except Exception as e:
print(f"❌ 服务器启动失败: {e}")
return False
def main():
"""主测试函数"""
print("🧪 MemOS Dashboard 测试")
print("=" * 50)
# 测试文件存在性
file_results = test_dashboard_files()
# 启动测试服务器
if start_test_servers():
# 测试API端点
api_results = test_api_endpoints()
# 测试Dashboard访问
dashboard_access = test_dashboard_access()
# 汇总结果
print("\n📊 测试结果汇总")
print("=" * 50)
# 文件测试结果
print("📁 文件测试:")
for filename, success, details in file_results:
status = "✅" if success else "❌"
print(f" {status} {filename}: {details}")
# API测试结果
print("\n🔌 API测试:")
api_success_count = sum(1 for _, success, _ in api_results if success)
print(f" 成功: {api_success_count}/{len(api_results)}")
for endpoint, success, details in api_results:
status = "✅" if success else "❌"
print(f" {status} {endpoint}: {details}")
# Dashboard访问测试
print(f"\n🌐 Dashboard访问: {'✅ 成功' if dashboard_access else '❌ 失败'}")
# 总体评估
total_tests = len(file_results) + len(api_results) + 1
passed_tests = (
sum(1 for _, success, _ in file_results if success) +
sum(1 for _, success, _ in api_results if success) +
(1 if dashboard_access else 0)
)
print(f"\n🎯 总体结果: {passed_tests}/{total_tests} 测试通过")
if passed_tests == total_tests:
print("🎉 所有测试通过！Dashboard功能正常")
return True
else:
print("⚠️ 部分测试失败，请检查相关功能")
return False
else:
print("❌ 无法启动测试服务器")
return False
if __name__ == "__main__":
try:
success = main()
sys.exit(0 if success else 1)
except KeyboardInterrupt:
print("\n🛑 测试被中断")
sys.exit(1)