""" FailureWatcher Service Tests - Phase 18 失敗自動修復閉環 ======================================================== 測試失敗監聽服務的核心功能 版本: v1.0 建立: 2026-03-31 (台北時區) 建立者: Claude Code (Phase 18.6 E2E 驗證) """ import pytest from src.services.failure_watcher import ( FAILURE_CLASSIFICATIONS, FailureWatcherService, ) class TestFailureClassification: """測試失敗分類功能""" def setup_method(self): self.service = FailureWatcherService() def test_classify_timeout(self): """測試超時錯誤分類""" result = self.service._classify_by_rules("Connection timed out after 30s") assert result == "TIMEOUT" def test_classify_timeout_deadline(self): """測試 deadline exceeded 分類""" result = self.service._classify_by_rules("context deadline exceeded") assert result == "TIMEOUT" def test_classify_k8s_error(self): """測試 K8s 錯誤分類""" result = self.service._classify_by_rules("Failed to get pod status from kubernetes") assert result == "K8S_ERROR" def test_classify_k8s_forbidden(self): """測試 K8s forbidden 分類""" result = self.service._classify_by_rules("Forbidden: pods is forbidden") assert result == "K8S_ERROR" def test_classify_network_error(self): """測試網路錯誤分類""" result = self.service._classify_by_rules("Connection refused: network unreachable") assert result == "NETWORK_ERROR" def test_classify_dns_error(self): """測試 DNS 錯誤分類""" result = self.service._classify_by_rules("Failed to resolve hostname") assert result == "NETWORK_ERROR" def test_classify_permission_denied(self): """測試權限錯誤分類""" result = self.service._classify_by_rules("Permission denied: unauthorized access") assert result == "PERMISSION_DENIED" def test_classify_401_error(self): """測試 401 錯誤分類""" result = self.service._classify_by_rules("HTTP 401: Unauthorized") assert result == "PERMISSION_DENIED" def test_classify_resource_error(self): """測試資源錯誤分類""" result = self.service._classify_by_rules("OOMKilled: memory limit exceeded") assert result == "RESOURCE_ERROR" def test_classify_unknown(self): """測試未知錯誤分類""" result = self.service._classify_by_rules("Something went wrong") assert result == "UNKNOWN" class TestRiskAssessment: """測試風險評估功能""" def setup_method(self): self.service = FailureWatcherService() def test_critical_delete_operation(self): """測試刪除操作為 CRITICAL""" result = self.service._assess_risk_level("DELETE_POD") assert result == "CRITICAL" def test_critical_drop_operation(self): """測試 drop 操作為 CRITICAL""" result = self.service._assess_risk_level("DROP_TABLE") assert result == "CRITICAL" def test_critical_force_operation(self): """測試 force 操作為 CRITICAL""" result = self.service._assess_risk_level("FORCE_RESTART") assert result == "CRITICAL" def test_medium_scale_operation(self): """測試 scale 操作為 MEDIUM""" result = self.service._assess_risk_level("SCALE_DEPLOYMENT") assert result == "MEDIUM" def test_medium_rollback_operation(self): """測試 rollback 操作為 MEDIUM""" result = self.service._assess_risk_level("ROLLBACK_DEPLOYMENT") assert result == "MEDIUM" def test_medium_update_operation(self): """測試 update 操作為 MEDIUM""" result = self.service._assess_risk_level("UPDATE_CONFIG") assert result == "MEDIUM" def test_low_restart_operation(self): """測試 restart 操作為 LOW""" result = self.service._assess_risk_level("RESTART_DEPLOYMENT") assert result == "LOW" def test_low_refresh_operation(self): """測試 refresh 操作為 LOW""" result = self.service._assess_risk_level("REFRESH_CACHE") assert result == "LOW" def test_low_clear_operation(self): """測試 clear 操作為 LOW""" result = self.service._assess_risk_level("CLEAR_CACHE") assert result == "LOW" def test_default_medium(self): """測試預設風險等級為 MEDIUM""" result = self.service._assess_risk_level("UNKNOWN_OPERATION") assert result == "MEDIUM" class TestRepairSuggestion: """測試修復建議功能""" def setup_method(self): self.service = FailureWatcherService() def test_suggest_timeout_repair(self): """測試超時修復建議""" result = self.service._suggest_repair("TIMEOUT") assert "超時" in result or "重試" in result def test_suggest_k8s_repair(self): """測試 K8s 修復建議""" result = self.service._suggest_repair("K8S_ERROR") assert "K8s" in result or "Pod" in result def test_suggest_network_repair(self): """測試網路修復建議""" result = self.service._suggest_repair("NETWORK_ERROR") assert "網路" in result or "DNS" in result def test_suggest_permission_repair(self): """測試權限修復建議""" result = self.service._suggest_repair("PERMISSION_DENIED") assert "權限" in result or "RBAC" in result def test_suggest_resource_repair(self): """測試資源修復建議""" result = self.service._suggest_repair("RESOURCE_ERROR") assert "資源" in result or "配額" in result def test_suggest_unknown_repair(self): """測試未知錯誤修復建議""" result = self.service._suggest_repair("UNKNOWN") assert "人工" in result class TestSeverityMapping: """測試嚴重度映射功能""" def setup_method(self): self.service = FailureWatcherService() def test_map_critical_severity(self): """測試 critical 映射""" assert self.service._map_severity_to_risk("critical") == "CRITICAL" assert self.service._map_severity_to_risk("CRITICAL") == "CRITICAL" assert self.service._map_severity_to_risk("高") == "CRITICAL" def test_map_medium_severity(self): """測試 medium 映射""" assert self.service._map_severity_to_risk("warning") == "MEDIUM" assert self.service._map_severity_to_risk("medium") == "MEDIUM" assert self.service._map_severity_to_risk("中") == "MEDIUM" def test_map_low_severity(self): """測試 low 映射""" assert self.service._map_severity_to_risk("low") == "LOW" assert self.service._map_severity_to_risk("info") == "LOW" class TestRepairActionExtraction: """測試修復動作提取功能""" def setup_method(self): self.service = FailureWatcherService() def test_extract_restart_deployment(self): """測試提取重啟 Deployment 動作""" result = self.service._extract_repair_action("建議重啟 deployment 以恢復服務") assert result == "restart_deployment" def test_extract_restart_pod(self): """測試提取重啟 Pod 動作""" result = self.service._extract_repair_action("需要 restart pod") assert result == "restart_pod" def test_extract_restart_chinese(self): """測試中文重啟動作""" result = self.service._extract_repair_action("重新啟動服務") assert result == "restart_pod" def test_extract_clear_cache(self): """測試提取清理快取動作""" result = self.service._extract_repair_action("清理 cache 後重試") assert result == "clear_cache" def test_extract_scale_up(self): """測試提取擴展動作""" result = self.service._extract_repair_action("需要 scale up replicas") assert result == "scale_up" def test_extract_unknown_action(self): """測試未知動作返回原始建議""" result = self.service._extract_repair_action("需要聯繫 DBA 檢查資料庫") assert "聯繫" in result or "DBA" in result class TestFailureClassificationKeywords: """測試失敗分類關鍵字配置""" def test_all_classifications_have_keywords(self): """測試所有分類都有關鍵字""" for classification, keywords in FAILURE_CLASSIFICATIONS.items(): assert len(keywords) > 0, f"{classification} 沒有關鍵字" def test_timeout_keywords(self): """測試超時關鍵字""" assert "timeout" in FAILURE_CLASSIFICATIONS["TIMEOUT"] assert "timed out" in FAILURE_CLASSIFICATIONS["TIMEOUT"] def test_k8s_keywords(self): """測試 K8s 關鍵字""" assert "kubernetes" in FAILURE_CLASSIFICATIONS["K8S_ERROR"] assert "pod" in FAILURE_CLASSIFICATIONS["K8S_ERROR"] def test_network_keywords(self): """測試網路關鍵字""" assert "connection" in FAILURE_CLASSIFICATIONS["NETWORK_ERROR"] assert "dns" in FAILURE_CLASSIFICATIONS["NETWORK_ERROR"] def test_permission_keywords(self): """測試權限關鍵字""" assert "permission" in FAILURE_CLASSIFICATIONS["PERMISSION_DENIED"] assert "401" in FAILURE_CLASSIFICATIONS["PERMISSION_DENIED"]