test_memory_ops_sdk.py•11.7 kB
#!/usr/bin/env python3
"""
MemOS Memory Ops SDK 测试脚本
全面测试SDK的功能和接口
"""
import os
import sys
import time
import tempfile
import shutil
from pathlib import Path
# 添加项目根目录到Python路径
sys.path.insert(0, str(Path(__file__).parent))
def test_basic_operations():
"""测试基础操作"""
print("🧪 测试基础操作")
print("-" * 30)
try:
from memory_ops_sdk import MemoryOps
# 使用临时目录
with tempfile.TemporaryDirectory() as temp_dir:
mem = MemoryOps(data_dir=temp_dir, verbose=True)
try:
# 测试添加记忆
print("测试添加记忆...")
success1 = mem.add("测试记忆1", tags=["测试"])
success2 = mem.add("测试记忆2", tags=["测试", "SDK"])
if success1 and success2:
print("✅ 添加记忆测试通过")
else:
print("❌ 添加记忆测试失败")
return False
# 测试查询记忆
print("测试查询记忆...")
results = mem.query("测试")
if len(results) >= 2:
print(f"✅ 查询记忆测试通过 (找到 {len(results)} 条)")
else:
print(f"❌ 查询记忆测试失败 (只找到 {len(results)} 条)")
return False
# 测试反馈
if results:
print("测试反馈功能...")
memory_id = results[0].get('id')
if memory_id:
feedback_success = mem.feedback(memory_id, "👍")
if feedback_success:
print("✅ 反馈功能测试通过")
else:
print("❌ 反馈功能测试失败")
return False
# 测试统计信息
print("测试统计信息...")
stats = mem.stats()
if stats and 'status' in stats:
print("✅ 统计信息测试通过")
else:
print("❌ 统计信息测试失败")
return False
# 测试健康检查
print("测试健康检查...")
is_healthy = mem.health()
if is_healthy:
print("✅ 健康检查测试通过")
else:
print("⚠️ 健康检查返回异常（可能是正常的）")
return True
finally:
mem.close()
except Exception as e:
print(f"❌ 基础操作测试失败: {e}")
return False
def test_context_manager():
"""测试上下文管理器"""
print("\n🔧 测试上下文管理器")
print("-" * 30)
try:
from memory_ops_sdk import MemoryOps
with tempfile.TemporaryDirectory() as temp_dir:
# 测试上下文管理器
with MemoryOps(data_dir=temp_dir, verbose=True) as mem:
success = mem.add("上下文管理器测试", tags=["上下文"])
results = mem.query("上下文")
if success and len(results) > 0:
print("✅ 上下文管理器测试通过")
return True
else:
print("❌ 上下文管理器测试失败")
return False
# 上下文管理器应该自动清理资源
print("✅ 资源自动清理完成")
except Exception as e:
print(f"❌ 上下文管理器测试失败: {e}")
return False
def test_convenience_functions():
"""测试便捷函数"""
print("\n⚡ 测试便捷函数")
print("-" * 30)
try:
from memory_ops_sdk import create_memory_ops, get_global_memory_ops
with tempfile.TemporaryDirectory() as temp_dir:
# 测试create_memory_ops
mem1 = create_memory_ops(temp_dir, verbose=True)
try:
success = mem1.add("便捷函数测试1", tags=["便捷"])
if success:
print("✅ create_memory_ops测试通过")
else:
print("❌ create_memory_ops测试失败")
return False
finally:
mem1.close()
# 测试get_global_memory_ops
mem2 = get_global_memory_ops(data_dir=temp_dir, verbose=True)
success = mem2.add("便捷函数测试2", tags=["全局"])
if success:
print("✅ get_global_memory_ops测试通过")
else:
print("❌ get_global_memory_ops测试失败")
return False
return True
except Exception as e:
print(f"❌ 便捷函数测试失败: {e}")
return False
def test_advanced_features():
"""测试高级功能"""
print("\n🚀 测试高级功能")
print("-" * 30)
try:
from memory_ops_sdk import MemoryOps
with tempfile.TemporaryDirectory() as temp_dir:
with MemoryOps(data_dir=temp_dir, verbose=True) as mem:
# 测试带元数据的添加
metadata = {
"source": "测试脚本",
"priority": "high",
"category": "测试"
}
success = mem.add("高级功能测试", tags=["高级", "元数据"], metadata=metadata)
if not success:
print("❌ 元数据添加测试失败")
return False
# 测试高级查询
results = mem.query(
"高级功能",
limit=3,
use_reranker=True,
use_time_decay=True,
use_topic_drift=False
)
if len(results) == 0:
print("❌ 高级查询测试失败")
return False
# 测试导出功能
memories = mem.dump()
if len(memories) == 0:
print("❌ 导出功能测试失败")
return False
# 测试优化功能
optimize_success = mem.optimize()
# 优化可能失败，这是正常的
print("✅ 高级功能测试通过")
return True
except Exception as e:
print(f"❌ 高级功能测试失败: {e}")
return False
def test_error_handling():
"""测试错误处理"""
print("\n🛡️ 测试错误处理")
print("-" * 30)
try:
from memory_ops_sdk import MemoryOps
# 测试未初始化的情况
mem = MemoryOps(auto_init=False, verbose=True)
try:
mem.add("这应该失败")
print("❌ 错误处理测试失败 - 应该抛出异常")
return False
except RuntimeError:
print("✅ 正确捕获未初始化异常")
# 测试手动初始化
if mem.init():
print("✅ 手动初始化成功")
mem.close()
else:
print("⚠️ 手动初始化失败（可能是环境问题）")
return True
except Exception as e:
print(f"❌ 错误处理测试失败: {e}")
return False
def test_performance():
"""测试性能"""
print("\n⚡ 测试性能")
print("-" * 30)
try:
from memory_ops_sdk import MemoryOps
with tempfile.TemporaryDirectory() as temp_dir:
with MemoryOps(data_dir=temp_dir, verbose=False) as mem:
# 测试批量添加性能
start_time = time.time()
for i in range(10):
mem.add(f"性能测试记忆 {i}", tags=["性能"])
add_time = time.time() - start_time
print(f"添加10条记忆耗时: {add_time:.2f}秒")
# 测试批量查询性能
start_time = time.time()
for i in range(5):
results = mem.query(f"性能测试 {i}")
query_time = time.time() - start_time
print(f"查询5次耗时: {query_time:.2f}秒")
# 性能阈值检查
if add_time < 30 and query_time < 15: # 宽松的阈值
print("✅ 性能测试通过")
return True
else:
print("⚠️ 性能可能需要优化")
return True # 不算失败，只是提醒
except Exception as e:
print(f"❌ 性能测试失败: {e}")
return False
def test_sdk_import():
"""测试SDK导入"""
print("📦 测试SDK导入")
print("-" * 30)
try:
# 测试主要导入
from memory_ops_sdk import MemoryOps
print("✅ MemoryOps导入成功")
from memory_ops_sdk import create_memory_ops
print("✅ create_memory_ops导入成功")
from memory_ops_sdk import get_global_memory_ops
print("✅ get_global_memory_ops导入成功")
# 测试类实例化
mem = MemoryOps(auto_init=False)
print(f"✅ MemoryOps实例化成功: {mem}")
return True
except Exception as e:
print(f"❌ SDK导入测试失败: {e}")
return False
def main():
"""主测试函数"""
print("🧠 MemOS Memory Ops SDK 测试")
print("=" * 50)
# 运行所有测试
tests = [
("SDK导入", test_sdk_import),
("基础操作", test_basic_operations),
("上下文管理器", test_context_manager),
("便捷函数", test_convenience_functions),
("高级功能", test_advanced_features),
("错误处理", test_error_handling),
("性能测试", test_performance)
]
passed = 0
total = len(tests)
for test_name, test_func in tests:
try:
if test_func():
passed += 1
print(f"✅ {test_name} 测试通过")
else:
print(f"❌ {test_name} 测试失败")
except Exception as e:
print(f"💥 {test_name} 测试异常: {e}")
# 汇总结果
print("\n📊 测试结果汇总")
print("=" * 50)
print(f"总测试数: {total}")
print(f"通过数: {passed}")
print(f"失败数: {total - passed}")
print(f"通过率: {passed/total*100:.1f}%")
if passed == total:
print("\n🎉 所有测试通过！SDK功能正常")
return True
else:
print(f"\n⚠️ {total - passed} 个测试失败，请检查相关功能")
return False
if __name__ == "__main__":
try:
success = main()
sys.exit(0 if success else 1)
except KeyboardInterrupt:
print("\n🛑 测试被中断")
sys.exit(1)