新增集成式內存監控系統,提供系統與進程內存使用監控、智能清理機制、內存洩漏檢測及性能優化建議。更新資源管理器與 Web UI,支持內存監控集成與警告回調,並新增測試模組以確保功能正確性。

This commit is contained in:
Minidoracat 2025-06-07 02:29:16 +08:00
parent ca6a73c3be
commit 90deebdee5
4 changed files with 1025 additions and 2 deletions

View File

@ -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

View File

@ -83,8 +83,70 @@ 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]:

View File

@ -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 靜態文件

View File

@ -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"])