""" Smart Router Tests - Phase 13.3 (更新: 2026-04-01 ogt) ======================================================= 測試意圖分類、複雜度評分、AI 路由 API 演進說明: - Phase 13.3 原始版: classify_sync() 返回 IntentType - 現在版: classify_sync() 返回 IntentResult (需取 .intent 欄位) - IntentType 正規化: ALERT_TRIAGE→DIAGNOSE, DEPLOYMENT→CONFIG, QUERY→DIAGNOSE - ComplexityScorer: features key 改為 resource_count (而非 service_count) - AIRouter: 預設使用 qwen2.5:7b-instruct (model_selection_strategy 更新) """ from src.services.ai_router import ( AIRouter, get_ai_router, ) from src.services.complexity_scorer import ( ComplexityScorer, get_complexity_scorer, ) from src.services.intent_classifier import ( IntentClassifier, IntentType, get_intent_classifier, ) class TestIntentClassifier: """測試意圖分類器""" def test_alert_keywords(self): """測試告警關鍵字匹配 → canonical: DIAGNOSE""" classifier = IntentClassifier() # 中文告警 → DIAGNOSE (ALERT_TRIAGE 已正規化) assert classifier.classify_sync("高負載警報").intent == IntentType.DIAGNOSE assert classifier.classify_sync("CPU 異常告警").intent == IntentType.DIAGNOSE assert classifier.classify_sync("OOM error detected").intent == IntentType.DIAGNOSE def test_deployment_keywords(self): """測試部署關鍵字匹配 → canonical: CONFIG""" classifier = IntentClassifier() # 部署 → CONFIG (DEPLOYMENT 已正規化) assert classifier.classify_sync("部署新版本").intent == IntentType.CONFIG assert classifier.classify_sync("kubectl apply -f manifest.yaml").intent == IntentType.CONFIG # rollout + deployment → 無關鍵字命中 (resource 偵測但不算意圖) assert classifier.classify_sync("rollout deployment api").intent == IntentType.UNKNOWN def test_query_keywords(self): """測試查詢關鍵字匹配""" classifier = IntentClassifier() # 查詢 Pod 狀態 → DIAGNOSE (match: 狀態) assert classifier.classify_sync("查詢 Pod 狀態").intent == IntentType.DIAGNOSE # kubectl get pods → DIAGNOSE assert classifier.classify_sync("kubectl get pods").intent == IntentType.DIAGNOSE # replicas → SCALE (match: replica) assert classifier.classify_sync("現在有多少 replicas").intent == IntentType.SCALE def test_maintenance_keywords(self): """測試維運關鍵字匹配""" classifier = IntentClassifier() # 重啟 → RESTART assert classifier.classify_sync("重啟服務").intent == IntentType.RESTART # scale → SCALE assert classifier.classify_sync("scale deployment to 5").intent == IntentType.SCALE # 回滾 → ROLLBACK assert classifier.classify_sync("回滾到上一版").intent == IntentType.ROLLBACK def test_code_review_keywords(self): """測試程式碼審查關鍵字匹配 → CODE_REVIEW 已移除，應返回 UNKNOWN""" classifier = IntentClassifier() # CODE_REVIEW 已不在 INTENT_KEYWORDS，預期為 UNKNOWN assert classifier.classify_sync("review this PR").intent == IntentType.UNKNOWN assert classifier.classify_sync("審查這個 commit").intent == IntentType.UNKNOWN def test_unknown_intent(self): """測試未知意圖""" classifier = IntentClassifier() assert classifier.classify_sync("hello world").intent == IntentType.UNKNOWN assert classifier.classify_sync("今天天氣如何").intent == IntentType.UNKNOWN def test_result_has_required_fields(self): """測試 IntentResult 包含所有必要欄位""" classifier = IntentClassifier() result = classifier.classify_sync("查詢 Pod 狀態") assert hasattr(result, "intent") assert hasattr(result, "confidence") assert hasattr(result, "method") assert hasattr(result, "risk_level") assert result.method == "keyword" # 關鍵字匹配信心度必須是 0.0 (非 AI 分析) assert result.confidence == 0.0 class TestComplexityScorer: """測試複雜度評分器""" def test_simple_context(self): """測試簡單上下文""" scorer = ComplexityScorer() result = scorer.score({}) assert result.score == 1 assert result.recommended_model == "llama3.2:3b" def test_multi_service_context(self): """測試多資源上下文 (feature: resource_count)""" scorer = ComplexityScorer() result = scorer.score({ "affected_services": ["api", "worker", "redis"], }) assert result.score >= 2 # 現在使用 resource_count (非 service_count) assert "resource_count" in result.features def test_code_analysis_context(self): """測試程式碼分析上下文""" scorer = ComplexityScorer() # 4個服務應觸發高複雜度 result = scorer.score({ "affected_services": ["api", "worker", "redis", "postgres"], }) assert result.score >= 3 def test_complex_context(self): """測試複雜上下文 (多資源)""" scorer = ComplexityScorer() result = scorer.score({ "affected_services": ["api", "worker", "redis", "postgres", "nginx"], "metrics": ["cpu", "memory", "latency", "error_rate", "rps"], }) assert result.score >= 3 # 高複雜度應使用較強模型 assert result.recommended_model != "llama3.2:3b" def test_score_increases_with_resources(self): """測試分數隨資源數量增加""" scorer = ComplexityScorer() r1 = scorer.score({}) r2 = scorer.score({"affected_services": ["api", "worker", "redis"]}) assert r2.score > r1.score class TestAIRouter: """測試 AI 路由器""" def test_query_routes_to_ollama(self): """測試查詢路由到 Ollama""" router = AIRouter() decision = router.route_sync("查詢 Pod 狀態", {}) # DIAGNOSE 意圖 → Ollama assert decision.intent == IntentType.DIAGNOSE assert decision.model is not None assert len(decision.fallback_models) >= 2 def test_alert_intent_classification(self): """測試告警意圖分類""" router = AIRouter() decision = router.route_sync("高負載告警", {}) # 告警 → DIAGNOSE assert decision.intent == IntentType.DIAGNOSE def test_complex_alert_routes_with_high_score(self): """測試複雜告警具備高複雜度分數""" router = AIRouter() decision = router.route_sync("高負載告警", { "affected_services": ["api", "worker", "redis", "postgres"], "metrics": ["cpu", "memory", "latency", "error_rate"], }) assert decision.intent == IntentType.DIAGNOSE assert decision.complexity.score >= 3 assert decision.model is not None def test_fallback_list(self): """測試 Fallback 列表""" router = AIRouter() decision = router.route_sync("查詢 Pod 狀態", {}) # Fallback 不應包含已選模型 assert decision.model not in decision.fallback_models # 應該有備援 assert len(decision.fallback_models) >= 2 class TestSingletons: """測試單例""" def test_intent_classifier_singleton(self): """測試 IntentClassifier 單例""" c1 = get_intent_classifier() c2 = get_intent_classifier() assert c1 is c2 def test_complexity_scorer_singleton(self): """測試 ComplexityScorer 單例""" s1 = get_complexity_scorer() s2 = get_complexity_scorer() assert s1 is s2 def test_ai_router_singleton(self): """測試 AIRouter 單例""" r1 = get_ai_router() r2 = get_ai_router() assert r1 is r2