MemOS-MCP

""" MemOS官方API测试脚本 全面测试API功能、性能和错误处理 """ import asyncio import time import json import requests from typing import Dict, Any, List from datetime import datetime class MemOSAPITester: """MemOS API测试器""" def __init__(self, base_url: str = "http://localhost:8000"): self.base_url = base_url self.headers = { "Content-Type": "application/json", "Accept": "application/json" } self.test_results = [] def log_test(self, test_name: str, success: bool, message: str, duration: float = 0): """记录测试结果""" result = { "test_name": test_name, "success": success, "message": message, "duration": duration, "timestamp": datetime.now().isoformat() } self.test_results.append(result) status = "✅" if success else "❌" print(f"{status} {test_name}: {message} ({duration:.3f}s)") def test_health_check(self) -> bool: """测试健康检查""" start_time = time.time() try: response = requests.get(f"{self.base_url}/health", headers=self.headers, timeout=10) duration = time.time() - start_time if response.status_code == 200: data = response.json() if data.get("success"): self.log_test("健康检查", True, f"系统状态: {data['data']['api_status']}", duration) return True else: self.log_test("健康检查", False, f"API返回失败: {data.get('message')}", duration) return False else: self.log_test("健康检查", False, f"HTTP错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("健康检查", False, f"请求异常: {str(e)}", duration) return False def test_system_status(self) -> bool: """测试系统状态""" start_time = time.time() try: response = requests.get(f"{self.base_url}/status", headers=self.headers, timeout=10) duration = time.time() - start_time if response.status_code == 200: data = response.json() if data.get("success"): status_data = data["data"] self.log_test("系统状态", True, f"模式: {status_data['mode']}, 模型: {status_data['model']}", duration) return True else: self.log_test("系统状态", False, f"获取失败: {data.get('message')}", duration) return False else: self.log_test("系统状态", False, f"HTTP错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("系统状态", False, f"请求异常: {str(e)}", duration) return False def test_create_memory(self) -> str: """测试创建记忆""" start_time = time.time() try: test_content = f"这是API测试记忆 - {datetime.now().strftime('%H:%M:%S')}" data = { "content": test_content, "tags": ["API测试", "自动化"], "metadata": {"test_type": "api_test"} } response = requests.post( f"{self.base_url}/memories", headers=self.headers, json=data, timeout=30 ) duration = time.time() - start_time if response.status_code == 200: result = response.json() if result.get("success"): self.log_test("创建记忆", True, f"记忆已创建: {result['data']['content_preview']}", duration) return test_content else: self.log_test("创建记忆", False, f"创建失败: {result.get('message')}", duration) return "" else: self.log_test("创建记忆", False, f"HTTP错误: {response.status_code}", duration) return "" except Exception as e: duration = time.time() - start_time self.log_test("创建记忆", False, f"请求异常: {str(e)}", duration) return "" def test_search_memory(self, query: str) -> List[Dict[str, Any]]: """测试搜索记忆""" start_time = time.time() try: data = { "query": query, "top_k": 5, "use_reranker": True, "use_feedback_boost": True } response = requests.post( f"{self.base_url}/memories/search", headers=self.headers, json=data, timeout=30 ) duration = time.time() - start_time if response.status_code == 200: result = response.json() if result.get("success"): results = result["data"]["results"] self.log_test("搜索记忆", True, f"找到 {len(results)} 条记忆", duration) return results else: self.log_test("搜索记忆", False, f"搜索失败: {result.get('message')}", duration) return [] else: self.log_test("搜索记忆", False, f"HTTP错误: {response.status_code}", duration) return [] except Exception as e: duration = time.time() - start_time self.log_test("搜索记忆", False, f"请求异常: {str(e)}", duration) return [] def test_memory_feedback(self, memory_id: str) -> bool: """测试记忆反馈""" start_time = time.time() try: data = { "memory_id": memory_id, "feedback_type": "thumbs_up" } response = requests.post( f"{self.base_url}/memories/feedback", headers=self.headers, json=data, timeout=10 ) duration = time.time() - start_time if response.status_code == 200: result = response.json() if result.get("success"): self.log_test("记忆反馈", True, f"反馈已保存: {result['data']['feedback_emoji']}", duration) return True else: self.log_test("记忆反馈", False, f"反馈失败: {result.get('message')}", duration) return False else: self.log_test("记忆反馈", False, f"HTTP错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("记忆反馈", False, f"请求异常: {str(e)}", duration) return False def test_performance_metrics(self) -> bool: """测试性能指标""" start_time = time.time() try: response = requests.get(f"{self.base_url}/metrics", headers=self.headers, timeout=10) duration = time.time() - start_time if response.status_code == 200: data = response.json() if data.get("success"): metrics = data["data"] self.log_test("性能指标", True, f"添加时间: {metrics['memory_add_avg_time']:.3f}s, " f"检索时间: {metrics['memory_search_avg_time']:.3f}s", duration) return True else: self.log_test("性能指标", False, f"获取失败: {data.get('message')}", duration) return False else: self.log_test("性能指标", False, f"HTTP错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("性能指标", False, f"请求异常: {str(e)}", duration) return False def test_capacity_report(self) -> bool: """测试容量报告""" start_time = time.time() try: response = requests.get(f"{self.base_url}/capacity/report", headers=self.headers, timeout=10) duration = time.time() - start_time if response.status_code == 200: data = response.json() if data.get("success"): report = data["data"] health = report.get("system_health", "unknown") self.log_test("容量报告", True, f"系统健康: {health}", duration) return True else: self.log_test("容量报告", False, f"获取失败: {data.get('message')}", duration) return False else: self.log_test("容量报告", False, f"HTTP错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("容量报告", False, f"请求异常: {str(e)}", duration) return False def test_system_optimization(self) -> bool: """测试系统优化""" start_time = time.time() try: response = requests.post(f"{self.base_url}/system/optimize", headers=self.headers, timeout=30) duration = time.time() - start_time if response.status_code == 200: data = response.json() if data.get("success"): self.log_test("系统优化", True, "优化完成", duration) return True else: self.log_test("系统优化", False, f"优化失败: {data.get('message')}", duration) return False else: self.log_test("系统优化", False, f"HTTP错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("系统优化", False, f"请求异常: {str(e)}", duration) return False def test_error_handling(self) -> bool: """测试错误处理""" start_time = time.time() try: # 测试无效的记忆创建请求 invalid_data = {"invalid_field": "test"} response = requests.post( f"{self.base_url}/memories", headers=self.headers, json=invalid_data, timeout=10 ) duration = time.time() - start_time if response.status_code == 422: # 验证错误 self.log_test("错误处理", True, "正确处理了验证错误", duration) return True else: self.log_test("错误处理", False, f"未正确处理错误: {response.status_code}", duration) return False except Exception as e: duration = time.time() - start_time self.log_test("错误处理", False, f"请求异常: {str(e)}", duration) return False def run_comprehensive_test(self) -> Dict[str, Any]: """运行综合测试""" print("🚀 开始MemOS官方API综合测试") print("=" * 60) total_start_time = time.time() # 基础功能测试 print("\n📋 基础功能测试:") health_ok = self.test_health_check() status_ok = self.test_system_status() # 记忆管理测试 print("\n🧠 记忆管理测试:") test_content = self.test_create_memory() search_results = [] if test_content: search_results = self.test_search_memory("API测试") # 反馈机制测试 print("\n👍 反馈机制测试:") feedback_ok = False if search_results: memory_id = search_results[0]["memory"]["id"] feedback_ok = self.test_memory_feedback(memory_id) # 性能监控测试 print("\n📊 性能监控测试:") metrics_ok = self.test_performance_metrics() capacity_ok = self.test_capacity_report() # 系统优化测试 print("\n⚡ 系统优化测试:") optimization_ok = self.test_system_optimization() # 错误处理测试 print("\n❌ 错误处理测试:") error_handling_ok = self.test_error_handling() total_duration = time.time() - total_start_time # 统计结果 passed_tests = sum(1 for result in self.test_results if result["success"]) total_tests = len(self.test_results) success_rate = (passed_tests / total_tests) * 100 if total_tests > 0 else 0 summary = { "total_tests": total_tests, "passed_tests": passed_tests, "failed_tests": total_tests - passed_tests, "success_rate": success_rate, "total_duration": total_duration, "test_results": self.test_results } print("\n" + "=" * 60) print("📊 测试结果汇总:") print(f" 总测试数: {total_tests}") print(f" 通过测试: {passed_tests}") print(f" 失败测试: {total_tests - passed_tests}") print(f" 成功率: {success_rate:.1f}%") print(f" 总耗时: {total_duration:.3f}s") if success_rate >= 80: print("✅ API测试整体通过！") else: print("❌ API测试存在问题，请检查失败的测试项") return summary def save_test_report(self, summary: Dict[str, Any], filename: str = "api_test_report.json"): """保存测试报告""" with open(filename, 'w', encoding='utf-8') as f: json.dump(summary, f, indent=2, ensure_ascii=False) print(f"\n📄 测试报告已保存到: {filename}") if __name__ == "__main__": # 运行API测试 tester = MemOSAPITester() try: summary = tester.run_comprehensive_test() tester.save_test_report(summary) except KeyboardInterrupt: print("\n\n⏹️ 测试被用户中断") except Exception as e: print(f"\n❌ 测试过程中发生错误: {e}") import traceback traceback.print_exc()