-
-
-
-
diff --git a/src/mcp_feedback_enhanced/web/utils/compression_config.py b/src/mcp_feedback_enhanced/web/utils/compression_config.py
new file mode 100644
index 0000000..5522c33
--- /dev/null
+++ b/src/mcp_feedback_enhanced/web/utils/compression_config.py
@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+壓縮配置管理器
+==============
+
+管理 Web UI 的 Gzip 壓縮配置和靜態文件緩存策略。
+支援可配置的壓縮參數和性能優化選項。
+"""
+
+import os
+from typing import Dict, List, Optional, Tuple
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass
+class CompressionConfig:
+ """壓縮配置類"""
+
+ # Gzip 壓縮設定
+ minimum_size: int = 1000 # 最小壓縮大小（bytes）
+ compression_level: int = 6 # 壓縮級別 (1-9, 6為平衡點)
+
+ # 緩存設定
+ static_cache_max_age: int = 3600 # 靜態文件緩存時間（秒）
+ api_cache_max_age: int = 0 # API 響應緩存時間（秒，0表示不緩存）
+
+ # 支援的 MIME 類型
+ compressible_types: List[str] = None
+
+ # 排除的路徑
+ exclude_paths: List[str] = None
+
+ def __post_init__(self):
+ """初始化後處理"""
+ if self.compressible_types is None:
+ self.compressible_types = [
+ 'text/html',
+ 'text/css',
+ 'text/javascript',
+ 'text/plain',
+ 'application/json',
+ 'application/javascript',
+ 'application/xml',
+ 'application/rss+xml',
+ 'application/atom+xml',
+ 'image/svg+xml'
+ ]
+
+ if self.exclude_paths is None:
+ self.exclude_paths = [
+ '/ws', # WebSocket 連接
+ '/api/ws', # WebSocket API
+ '/health', # 健康檢查
+ ]
+
+ @classmethod
+ def from_env(cls) -> 'CompressionConfig':
+ """從環境變數創建配置"""
+ return cls(
+ minimum_size=int(os.getenv('MCP_GZIP_MIN_SIZE', '1000')),
+ compression_level=int(os.getenv('MCP_GZIP_LEVEL', '6')),
+ static_cache_max_age=int(os.getenv('MCP_STATIC_CACHE_AGE', '3600')),
+ api_cache_max_age=int(os.getenv('MCP_API_CACHE_AGE', '0'))
+ )
+
+ def should_compress(self, content_type: str, content_length: int) -> bool:
+ """判斷是否應該壓縮"""
+ if content_length < self.minimum_size:
+ return False
+
+ if not content_type:
+ return False
+
+ # 檢查 MIME 類型
+ for mime_type in self.compressible_types:
+ if content_type.startswith(mime_type):
+ return True
+
+ return False
+
+ def should_exclude_path(self, path: str) -> bool:
+ """判斷路徑是否應該排除壓縮"""
+ for exclude_path in self.exclude_paths:
+ if path.startswith(exclude_path):
+ return True
+ return False
+
+ def get_cache_headers(self, path: str) -> Dict[str, str]:
+ """獲取緩存頭"""
+ headers = {}
+
+ if path.startswith('/static/'):
+ # 靜態文件緩存
+ headers['Cache-Control'] = f'public, max-age={self.static_cache_max_age}'
+ headers['Expires'] = self._get_expires_header(self.static_cache_max_age)
+ elif path.startswith('/api/') and self.api_cache_max_age > 0:
+ # API 緩存（如果啟用）
+ headers['Cache-Control'] = f'public, max-age={self.api_cache_max_age}'
+ headers['Expires'] = self._get_expires_header(self.api_cache_max_age)
+ else:
+ # 其他路徑不緩存
+ headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
+ headers['Pragma'] = 'no-cache'
+ headers['Expires'] = '0'
+
+ return headers
+
+ def _get_expires_header(self, max_age: int) -> str:
+ """生成 Expires 頭"""
+ from datetime import datetime, timedelta
+ expires_time = datetime.utcnow() + timedelta(seconds=max_age)
+ return expires_time.strftime('%a, %d %b %Y %H:%M:%S GMT')
+
+ def get_compression_stats(self) -> Dict[str, any]:
+ """獲取壓縮配置統計"""
+ return {
+ 'minimum_size': self.minimum_size,
+ 'compression_level': self.compression_level,
+ 'static_cache_max_age': self.static_cache_max_age,
+ 'compressible_types_count': len(self.compressible_types),
+ 'exclude_paths_count': len(self.exclude_paths),
+ 'compressible_types': self.compressible_types,
+ 'exclude_paths': self.exclude_paths
+ }
+
+
+class CompressionManager:
+ """壓縮管理器"""
+
+ def __init__(self, config: Optional[CompressionConfig] = None):
+ self.config = config or CompressionConfig.from_env()
+ self._stats = {
+ 'requests_total': 0,
+ 'requests_compressed': 0,
+ 'bytes_original': 0,
+ 'bytes_compressed': 0,
+ 'compression_ratio': 0.0
+ }
+
+ def update_stats(self, original_size: int, compressed_size: int, was_compressed: bool):
+ """更新壓縮統計"""
+ self._stats['requests_total'] += 1
+ self._stats['bytes_original'] += original_size
+
+ if was_compressed:
+ self._stats['requests_compressed'] += 1
+ self._stats['bytes_compressed'] += compressed_size
+ else:
+ self._stats['bytes_compressed'] += original_size
+
+ # 計算壓縮比率
+ if self._stats['bytes_original'] > 0:
+ self._stats['compression_ratio'] = (
+ 1 - self._stats['bytes_compressed'] / self._stats['bytes_original']
+ ) * 100
+
+ def get_stats(self) -> Dict[str, any]:
+ """獲取壓縮統計"""
+ stats = self._stats.copy()
+ stats['compression_percentage'] = (
+ self._stats['requests_compressed'] / max(self._stats['requests_total'], 1) * 100
+ )
+ return stats
+
+ def reset_stats(self):
+ """重置統計"""
+ self._stats = {
+ 'requests_total': 0,
+ 'requests_compressed': 0,
+ 'bytes_original': 0,
+ 'bytes_compressed': 0,
+ 'compression_ratio': 0.0
+ }
+
+
+# 全域壓縮管理器實例
+_compression_manager: Optional[CompressionManager] = None
+
+
+def get_compression_manager() -> CompressionManager:
+ """獲取全域壓縮管理器實例"""
+ global _compression_manager
+ if _compression_manager is None:
+ _compression_manager = CompressionManager()
+ return _compression_manager
diff --git a/src/mcp_feedback_enhanced/web/utils/compression_monitor.py b/src/mcp_feedback_enhanced/web/utils/compression_monitor.py
new file mode 100644
index 0000000..475bada
--- /dev/null
+++ b/src/mcp_feedback_enhanced/web/utils/compression_monitor.py
@@ -0,0 +1,290 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+壓縮性能監控工具
+================
+
+監控 Gzip 壓縮的性能效果，包括壓縮比率、響應時間和文件大小統計。
+提供實時性能數據和優化建議。
+"""
+
+import time
+import threading
+from typing import Dict, List, Optional, Tuple
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+import json
+
+
+@dataclass
+class CompressionMetrics:
+ """壓縮指標數據類"""
+ timestamp: datetime
+ path: str
+ original_size: int
+ compressed_size: int
+ compression_ratio: float
+ response_time: float
+ content_type: str
+ was_compressed: bool
+
+
+@dataclass
+class CompressionSummary:
+ """壓縮摘要統計"""
+ total_requests: int = 0
+ compressed_requests: int = 0
+ total_original_bytes: int = 0
+ total_compressed_bytes: int = 0
+ average_compression_ratio: float = 0.0
+ average_response_time: float = 0.0
+ compression_percentage: float = 0.0
+ bandwidth_saved: int = 0
+ top_compressed_paths: List[Tuple[str, float]] = field(default_factory=list)
+
+
+class CompressionMonitor:
+ """壓縮性能監控器"""
+
+ def __init__(self, max_metrics: int = 1000):
+ self.max_metrics = max_metrics
+ self.metrics: List[CompressionMetrics] = []
+ self.lock = threading.Lock()
+ self._start_time = datetime.now()
+
+ # 路徑統計
+ self.path_stats: Dict[str, Dict] = {}
+
+ # 內容類型統計
+ self.content_type_stats: Dict[str, Dict] = {}
+
+ def record_request(self,
+ path: str,
+ original_size: int,
+ compressed_size: int,
+ response_time: float,
+ content_type: str = "",
+ was_compressed: bool = False):
+ """記錄請求的壓縮數據"""
+
+ compression_ratio = 0.0
+ if original_size > 0 and was_compressed:
+ compression_ratio = (1 - compressed_size / original_size) * 100
+
+ metric = CompressionMetrics(
+ timestamp=datetime.now(),
+ path=path,
+ original_size=original_size,
+ compressed_size=compressed_size,
+ compression_ratio=compression_ratio,
+ response_time=response_time,
+ content_type=content_type,
+ was_compressed=was_compressed
+ )
+
+ with self.lock:
+ self.metrics.append(metric)
+
+ # 限制記錄數量
+ if len(self.metrics) > self.max_metrics:
+ self.metrics = self.metrics[-self.max_metrics:]
+
+ # 更新路徑統計
+ self._update_path_stats(metric)
+
+ # 更新內容類型統計
+ self._update_content_type_stats(metric)
+
+ def _update_path_stats(self, metric: CompressionMetrics):
+ """更新路徑統計"""
+ path = metric.path
+ if path not in self.path_stats:
+ self.path_stats[path] = {
+ 'requests': 0,
+ 'compressed_requests': 0,
+ 'total_original_bytes': 0,
+ 'total_compressed_bytes': 0,
+ 'total_response_time': 0.0,
+ 'best_compression_ratio': 0.0
+ }
+
+ stats = self.path_stats[path]
+ stats['requests'] += 1
+ stats['total_original_bytes'] += metric.original_size
+ stats['total_compressed_bytes'] += metric.compressed_size
+ stats['total_response_time'] += metric.response_time
+
+ if metric.was_compressed:
+ stats['compressed_requests'] += 1
+ stats['best_compression_ratio'] = max(
+ stats['best_compression_ratio'],
+ metric.compression_ratio
+ )
+
+ def _update_content_type_stats(self, metric: CompressionMetrics):
+ """更新內容類型統計"""
+ content_type = metric.content_type or 'unknown'
+ if content_type not in self.content_type_stats:
+ self.content_type_stats[content_type] = {
+ 'requests': 0,
+ 'compressed_requests': 0,
+ 'total_original_bytes': 0,
+ 'total_compressed_bytes': 0,
+ 'average_compression_ratio': 0.0
+ }
+
+ stats = self.content_type_stats[content_type]
+ stats['requests'] += 1
+ stats['total_original_bytes'] += metric.original_size
+ stats['total_compressed_bytes'] += metric.compressed_size
+
+ if metric.was_compressed:
+ stats['compressed_requests'] += 1
+
+ # 重新計算平均壓縮比
+ if stats['total_original_bytes'] > 0:
+ stats['average_compression_ratio'] = (
+ 1 - stats['total_compressed_bytes'] / stats['total_original_bytes']
+ ) * 100
+
+ def get_summary(self, time_window: Optional[timedelta] = None) -> CompressionSummary:
+ """獲取壓縮摘要統計"""
+ with self.lock:
+ metrics = self.metrics
+
+ # 如果指定時間窗口，過濾數據
+ if time_window:
+ cutoff_time = datetime.now() - time_window
+ metrics = [m for m in metrics if m.timestamp >= cutoff_time]
+
+ if not metrics:
+ return CompressionSummary()
+
+ total_requests = len(metrics)
+ compressed_requests = sum(1 for m in metrics if m.was_compressed)
+ total_original_bytes = sum(m.original_size for m in metrics)
+ total_compressed_bytes = sum(m.compressed_size for m in metrics)
+ total_response_time = sum(m.response_time for m in metrics)
+
+ # 計算統計數據
+ compression_percentage = (compressed_requests / total_requests * 100) if total_requests > 0 else 0
+ average_compression_ratio = 0.0
+ bandwidth_saved = 0
+
+ if total_original_bytes > 0:
+ average_compression_ratio = (1 - total_compressed_bytes / total_original_bytes) * 100
+ bandwidth_saved = total_original_bytes - total_compressed_bytes
+
+ average_response_time = total_response_time / total_requests if total_requests > 0 else 0
+
+ # 獲取壓縮效果最好的路徑
+ top_compressed_paths = self._get_top_compressed_paths()
+
+ return CompressionSummary(
+ total_requests=total_requests,
+ compressed_requests=compressed_requests,
+ total_original_bytes=total_original_bytes,
+ total_compressed_bytes=total_compressed_bytes,
+ average_compression_ratio=average_compression_ratio,
+ average_response_time=average_response_time,
+ compression_percentage=compression_percentage,
+ bandwidth_saved=bandwidth_saved,
+ top_compressed_paths=top_compressed_paths
+ )
+
+ def _get_top_compressed_paths(self, limit: int = 5) -> List[Tuple[str, float]]:
+ """獲取壓縮效果最好的路徑"""
+ path_ratios = []
+
+ for path, stats in self.path_stats.items():
+ if stats['compressed_requests'] > 0 and stats['total_original_bytes'] > 0:
+ compression_ratio = (
+ 1 - stats['total_compressed_bytes'] / stats['total_original_bytes']
+ ) * 100
+ path_ratios.append((path, compression_ratio))
+
+ # 按壓縮比排序
+ path_ratios.sort(key=lambda x: x[1], reverse=True)
+ return path_ratios[:limit]
+
+ def get_path_stats(self) -> Dict[str, Dict]:
+ """獲取路徑統計"""
+ with self.lock:
+ return self.path_stats.copy()
+
+ def get_content_type_stats(self) -> Dict[str, Dict]:
+ """獲取內容類型統計"""
+ with self.lock:
+ return self.content_type_stats.copy()
+
+ def get_recent_metrics(self, limit: int = 100) -> List[CompressionMetrics]:
+ """獲取最近的指標數據"""
+ with self.lock:
+ return self.metrics[-limit:] if self.metrics else []
+
+ def reset_stats(self):
+ """重置統計數據"""
+ with self.lock:
+ self.metrics.clear()
+ self.path_stats.clear()
+ self.content_type_stats.clear()
+ self._start_time = datetime.now()
+
+ def export_stats(self) -> Dict:
+ """導出統計數據為字典格式"""
+ summary = self.get_summary()
+
+ return {
+ 'summary': {
+ 'total_requests': summary.total_requests,
+ 'compressed_requests': summary.compressed_requests,
+ 'compression_percentage': round(summary.compression_percentage, 2),
+ 'average_compression_ratio': round(summary.average_compression_ratio, 2),
+ 'bandwidth_saved_mb': round(summary.bandwidth_saved / (1024 * 1024), 2),
+ 'average_response_time_ms': round(summary.average_response_time * 1000, 2),
+ 'monitoring_duration_hours': round(
+ (datetime.now() - self._start_time).total_seconds() / 3600, 2
+ )
+ },
+ 'top_compressed_paths': [
+ {'path': path, 'compression_ratio': round(ratio, 2)}
+ for path, ratio in summary.top_compressed_paths
+ ],
+ 'path_stats': {
+ path: {
+ 'requests': stats['requests'],
+ 'compression_percentage': round(
+ stats['compressed_requests'] / stats['requests'] * 100, 2
+ ) if stats['requests'] > 0 else 0,
+ 'average_response_time_ms': round(
+ stats['total_response_time'] / stats['requests'] * 1000, 2
+ ) if stats['requests'] > 0 else 0,
+ 'bandwidth_saved_kb': round(
+ (stats['total_original_bytes'] - stats['total_compressed_bytes']) / 1024, 2
+ )
+ }
+ for path, stats in self.path_stats.items()
+ },
+ 'content_type_stats': {
+ content_type: {
+ 'requests': stats['requests'],
+ 'compression_percentage': round(
+ stats['compressed_requests'] / stats['requests'] * 100, 2
+ ) if stats['requests'] > 0 else 0,
+ 'average_compression_ratio': round(stats['average_compression_ratio'], 2)
+ }
+ for content_type, stats in self.content_type_stats.items()
+ }
+ }
+
+
+# 全域監控器實例
+_compression_monitor: Optional[CompressionMonitor] = None
+
+
+def get_compression_monitor() -> CompressionMonitor:
+ """獲取全域壓縮監控器實例"""
+ global _compression_monitor
+ if _compression_monitor is None:
+ _compression_monitor = CompressionMonitor()
+ return _compression_monitor
diff --git a/src/mcp_feedback_enhanced/web/utils/port_manager.py b/src/mcp_feedback_enhanced/web/utils/port_manager.py
new file mode 100644
index 0000000..635909d
--- /dev/null
+++ b/src/mcp_feedback_enhanced/web/utils/port_manager.py
@@ -0,0 +1,307 @@
+"""
+端口管理工具模組
+
+提供增強的端口管理功能，包括：
+- 智能端口查找
+- 進程檢測和清理
+- 端口衝突解決
+"""
+
+import socket
+import subprocess
+import platform
+import psutil
+import time
+from typing import Optional, Dict, Any, List
+from ...debug import debug_log
+
+
+class PortManager:
+ """端口管理器 - 提供增強的端口管理功能"""
+
+ @staticmethod
+ def find_process_using_port(port: int) -> Optional[Dict[str, Any]]:
+ """
+ 查找占用指定端口的進程
+
+ Args:
+ port: 要檢查的端口號
+
+ Returns:
+ Dict[str, Any]: 進程信息字典，包含 pid, name, cmdline 等
+ None: 如果沒有進程占用該端口
+ """
+ try:
+ for conn in psutil.net_connections(kind='inet'):
+ if conn.laddr.port == port and conn.status == psutil.CONN_LISTEN:
+ try:
+ process = psutil.Process(conn.pid)
+ return {
+ 'pid': conn.pid,
+ 'name': process.name(),
+ 'cmdline': ' '.join(process.cmdline()),
+ 'create_time': process.create_time(),
+ 'status': process.status()
+ }
+ except (psutil.NoSuchProcess, psutil.AccessDenied):
+ # 進程可能已經結束或無權限訪問
+ continue
+ except Exception as e:
+ debug_log(f"查找端口 {port} 占用進程時發生錯誤: {e}")
+
+ return None
+
+ @staticmethod
+ def kill_process_on_port(port: int, force: bool = False) -> bool:
+ """
+ 終止占用指定端口的進程
+
+ Args:
+ port: 要清理的端口號
+ force: 是否強制終止進程
+
+ Returns:
+ bool: 是否成功終止進程
+ """
+ process_info = PortManager.find_process_using_port(port)
+ if not process_info:
+ debug_log(f"端口 {port} 沒有被任何進程占用")
+ return True
+
+ try:
+ pid = process_info['pid']
+ process = psutil.Process(pid)
+ process_name = process_info['name']
+
+ debug_log(f"發現進程 {process_name} (PID: {pid}) 占用端口 {port}")
+
+ # 檢查是否是自己的進程（避免誤殺）
+ if 'mcp-feedback-enhanced' in process_info['cmdline'].lower():
+ debug_log(f"檢測到 MCP Feedback Enhanced 相關進程，嘗試優雅終止")
+
+ if force:
+ debug_log(f"強制終止進程 {process_name} (PID: {pid})")
+ process.kill()
+ else:
+ debug_log(f"優雅終止進程 {process_name} (PID: {pid})")
+ process.terminate()
+
+ # 等待進程結束
+ try:
+ process.wait(timeout=5)
+ debug_log(f"成功終止進程 {process_name} (PID: {pid})")
+ return True
+ except psutil.TimeoutExpired:
+ if not force:
+ debug_log(f"優雅終止超時，強制終止進程 {process_name} (PID: {pid})")
+ process.kill()
+ process.wait(timeout=3)
+ return True
+ else:
+ debug_log(f"強制終止進程 {process_name} (PID: {pid}) 失敗")
+ return False
+
+ except (psutil.NoSuchProcess, psutil.AccessDenied) as e:
+ debug_log(f"無法終止進程 (PID: {process_info['pid']}): {e}")
+ return False
+ except Exception as e:
+ debug_log(f"終止端口 {port} 占用進程時發生錯誤: {e}")
+ return False
+
+ @staticmethod
+ def is_port_available(host: str, port: int) -> bool:
+ """
+ 檢查端口是否可用
+
+ Args:
+ host: 主機地址
+ port: 端口號
+
+ Returns:
+ bool: 端口是否可用
+ """
+ try:
+ # 首先嘗試不使用 SO_REUSEADDR 來檢測端口
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+ sock.bind((host, port))
+ return True
+ except OSError:
+ # 如果綁定失敗，再檢查是否真的有進程在監聽
+ # 使用 psutil 檢查是否有進程在監聽該端口
+ try:
+ import psutil
+ for conn in psutil.net_connections(kind='inet'):
+ if (conn.laddr.port == port and
+ conn.laddr.ip in [host, '0.0.0.0', '::'] and
+ conn.status == psutil.CONN_LISTEN):
+ return False
+ # 沒有找到監聽的進程，可能是臨時占用，認為可用
+ return True
+ except Exception:
+ # 如果 psutil 檢查失敗，保守地認為端口不可用
+ return False
+
+ @staticmethod
+ def find_free_port_enhanced(
+ preferred_port: int = 8765,
+ auto_cleanup: bool = True,
+ host: str = "127.0.0.1",
+ max_attempts: int = 100
+ ) -> int:
+ """
+ 增強的端口查找功能
+
+ Args:
+ preferred_port: 偏好端口號
+ auto_cleanup: 是否自動清理占用端口的進程
+ host: 主機地址
+ max_attempts: 最大嘗試次數
+
+ Returns:
+ int: 可用的端口號
+
+ Raises:
+ RuntimeError: 如果找不到可用端口
+ """
+ # 首先嘗試偏好端口
+ if PortManager.is_port_available(host, preferred_port):
+ debug_log(f"偏好端口 {preferred_port} 可用")
+ return preferred_port
+
+ # 如果偏好端口被占用且啟用自動清理
+ if auto_cleanup:
+ debug_log(f"偏好端口 {preferred_port} 被占用，嘗試清理占用進程")
+ process_info = PortManager.find_process_using_port(preferred_port)
+
+ if process_info:
+ debug_log(f"端口 {preferred_port} 被進程 {process_info['name']} (PID: {process_info['pid']}) 占用")
+
+ # 詢問用戶是否清理（在實際使用中可能需要配置選項）
+ if PortManager._should_cleanup_process(process_info):
+ if PortManager.kill_process_on_port(preferred_port):
+ # 等待一下讓端口釋放
+ time.sleep(1)
+ if PortManager.is_port_available(host, preferred_port):
+ debug_log(f"成功清理端口 {preferred_port}，現在可用")
+ return preferred_port
+
+ # 如果偏好端口仍不可用，尋找其他端口
+ debug_log(f"偏好端口 {preferred_port} 不可用，尋找其他可用端口")
+
+ for i in range(max_attempts):
+ port = preferred_port + i + 1
+ if PortManager.is_port_available(host, port):
+ debug_log(f"找到可用端口: {port}")
+ return port
+
+ # 如果向上查找失敗，嘗試向下查找
+ for i in range(1, min(preferred_port - 1024, max_attempts)):
+ port = preferred_port - i
+ if port < 1024: # 避免使用系統保留端口
+ break
+ if PortManager.is_port_available(host, port):
+ debug_log(f"找到可用端口: {port}")
+ return port
+
+ raise RuntimeError(
+ f"無法在 {preferred_port}±{max_attempts} 範圍內找到可用端口。"
+ f"請檢查是否有過多進程占用端口，或手動指定其他端口。"
+ )
+
+ @staticmethod
+ def _should_cleanup_process(process_info: Dict[str, Any]) -> bool:
+ """
+ 判斷是否應該清理指定進程
+
+ Args:
+ process_info: 進程信息字典
+
+ Returns:
+ bool: 是否應該清理該進程
+ """
+ # 檢查是否是 MCP Feedback Enhanced 相關進程
+ cmdline = process_info.get('cmdline', '').lower()
+ process_name = process_info.get('name', '').lower()
+
+ # 如果是自己的進程，允許清理
+ if any(keyword in cmdline for keyword in ['mcp-feedback-enhanced', 'mcp_feedback_enhanced']):
+ return True
+
+ # 如果是 Python 進程且命令行包含相關關鍵字
+ if 'python' in process_name and any(keyword in cmdline for keyword in ['uvicorn', 'fastapi']):
+ return True
+
+ # 其他情況下，為了安全起見，不自動清理
+ debug_log(f"進程 {process_info['name']} (PID: {process_info['pid']}) 不是 MCP 相關進程，跳過自動清理")
+ return False
+
+ @staticmethod
+ def get_port_status(port: int, host: str = "127.0.0.1") -> Dict[str, Any]:
+ """
+ 獲取端口狀態信息
+
+ Args:
+ port: 端口號
+ host: 主機地址
+
+ Returns:
+ Dict[str, Any]: 端口狀態信息
+ """
+ status = {
+ 'port': port,
+ 'host': host,
+ 'available': False,
+ 'process': None,
+ 'error': None
+ }
+
+ try:
+ # 檢查端口是否可用
+ status['available'] = PortManager.is_port_available(host, port)
+
+ # 如果不可用，查找占用進程
+ if not status['available']:
+ status['process'] = PortManager.find_process_using_port(port)
+
+ except Exception as e:
+ status['error'] = str(e)
+ debug_log(f"獲取端口 {port} 狀態時發生錯誤: {e}")
+
+ return status
+
+ @staticmethod
+ def list_listening_ports(start_port: int = 8000, end_port: int = 9000) -> List[Dict[str, Any]]:
+ """
+ 列出指定範圍內正在監聽的端口
+
+ Args:
+ start_port: 起始端口
+ end_port: 結束端口
+
+ Returns:
+ List[Dict[str, Any]]: 監聽端口列表
+ """
+ listening_ports = []
+
+ try:
+ for conn in psutil.net_connections(kind='inet'):
+ if (conn.status == psutil.CONN_LISTEN and
+ start_port <= conn.laddr.port <= end_port):
+
+ try:
+ process = psutil.Process(conn.pid)
+ port_info = {
+ 'port': conn.laddr.port,
+ 'host': conn.laddr.ip,
+ 'pid': conn.pid,
+ 'process_name': process.name(),
+ 'cmdline': ' '.join(process.cmdline())
+ }
+ listening_ports.append(port_info)
+ except (psutil.NoSuchProcess, psutil.AccessDenied):
+ continue
+
+ except Exception as e:
+ debug_log(f"列出監聽端口時發生錯誤: {e}")
+
+ return listening_ports
diff --git a/src/mcp_feedback_enhanced/web/utils/session_cleanup_manager.py b/src/mcp_feedback_enhanced/web/utils/session_cleanup_manager.py
new file mode 100644
index 0000000..5534db3
--- /dev/null
+++ b/src/mcp_feedback_enhanced/web/utils/session_cleanup_manager.py
@@ -0,0 +1,504 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+會話清理管理器
+==============
+
+統一管理 Web 會話的清理策略、統計和性能監控。
+與內存監控系統深度集成，提供智能清理決策。
+"""
+
+import time
+import threading
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Callable, Any
+from dataclasses import dataclass, field
+from enum import Enum
+
+from ...debug import web_debug_log as debug_log
+from ...utils.error_handler import ErrorHandler, ErrorType
+from ..models.feedback_session import CleanupReason, SessionStatus
+
+
+@dataclass
+class CleanupPolicy:
+ """清理策略配置"""
+ max_idle_time: int = 1800 # 最大空閒時間（秒）
+ max_session_age: int = 7200 # 最大會話年齡（秒）
+ max_sessions: int = 10 # 最大會話數量
+ cleanup_interval: int = 300 # 清理間隔（秒）
+ memory_pressure_threshold: float = 0.8 # 內存壓力閾值
+ enable_auto_cleanup: bool = True # 啟用自動清理
+ preserve_active_session: bool = True # 保護活躍會話
+
+
+@dataclass
+class CleanupStats:
+ """清理統計數據"""
+ total_cleanups: int = 0
+ expired_cleanups: int = 0
+ memory_pressure_cleanups: int = 0
+ manual_cleanups: int = 0
+ auto_cleanups: int = 0
+ total_sessions_cleaned: int = 0
+ total_cleanup_time: float = 0.0
+ average_cleanup_time: float = 0.0
+ last_cleanup_time: Optional[datetime] = None
+ cleanup_efficiency: float = 0.0 # 清理效率（清理的會話數/總會話數）
+
+
+class CleanupTrigger(Enum):
+ """清理觸發器類型"""
+ AUTO = "auto" # 自動清理
+ MEMORY_PRESSURE = "memory_pressure" # 內存壓力
+ MANUAL = "manual" # 手動清理
+ EXPIRED = "expired" # 過期清理
+ CAPACITY = "capacity" # 容量限制
+
+
+class SessionCleanupManager:
+ """會話清理管理器"""
+
+ def __init__(self, web_ui_manager, policy: CleanupPolicy = None):
+ """
+ 初始化會話清理管理器
+
+ Args:
+ web_ui_manager: WebUIManager 實例
+ policy: 清理策略配置
+ """
+ self.web_ui_manager = web_ui_manager
+ self.policy = policy or CleanupPolicy()
+ self.stats = CleanupStats()
+
+ # 清理狀態
+ self.is_running = False
+ self.cleanup_thread: Optional[threading.Thread] = None
+ self._stop_event = threading.Event()
+
+ # 回調函數
+ self.cleanup_callbacks: List[Callable] = []
+ self.stats_callbacks: List[Callable] = []
+
+ # 清理歷史記錄
+ self.cleanup_history: List[Dict[str, Any]] = []
+ self.max_history = 100
+
+ debug_log("SessionCleanupManager 初始化完成")
+
+ def start_auto_cleanup(self) -> bool:
+ """啟動自動清理"""
+ if not self.policy.enable_auto_cleanup:
+ debug_log("自動清理已禁用")
+ return False
+
+ if self.is_running:
+ debug_log("自動清理已在運行")
+ return True
+
+ try:
+ self.is_running = True
+ self._stop_event.clear()
+
+ self.cleanup_thread = threading.Thread(
+ target=self._auto_cleanup_loop,
+ name="SessionCleanupManager",
+ daemon=True
+ )
+ self.cleanup_thread.start()
+
+ debug_log(f"自動清理已啟動，間隔 {self.policy.cleanup_interval} 秒")
+ return True
+
+ except Exception as e:
+ self.is_running = 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_auto_cleanup(self) -> bool:
+ """停止自動清理"""
+ if not self.is_running:
+ debug_log("自動清理未在運行")
+ return True
+
+ try:
+ self.is_running = False
+ self._stop_event.set()
+
+ if self.cleanup_thread and self.cleanup_thread.is_alive():
+ self.cleanup_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 _auto_cleanup_loop(self):
+ """自動清理主循環"""
+ debug_log("自動清理循環開始")
+
+ while not self._stop_event.is_set():
+ try:
+ # 執行清理檢查
+ self._perform_auto_cleanup()
+
+ # 等待下次清理
+ if self._stop_event.wait(self.policy.cleanup_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(30):
+ break
+
+ debug_log("自動清理循環結束")
+
+ def _perform_auto_cleanup(self):
+ """執行自動清理"""
+ cleanup_start_time = time.time()
+ cleaned_sessions = 0
+
+ try:
+ # 1. 檢查會話數量限制
+ if len(self.web_ui_manager.sessions) > self.policy.max_sessions:
+ cleaned = self._cleanup_by_capacity()
+ cleaned_sessions += cleaned
+ debug_log(f"容量限制清理了 {cleaned} 個會話")
+
+ # 2. 清理過期會話
+ cleaned = self._cleanup_expired_sessions()
+ cleaned_sessions += cleaned
+
+ # 3. 清理空閒會話
+ cleaned = self._cleanup_idle_sessions()
+ cleaned_sessions += cleaned
+
+ # 4. 更新統計
+ cleanup_duration = time.time() - cleanup_start_time
+ self._update_cleanup_stats(
+ CleanupTrigger.AUTO,
+ cleaned_sessions,
+ cleanup_duration
+ )
+
+ if cleaned_sessions > 0:
+ debug_log(f"自動清理完成，清理了 {cleaned_sessions} 個會話，耗時: {cleanup_duration:.2f}秒")
+
+ 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 trigger_cleanup(self, trigger: CleanupTrigger, force: bool = False) -> int:
+ """觸發清理操作"""
+ cleanup_start_time = time.time()
+ cleaned_sessions = 0
+
+ try:
+ debug_log(f"觸發清理操作，觸發器: {trigger.value}，強制: {force}")
+
+ if trigger == CleanupTrigger.MEMORY_PRESSURE:
+ cleaned_sessions = self.web_ui_manager.cleanup_sessions_by_memory_pressure(force)
+ elif trigger == CleanupTrigger.EXPIRED:
+ cleaned_sessions = self.web_ui_manager.cleanup_expired_sessions()
+ elif trigger == CleanupTrigger.CAPACITY:
+ cleaned_sessions = self._cleanup_by_capacity()
+ elif trigger == CleanupTrigger.MANUAL:
+ # 手動清理：組合多種策略
+ cleaned_sessions += self.web_ui_manager.cleanup_expired_sessions()
+ if force:
+ cleaned_sessions += self.web_ui_manager.cleanup_sessions_by_memory_pressure(force)
+ else:
+ # 自動清理
+ self._perform_auto_cleanup()
+ return 0 # 統計已在 _perform_auto_cleanup 中更新
+
+ # 更新統計
+ cleanup_duration = time.time() - cleanup_start_time
+ self._update_cleanup_stats(trigger, cleaned_sessions, cleanup_duration)
+
+ debug_log(f"清理操作完成，清理了 {cleaned_sessions} 個會話，耗時: {cleanup_duration:.2f}秒")
+ return cleaned_sessions
+
+ except Exception as e:
+ error_id = ErrorHandler.log_error_with_context(
+ e,
+ context={"operation": "觸發清理", "trigger": trigger.value, "force": force},
+ error_type=ErrorType.SYSTEM
+ )
+ debug_log(f"觸發清理操作失敗 [錯誤ID: {error_id}]: {e}")
+ return 0
+
+ def _cleanup_by_capacity(self) -> int:
+ """根據容量限制清理會話"""
+ sessions = self.web_ui_manager.sessions
+ if len(sessions) <= self.policy.max_sessions:
+ return 0
+
+ # 計算需要清理的會話數量
+ excess_count = len(sessions) - self.policy.max_sessions
+
+ # 按優先級排序會話（優先清理舊的、非活躍的會話）
+ session_priorities = []
+ for session_id, session in sessions.items():
+ # 跳過當前活躍會話（如果啟用保護）
+ if (self.policy.preserve_active_session and
+ self.web_ui_manager.current_session and
+ session.session_id == self.web_ui_manager.current_session.session_id):
+ continue
+
+ # 計算優先級分數（分數越高越優先清理）
+ priority_score = 0
+
+ # 狀態優先級
+ if session.status in [SessionStatus.COMPLETED, SessionStatus.ERROR, SessionStatus.TIMEOUT]:
+ priority_score += 100
+ elif session.status == SessionStatus.FEEDBACK_SUBMITTED:
+ priority_score += 50
+
+ # 年齡優先級
+ age = session.get_age()
+ priority_score += age / 60 # 每分鐘加1分
+
+ # 空閒時間優先級
+ idle_time = session.get_idle_time()
+ priority_score += idle_time / 30 # 每30秒加1分
+
+ session_priorities.append((session_id, session, priority_score))
+
+ # 按優先級排序並清理
+ session_priorities.sort(key=lambda x: x[2], reverse=True)
+ cleaned_count = 0
+
+ for i in range(min(excess_count, len(session_priorities))):
+ session_id, session, _ = session_priorities[i]
+ try:
+ session._cleanup_sync_enhanced(CleanupReason.MANUAL)
+ del self.web_ui_manager.sessions[session_id]
+ cleaned_count += 1
+ except Exception as e:
+ debug_log(f"容量清理會話 {session_id} 失敗: {e}")
+
+ return cleaned_count
+
+ def _cleanup_expired_sessions(self) -> int:
+ """清理過期會話"""
+ expired_sessions = []
+ current_time = time.time()
+
+ for session_id, session in self.web_ui_manager.sessions.items():
+ # 檢查是否過期
+ if session.is_expired():
+ expired_sessions.append(session_id)
+ # 檢查是否超過最大年齡
+ elif session.get_age() > self.policy.max_session_age:
+ expired_sessions.append(session_id)
+
+ # 清理過期會話
+ cleaned_count = 0
+ for session_id in expired_sessions:
+ try:
+ session = self.web_ui_manager.sessions.get(session_id)
+ if session:
+ session._cleanup_sync_enhanced(CleanupReason.EXPIRED)
+ del self.web_ui_manager.sessions[session_id]
+ cleaned_count += 1
+
+ # 如果清理的是當前活躍會話，清空當前會話
+ if (self.web_ui_manager.current_session and
+ self.web_ui_manager.current_session.session_id == session_id):
+ self.web_ui_manager.current_session = None
+
+ except Exception as e:
+ debug_log(f"清理過期會話 {session_id} 失敗: {e}")
+
+ return cleaned_count
+
+ def _cleanup_idle_sessions(self) -> int:
+ """清理空閒會話"""
+ idle_sessions = []
+
+ for session_id, session in self.web_ui_manager.sessions.items():
+ # 跳過當前活躍會話（如果啟用保護）
+ if (self.policy.preserve_active_session and
+ self.web_ui_manager.current_session and
+ session.session_id == self.web_ui_manager.current_session.session_id):
+ continue
+
+ # 檢查是否空閒時間過長
+ if session.get_idle_time() > self.policy.max_idle_time:
+ idle_sessions.append(session_id)
+
+ # 清理空閒會話
+ cleaned_count = 0
+ for session_id in idle_sessions:
+ try:
+ session = self.web_ui_manager.sessions.get(session_id)
+ if session:
+ session._cleanup_sync_enhanced(CleanupReason.EXPIRED)
+ del self.web_ui_manager.sessions[session_id]
+ cleaned_count += 1
+
+ except Exception as e:
+ debug_log(f"清理空閒會話 {session_id} 失敗: {e}")
+
+ return cleaned_count
+
+ def _update_cleanup_stats(self, trigger: CleanupTrigger, cleaned_count: int, duration: float):
+ """更新清理統計"""
+ self.stats.total_cleanups += 1
+ self.stats.total_sessions_cleaned += cleaned_count
+ self.stats.total_cleanup_time += duration
+ self.stats.last_cleanup_time = datetime.now()
+
+ # 更新平均清理時間
+ if self.stats.total_cleanups > 0:
+ self.stats.average_cleanup_time = self.stats.total_cleanup_time / self.stats.total_cleanups
+
+ # 更新清理效率
+ total_sessions = len(self.web_ui_manager.sessions) + cleaned_count
+ if total_sessions > 0:
+ self.stats.cleanup_efficiency = cleaned_count / total_sessions
+
+ # 根據觸發器類型更新統計
+ if trigger == CleanupTrigger.AUTO:
+ self.stats.auto_cleanups += 1
+ elif trigger == CleanupTrigger.MEMORY_PRESSURE:
+ self.stats.memory_pressure_cleanups += 1
+ elif trigger == CleanupTrigger.EXPIRED:
+ self.stats.expired_cleanups += 1
+ elif trigger == CleanupTrigger.MANUAL:
+ self.stats.manual_cleanups += 1
+
+ # 記錄清理歷史
+ cleanup_record = {
+ "timestamp": datetime.now().isoformat(),
+ "trigger": trigger.value,
+ "cleaned_count": cleaned_count,
+ "duration": duration,
+ "total_sessions_before": total_sessions,
+ "total_sessions_after": len(self.web_ui_manager.sessions)
+ }
+
+ self.cleanup_history.append(cleanup_record)
+
+ # 限制歷史記錄數量
+ if len(self.cleanup_history) > self.max_history:
+ self.cleanup_history = self.cleanup_history[-self.max_history:]
+
+ # 調用統計回調
+ for callback in self.stats_callbacks:
+ try:
+ callback(self.stats, cleanup_record)
+ except Exception as e:
+ debug_log(f"統計回調執行失敗: {e}")
+
+ def get_cleanup_statistics(self) -> Dict[str, Any]:
+ """獲取清理統計數據"""
+ stats_dict = {
+ "total_cleanups": self.stats.total_cleanups,
+ "expired_cleanups": self.stats.expired_cleanups,
+ "memory_pressure_cleanups": self.stats.memory_pressure_cleanups,
+ "manual_cleanups": self.stats.manual_cleanups,
+ "auto_cleanups": self.stats.auto_cleanups,
+ "total_sessions_cleaned": self.stats.total_sessions_cleaned,
+ "total_cleanup_time": round(self.stats.total_cleanup_time, 2),
+ "average_cleanup_time": round(self.stats.average_cleanup_time, 2),
+ "cleanup_efficiency": round(self.stats.cleanup_efficiency, 3),
+ "last_cleanup_time": self.stats.last_cleanup_time.isoformat() if self.stats.last_cleanup_time else None,
+ "is_auto_cleanup_running": self.is_running,
+ "current_sessions": len(self.web_ui_manager.sessions),
+ "policy": {
+ "max_idle_time": self.policy.max_idle_time,
+ "max_session_age": self.policy.max_session_age,
+ "max_sessions": self.policy.max_sessions,
+ "cleanup_interval": self.policy.cleanup_interval,
+ "enable_auto_cleanup": self.policy.enable_auto_cleanup,
+ "preserve_active_session": self.policy.preserve_active_session
+ }
+ }
+
+ return stats_dict
+
+ def get_cleanup_history(self, limit: int = 20) -> List[Dict[str, Any]]:
+ """獲取清理歷史記錄"""
+ return self.cleanup_history[-limit:] if self.cleanup_history else []
+
+ def add_cleanup_callback(self, callback: Callable):
+ """添加清理回調函數"""
+ if callback not in self.cleanup_callbacks:
+ self.cleanup_callbacks.append(callback)
+ debug_log("添加清理回調函數")
+
+ def add_stats_callback(self, callback: Callable):
+ """添加統計回調函數"""
+ if callback not in self.stats_callbacks:
+ self.stats_callbacks.append(callback)
+ debug_log("添加統計回調函數")
+
+ def update_policy(self, **kwargs):
+ """更新清理策略"""
+ for key, value in kwargs.items():
+ if hasattr(self.policy, key):
+ setattr(self.policy, key, value)
+ debug_log(f"更新清理策略 {key} = {value}")
+ else:
+ debug_log(f"未知的策略參數: {key}")
+
+ def reset_stats(self):
+ """重置統計數據"""
+ self.stats = CleanupStats()
+ self.cleanup_history.clear()
+ debug_log("清理統計數據已重置")
+
+ def force_cleanup_all(self, exclude_current: bool = True) -> int:
+ """強制清理所有會話"""
+ sessions_to_clean = []
+
+ for session_id, session in self.web_ui_manager.sessions.items():
+ # 是否排除當前活躍會話
+ if (exclude_current and
+ self.web_ui_manager.current_session and
+ session.session_id == self.web_ui_manager.current_session.session_id):
+ continue
+ sessions_to_clean.append(session_id)
+
+ # 清理會話
+ cleaned_count = 0
+ for session_id in sessions_to_clean:
+ try:
+ session = self.web_ui_manager.sessions.get(session_id)
+ if session:
+ session._cleanup_sync_enhanced(CleanupReason.MANUAL)
+ del self.web_ui_manager.sessions[session_id]
+ cleaned_count += 1
+ except Exception as e:
+ debug_log(f"強制清理會話 {session_id} 失敗: {e}")
+
+ # 更新統計
+ self._update_cleanup_stats(CleanupTrigger.MANUAL, cleaned_count, 0.0)
+
+ debug_log(f"強制清理完成，清理了 {cleaned_count} 個會話")
+ return cleaned_count
diff --git a/tests/test_error_handler.py b/tests/test_error_handler.py
new file mode 100644
index 0000000..3a973ac
--- /dev/null
+++ b/tests/test_error_handler.py
@@ -0,0 +1,253 @@
+"""
+錯誤處理框架測試模組
+
+測試 ErrorHandler 類的各項功能，包括：
+- 錯誤類型自動分類
+- 用戶友好錯誤信息生成
+- 國際化支持
+- 錯誤上下文記錄
+"""
+
+import pytest
+import sys
+import os
+from unittest.mock import patch, MagicMock
+
+# 添加 src 目錄到 Python 路徑
+sys.path.insert(0, 'src')
+
+from mcp_feedback_enhanced.utils.error_handler import ErrorHandler, ErrorType, ErrorSeverity
+
+
+class TestErrorHandler:
+ """錯誤處理器測試類"""
+
+ def test_classify_error_network(self):
+ """測試網絡錯誤分類"""
+ # 測試 ConnectionError
+ error = ConnectionError("Connection failed")
+ assert ErrorHandler.classify_error(error) == ErrorType.NETWORK
+
+ # 測試包含網絡關鍵字的錯誤（不包含 timeout）
+ error = Exception("socket connection failed")
+ assert ErrorHandler.classify_error(error) == ErrorType.NETWORK
+
+ def test_classify_error_file_io(self):
+ """測試文件 I/O 錯誤分類"""
+ # 測試 FileNotFoundError
+ error = FileNotFoundError("No such file or directory")
+ assert ErrorHandler.classify_error(error) == ErrorType.FILE_IO
+
+ # 測試包含文件關鍵字的錯誤（不包含權限關鍵字）
+ error = Exception("file not found")
+ assert ErrorHandler.classify_error(error) == ErrorType.FILE_IO
+
+ def test_classify_error_timeout(self):
+ """測試超時錯誤分類"""
+ error = TimeoutError("Operation timed out")
+ assert ErrorHandler.classify_error(error) == ErrorType.TIMEOUT
+
+ error = Exception("timeout occurred")
+ assert ErrorHandler.classify_error(error) == ErrorType.TIMEOUT
+
+ def test_classify_error_permission(self):
+ """測試權限錯誤分類"""
+ error = PermissionError("Access denied")
+ assert ErrorHandler.classify_error(error) == ErrorType.PERMISSION
+
+ error = Exception("access denied")
+ assert ErrorHandler.classify_error(error) == ErrorType.PERMISSION
+
+ def test_classify_error_validation(self):
+ """測試驗證錯誤分類"""
+ error = ValueError("Invalid value")
+ assert ErrorHandler.classify_error(error) == ErrorType.VALIDATION
+
+ error = TypeError("Wrong type")
+ assert ErrorHandler.classify_error(error) == ErrorType.VALIDATION
+
+ def test_classify_error_default_system(self):
+ """測試默認系統錯誤分類"""
+ error = Exception("Some completely unknown issue")
+ assert ErrorHandler.classify_error(error) == ErrorType.SYSTEM
+
+ def test_format_user_error_basic(self):
+ """測試基本用戶友好錯誤信息生成"""
+ error = ConnectionError("Connection failed")
+ result = ErrorHandler.format_user_error(error)
+
+ assert "❌" in result
+ assert "網絡連接出現問題" in result or "网络连接出现问题" in result or "Network connection issue" in result
+
+ def test_format_user_error_with_context(self):
+ """測試帶上下文的錯誤信息生成"""
+ error = FileNotFoundError("File not found")
+ context = {
+ "operation": "文件讀取",
+ "file_path": "/path/to/file.txt"
+ }
+
+ result = ErrorHandler.format_user_error(error, context=context)
+
+ assert "❌" in result
+ assert "文件讀取" in result or "文件读取" in result or "文件讀取" in result
+ assert "/path/to/file.txt" in result
+
+ def test_format_user_error_with_technical_details(self):
+ """測試包含技術細節的錯誤信息"""
+ error = ValueError("Invalid input")
+ result = ErrorHandler.format_user_error(error, include_technical=True)
+
+ assert "❌" in result
+ assert "ValueError" in result
+ assert "Invalid input" in result
+
+ def test_get_error_solutions(self):
+ """測試獲取錯誤解決方案"""
+ solutions = ErrorHandler.get_error_solutions(ErrorType.NETWORK)
+
+ assert isinstance(solutions, list)
+ assert len(solutions) > 0
+ # 應該包含網絡相關的解決方案
+ solutions_text = " ".join(solutions).lower()
+ assert any(keyword in solutions_text for keyword in ["網絡", "网络", "network", "連接", "连接", "connection"])
+
+ def test_log_error_with_context(self):
+ """測試帶上下文的錯誤記錄"""
+ error = Exception("Test error")
+ context = {"operation": "測試操作", "user": "test_user"}
+
+ error_id = ErrorHandler.log_error_with_context(error, context=context)
+
+ assert isinstance(error_id, str)
+ assert error_id.startswith("ERR_")
+ assert len(error_id.split("_")) == 3 # ERR_timestamp_id
+
+ def test_create_error_response(self):
+ """測試創建標準化錯誤響應"""
+ error = ConnectionError("Network error")
+ context = {"operation": "網絡請求"}
+
+ response = ErrorHandler.create_error_response(error, context=context)
+
+ assert isinstance(response, dict)
+ assert response["success"] is False
+ assert "error_id" in response
+ assert "error_type" in response
+ assert "message" in response
+ assert response["error_type"] == ErrorType.NETWORK.value
+ assert "solutions" in response
+
+ def test_create_error_response_for_user(self):
+ """測試為用戶界面創建錯誤響應"""
+ error = FileNotFoundError("File not found")
+
+ response = ErrorHandler.create_error_response(error, for_user=True)
+
+ assert response["success"] is False
+ assert "context" not in response # 用戶界面不應包含技術上下文
+ assert "❌" in response["message"] # 應該包含用戶友好的格式
+
+ @patch('mcp_feedback_enhanced.utils.error_handler.ErrorHandler.get_i18n_error_message')
+ def test_language_support(self, mock_get_message):
+ """測試多語言支持"""
+ error = ConnectionError("Network error")
+
+ # 測試繁體中文
+ mock_get_message.return_value = "網絡連接出現問題"
+ result = ErrorHandler.format_user_error(error)
+ assert "網絡連接出現問題" in result
+
+ # 測試簡體中文
+ mock_get_message.return_value = "网络连接出现问题"
+ result = ErrorHandler.format_user_error(error)
+ assert "网络连接出现问题" in result
+
+ # 測試英文
+ mock_get_message.return_value = "Network connection issue"
+ result = ErrorHandler.format_user_error(error)
+ assert "Network connection issue" in result
+
+ def test_error_severity_logging(self):
+ """測試錯誤嚴重程度記錄"""
+ error = Exception("Critical system error")
+
+ # 測試高嚴重程度錯誤
+ error_id = ErrorHandler.log_error_with_context(
+ error,
+ severity=ErrorSeverity.CRITICAL
+ )
+
+ assert isinstance(error_id, str)
+ assert error_id.startswith("ERR_")
+
+ def test_get_current_language_fallback(self):
+ """測試語言獲取回退機制"""
+ # 由於 i18n 系統可能會覆蓋環境變數，我們主要測試函數不會拋出異常
+ language = ErrorHandler.get_current_language()
+ assert isinstance(language, str)
+ assert len(language) > 0
+
+ # 測試語言代碼格式
+ assert language in ["zh-TW", "zh-CN", "en"] or "-" in language
+
+ def test_i18n_integration(self):
+ """測試國際化系統集成"""
+ # 測試當 i18n 系統不可用時的回退
+ error_type = ErrorType.NETWORK
+
+ # 測試獲取錯誤信息
+ message = ErrorHandler.get_i18n_error_message(error_type)
+ assert isinstance(message, str)
+ assert len(message) > 0
+
+ # 測試獲取解決方案
+ solutions = ErrorHandler.get_i18n_error_solutions(error_type)
+ assert isinstance(solutions, list)
+
+ def test_error_context_preservation(self):
+ """測試錯誤上下文保存"""
+ error = Exception("Test error")
+ context = {
+ "operation": "測試操作",
+ "file_path": "/test/path",
+ "user_id": "test_user",
+ "timestamp": "2025-01-05"
+ }
+
+ error_id = ErrorHandler.log_error_with_context(error, context=context)
+
+ # 驗證錯誤 ID 格式
+ assert isinstance(error_id, str)
+ assert error_id.startswith("ERR_")
+
+ # 上下文應該被記錄到調試日誌中（通過 debug_log）
+ # 這裡我們主要驗證函數不會拋出異常
+
+ def test_json_rpc_safety(self):
+ """測試不影響 JSON RPC 通信"""
+ # 錯誤處理應該只記錄到 stderr（通過 debug_log）
+ # 不應該影響 stdout 或 JSON RPC 響應
+
+ error = Exception("Test error for JSON RPC safety")
+ context = {"operation": "JSON RPC 測試"}
+
+ # 這些操作不應該影響 stdout
+ error_id = ErrorHandler.log_error_with_context(error, context=context)
+ user_message = ErrorHandler.format_user_error(error)
+ response = ErrorHandler.create_error_response(error)
+
+ # 驗證返回值類型正確
+ assert isinstance(error_id, str)
+ assert isinstance(user_message, str)
+ assert isinstance(response, dict)
+
+ # 驗證不會拋出異常
+ assert error_id.startswith("ERR_")
+ assert "❌" in user_message
+ assert response["success"] is False
+
+
+if __name__ == '__main__':
+ # 運行測試
+ pytest.main([__file__, '-v'])
diff --git a/tests/test_gzip_compression.py b/tests/test_gzip_compression.py
new file mode 100644
index 0000000..fc1d337
--- /dev/null
+++ b/tests/test_gzip_compression.py
@@ -0,0 +1,346 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Gzip 壓縮功能測試
+================
+
+測試 FastAPI Gzip 壓縮中間件的功能，包括：
+- 壓縮效果驗證
+- WebSocket 兼容性
+- 靜態文件緩存
+- 性能提升測試
+"""
+
+import pytest
+import asyncio
+import gzip
+import json
+from unittest.mock import Mock, patch
+from fastapi.testclient import TestClient
+from fastapi import FastAPI, Response
+from fastapi.middleware.gzip import GZipMiddleware
+
+from src.mcp_feedback_enhanced.web.utils.compression_config import (
+ CompressionConfig, CompressionManager, get_compression_manager
+)
+from src.mcp_feedback_enhanced.web.utils.compression_monitor import (
+ CompressionMonitor, get_compression_monitor
+)
+
+
+class TestCompressionConfig:
+ """測試壓縮配置類"""
+
+ def test_default_config(self):
+ """測試預設配置"""
+ config = CompressionConfig()
+
+ assert config.minimum_size == 1000
+ assert config.compression_level == 6
+ assert config.static_cache_max_age == 3600
+ assert config.api_cache_max_age == 0
+ assert 'text/html' in config.compressible_types
+ assert 'application/json' in config.compressible_types
+ assert '/ws' in config.exclude_paths
+
+ def test_from_env(self):
+ """測試從環境變數創建配置"""
+ with patch.dict('os.environ', {
+ 'MCP_GZIP_MIN_SIZE': '2000',
+ 'MCP_GZIP_LEVEL': '9',
+ 'MCP_STATIC_CACHE_AGE': '7200'
+ }):
+ config = CompressionConfig.from_env()
+
+ assert config.minimum_size == 2000
+ assert config.compression_level == 9
+ assert config.static_cache_max_age == 7200
+
+ def test_should_compress(self):
+ """測試壓縮判斷邏輯"""
+ config = CompressionConfig()
+
+ # 應該壓縮的情況
+ assert config.should_compress('text/html', 2000) == True
+ assert config.should_compress('application/json', 1500) == True
+
+ # 不應該壓縮的情況
+ assert config.should_compress('text/html', 500) == False # 太小
+ assert config.should_compress('image/jpeg', 2000) == False # 不支援的類型
+ assert config.should_compress('', 2000) == False # 無內容類型
+
+ def test_should_exclude_path(self):
+ """測試路徑排除邏輯"""
+ config = CompressionConfig()
+
+ assert config.should_exclude_path('/ws') == True
+ assert config.should_exclude_path('/api/ws') == True
+ assert config.should_exclude_path('/health') == True
+ assert config.should_exclude_path('/static/css/style.css') == False
+ assert config.should_exclude_path('/api/feedback') == False
+
+ def test_get_cache_headers(self):
+ """測試緩存頭生成"""
+ config = CompressionConfig()
+
+ # 靜態文件
+ static_headers = config.get_cache_headers('/static/css/style.css')
+ assert 'Cache-Control' in static_headers
+ assert 'public, max-age=3600' in static_headers['Cache-Control']
+
+ # API 路徑（預設不緩存）
+ api_headers = config.get_cache_headers('/api/feedback')
+ assert 'no-cache' in api_headers['Cache-Control']
+
+ # 其他路徑
+ other_headers = config.get_cache_headers('/feedback')
+ assert 'no-cache' in other_headers['Cache-Control']
+
+
+class TestCompressionManager:
+ """測試壓縮管理器"""
+
+ def test_manager_initialization(self):
+ """測試管理器初始化"""
+ manager = CompressionManager()
+
+ assert manager.config is not None
+ assert manager._stats['requests_total'] == 0
+ assert manager._stats['requests_compressed'] == 0
+
+ def test_update_stats(self):
+ """測試統計更新"""
+ manager = CompressionManager()
+
+ # 測試壓縮請求
+ manager.update_stats(1000, 600, True)
+ stats = manager.get_stats()
+
+ assert stats['requests_total'] == 1
+ assert stats['requests_compressed'] == 1
+ assert stats['bytes_original'] == 1000
+ assert stats['bytes_compressed'] == 600
+ assert stats['compression_ratio'] == 40.0 # (1000-600)/1000 * 100
+
+ # 測試未壓縮請求
+ manager.update_stats(500, 500, False)
+ stats = manager.get_stats()
+
+ assert stats['requests_total'] == 2
+ assert stats['requests_compressed'] == 1
+ assert stats['compression_percentage'] == 50.0 # 1/2 * 100
+
+ def test_reset_stats(self):
+ """測試統計重置"""
+ manager = CompressionManager()
+ manager.update_stats(1000, 600, True)
+
+ manager.reset_stats()
+ stats = manager.get_stats()
+
+ assert stats['requests_total'] == 0
+ assert stats['requests_compressed'] == 0
+ assert stats['compression_ratio'] == 0.0
+
+
+class TestCompressionMonitor:
+ """測試壓縮監控器"""
+
+ def test_monitor_initialization(self):
+ """測試監控器初始化"""
+ monitor = CompressionMonitor()
+
+ assert monitor.max_metrics == 1000
+ assert len(monitor.metrics) == 0
+ assert len(monitor.path_stats) == 0
+
+ def test_record_request(self):
+ """測試請求記錄"""
+ monitor = CompressionMonitor()
+
+ monitor.record_request(
+ path='/static/css/style.css',
+ original_size=2000,
+ compressed_size=1200,
+ response_time=0.05,
+ content_type='text/css',
+ was_compressed=True
+ )
+
+ assert len(monitor.metrics) == 1
+ metric = monitor.metrics[0]
+ assert metric.path == '/static/css/style.css'
+ assert metric.compression_ratio == 40.0 # (2000-1200)/2000 * 100
+
+ # 檢查路徑統計
+ path_stats = monitor.get_path_stats()
+ assert '/static/css/style.css' in path_stats
+ assert path_stats['/static/css/style.css']['requests'] == 1
+ assert path_stats['/static/css/style.css']['compressed_requests'] == 1
+
+ def test_get_summary(self):
+ """測試摘要統計"""
+ monitor = CompressionMonitor()
+
+ # 記錄多個請求
+ monitor.record_request('/static/css/style.css', 2000, 1200, 0.05, 'text/css', True)
+ monitor.record_request('/static/js/app.js', 3000, 1800, 0.08, 'application/javascript', True)
+ monitor.record_request('/api/feedback', 500, 500, 0.02, 'application/json', False)
+
+ summary = monitor.get_summary()
+
+ assert summary.total_requests == 3
+ assert summary.compressed_requests == 2
+ assert abs(summary.compression_percentage - 66.67) < 0.01 # 2/3 * 100 (約)
+ assert summary.bandwidth_saved == 2000 # (2000-1200) + (3000-1800) + 0 = 800 + 1200 + 0 = 2000
+
+ def test_export_stats(self):
+ """測試統計導出"""
+ monitor = CompressionMonitor()
+
+ monitor.record_request('/static/css/style.css', 2000, 1200, 0.05, 'text/css', True)
+
+ exported = monitor.export_stats()
+
+ assert 'summary' in exported
+ assert 'top_compressed_paths' in exported
+ assert 'path_stats' in exported
+ assert 'content_type_stats' in exported
+
+ assert exported['summary']['total_requests'] == 1
+ assert exported['summary']['compressed_requests'] == 1
+
+
+class TestGzipIntegration:
+ """測試 Gzip 壓縮集成"""
+
+ def create_test_app(self):
+ """創建測試應用"""
+ app = FastAPI()
+
+ # 添加 Gzip 中間件
+ app.add_middleware(GZipMiddleware, minimum_size=100)
+
+ @app.get("/test-large")
+ async def test_large():
+ # 返回大於最小壓縮大小的內容
+ return {"data": "x" * 1000}
+
+ @app.get("/test-small")
+ async def test_small():
+ # 返回小於最小壓縮大小的內容
+ return {"data": "small"}
+
+ @app.get("/test-html")
+ async def test_html():
+ html_content = "" + "content " * 100 + ""
+ return Response(content=html_content, media_type="text/html")
+
+ return app
+
+ def test_gzip_compression_large_content(self):
+ """測試大內容的 Gzip 壓縮"""
+ app = self.create_test_app()
+ client = TestClient(app)
+
+ # 請求壓縮
+ response = client.get("/test-large", headers={"Accept-Encoding": "gzip"})
+
+ assert response.status_code == 200
+ assert response.headers.get("content-encoding") == "gzip"
+
+ # 驗證內容正確性
+ data = response.json()
+ assert "data" in data
+ assert len(data["data"]) == 1000
+
+ def test_gzip_compression_small_content(self):
+ """測試小內容不壓縮"""
+ app = self.create_test_app()
+ client = TestClient(app)
+
+ response = client.get("/test-small", headers={"Accept-Encoding": "gzip"})
+
+ assert response.status_code == 200
+ # 小內容不應該被壓縮
+ assert response.headers.get("content-encoding") != "gzip"
+
+ def test_gzip_compression_html_content(self):
+ """測試 HTML 內容壓縮"""
+ app = self.create_test_app()
+ client = TestClient(app)
+
+ response = client.get("/test-html", headers={"Accept-Encoding": "gzip"})
+
+ assert response.status_code == 200
+ assert response.headers.get("content-encoding") == "gzip"
+ assert response.headers.get("content-type") == "text/html; charset=utf-8"
+
+ def test_no_compression_without_accept_encoding(self):
+ """測試不支援壓縮的客戶端"""
+ app = self.create_test_app()
+ client = TestClient(app)
+
+ # FastAPI 的 TestClient 預設會添加 Accept-Encoding，所以我們測試明確拒絕壓縮
+ response = client.get("/test-large", headers={"Accept-Encoding": "identity"})
+
+ assert response.status_code == 200
+ # 當明確要求不壓縮時，應該不會有 gzip 編碼
+ # 注意：某些情況下 FastAPI 仍可能壓縮，這是正常行為
+
+
+class TestWebSocketCompatibility:
+ """測試 WebSocket 兼容性"""
+
+ def test_websocket_not_compressed(self):
+ """測試 WebSocket 連接不受壓縮影響"""
+ # 這個測試確保 WebSocket 路徑被正確排除
+ config = CompressionConfig()
+
+ # WebSocket 路徑應該被排除
+ assert config.should_exclude_path('/ws') == True
+ assert config.should_exclude_path('/api/ws') == True
+
+ # 確保 WebSocket 不會被壓縮配置影響
+ assert not config.should_compress('application/json', 1000) or config.should_exclude_path('/ws')
+
+
+@pytest.mark.asyncio
+async def test_compression_performance():
+ """測試壓縮性能"""
+ # 創建測試數據
+ test_data = {"message": "test " * 1000} # 大約 5KB 的 JSON
+ json_data = json.dumps(test_data)
+
+ # 手動壓縮測試
+ compressed_data = gzip.compress(json_data.encode('utf-8'))
+
+ # 驗證壓縮效果
+ original_size = len(json_data.encode('utf-8'))
+ compressed_size = len(compressed_data)
+ compression_ratio = (1 - compressed_size / original_size) * 100
+
+ # 壓縮比應該大於 50%（JSON 數據通常壓縮效果很好）
+ assert compression_ratio > 50
+ assert compressed_size < original_size
+
+ # 驗證解壓縮正確性
+ decompressed_data = gzip.decompress(compressed_data).decode('utf-8')
+ assert decompressed_data == json_data
+
+
+def test_global_instances():
+ """測試全域實例"""
+ # 測試壓縮管理器全域實例
+ manager1 = get_compression_manager()
+ manager2 = get_compression_manager()
+ assert manager1 is manager2
+
+ # 測試壓縮監控器全域實例
+ monitor1 = get_compression_monitor()
+ monitor2 = get_compression_monitor()
+ assert monitor1 is monitor2
+
+
+if __name__ == "__main__":
+ pytest.main([__file__, "-v"])
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"])
diff --git a/tests/test_port_manager.py b/tests/test_port_manager.py
new file mode 100644
index 0000000..d1ccc61
--- /dev/null
+++ b/tests/test_port_manager.py
@@ -0,0 +1,249 @@
+"""
+端口管理器測試模組
+
+測試 PortManager 類的各項功能，包括：
+- 端口可用性檢測
+- 進程查找和清理
+- 增強端口查找
+"""
+
+import pytest
+import socket
+import time
+import threading
+import subprocess
+import sys
+from unittest.mock import patch, MagicMock
+
+# 添加 src 目錄到 Python 路徑
+sys.path.insert(0, 'src')
+
+from mcp_feedback_enhanced.web.utils.port_manager import PortManager
+
+
+class TestPortManager:
+ """端口管理器測試類"""
+
+ def test_is_port_available_free_port(self):
+ """測試檢測空閒端口"""
+ # 找一個肯定空閒的端口
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(('127.0.0.1', 0))
+ free_port = s.getsockname()[1]
+
+ # 測試該端口是否被檢測為可用
+ assert PortManager.is_port_available('127.0.0.1', free_port) is True
+
+ def test_is_port_available_occupied_port(self):
+ """測試檢測被占用的端口"""
+ # 創建一個占用端口的 socket
+ server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ server_socket.bind(('127.0.0.1', 0))
+ occupied_port = server_socket.getsockname()[1]
+ server_socket.listen(1)
+
+ try:
+ # 測試該端口是否被檢測為不可用
+ assert PortManager.is_port_available('127.0.0.1', occupied_port) is False
+ finally:
+ server_socket.close()
+
+ def test_find_free_port_enhanced_preferred_available(self):
+ """測試當偏好端口可用時的行為"""
+ # 找一個空閒端口作為偏好端口
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(('127.0.0.1', 0))
+ preferred_port = s.getsockname()[1]
+
+ # 測試是否返回偏好端口
+ result_port = PortManager.find_free_port_enhanced(
+ preferred_port=preferred_port,
+ auto_cleanup=False
+ )
+ assert result_port == preferred_port
+
+ def test_find_free_port_enhanced_preferred_occupied(self):
+ """測試當偏好端口被占用時的行為"""
+ # 創建一個占用端口的 socket
+ server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ server_socket.bind(('127.0.0.1', 0))
+ occupied_port = server_socket.getsockname()[1]
+ server_socket.listen(1)
+
+ try:
+ # 測試是否返回其他可用端口
+ result_port = PortManager.find_free_port_enhanced(
+ preferred_port=occupied_port,
+ auto_cleanup=False
+ )
+ assert result_port != occupied_port
+ assert result_port > occupied_port # 應該向上查找
+
+ # 驗證返回的端口確實可用
+ assert PortManager.is_port_available('127.0.0.1', result_port) is True
+ finally:
+ server_socket.close()
+
+ def test_find_process_using_port_no_process(self):
+ """測試查找沒有進程占用的端口"""
+ # 找一個空閒端口
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(('127.0.0.1', 0))
+ free_port = s.getsockname()[1]
+
+ # 測試是否正確返回 None
+ result = PortManager.find_process_using_port(free_port)
+ assert result is None
+
+ def test_find_process_using_port_with_process(self):
+ """測試查找有進程占用的端口"""
+ # 創建一個簡單的測試服務器
+ server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ server_socket.bind(('127.0.0.1', 0))
+ test_port = server_socket.getsockname()[1]
+ server_socket.listen(1)
+
+ try:
+ # 測試是否能找到進程信息
+ result = PortManager.find_process_using_port(test_port)
+
+ if result: # 如果找到了進程（在某些環境下可能找不到）
+ assert isinstance(result, dict)
+ assert 'pid' in result
+ assert 'name' in result
+ assert 'cmdline' in result
+ assert isinstance(result['pid'], int)
+ assert result['pid'] > 0
+ finally:
+ server_socket.close()
+
+ def test_get_port_status_available(self):
+ """測試獲取可用端口的狀態"""
+ # 找一個空閒端口
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(('127.0.0.1', 0))
+ free_port = s.getsockname()[1]
+
+ status = PortManager.get_port_status(free_port)
+
+ assert status['port'] == free_port
+ assert status['host'] == '127.0.0.1'
+ assert status['available'] is True
+ assert status['process'] is None
+ assert status['error'] is None
+
+ def test_get_port_status_occupied(self):
+ """測試獲取被占用端口的狀態"""
+ # 創建一個占用端口的 socket
+ server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ server_socket.bind(('127.0.0.1', 0))
+ occupied_port = server_socket.getsockname()[1]
+ server_socket.listen(1)
+
+ try:
+ status = PortManager.get_port_status(occupied_port)
+
+ assert status['port'] == occupied_port
+ assert status['host'] == '127.0.0.1'
+ assert status['available'] is False
+ # process 可能為 None（取決於系統權限）
+ assert status['error'] is None
+ finally:
+ server_socket.close()
+
+ def test_list_listening_ports(self):
+ """測試列出監聽端口"""
+ # 創建幾個測試服務器
+ servers = []
+ test_ports = []
+
+ try:
+ for i in range(2):
+ server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ server_socket.bind(('127.0.0.1', 0))
+ port = server_socket.getsockname()[1]
+ server_socket.listen(1)
+
+ servers.append(server_socket)
+ test_ports.append(port)
+
+ # 測試列出監聽端口
+ min_port = min(test_ports) - 10
+ max_port = max(test_ports) + 10
+
+ listening_ports = PortManager.list_listening_ports(min_port, max_port)
+
+ # 驗證結果
+ assert isinstance(listening_ports, list)
+
+ # 檢查我們的測試端口是否在列表中
+ found_ports = [p['port'] for p in listening_ports]
+ for test_port in test_ports:
+ if test_port in found_ports:
+ # 找到了我們的端口，驗證信息完整性
+ port_info = next(p for p in listening_ports if p['port'] == test_port)
+ assert 'host' in port_info
+ assert 'pid' in port_info
+ assert 'process_name' in port_info
+ assert 'cmdline' in port_info
+
+ finally:
+ # 清理測試服務器
+ for server in servers:
+ server.close()
+
+ @patch('mcp_feedback_enhanced.web.utils.port_manager.psutil.Process')
+ def test_should_cleanup_process_mcp_process(self, mock_process):
+ """測試是否應該清理 MCP 相關進程"""
+ # 模擬 MCP 相關進程
+ process_info = {
+ 'pid': 1234,
+ 'name': 'python.exe',
+ 'cmdline': 'python -m mcp-feedback-enhanced test --web',
+ 'create_time': time.time(),
+ 'status': 'running'
+ }
+
+ result = PortManager._should_cleanup_process(process_info)
+ assert result is True
+
+ @patch('mcp_feedback_enhanced.web.utils.port_manager.psutil.Process')
+ def test_should_cleanup_process_other_process(self, mock_process):
+ """測試是否應該清理其他進程"""
+ # 模擬其他進程
+ process_info = {
+ 'pid': 5678,
+ 'name': 'chrome.exe',
+ 'cmdline': 'chrome --new-window',
+ 'create_time': time.time(),
+ 'status': 'running'
+ }
+
+ result = PortManager._should_cleanup_process(process_info)
+ assert result is False
+
+ def test_find_free_port_enhanced_max_attempts(self):
+ """測試最大嘗試次數限制"""
+ # 這個測試比較難實現，因為需要占用大量連續端口
+ # 我們只測試參數是否正確傳遞
+ try:
+ result = PortManager.find_free_port_enhanced(
+ preferred_port=65000, # 使用高端口減少衝突
+ auto_cleanup=False,
+ max_attempts=10
+ )
+ assert isinstance(result, int)
+ assert 65000 <= result <= 65535
+ except RuntimeError:
+ # 如果真的找不到端口，這也是正常的
+ pass
+
+
+if __name__ == '__main__':
+ # 運行測試
+ pytest.main([__file__, '-v'])
diff --git a/tests/test_resource_manager.py b/tests/test_resource_manager.py
new file mode 100644
index 0000000..bbe10ab
--- /dev/null
+++ b/tests/test_resource_manager.py
@@ -0,0 +1,394 @@
+"""
+資源管理器測試模組
+
+測試 ResourceManager 類的各項功能，包括：
+- 臨時文件和目錄管理
+- 進程註冊和清理
+- 自動清理機制
+- 資源統計和監控
+"""
+
+import pytest
+import os
+import sys
+import time
+import tempfile
+import subprocess
+import threading
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+# 添加 src 目錄到 Python 路徑
+sys.path.insert(0, 'src')
+
+from mcp_feedback_enhanced.utils.resource_manager import (
+ ResourceManager,
+ get_resource_manager,
+ create_temp_file,
+ create_temp_dir,
+ register_process,
+ cleanup_all_resources
+)
+
+
+class TestResourceManager:
+ """資源管理器測試類"""
+
+ def setup_method(self):
+ """每個測試方法前的設置"""
+ # 重置單例實例
+ ResourceManager._instance = None
+
+ def test_singleton_pattern(self):
+ """測試單例模式"""
+ rm1 = ResourceManager()
+ rm2 = ResourceManager()
+ rm3 = get_resource_manager()
+
+ assert rm1 is rm2
+ assert rm2 is rm3
+ assert id(rm1) == id(rm2) == id(rm3)
+
+ def test_create_temp_file(self):
+ """測試創建臨時文件"""
+ rm = get_resource_manager()
+
+ # 測試基本創建
+ temp_file = rm.create_temp_file(suffix=".txt", prefix="test_")
+
+ assert isinstance(temp_file, str)
+ assert os.path.exists(temp_file)
+ assert temp_file.endswith(".txt")
+ assert "test_" in os.path.basename(temp_file)
+ assert temp_file in rm.temp_files
+
+ # 清理
+ os.remove(temp_file)
+
+ def test_create_temp_dir(self):
+ """測試創建臨時目錄"""
+ rm = get_resource_manager()
+
+ # 測試基本創建
+ temp_dir = rm.create_temp_dir(suffix="_test", prefix="test_")
+
+ assert isinstance(temp_dir, str)
+ assert os.path.exists(temp_dir)
+ assert os.path.isdir(temp_dir)
+ assert temp_dir.endswith("_test")
+ assert "test_" in os.path.basename(temp_dir)
+ assert temp_dir in rm.temp_dirs
+
+ # 清理
+ os.rmdir(temp_dir)
+
+ def test_convenience_functions(self):
+ """測試便捷函數"""
+ # 測試 create_temp_file 便捷函數
+ temp_file = create_temp_file(suffix=".log", prefix="conv_")
+ assert isinstance(temp_file, str)
+ assert os.path.exists(temp_file)
+ assert temp_file.endswith(".log")
+
+ # 測試 create_temp_dir 便捷函數
+ temp_dir = create_temp_dir(suffix="_conv", prefix="conv_")
+ assert isinstance(temp_dir, str)
+ assert os.path.exists(temp_dir)
+ assert os.path.isdir(temp_dir)
+
+ # 清理
+ os.remove(temp_file)
+ os.rmdir(temp_dir)
+
+ def test_register_process_with_popen(self):
+ """測試註冊 Popen 進程"""
+ rm = get_resource_manager()
+
+ # 創建一個簡單的進程
+ process = subprocess.Popen(
+ ["python", "-c", "import time; time.sleep(0.1)"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE
+ )
+
+ # 註冊進程
+ pid = rm.register_process(process, description="測試進程")
+
+ assert pid == process.pid
+ assert pid in rm.processes
+ assert rm.processes[pid]["description"] == "測試進程"
+ assert rm.processes[pid]["process"] is process
+
+ # 等待進程結束
+ process.wait()
+
+ def test_register_process_with_pid(self):
+ """測試註冊 PID"""
+ rm = get_resource_manager()
+
+ # 使用當前進程的 PID
+ current_pid = os.getpid()
+
+ # 註冊 PID
+ registered_pid = rm.register_process(current_pid, description="當前進程")
+
+ assert registered_pid == current_pid
+ assert current_pid in rm.processes
+ assert rm.processes[current_pid]["description"] == "當前進程"
+ assert rm.processes[current_pid]["process"] is None
+
+ def test_unregister_temp_file(self):
+ """測試取消臨時文件追蹤"""
+ rm = get_resource_manager()
+
+ # 創建臨時文件
+ temp_file = rm.create_temp_file()
+ assert temp_file in rm.temp_files
+
+ # 取消追蹤
+ result = rm.unregister_temp_file(temp_file)
+ assert result is True
+ assert temp_file not in rm.temp_files
+
+ # 再次取消追蹤（應該返回 False）
+ result = rm.unregister_temp_file(temp_file)
+ assert result is False
+
+ # 清理
+ if os.path.exists(temp_file):
+ os.remove(temp_file)
+
+ def test_unregister_process(self):
+ """測試取消進程追蹤"""
+ rm = get_resource_manager()
+
+ # 註冊進程
+ current_pid = os.getpid()
+ rm.register_process(current_pid, description="測試進程")
+ assert current_pid in rm.processes
+
+ # 取消追蹤
+ result = rm.unregister_process(current_pid)
+ assert result is True
+ assert current_pid not in rm.processes
+
+ # 再次取消追蹤（應該返回 False）
+ result = rm.unregister_process(current_pid)
+ assert result is False
+
+ def test_cleanup_temp_files(self):
+ """測試清理臨時文件"""
+ rm = get_resource_manager()
+
+ # 創建多個臨時文件
+ temp_files = []
+ for i in range(3):
+ temp_file = rm.create_temp_file(prefix=f"cleanup_test_{i}_")
+ temp_files.append(temp_file)
+
+ # 確認文件都存在
+ for temp_file in temp_files:
+ assert os.path.exists(temp_file)
+ assert temp_file in rm.temp_files
+
+ # 執行清理（max_age=0 清理所有文件）
+ cleaned_count = rm.cleanup_temp_files(max_age=0)
+
+ assert cleaned_count == 3
+ for temp_file in temp_files:
+ assert not os.path.exists(temp_file)
+ assert temp_file not in rm.temp_files
+
+ def test_cleanup_temp_dirs(self):
+ """測試清理臨時目錄"""
+ rm = get_resource_manager()
+
+ # 創建多個臨時目錄
+ temp_dirs = []
+ for i in range(2):
+ temp_dir = rm.create_temp_dir(prefix=f"cleanup_test_{i}_")
+ temp_dirs.append(temp_dir)
+
+ # 確認目錄都存在
+ for temp_dir in temp_dirs:
+ assert os.path.exists(temp_dir)
+ assert temp_dir in rm.temp_dirs
+
+ # 執行清理
+ cleaned_count = rm.cleanup_temp_dirs()
+
+ assert cleaned_count == 2
+ for temp_dir in temp_dirs:
+ assert not os.path.exists(temp_dir)
+ assert temp_dir not in rm.temp_dirs
+
+ def test_cleanup_all(self):
+ """測試全面清理"""
+ rm = get_resource_manager()
+
+ # 創建各種資源
+ temp_file = rm.create_temp_file(prefix="cleanup_all_")
+ temp_dir = rm.create_temp_dir(prefix="cleanup_all_")
+
+ # 註冊進程
+ current_pid = os.getpid()
+ rm.register_process(current_pid, description="測試進程", auto_cleanup=False)
+
+ # 執行全面清理
+ results = rm.cleanup_all()
+
+ assert isinstance(results, dict)
+ assert "temp_files" in results
+ assert "temp_dirs" in results
+ assert "processes" in results
+ assert "file_handles" in results
+
+ # 檢查文件和目錄是否被清理
+ assert not os.path.exists(temp_file)
+ assert not os.path.exists(temp_dir)
+ assert temp_file not in rm.temp_files
+ assert temp_dir not in rm.temp_dirs
+
+ # 進程不應該被清理（auto_cleanup=False）
+ assert current_pid in rm.processes
+
+ def test_get_resource_stats(self):
+ """測試獲取資源統計"""
+ rm = get_resource_manager()
+
+ # 創建一些資源
+ temp_file = rm.create_temp_file()
+ temp_dir = rm.create_temp_dir()
+ rm.register_process(os.getpid(), description="統計測試")
+
+ # 獲取統計
+ stats = rm.get_resource_stats()
+
+ assert isinstance(stats, dict)
+ assert "current_temp_files" in stats
+ assert "current_temp_dirs" in stats
+ assert "current_processes" in stats
+ assert "temp_files_created" in stats
+ assert "temp_dirs_created" in stats
+ assert "auto_cleanup_enabled" in stats
+
+ assert stats["current_temp_files"] >= 1
+ assert stats["current_temp_dirs"] >= 1
+ assert stats["current_processes"] >= 1
+
+ # 清理
+ os.remove(temp_file)
+ os.rmdir(temp_dir)
+
+ def test_get_detailed_info(self):
+ """測試獲取詳細信息"""
+ rm = get_resource_manager()
+
+ # 創建一些資源
+ temp_file = rm.create_temp_file(prefix="detail_test_")
+ rm.register_process(os.getpid(), description="詳細信息測試")
+
+ # 獲取詳細信息
+ info = rm.get_detailed_info()
+
+ assert isinstance(info, dict)
+ assert "temp_files" in info
+ assert "temp_dirs" in info
+ assert "processes" in info
+ assert "stats" in info
+
+ assert temp_file in info["temp_files"]
+ assert os.getpid() in info["processes"]
+ assert info["processes"][os.getpid()]["description"] == "詳細信息測試"
+
+ # 清理
+ os.remove(temp_file)
+
+ def test_configure(self):
+ """測試配置功能"""
+ rm = get_resource_manager()
+
+ # 測試配置更新
+ rm.configure(
+ auto_cleanup_enabled=False,
+ cleanup_interval=120,
+ temp_file_max_age=1800
+ )
+
+ assert rm.auto_cleanup_enabled is False
+ assert rm.cleanup_interval == 120
+ assert rm.temp_file_max_age == 1800
+
+ # 測試最小值限制
+ rm.configure(
+ cleanup_interval=30, # 小於最小值 60
+ temp_file_max_age=100 # 小於最小值 300
+ )
+
+ assert rm.cleanup_interval == 60 # 應該被限制為最小值
+ assert rm.temp_file_max_age == 300 # 應該被限制為最小值
+
+ def test_cleanup_all_convenience_function(self):
+ """測試全面清理便捷函數"""
+ # 創建一些資源
+ temp_file = create_temp_file(prefix="conv_cleanup_")
+ temp_dir = create_temp_dir(prefix="conv_cleanup_")
+
+ # 執行清理
+ results = cleanup_all_resources()
+
+ assert isinstance(results, dict)
+ assert not os.path.exists(temp_file)
+ assert not os.path.exists(temp_dir)
+
+ def test_error_handling(self):
+ """測試錯誤處理"""
+ rm = get_resource_manager()
+
+ # 測試創建臨時文件時的錯誤處理
+ with patch('tempfile.mkstemp', side_effect=OSError("Mock error")):
+ with pytest.raises(OSError):
+ rm.create_temp_file()
+
+ # 測試創建臨時目錄時的錯誤處理
+ with patch('tempfile.mkdtemp', side_effect=OSError("Mock error")):
+ with pytest.raises(OSError):
+ rm.create_temp_dir()
+
+ def test_file_handle_registration(self):
+ """測試文件句柄註冊"""
+ rm = get_resource_manager()
+
+ # 創建一個文件句柄
+ temp_file = rm.create_temp_file()
+ with open(temp_file, 'w') as f:
+ f.write("test")
+ rm.register_file_handle(f)
+
+ # 檢查是否註冊成功
+ assert len(rm.file_handles) > 0
+
+ # 清理
+ os.remove(temp_file)
+
+ def test_auto_cleanup_thread(self):
+ """測試自動清理線程"""
+ rm = get_resource_manager()
+
+ # 確保自動清理已啟動
+ assert rm.auto_cleanup_enabled is True
+ assert rm._cleanup_thread is not None
+ assert rm._cleanup_thread.is_alive()
+
+ # 測試停止自動清理
+ rm.stop_auto_cleanup()
+ assert rm._cleanup_thread is None
+
+ # 重新啟動
+ rm.configure(auto_cleanup_enabled=True)
+ assert rm._cleanup_thread is not None
+
+
+if __name__ == '__main__':
+ # 運行測試
+ pytest.main([__file__, '-v'])
diff --git a/tests/test_session_cleanup.py b/tests/test_session_cleanup.py
new file mode 100644
index 0000000..96900ca
--- /dev/null
+++ b/tests/test_session_cleanup.py
@@ -0,0 +1,366 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+會話清理優化測試
+================
+
+測試 WebFeedbackSession 和 SessionCleanupManager 的清理功能。
+"""
+
+import asyncio
+import pytest
+import time
+import threading
+from unittest.mock import Mock, patch, MagicMock
+from datetime import datetime, timedelta
+
+import sys
+import os
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from src.mcp_feedback_enhanced.web.models.feedback_session import (
+ WebFeedbackSession, SessionStatus, CleanupReason
+)
+from src.mcp_feedback_enhanced.web.utils.session_cleanup_manager import (
+ SessionCleanupManager, CleanupPolicy, CleanupTrigger
+)
+
+
+class TestWebFeedbackSessionCleanup:
+ """測試 WebFeedbackSession 清理功能"""
+
+ def setup_method(self):
+ """測試前設置"""
+ self.session_id = "test_session_001"
+ self.project_dir = "/tmp/test_project"
+ self.summary = "測試會話摘要"
+
+ # 創建測試會話
+ self.session = WebFeedbackSession(
+ self.session_id,
+ self.project_dir,
+ self.summary,
+ auto_cleanup_delay=60, # 1分鐘自動清理
+ max_idle_time=30 # 30秒最大空閒時間
+ )
+
+ def teardown_method(self):
+ """測試後清理"""
+ if hasattr(self, 'session') and self.session:
+ try:
+ self.session._cleanup_sync_enhanced(CleanupReason.MANUAL)
+ except:
+ pass
+
+ def test_session_initialization(self):
+ """測試會話初始化"""
+ assert self.session.session_id == self.session_id
+ assert self.session.project_directory == self.project_dir
+ assert self.session.summary == self.summary
+ assert self.session.status == SessionStatus.WAITING
+ assert self.session.auto_cleanup_delay == 60
+ assert self.session.max_idle_time == 30
+ assert self.session.cleanup_timer is not None
+ assert len(self.session.cleanup_stats) > 0
+
+ def test_is_expired_by_idle_time(self):
+ """測試空閒時間過期檢測"""
+ # 新創建的會話不應該過期
+ assert not self.session.is_expired()
+
+ # 模擬空閒時間過長
+ self.session.last_activity = time.time() - 40 # 40秒前
+ assert self.session.is_expired()
+
+ def test_is_expired_by_status(self):
+ """測試狀態過期檢測"""
+ # 設置為錯誤狀態
+ self.session.status = SessionStatus.ERROR
+ self.session.last_activity = time.time() - 400 # 400秒前
+ assert self.session.is_expired()
+
+ # 設置為已過期狀態
+ self.session.status = SessionStatus.EXPIRED
+ assert self.session.is_expired()
+
+ def test_get_age_and_idle_time(self):
+ """測試年齡和空閒時間計算"""
+ # 測試年齡
+ age = self.session.get_age()
+ assert age >= 0
+ assert age < 1 # 剛創建，應該小於1秒
+
+ # 測試空閒時間
+ idle_time = self.session.get_idle_time()
+ assert idle_time >= 0
+ assert idle_time < 1 # 剛創建，應該小於1秒
+
+ def test_cleanup_timer_scheduling(self):
+ """測試清理定時器調度"""
+ # 檢查定時器是否已設置
+ assert self.session.cleanup_timer is not None
+ assert self.session.cleanup_timer.is_alive()
+
+ # 測試延長定時器
+ old_timer = self.session.cleanup_timer
+ self.session.extend_cleanup_timer(120)
+
+ # 應該創建新的定時器
+ assert self.session.cleanup_timer != old_timer
+ assert self.session.cleanup_timer.is_alive()
+
+ def test_cleanup_callbacks(self):
+ """測試清理回調函數"""
+ callback_called = False
+ callback_session = None
+ callback_reason = None
+
+ def test_callback(session, reason):
+ nonlocal callback_called, callback_session, callback_reason
+ callback_called = True
+ callback_session = session
+ callback_reason = reason
+
+ # 添加回調
+ self.session.add_cleanup_callback(test_callback)
+ assert len(self.session.cleanup_callbacks) == 1
+
+ # 執行清理
+ self.session._cleanup_sync_enhanced(CleanupReason.MANUAL)
+
+ # 檢查回調是否被調用
+ assert callback_called
+ assert callback_session == self.session
+ assert callback_reason == CleanupReason.MANUAL
+
+ # 移除回調
+ self.session.remove_cleanup_callback(test_callback)
+ assert len(self.session.cleanup_callbacks) == 0
+
+ def test_cleanup_stats(self):
+ """測試清理統計"""
+ # 初始統計
+ stats = self.session.get_cleanup_stats()
+ assert stats["cleanup_count"] == 0
+ assert stats["session_id"] == self.session_id
+ assert stats["is_active"] == True
+
+ # 執行清理
+ self.session._cleanup_sync_enhanced(CleanupReason.EXPIRED)
+
+ # 檢查統計更新
+ stats = self.session.get_cleanup_stats()
+ assert stats["cleanup_count"] == 1
+ assert stats["cleanup_reason"] == CleanupReason.EXPIRED.value
+ assert stats["last_cleanup_time"] is not None
+ assert stats["cleanup_duration"] >= 0
+
+ @pytest.mark.asyncio
+ async def test_async_cleanup(self):
+ """測試異步清理"""
+ # 模擬 WebSocket 連接
+ mock_websocket = Mock()
+ mock_websocket.send_json = Mock(return_value=asyncio.Future())
+ mock_websocket.send_json.return_value.set_result(None)
+ mock_websocket.close = Mock(return_value=asyncio.Future())
+ mock_websocket.close.return_value.set_result(None)
+ mock_websocket.client_state.DISCONNECTED = False
+
+ self.session.websocket = mock_websocket
+
+ # 執行異步清理
+ await self.session._cleanup_resources_enhanced(CleanupReason.TIMEOUT)
+
+ # 檢查 WebSocket 是否被正確處理
+ mock_websocket.send_json.assert_called_once()
+
+ # 檢查清理統計
+ stats = self.session.get_cleanup_stats()
+ assert stats["cleanup_count"] == 1
+ assert stats["cleanup_reason"] == CleanupReason.TIMEOUT.value
+
+ def test_status_update_resets_timer(self):
+ """測試狀態更新重置定時器"""
+ old_timer = self.session.cleanup_timer
+
+ # 更新狀態為活躍
+ self.session.update_status(SessionStatus.ACTIVE, "測試活躍狀態")
+
+ # 檢查定時器是否被重置
+ assert self.session.cleanup_timer != old_timer
+ assert self.session.cleanup_timer.is_alive()
+ assert self.session.status == SessionStatus.ACTIVE
+
+
+class TestSessionCleanupManager:
+ """測試 SessionCleanupManager 功能"""
+
+ def setup_method(self):
+ """測試前設置"""
+ # 創建模擬的 WebUIManager
+ self.mock_web_ui_manager = Mock()
+ self.mock_web_ui_manager.sessions = {}
+ self.mock_web_ui_manager.current_session = None
+ self.mock_web_ui_manager.cleanup_expired_sessions = Mock(return_value=0)
+ self.mock_web_ui_manager.cleanup_sessions_by_memory_pressure = Mock(return_value=0)
+
+ # 創建清理策略
+ self.policy = CleanupPolicy(
+ max_idle_time=30,
+ max_session_age=300,
+ max_sessions=5,
+ cleanup_interval=10,
+ enable_auto_cleanup=True
+ )
+
+ # 創建清理管理器
+ self.cleanup_manager = SessionCleanupManager(
+ self.mock_web_ui_manager,
+ self.policy
+ )
+
+ def teardown_method(self):
+ """測試後清理"""
+ if hasattr(self, 'cleanup_manager'):
+ self.cleanup_manager.stop_auto_cleanup()
+
+ def test_cleanup_manager_initialization(self):
+ """測試清理管理器初始化"""
+ assert self.cleanup_manager.web_ui_manager == self.mock_web_ui_manager
+ assert self.cleanup_manager.policy == self.policy
+ assert not self.cleanup_manager.is_running
+ assert self.cleanup_manager.cleanup_thread is None
+ assert len(self.cleanup_manager.cleanup_callbacks) == 0
+ assert len(self.cleanup_manager.cleanup_history) == 0
+
+ def test_auto_cleanup_start_stop(self):
+ """測試自動清理啟動和停止"""
+ # 啟動自動清理
+ result = self.cleanup_manager.start_auto_cleanup()
+ assert result == True
+ assert self.cleanup_manager.is_running == True
+ assert self.cleanup_manager.cleanup_thread is not None
+ assert self.cleanup_manager.cleanup_thread.is_alive()
+
+ # 停止自動清理
+ result = self.cleanup_manager.stop_auto_cleanup()
+ assert result == True
+ assert self.cleanup_manager.is_running == False
+
+ def test_trigger_cleanup_memory_pressure(self):
+ """測試內存壓力清理觸發"""
+ # 設置模擬返回值
+ self.mock_web_ui_manager.cleanup_sessions_by_memory_pressure.return_value = 3
+
+ # 觸發內存壓力清理
+ cleaned = self.cleanup_manager.trigger_cleanup(CleanupTrigger.MEMORY_PRESSURE, force=True)
+
+ # 檢查結果
+ assert cleaned == 3
+ self.mock_web_ui_manager.cleanup_sessions_by_memory_pressure.assert_called_once_with(True)
+
+ # 檢查統計更新
+ stats = self.cleanup_manager.get_cleanup_statistics()
+ assert stats["total_cleanups"] == 1
+ assert stats["memory_pressure_cleanups"] == 1
+ assert stats["total_sessions_cleaned"] == 3
+
+ def test_trigger_cleanup_expired(self):
+ """測試過期清理觸發"""
+ # 設置模擬返回值
+ self.mock_web_ui_manager.cleanup_expired_sessions.return_value = 2
+
+ # 觸發過期清理
+ cleaned = self.cleanup_manager.trigger_cleanup(CleanupTrigger.EXPIRED)
+
+ # 檢查結果
+ assert cleaned == 2
+ self.mock_web_ui_manager.cleanup_expired_sessions.assert_called_once()
+
+ # 檢查統計更新
+ stats = self.cleanup_manager.get_cleanup_statistics()
+ assert stats["total_cleanups"] == 1
+ assert stats["expired_cleanups"] == 1
+ assert stats["total_sessions_cleaned"] == 2
+
+ def test_cleanup_statistics(self):
+ """測試清理統計功能"""
+ # 初始統計
+ stats = self.cleanup_manager.get_cleanup_statistics()
+ assert stats["total_cleanups"] == 0
+ assert stats["total_sessions_cleaned"] == 0
+ assert stats["is_auto_cleanup_running"] == False
+
+ # 執行一些清理操作
+ self.mock_web_ui_manager.cleanup_expired_sessions.return_value = 1
+ self.cleanup_manager.trigger_cleanup(CleanupTrigger.EXPIRED)
+
+ self.mock_web_ui_manager.cleanup_sessions_by_memory_pressure.return_value = 2
+ self.cleanup_manager.trigger_cleanup(CleanupTrigger.MEMORY_PRESSURE)
+
+ # 檢查統計
+ stats = self.cleanup_manager.get_cleanup_statistics()
+ assert stats["total_cleanups"] == 2
+ assert stats["expired_cleanups"] == 1
+ assert stats["memory_pressure_cleanups"] == 1
+ assert stats["total_sessions_cleaned"] == 3
+ assert stats["average_cleanup_time"] >= 0
+
+ def test_cleanup_history(self):
+ """測試清理歷史記錄"""
+ # 初始歷史為空
+ history = self.cleanup_manager.get_cleanup_history()
+ assert len(history) == 0
+
+ # 執行清理操作
+ self.mock_web_ui_manager.cleanup_expired_sessions.return_value = 1
+ self.cleanup_manager.trigger_cleanup(CleanupTrigger.EXPIRED)
+
+ # 檢查歷史記錄
+ history = self.cleanup_manager.get_cleanup_history()
+ assert len(history) == 1
+
+ record = history[0]
+ assert record["trigger"] == CleanupTrigger.EXPIRED.value
+ assert record["cleaned_count"] == 1
+ assert "timestamp" in record
+ assert "duration" in record
+
+ def test_policy_update(self):
+ """測試策略更新"""
+ # 更新策略
+ self.cleanup_manager.update_policy(
+ max_idle_time=60,
+ max_sessions=10,
+ enable_auto_cleanup=False
+ )
+
+ # 檢查策略是否更新
+ assert self.cleanup_manager.policy.max_idle_time == 60
+ assert self.cleanup_manager.policy.max_sessions == 10
+ assert self.cleanup_manager.policy.enable_auto_cleanup == False
+
+ def test_stats_reset(self):
+ """測試統計重置"""
+ # 執行一些操作產生統計
+ self.mock_web_ui_manager.cleanup_expired_sessions.return_value = 1
+ self.cleanup_manager.trigger_cleanup(CleanupTrigger.EXPIRED)
+
+ # 檢查有統計數據
+ stats = self.cleanup_manager.get_cleanup_statistics()
+ assert stats["total_cleanups"] > 0
+
+ # 重置統計
+ self.cleanup_manager.reset_stats()
+
+ # 檢查統計已重置
+ stats = self.cleanup_manager.get_cleanup_statistics()
+ assert stats["total_cleanups"] == 0
+ assert stats["total_sessions_cleaned"] == 0
+
+ history = self.cleanup_manager.get_cleanup_history()
+ assert len(history) == 0
+
+
+if __name__ == "__main__":
+ pytest.main([__file__, "-v"])