""" Context Gatherer Unit Tests ============================ Phase 5.2.1: 日誌清洗模組測試 Gate 2 Checkpoint: 驗證 ERROR Only 過濾邏輯 - 確保餵給 Ollama 的是純淨的戰訊，不含雜訊 """ import pytest from src.services.context_gatherer import LogLevelFilter class TestLogLevelFilter: """LogLevelFilter 單元測試 - ERROR Only 原則驗證""" # ========================================================================= # 測試案例 1: 禁止的日誌等級 (必須過濾) # ========================================================================= @pytest.mark.parametrize("line", [ "[DEBUG] Starting application initialization", "[INFO] Server listening on port 8080", "[TRACE] Request ID: abc123 processing", "[VERBOSE] Memory allocation details", "DEBUG: Connection pool initialized", "INFO: Health check passed", "TRACE: Stack trace dump", 'level=DEBUG msg="Processing request"', 'level="INFO" service=api status=healthy', 'level=info component="scheduler"', ]) def test_forbidden_levels_are_filtered(self, line: str): """禁止等級 (DEBUG/INFO/TRACE/VERBOSE) 必須被過濾""" assert LogLevelFilter.is_allowed(line) is False, f"Should filter: {line}" # ========================================================================= # 測試案例 2: 允許的日誌等級 (必須保留) # ========================================================================= @pytest.mark.parametrize("line", [ "[ERROR] Database connection failed", "[FATAL] Out of memory, shutting down", "[CRITICAL] SSL certificate expired", "[WARN] High CPU usage detected (95%)", "[WARNING] Disk space low on /var/log", "ERROR: Unable to connect to Redis", "FATAL: Unrecoverable state", "CRITICAL: Data corruption detected", "WARN: Response time degraded", "WARNING: Connection pool exhausted", 'level=ERROR msg="Request failed"', 'level="CRITICAL" service=db error="timeout"', 'level=warning component="cache" status=degraded', ]) def test_allowed_levels_are_preserved(self, line: str): """允許等級 (ERROR/FATAL/CRITICAL/WARN/WARNING) 必須保留""" assert LogLevelFilter.is_allowed(line) is True, f"Should preserve: {line}" # ========================================================================= # 測試案例 3: Stacktrace 保留 # ========================================================================= @pytest.mark.parametrize("line", [ "Traceback (most recent call last):", ' File "/app/main.py", line 42, in handle_request', " at com.example.Service.process(Service.java:123)", " at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)", "panic: runtime error: index out of range", " 0: 0x7fff5fbff8c0 main.main+0x20", ]) def test_stacktrace_lines_are_preserved(self, line: str): """Stacktrace 行必須保留 (包括 Python/Java/Go)""" assert LogLevelFilter.is_allowed(line) is True, f"Should preserve stacktrace: {line}" # ========================================================================= # 測試案例 4: K8s 事件格式 # ========================================================================= @pytest.mark.parametrize("line", [ "Warning BackOff 2m30s kubelet Back-off restarting failed container", "Error Failed 5m kubelet Error: ImagePullBackOff", ]) def test_k8s_warning_error_events_preserved(self, line: str): """K8s Warning/Error 事件必須保留""" assert LogLevelFilter.is_allowed(line) is True, f"Should preserve K8s event: {line}" @pytest.mark.parametrize("line", [ "Normal Scheduled 10m default-scheduler Successfully assigned", "Normal Pulled 8m kubelet Container image pulled", ]) def test_k8s_normal_events_filtered(self, line: str): """K8s Normal 事件應該被過濾""" assert LogLevelFilter.is_allowed(line) is False, f"Should filter K8s Normal: {line}" # ========================================================================= # 測試案例 5: 空行與邊界情況 # ========================================================================= @pytest.mark.parametrize("line", [ "", " ", "\t\t", ]) def test_empty_lines_are_filtered(self, line: str): """空行必須被過濾""" assert LogLevelFilter.is_allowed(line) is False # ========================================================================= # 測試案例 6: 完整日誌過濾 (多行) # ========================================================================= def test_filter_logs_multiline(self): """測試多行日誌過濾 - ERROR Only 原則""" raw_logs = """ [INFO] Application started successfully [DEBUG] Loading configuration from /etc/app/config.yaml [INFO] Connected to database [ERROR] Failed to connect to Redis: Connection refused [INFO] Retrying connection... [ERROR] Redis connection failed after 3 retries Traceback (most recent call last): File "/app/redis_client.py", line 45, in connect raise ConnectionError("Unable to connect") [DEBUG] Cleanup initiated [WARN] Memory usage high: 85% [INFO] Health check passed [CRITICAL] Service degraded, entering maintenance mode """.strip() filtered = LogLevelFilter.filter_logs(raw_logs) _lines = [line for line in filtered.split("\n") if line.strip()] # 驗證: 只有 ERROR/WARN/CRITICAL 和 Stacktrace 被保留 assert "[INFO]" not in filtered, "INFO should be filtered" assert "[DEBUG]" not in filtered, "DEBUG should be filtered" assert "[ERROR] Failed to connect to Redis" in filtered assert "[ERROR] Redis connection failed" in filtered assert "Traceback (most recent call last):" in filtered assert "[WARN] Memory usage high" in filtered assert "[CRITICAL] Service degraded" in filtered # 計算過濾效果 stats = LogLevelFilter.get_filter_stats(raw_logs, filtered) assert stats["filtered_lines"] < stats["original_lines"] assert stats["removal_rate_percent"] > 0 def test_filter_stats_calculation(self): """測試過濾統計計算""" original = "[INFO] line1\n[ERROR] line2\n[DEBUG] line3" filtered = "[ERROR] line2" stats = LogLevelFilter.get_filter_stats(original, filtered) assert stats["original_lines"] == 3 assert stats["filtered_lines"] == 1 assert stats["removed_lines"] == 2 assert stats["removal_rate_percent"] == pytest.approx(66.7, rel=0.1) # ========================================================================= # 測試案例 7: 真實 K8s Pod 日誌模擬 # ========================================================================= def test_real_world_k8s_pod_logs(self): """模擬真實 K8s Pod 日誌 - 驗證雜訊過濾效果""" # 模擬 Harbor Core Pod 崩潰日誌 k8s_logs = """ 2024-03-21T10:15:23.456Z INFO [harbor.core] Starting Harbor Core v2.9.0 2024-03-21T10:15:24.789Z DEBUG [harbor.core.db] Initializing database connection pool 2024-03-21T10:15:25.123Z INFO [harbor.core.db] Connected to PostgreSQL 2024-03-21T10:15:26.456Z DEBUG [harbor.core.cache] Redis client initialized 2024-03-21T10:15:27.789Z INFO [harbor.core.api] HTTP server listening on :8080 2024-03-21T10:16:45.123Z ERROR [harbor.core.db] Connection lost to PostgreSQL 2024-03-21T10:16:45.456Z FATAL [harbor.core] Database connection unrecoverable Traceback (most recent call last): File "/harbor/core/db.py", line 234, in connect raise DatabaseConnectionError("Max retries exceeded") 2024-03-21T10:16:46.789Z INFO [harbor.core] Graceful shutdown initiated 2024-03-21T10:16:47.123Z DEBUG [harbor.core] Cleanup completed """.strip() filtered = LogLevelFilter.filter_logs(k8s_logs) stats = LogLevelFilter.get_filter_stats(k8s_logs, filtered) # 驗證: 只保留 ERROR, FATAL 和 Stacktrace assert "ERROR" in filtered assert "FATAL" in filtered assert "Traceback" in filtered assert "INFO" not in filtered.replace("Co", "") # 避免誤判 assert "DEBUG" not in filtered # 驗證: 過濾率應該很高 (約 60-70%) assert stats["removal_rate_percent"] > 50, f"Should filter >50%, got {stats['removal_rate_percent']}%" print("\n📊 K8s Log Filter Stats:") print(f" Original: {stats['original_lines']} lines") print(f" Filtered: {stats['filtered_lines']} lines") print(f" Removed: {stats['removed_lines']} lines ({stats['removal_rate_percent']}%)") print(f"\n✅ 純淨戰訊 (ERROR Only):\n{filtered}") # ============================================================================= # CLI 測試入口 # ============================================================================= if __name__ == "__main__": # 快速驗證測試 print("=" * 60) print("Phase 5.2.1 - Context Gatherer Unit Tests") print("Gate 2 Checkpoint: ERROR Only 過濾邏輯驗證") print("=" * 60) test = TestLogLevelFilter() # 執行關鍵測試 print("\n🔍 測試 1: 禁止等級過濾...") for line in [ "[DEBUG] test", "[INFO] test", "[TRACE] test", "level=DEBUG msg=test", "INFO: application started", ]: result = LogLevelFilter.is_allowed(line) status = "❌ 過濾" if not result else "⚠️ 錯誤保留" print(f" {status}: {line[:50]}") print("\n🔍 測試 2: 允許等級保留...") for line in [ "[ERROR] Database connection failed", "[FATAL] Out of memory", "[CRITICAL] SSL expired", "[WARN] High CPU", "[WARNING] Disk low", ]: result = LogLevelFilter.is_allowed(line) status = "✅ 保留" if result else "⚠️ 錯誤過濾" print(f" {status}: {line[:50]}") print("\n🔍 測試 3: 多行日誌過濾效果...") test.test_real_world_k8s_pod_logs() print("\n" + "=" * 60) print("✅ Gate 2 Checkpoint: ERROR Only 過濾邏輯驗證完成") print("=" * 60)