diff --git a/src/mcp_feedback_enhanced/utils/memory_monitor.py b/src/mcp_feedback_enhanced/utils/memory_monitor.py new file mode 100644 index 0000000..644161a --- /dev/null +++ b/src/mcp_feedback_enhanced/utils/memory_monitor.py @@ -0,0 +1,526 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +集成式內存監控系統 +================== + +提供與資源管理器深度集成的內存監控功能，包括： +- 系統和進程內存使用監控 +- 智能清理觸發機制 +- 內存洩漏檢測和趨勢分析 +- 性能優化建議 +""" + +import os +import gc +import time +import threading +import psutil +from typing import Dict, List, Optional, Callable, Any +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from collections import deque +from ..debug import debug_log +from .error_handler import ErrorHandler, ErrorType + + +@dataclass +class MemorySnapshot: + """內存快照數據類""" + timestamp: datetime + system_total: int # 系統總內存 (bytes) + system_available: int # 系統可用內存 (bytes) + system_used: int # 系統已用內存 (bytes) + system_percent: float # 系統內存使用率 (%) + process_rss: int # 進程常駐內存 (bytes) + process_vms: int # 進程虛擬內存 (bytes) + process_percent: float # 進程內存使用率 (%) + gc_objects: int # Python 垃圾回收對象數量 + + +@dataclass +class MemoryAlert: + """內存警告數據類""" + level: str # warning, critical, emergency + message: str + timestamp: datetime + memory_percent: float + recommended_action: str + + +@dataclass +class MemoryStats: + """內存統計數據類""" + monitoring_duration: float # 監控持續時間 (秒) + snapshots_count: int # 快照數量 + average_system_usage: float # 平均系統內存使用率 + peak_system_usage: float # 峰值系統內存使用率 + average_process_usage: float # 平均進程內存使用率 + peak_process_usage: float # 峰值進程內存使用率 + alerts_count: int # 警告數量 + cleanup_triggers: int # 清理觸發次數 + memory_trend: str # 內存趨勢 (stable, increasing, decreasing) + + +class MemoryMonitor: + """集成式內存監控器""" + + def __init__(self, + warning_threshold: float = 0.8, + critical_threshold: float = 0.9, + emergency_threshold: float = 0.95, + monitoring_interval: int = 30, + max_snapshots: int = 1000): + """ + 初始化內存監控器 + + Args: + warning_threshold: 警告閾值 (0.0-1.0) + critical_threshold: 危險閾值 (0.0-1.0) + emergency_threshold: 緊急閾值 (0.0-1.0) + monitoring_interval: 監控間隔 (秒) + max_snapshots: 最大快照數量 + """ + self.warning_threshold = warning_threshold + self.critical_threshold = critical_threshold + self.emergency_threshold = emergency_threshold + self.monitoring_interval = monitoring_interval + self.max_snapshots = max_snapshots + + # 監控狀態 + self.is_monitoring = False + self.monitor_thread: Optional[threading.Thread] = None + self._stop_event = threading.Event() + + # 數據存儲 + self.snapshots: deque = deque(maxlen=max_snapshots) + self.alerts: List[MemoryAlert] = [] + self.max_alerts = 100 + + # 回調函數 + self.cleanup_callbacks: List[Callable] = [] + self.alert_callbacks: List[Callable[[MemoryAlert], None]] = [] + + # 統計數據 + self.start_time: Optional[datetime] = None + self.cleanup_triggers_count = 0 + + # 進程信息 + self.process = psutil.Process() + + debug_log("MemoryMonitor 初始化完成") + + def start_monitoring(self) -> bool: + """ + 開始內存監控 + + Returns: + bool: 是否成功啟動 + """ + if self.is_monitoring: + debug_log("內存監控已在運行") + return True + + try: + self.is_monitoring = True + self.start_time = datetime.now() + self._stop_event.clear() + + self.monitor_thread = threading.Thread( + target=self._monitoring_loop, + name="MemoryMonitor", + daemon=True + ) + self.monitor_thread.start() + + debug_log(f"內存監控已啟動，間隔 {self.monitoring_interval} 秒") + return True + + except Exception as e: + self.is_monitoring = False + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "啟動內存監控"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"啟動內存監控失敗 [錯誤ID: {error_id}]: {e}") + return False + + def stop_monitoring(self) -> bool: + """ + 停止內存監控 + + Returns: + bool: 是否成功停止 + """ + if not self.is_monitoring: + debug_log("內存監控未在運行") + return True + + try: + self.is_monitoring = False + self._stop_event.set() + + if self.monitor_thread and self.monitor_thread.is_alive(): + self.monitor_thread.join(timeout=5) + + debug_log("內存監控已停止") + return True + + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "停止內存監控"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"停止內存監控失敗 [錯誤ID: {error_id}]: {e}") + return False + + def _monitoring_loop(self): + """內存監控主循環""" + debug_log("內存監控循環開始") + + while not self._stop_event.is_set(): + try: + # 收集內存快照 + snapshot = self._collect_memory_snapshot() + self.snapshots.append(snapshot) + + # 檢查內存使用情況 + self._check_memory_usage(snapshot) + + # 等待下次監控 + if self._stop_event.wait(self.monitoring_interval): + break + + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "內存監控循環"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"內存監控循環錯誤 [錯誤ID: {error_id}]: {e}") + + # 發生錯誤時等待較短時間後重試 + if self._stop_event.wait(5): + break + + debug_log("內存監控循環結束") + + def _collect_memory_snapshot(self) -> MemorySnapshot: + """收集內存快照""" + try: + # 系統內存信息 + system_memory = psutil.virtual_memory() + + # 進程內存信息 + process_memory = self.process.memory_info() + process_percent = self.process.memory_percent() + + # Python 垃圾回收信息 + gc_objects = len(gc.get_objects()) + + return MemorySnapshot( + timestamp=datetime.now(), + system_total=system_memory.total, + system_available=system_memory.available, + system_used=system_memory.used, + system_percent=system_memory.percent, + process_rss=process_memory.rss, + process_vms=process_memory.vms, + process_percent=process_percent, + gc_objects=gc_objects + ) + + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "收集內存快照"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"收集內存快照失敗 [錯誤ID: {error_id}]: {e}") + raise + + def _check_memory_usage(self, snapshot: MemorySnapshot): + """檢查內存使用情況並觸發相應動作""" + usage_percent = snapshot.system_percent / 100.0 + + # 檢查緊急閾值 + if usage_percent >= self.emergency_threshold: + alert = MemoryAlert( + level="emergency", + message=f"內存使用率達到緊急水平: {snapshot.system_percent:.1f}%", + timestamp=snapshot.timestamp, + memory_percent=snapshot.system_percent, + recommended_action="立即執行強制清理和垃圾回收" + ) + self._handle_alert(alert) + self._trigger_emergency_cleanup() + + # 檢查危險閾值 + elif usage_percent >= self.critical_threshold: + alert = MemoryAlert( + level="critical", + message=f"內存使用率達到危險水平: {snapshot.system_percent:.1f}%", + timestamp=snapshot.timestamp, + memory_percent=snapshot.system_percent, + recommended_action="執行資源清理和垃圾回收" + ) + self._handle_alert(alert) + self._trigger_cleanup() + + # 檢查警告閾值 + elif usage_percent >= self.warning_threshold: + alert = MemoryAlert( + level="warning", + message=f"內存使用率較高: {snapshot.system_percent:.1f}%", + timestamp=snapshot.timestamp, + memory_percent=snapshot.system_percent, + recommended_action="考慮執行輕量級清理" + ) + self._handle_alert(alert) + + def _handle_alert(self, alert: MemoryAlert): + """處理內存警告""" + # 添加到警告列表 + self.alerts.append(alert) + + # 限制警告數量 + if len(self.alerts) > self.max_alerts: + self.alerts = self.alerts[-self.max_alerts:] + + # 調用警告回調 + for callback in self.alert_callbacks: + try: + callback(alert) + except Exception as e: + debug_log(f"警告回調執行失敗: {e}") + + debug_log(f"內存警告 [{alert.level}]: {alert.message}") + + def _trigger_cleanup(self): + """觸發清理操作""" + self.cleanup_triggers_count += 1 + debug_log("觸發內存清理操作") + + # 執行 Python 垃圾回收 + collected = gc.collect() + debug_log(f"垃圾回收清理了 {collected} 個對象") + + # 調用清理回調 + for callback in self.cleanup_callbacks: + try: + callback() + except Exception as e: + debug_log(f"清理回調執行失敗: {e}") + + def _trigger_emergency_cleanup(self): + """觸發緊急清理操作""" + debug_log("觸發緊急內存清理操作") + + # 執行強制垃圾回收 + for _ in range(3): + collected = gc.collect() + debug_log(f"強制垃圾回收清理了 {collected} 個對象") + + # 調用清理回調（強制模式） + for callback in self.cleanup_callbacks: + try: + if hasattr(callback, '__call__'): + # 嘗試傳遞 force 參數 + import inspect + sig = inspect.signature(callback) + if 'force' in sig.parameters: + callback(force=True) + else: + callback() + else: + callback() + except Exception as e: + debug_log(f"緊急清理回調執行失敗: {e}") + + + def add_cleanup_callback(self, callback: Callable): + """添加清理回調函數""" + if callback not in self.cleanup_callbacks: + self.cleanup_callbacks.append(callback) + debug_log("添加清理回調函數") + + def add_alert_callback(self, callback: Callable[[MemoryAlert], None]): + """添加警告回調函數""" + if callback not in self.alert_callbacks: + self.alert_callbacks.append(callback) + debug_log("添加警告回調函數") + + def remove_cleanup_callback(self, callback: Callable): + """移除清理回調函數""" + if callback in self.cleanup_callbacks: + self.cleanup_callbacks.remove(callback) + debug_log("移除清理回調函數") + + def remove_alert_callback(self, callback: Callable[[MemoryAlert], None]): + """移除警告回調函數""" + if callback in self.alert_callbacks: + self.alert_callbacks.remove(callback) + debug_log("移除警告回調函數") + + def get_current_memory_info(self) -> Dict[str, Any]: + """獲取當前內存信息""" + try: + snapshot = self._collect_memory_snapshot() + return { + "timestamp": snapshot.timestamp.isoformat(), + "system": { + "total_gb": round(snapshot.system_total / (1024**3), 2), + "available_gb": round(snapshot.system_available / (1024**3), 2), + "used_gb": round(snapshot.system_used / (1024**3), 2), + "usage_percent": round(snapshot.system_percent, 1) + }, + "process": { + "rss_mb": round(snapshot.process_rss / (1024**2), 2), + "vms_mb": round(snapshot.process_vms / (1024**2), 2), + "usage_percent": round(snapshot.process_percent, 1) + }, + "gc_objects": snapshot.gc_objects, + "status": self._get_memory_status(snapshot.system_percent / 100.0) + } + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "獲取當前內存信息"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"獲取內存信息失敗 [錯誤ID: {error_id}]: {e}") + return {} + + def get_memory_stats(self) -> MemoryStats: + """獲取內存統計數據""" + if not self.snapshots: + return MemoryStats( + monitoring_duration=0.0, + snapshots_count=0, + average_system_usage=0.0, + peak_system_usage=0.0, + average_process_usage=0.0, + peak_process_usage=0.0, + alerts_count=0, + cleanup_triggers=0, + memory_trend="unknown" + ) + + # 計算統計數據 + system_usages = [s.system_percent for s in self.snapshots] + process_usages = [s.process_percent for s in self.snapshots] + + duration = 0.0 + if self.start_time: + duration = (datetime.now() - self.start_time).total_seconds() + + return MemoryStats( + monitoring_duration=duration, + snapshots_count=len(self.snapshots), + average_system_usage=sum(system_usages) / len(system_usages), + peak_system_usage=max(system_usages), + average_process_usage=sum(process_usages) / len(process_usages), + peak_process_usage=max(process_usages), + alerts_count=len(self.alerts), + cleanup_triggers=self.cleanup_triggers_count, + memory_trend=self._analyze_memory_trend() + ) + + def get_recent_alerts(self, limit: int = 10) -> List[MemoryAlert]: + """獲取最近的警告""" + return self.alerts[-limit:] if self.alerts else [] + + def _get_memory_status(self, usage_percent: float) -> str: + """獲取內存狀態描述""" + if usage_percent >= self.emergency_threshold: + return "emergency" + elif usage_percent >= self.critical_threshold: + return "critical" + elif usage_percent >= self.warning_threshold: + return "warning" + else: + return "normal" + + def _analyze_memory_trend(self) -> str: + """分析內存使用趨勢""" + if len(self.snapshots) < 10: + return "insufficient_data" + + # 取最近的快照進行趨勢分析 + recent_snapshots = list(self.snapshots)[-10:] + usages = [s.system_percent for s in recent_snapshots] + + # 簡單的線性趨勢分析 + first_half = usages[:5] + second_half = usages[5:] + + avg_first = sum(first_half) / len(first_half) + avg_second = sum(second_half) / len(second_half) + + diff = avg_second - avg_first + + if abs(diff) < 2.0: # 變化小於 2% + return "stable" + elif diff > 0: + return "increasing" + else: + return "decreasing" + + def force_cleanup(self): + """手動觸發清理操作""" + debug_log("手動觸發內存清理") + self._trigger_cleanup() + + def force_emergency_cleanup(self): + """手動觸發緊急清理操作""" + debug_log("手動觸發緊急內存清理") + self._trigger_emergency_cleanup() + + def reset_stats(self): + """重置統計數據""" + self.snapshots.clear() + self.alerts.clear() + self.cleanup_triggers_count = 0 + self.start_time = datetime.now() if self.is_monitoring else None + debug_log("內存監控統計數據已重置") + + def export_memory_data(self) -> Dict[str, Any]: + """導出內存數據""" + return { + "config": { + "warning_threshold": self.warning_threshold, + "critical_threshold": self.critical_threshold, + "emergency_threshold": self.emergency_threshold, + "monitoring_interval": self.monitoring_interval + }, + "current_info": self.get_current_memory_info(), + "stats": self.get_memory_stats().__dict__, + "recent_alerts": [ + { + "level": alert.level, + "message": alert.message, + "timestamp": alert.timestamp.isoformat(), + "memory_percent": alert.memory_percent, + "recommended_action": alert.recommended_action + } + for alert in self.get_recent_alerts() + ], + "is_monitoring": self.is_monitoring + } + + +# 全域內存監控器實例 +_memory_monitor: Optional[MemoryMonitor] = None +_monitor_lock = threading.Lock() + + +def get_memory_monitor() -> MemoryMonitor: + """獲取全域內存監控器實例""" + global _memory_monitor + if _memory_monitor is None: + with _monitor_lock: + if _memory_monitor is None: + _memory_monitor = MemoryMonitor() + return _memory_monitor diff --git a/src/mcp_feedback_enhanced/utils/resource_manager.py b/src/mcp_feedback_enhanced/utils/resource_manager.py index 205c180..8bd200c 100644 --- a/src/mcp_feedback_enhanced/utils/resource_manager.py +++ b/src/mcp_feedback_enhanced/utils/resource_manager.py @@ -82,9 +82,71 @@ class ResourceManager: # 啟動自動清理 self._start_auto_cleanup() - + + # 集成內存監控 + self._setup_memory_monitoring() + debug_log("ResourceManager 初始化完成") - + + def _setup_memory_monitoring(self): + """設置內存監控集成""" + try: + # 延遲導入避免循環依賴 + from .memory_monitor import get_memory_monitor + + self.memory_monitor = get_memory_monitor() + + # 註冊清理回調 + self.memory_monitor.add_cleanup_callback(self._memory_triggered_cleanup) + + # 啟動內存監控 + if self.memory_monitor.start_monitoring(): + debug_log("內存監控已集成到資源管理器") + else: + debug_log("內存監控啟動失敗") + + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "設置內存監控"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"設置內存監控失敗 [錯誤ID: {error_id}]: {e}") + + def _memory_triggered_cleanup(self, force: bool = False): + """內存監控觸發的清理操作""" + debug_log(f"內存監控觸發清理操作 (force={force})") + + try: + # 清理臨時文件 + cleaned_files = self.cleanup_temp_files() + + # 清理臨時目錄 + cleaned_dirs = self.cleanup_temp_dirs() + + # 清理文件句柄 + cleaned_handles = self.cleanup_file_handles() + + # 如果是強制清理，也清理進程 + cleaned_processes = 0 + if force: + cleaned_processes = self.cleanup_processes(force=True) + + debug_log(f"內存觸發清理完成: 文件={cleaned_files}, 目錄={cleaned_dirs}, " + f"句柄={cleaned_handles}, 進程={cleaned_processes}") + + # 更新統計 + self.stats["cleanup_runs"] += 1 + self.stats["last_cleanup"] = time.time() + + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "內存觸發清理", "force": force}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"內存觸發清理失敗 [錯誤ID: {error_id}]: {e}") + def create_temp_file( self, suffix: str = "", @@ -620,6 +682,22 @@ class ResourceManager: "temp_file_max_age": self.temp_file_max_age }) + # 添加內存監控統計 + try: + if hasattr(self, 'memory_monitor') and self.memory_monitor: + memory_info = self.memory_monitor.get_current_memory_info() + memory_stats = self.memory_monitor.get_memory_stats() + + current_stats.update({ + "memory_monitoring_enabled": self.memory_monitor.is_monitoring, + "current_memory_usage": memory_info.get("system", {}).get("usage_percent", 0), + "memory_status": memory_info.get("status", "unknown"), + "memory_cleanup_triggers": memory_stats.cleanup_triggers, + "memory_alerts_count": memory_stats.alerts_count + }) + except Exception as e: + debug_log(f"獲取內存統計失敗: {e}") + return current_stats def get_detailed_info(self) -> Dict[str, Any]: diff --git a/src/mcp_feedback_enhanced/web/main.py b/src/mcp_feedback_enhanced/web/main.py index a837aed..05b50fc 100644 --- a/src/mcp_feedback_enhanced/web/main.py +++ b/src/mcp_feedback_enhanced/web/main.py @@ -32,6 +32,7 @@ from .utils import find_free_port, get_browser_opener from .utils.port_manager import PortManager from .utils.compression_config import get_compression_manager from ..utils.error_handler import ErrorHandler, ErrorType +from ..utils.memory_monitor import get_memory_monitor from ..debug import web_debug_log as debug_log from ..i18n import get_i18n_manager @@ -71,6 +72,9 @@ class WebUIManager: # 設置壓縮和緩存中間件 self._setup_compression_middleware() + # 設置內存監控 + self._setup_memory_monitoring() + # 重構：使用單一活躍會話而非會話字典 self.current_session: Optional[WebFeedbackSession] = None self.sessions: Dict[str, WebFeedbackSession] = {} # 保留用於向後兼容 @@ -136,6 +140,33 @@ class WebUIManager: debug_log("壓縮和緩存中間件設置完成") + def _setup_memory_monitoring(self): + """設置內存監控""" + try: + self.memory_monitor = get_memory_monitor() + + # 添加 Web 應用特定的警告回調 + def web_memory_alert(alert): + debug_log(f"Web UI 內存警告 [{alert.level}]: {alert.message}") + # 可以在這裡添加更多 Web 特定的處理邏輯 + # 例如：通過 WebSocket 通知前端、記錄到特定日誌等 + + self.memory_monitor.add_alert_callback(web_memory_alert) + + # 確保內存監控已啟動（ResourceManager 可能已經啟動了） + if not self.memory_monitor.is_monitoring: + self.memory_monitor.start_monitoring() + + debug_log("Web UI 內存監控設置完成") + + except Exception as e: + error_id = ErrorHandler.log_error_with_context( + e, + context={"operation": "設置 Web UI 內存監控"}, + error_type=ErrorType.SYSTEM + ) + debug_log(f"設置 Web UI 內存監控失敗 [錯誤ID: {error_id}]: {e}") + def _setup_static_files(self): """設置靜態文件服務""" # Web UI 靜態文件 diff --git a/tests/test_memory_monitor.py b/tests/test_memory_monitor.py new file mode 100644 index 0000000..d484c8e --- /dev/null +++ b/tests/test_memory_monitor.py @@ -0,0 +1,388 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +內存監控系統測試 +================ + +測試集成式內存監控系統的功能，包括： +- 內存監控準確性 +- 警告機制 +- 清理觸發 +- 統計和分析功能 +""" + +import pytest +import time +import threading +from unittest.mock import Mock, patch, MagicMock +from datetime import datetime, timedelta + +from src.mcp_feedback_enhanced.utils.memory_monitor import ( + MemoryMonitor, MemorySnapshot, MemoryAlert, MemoryStats, + get_memory_monitor +) + + +class TestMemorySnapshot: + """測試內存快照數據類""" + + def test_memory_snapshot_creation(self): + """測試內存快照創建""" + snapshot = MemorySnapshot( + timestamp=datetime.now(), + system_total=8 * 1024**3, # 8GB + system_available=4 * 1024**3, # 4GB + system_used=4 * 1024**3, # 4GB + system_percent=50.0, + process_rss=100 * 1024**2, # 100MB + process_vms=200 * 1024**2, # 200MB + process_percent=1.25, + gc_objects=10000 + ) + + assert snapshot.system_total == 8 * 1024**3 + assert snapshot.system_percent == 50.0 + assert snapshot.process_rss == 100 * 1024**2 + assert snapshot.gc_objects == 10000 + + +class TestMemoryAlert: + """測試內存警告數據類""" + + def test_memory_alert_creation(self): + """測試內存警告創建""" + alert = MemoryAlert( + level="warning", + message="內存使用率較高: 85.0%", + timestamp=datetime.now(), + memory_percent=85.0, + recommended_action="考慮執行輕量級清理" + ) + + assert alert.level == "warning" + assert alert.memory_percent == 85.0 + assert "85.0%" in alert.message + + +class TestMemoryMonitor: + """測試內存監控器""" + + def test_monitor_initialization(self): + """測試監控器初始化""" + monitor = MemoryMonitor( + warning_threshold=0.7, + critical_threshold=0.85, + emergency_threshold=0.95, + monitoring_interval=10 + ) + + assert monitor.warning_threshold == 0.7 + assert monitor.critical_threshold == 0.85 + assert monitor.emergency_threshold == 0.95 + assert monitor.monitoring_interval == 10 + assert not monitor.is_monitoring + assert len(monitor.snapshots) == 0 + assert len(monitor.alerts) == 0 + + @patch('src.mcp_feedback_enhanced.utils.memory_monitor.psutil') + def test_collect_memory_snapshot(self, mock_psutil): + """測試內存快照收集""" + # 模擬 psutil 返回值 + mock_virtual_memory = Mock() + mock_virtual_memory.total = 8 * 1024**3 + mock_virtual_memory.available = 4 * 1024**3 + mock_virtual_memory.used = 4 * 1024**3 + mock_virtual_memory.percent = 50.0 + + mock_memory_info = Mock() + mock_memory_info.rss = 100 * 1024**2 + mock_memory_info.vms = 200 * 1024**2 + + mock_process = Mock() + mock_process.memory_info.return_value = mock_memory_info + mock_process.memory_percent.return_value = 1.25 + + mock_psutil.virtual_memory.return_value = mock_virtual_memory + mock_psutil.Process.return_value = mock_process + + monitor = MemoryMonitor() + snapshot = monitor._collect_memory_snapshot() + + assert snapshot.system_total == 8 * 1024**3 + assert snapshot.system_percent == 50.0 + assert snapshot.process_rss == 100 * 1024**2 + assert snapshot.process_percent == 1.25 + + def test_memory_status_classification(self): + """測試內存狀態分類""" + monitor = MemoryMonitor( + warning_threshold=0.8, + critical_threshold=0.9, + emergency_threshold=0.95 + ) + + assert monitor._get_memory_status(0.5) == "normal" + assert monitor._get_memory_status(0.85) == "warning" + assert monitor._get_memory_status(0.92) == "critical" + assert monitor._get_memory_status(0.97) == "emergency" + + def test_callback_management(self): + """測試回調函數管理""" + monitor = MemoryMonitor() + + cleanup_callback = Mock() + alert_callback = Mock() + + # 添加回調 + monitor.add_cleanup_callback(cleanup_callback) + monitor.add_alert_callback(alert_callback) + + assert cleanup_callback in monitor.cleanup_callbacks + assert alert_callback in monitor.alert_callbacks + + # 移除回調 + monitor.remove_cleanup_callback(cleanup_callback) + monitor.remove_alert_callback(alert_callback) + + assert cleanup_callback not in monitor.cleanup_callbacks + assert alert_callback not in monitor.alert_callbacks + + @patch('src.mcp_feedback_enhanced.utils.memory_monitor.gc') + def test_cleanup_triggering(self, mock_gc): + """測試清理觸發""" + monitor = MemoryMonitor() + cleanup_callback = Mock() + monitor.add_cleanup_callback(cleanup_callback) + + mock_gc.collect.return_value = 42 + + # 測試普通清理 + monitor._trigger_cleanup() + + assert monitor.cleanup_triggers_count == 1 + cleanup_callback.assert_called_once() + mock_gc.collect.assert_called() + + # 測試緊急清理 + cleanup_callback.reset_mock() + mock_gc.collect.reset_mock() + + monitor._trigger_emergency_cleanup() + + # 緊急清理會調用多次垃圾回收 + assert mock_gc.collect.call_count == 3 + + @patch('src.mcp_feedback_enhanced.utils.memory_monitor.psutil') + def test_memory_usage_checking(self, mock_psutil): + """測試內存使用檢查和警告觸發""" + monitor = MemoryMonitor( + warning_threshold=0.8, + critical_threshold=0.9, + emergency_threshold=0.95 + ) + + alert_callback = Mock() + cleanup_callback = Mock() + monitor.add_alert_callback(alert_callback) + monitor.add_cleanup_callback(cleanup_callback) + + # 模擬不同的內存使用情況 + test_cases = [ + (75.0, "normal", 0, 0), # 正常情況 + (85.0, "warning", 1, 0), # 警告情況 + (92.0, "critical", 1, 1), # 危險情況 + (97.0, "emergency", 1, 1), # 緊急情況 + ] + + for memory_percent, expected_status, expected_alerts, expected_cleanups in test_cases: + # 重置計數器 + alert_callback.reset_mock() + cleanup_callback.reset_mock() + monitor.alerts.clear() + monitor.cleanup_triggers_count = 0 + + # 創建模擬快照 + snapshot = MemorySnapshot( + timestamp=datetime.now(), + system_total=8 * 1024**3, + system_available=int(8 * 1024**3 * (100 - memory_percent) / 100), + system_used=int(8 * 1024**3 * memory_percent / 100), + system_percent=memory_percent, + process_rss=100 * 1024**2, + process_vms=200 * 1024**2, + process_percent=1.25, + gc_objects=10000 + ) + + # 檢查內存使用 + monitor._check_memory_usage(snapshot) + + # 驗證結果 + assert monitor._get_memory_status(memory_percent / 100.0) == expected_status + + if expected_alerts > 0: + assert len(monitor.alerts) == expected_alerts + assert alert_callback.call_count == expected_alerts + + if expected_cleanups > 0: + assert cleanup_callback.call_count == expected_cleanups + + def test_memory_trend_analysis(self): + """測試內存趨勢分析""" + monitor = MemoryMonitor() + + # 測試數據不足的情況 + assert monitor._analyze_memory_trend() == "insufficient_data" + + # 添加穩定趨勢的快照 + base_time = datetime.now() + for i in range(10): + snapshot = MemorySnapshot( + timestamp=base_time + timedelta(seconds=i * 30), + system_total=8 * 1024**3, + system_available=4 * 1024**3, + system_used=4 * 1024**3, + system_percent=50.0 + (i % 2), # 輕微波動 + process_rss=100 * 1024**2, + process_vms=200 * 1024**2, + process_percent=1.25, + gc_objects=10000 + ) + monitor.snapshots.append(snapshot) + + assert monitor._analyze_memory_trend() == "stable" + + # 清空並添加遞增趨勢的快照 + monitor.snapshots.clear() + for i in range(10): + snapshot = MemorySnapshot( + timestamp=base_time + timedelta(seconds=i * 30), + system_total=8 * 1024**3, + system_available=4 * 1024**3, + system_used=4 * 1024**3, + system_percent=50.0 + i * 2, # 遞增趨勢 + process_rss=100 * 1024**2, + process_vms=200 * 1024**2, + process_percent=1.25, + gc_objects=10000 + ) + monitor.snapshots.append(snapshot) + + assert monitor._analyze_memory_trend() == "increasing" + + @patch('src.mcp_feedback_enhanced.utils.memory_monitor.psutil') + def test_get_current_memory_info(self, mock_psutil): + """測試獲取當前內存信息""" + # 模擬 psutil 返回值 + mock_virtual_memory = Mock() + mock_virtual_memory.total = 8 * 1024**3 + mock_virtual_memory.available = 4 * 1024**3 + mock_virtual_memory.used = 4 * 1024**3 + mock_virtual_memory.percent = 50.0 + + mock_memory_info = Mock() + mock_memory_info.rss = 100 * 1024**2 + mock_memory_info.vms = 200 * 1024**2 + + mock_process = Mock() + mock_process.memory_info.return_value = mock_memory_info + mock_process.memory_percent.return_value = 1.25 + + mock_psutil.virtual_memory.return_value = mock_virtual_memory + mock_psutil.Process.return_value = mock_process + + monitor = MemoryMonitor() + info = monitor.get_current_memory_info() + + assert "system" in info + assert "process" in info + assert info["system"]["total_gb"] == 8.0 + assert info["system"]["usage_percent"] == 50.0 + assert info["process"]["rss_mb"] == 100.0 + assert info["status"] == "normal" + + def test_memory_stats_calculation(self): + """測試內存統計計算""" + monitor = MemoryMonitor() + monitor.start_time = datetime.now() - timedelta(minutes=5) + + # 添加一些測試快照 + base_time = datetime.now() + for i in range(5): + snapshot = MemorySnapshot( + timestamp=base_time + timedelta(seconds=i * 30), + system_total=8 * 1024**3, + system_available=4 * 1024**3, + system_used=4 * 1024**3, + system_percent=50.0 + i * 5, # 50%, 55%, 60%, 65%, 70% + process_rss=100 * 1024**2, + process_vms=200 * 1024**2, + process_percent=1.0 + i * 0.2, # 1.0%, 1.2%, 1.4%, 1.6%, 1.8% + gc_objects=10000 + ) + monitor.snapshots.append(snapshot) + + # 添加一些警告 + monitor.alerts.append(MemoryAlert( + level="warning", + message="Test warning", + timestamp=datetime.now(), + memory_percent=85.0, + recommended_action="Test action" + )) + + monitor.cleanup_triggers_count = 2 + + stats = monitor.get_memory_stats() + + assert stats.snapshots_count == 5 + assert stats.average_system_usage == 60.0 # (50+55+60+65+70)/5 + assert stats.peak_system_usage == 70.0 + assert stats.average_process_usage == 1.4 # (1.0+1.2+1.4+1.6+1.8)/5 + assert stats.peak_process_usage == 1.8 + assert stats.alerts_count == 1 + assert stats.cleanup_triggers == 2 + assert stats.monitoring_duration > 0 + + def test_export_memory_data(self): + """測試內存數據導出""" + monitor = MemoryMonitor() + + # 添加一些測試數據 + monitor.alerts.append(MemoryAlert( + level="warning", + message="Test warning", + timestamp=datetime.now(), + memory_percent=85.0, + recommended_action="Test action" + )) + + with patch.object(monitor, 'get_current_memory_info') as mock_info: + mock_info.return_value = { + "system": {"usage_percent": 75.0}, + "status": "warning" + } + + exported_data = monitor.export_memory_data() + + assert "config" in exported_data + assert "current_info" in exported_data + assert "stats" in exported_data + assert "recent_alerts" in exported_data + assert "is_monitoring" in exported_data + + assert exported_data["config"]["warning_threshold"] == 0.8 + assert len(exported_data["recent_alerts"]) == 1 + + +def test_global_memory_monitor_singleton(): + """測試全域內存監控器單例模式""" + monitor1 = get_memory_monitor() + monitor2 = get_memory_monitor() + + assert monitor1 is monitor2 + assert isinstance(monitor1, MemoryMonitor) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])