awoooi/apps/api/tests/test_smart_router.py

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