688146ef9c
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled
DIAGNOSE local chain 選 Nemotron 後 fallback 只剩 Ollama 一個 >= 2 斷言過嚴，與 test_query_routes_to_ollama 同樣修正 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
222 lines
8.2 KiB
Python
222 lines
8.2 KiB
Python
"""
|
||
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):
|
||
"""測試查詢路由到 Nemotron（Phase 25 P0: DIAGNOSE 升級至 Nemotron）"""
|
||
router = AIRouter()
|
||
|
||
decision = router.route_sync("查詢 Pod 狀態", {})
|
||
# DIAGNOSE 意圖 → Nemotron (P0 升級)
|
||
assert decision.intent == IntentType.DIAGNOSE
|
||
assert decision.model is not None
|
||
# DIAGNOSE 使用 _local_fallback_chain [NEMOTRON, OLLAMA]
|
||
# 選了 NEMOTRON 後，fallback 只剩 OLLAMA 一個 — 此為正確的隱私邊界設計
|
||
assert len(decision.fallback_models) >= 1
|
||
|
||
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
|
||
# 應該有備援 (DIAGNOSE local chain 選 Nemotron 後只剩 Ollama 一個，>=1 正確)
|
||
assert len(decision.fallback_models) >= 1
|
||
|
||
|
||
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
|