test_status_fix.py•6.58 kB
#!/usr/bin/env python3
"""
测试状态修复效果
验证MVP管理器和MCP服务器的状态显示一致性
"""
import os
import sys
import time
import asyncio
from pathlib import Path
# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent))
def test_mvp_status_consistency():
"""测试MVP管理器状态一致性"""
print("=" * 60)
print("🔍 测试MVP管理器状态一致性")
print("=" * 60)
try:
from mvp_memory import MVPMemoryManager
# 创建MVP管理器实例
print("🚀 创建MVP管理器实例...")
mvp_manager = MVPMemoryManager(data_dir="./mcp_memos_data", use_enhanced=True)
# 检查新的状态属性
print(f"✅ desired_mode (期望模式): {mvp_manager.desired_mode}")
print(f"✅ current_mode (当前模式): {mvp_manager.current_mode}")
print(f"✅ use_enhanced (兼容属性): {mvp_manager.use_enhanced}")
print(f"✅ fallback_reason (降级原因): {mvp_manager.fallback_reason}")
# 获取详细状态信息
status_info = mvp_manager.get_status_info()
print(f"✅ 状态信息: {status_info}")
return mvp_manager, status_info
except Exception as e:
print(f"❌ MVP状态测试失败: {e}")
import traceback
traceback.print_exc()
return None, None
async def test_mcp_status_display():
"""测试MCP服务器状态显示"""
print("\n" + "=" * 60)
print("🔍 测试MCP服务器状态显示")
print("=" * 60)
try:
from memos_mcp_server import MemOSMCPServer
print("🚀 创建MCP服务器实例...")
server = MemOSMCPServer()
# 检查MVP管理器状态
print(f"✅ MCP中desired_mode: {server.mvp_manager.desired_mode}")
print(f"✅ MCP中current_mode: {server.mvp_manager.current_mode}")
print(f"✅ MCP中use_enhanced: {server.mvp_manager.use_enhanced}")
# 获取状态信息
status_info = server.mvp_manager.get_status_info()
print(f"✅ MCP状态信息: {status_info}")
# 模拟添加记忆的MCP调用
print("\n🧪 模拟MCP添加记忆调用...")
test_args = {
"content": f"状态修复测试记忆 - {time.time()}",
"tags": ["状态修复", "测试"]
}
response = await server.add_memos_memory(test_args)
print(f"✅ MCP响应: {response}")
# 检查响应中的模型信息
if "result" in response and "content" in response["result"]:
content = response["result"]["content"][0]["text"]
print(f"📋 响应内容: {content}")
# 提取模型信息
if "使用" in content:
model_part = content.split("使用")[1].split(")")[0]
print(f"🎯 显示的模型: {model_part}")
return server, status_info
except Exception as e:
print(f"❌ MCP状态测试失败: {e}")
import traceback
traceback.print_exc()
return None, None
def test_fallback_scenario():
"""测试降级场景"""
print("\n" + "=" * 60)
print("🔍 测试降级场景")
print("=" * 60)
try:
from mvp_memory import MVPMemoryManager
# 使用不存在的数据目录触发可能的错误
print("🚀 测试降级场景...")
mvp_manager = MVPMemoryManager(data_dir="./test_fallback_data", use_enhanced=True)
print(f"✅ 降级测试 - desired_mode: {mvp_manager.desired_mode}")
print(f"✅ 降级测试 - current_mode: {mvp_manager.current_mode}")
print(f"✅ 降级测试 - fallback_reason: {mvp_manager.fallback_reason}")
status_info = mvp_manager.get_status_info()
print(f"✅ 降级状态信息: {status_info}")
return True
except Exception as e:
print(f"❌ 降级测试失败: {e}")
return False
async def test_health_check():
"""测试健康检查功能"""
print("\n" + "=" * 60)
print("🔍 测试健康检查功能")
print("=" * 60)
try:
from mvp_memory import MVPMemoryManager
mvp_manager = MVPMemoryManager(data_dir="./mcp_memos_data", use_enhanced=True)
print("🚀 执行健康检查...")
health_result = await mvp_manager.health_check()
print(f"✅ 健康检查结果: {health_result}")
return health_result
except Exception as e:
print(f"❌ 健康检查失败: {e}")
return None
async def main():
"""主测试流程"""
print("🩺 MemOS状态修复效果测试")
print("=" * 80)
# 1. 测试MVP管理器状态一致性
mvp_manager, mvp_status = test_mvp_status_consistency()
# 2. 测试MCP服务器状态显示
mcp_server, mcp_status = await test_mcp_status_display()
# 3. 测试降级场景
fallback_ok = test_fallback_scenario()
# 4. 测试健康检查
health_result = await test_health_check()
# 5. 对比结果
print("\n" + "=" * 60)
print("📊 修复效果验证")
print("=" * 60)
if mvp_manager and mcp_server:
print("🔍 状态一致性检查:")
print(f" MVP desired_mode: {mvp_manager.desired_mode}")
print(f" MVP current_mode: {mvp_manager.current_mode}")
print(f" MCP desired_mode: {mcp_server.mvp_manager.desired_mode}")
print(f" MCP current_mode: {mcp_server.mvp_manager.current_mode}")
if (mvp_manager.desired_mode == mcp_server.mvp_manager.desired_mode and
mvp_manager.current_mode == mcp_server.mvp_manager.current_mode):
print("✅ 状态完全一致！")
else:
print("❌ 状态仍有不一致")
if mvp_status and mcp_status:
print(f"\n🔍 状态信息对比:")
print(f" MVP状态: {mvp_status}")
print(f" MCP状态: {mcp_status}")
if mvp_status == mcp_status:
print("✅ 状态信息完全一致！")
else:
print("❌ 状态信息有差异")
print(f"\n🔍 健康检查: {'✅ 通过' if health_result and health_result.get('healthy') else '❌ 失败'}")
print(f"🔍 降级测试: {'✅ 通过' if fallback_ok else '❌ 失败'}")
print("\n🎯 修复效果测试完成")
if __name__ == "__main__":
asyncio.run(main())