289 lines
9.9 KiB
Python
289 lines
9.9 KiB
Python
"""
|
||
Intent Classifier Tests - Phase 13.4 Ollama 整合
|
||
================================================
|
||
2026-03-30 Claude Code: Intent Classifier LLM 整合測試
|
||
|
||
測試範圍:
|
||
- _llm_classify: LLM 分類邏輯
|
||
- _parse_intent_type: 意圖解析
|
||
- _llm_fallback_result: 失敗回退
|
||
- classify: 完整分類流程
|
||
"""
|
||
|
||
import pytest
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
import json
|
||
|
||
from src.services.intent_classifier import (
|
||
IntentClassifier,
|
||
IntentType,
|
||
IntentResult,
|
||
RiskLevel,
|
||
get_intent_classifier,
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# Fixtures
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.fixture
|
||
def classifier():
|
||
"""測試用 IntentClassifier"""
|
||
return IntentClassifier()
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_ollama_response():
|
||
"""模擬 Ollama 回應"""
|
||
return {
|
||
"response": json.dumps({
|
||
"intent": "restart",
|
||
"confidence": 0.85,
|
||
"reasoning": "用戶要求重啟 Pod"
|
||
})
|
||
}
|
||
|
||
|
||
# =============================================================================
|
||
# Test Cases
|
||
# =============================================================================
|
||
|
||
|
||
class TestParseIntentType:
|
||
"""_parse_intent_type 測試"""
|
||
|
||
def test_parse_restart(self, classifier):
|
||
"""測試解析 restart"""
|
||
assert classifier._parse_intent_type("restart") == IntentType.RESTART
|
||
assert classifier._parse_intent_type("RESTART") == IntentType.RESTART
|
||
|
||
def test_parse_scale(self, classifier):
|
||
"""測試解析 scale"""
|
||
assert classifier._parse_intent_type("scale") == IntentType.SCALE
|
||
|
||
def test_parse_config(self, classifier):
|
||
"""測試解析 config"""
|
||
assert classifier._parse_intent_type("config") == IntentType.CONFIG
|
||
|
||
def test_parse_diagnose(self, classifier):
|
||
"""測試解析 diagnose"""
|
||
assert classifier._parse_intent_type("diagnose") == IntentType.DIAGNOSE
|
||
|
||
def test_parse_delete(self, classifier):
|
||
"""測試解析 delete"""
|
||
assert classifier._parse_intent_type("delete") == IntentType.DELETE
|
||
|
||
def test_parse_rollback(self, classifier):
|
||
"""測試解析 rollback"""
|
||
assert classifier._parse_intent_type("rollback") == IntentType.ROLLBACK
|
||
|
||
def test_parse_unknown(self, classifier):
|
||
"""測試解析 unknown"""
|
||
assert classifier._parse_intent_type("unknown") == IntentType.UNKNOWN
|
||
assert classifier._parse_intent_type("invalid") == IntentType.UNKNOWN
|
||
assert classifier._parse_intent_type("") == IntentType.UNKNOWN
|
||
|
||
|
||
class TestLlmFallbackResult:
|
||
"""_llm_fallback_result 測試"""
|
||
|
||
def test_fallback_result(self, classifier):
|
||
"""測試 fallback 結果"""
|
||
result = classifier._llm_fallback_result("test error")
|
||
|
||
assert result.intent == IntentType.UNKNOWN
|
||
assert result.confidence == 0.0
|
||
assert result.method == "llm"
|
||
assert result.reasoning == "test error"
|
||
assert result.matched_keywords == []
|
||
assert result.detected_resources == []
|
||
|
||
|
||
class TestKeywordClassify:
|
||
"""關鍵字分類測試"""
|
||
|
||
def test_restart_keywords(self, classifier):
|
||
"""測試重啟關鍵字"""
|
||
result = classifier.classify_sync("重啟 api-server pod")
|
||
assert result.intent == IntentType.RESTART
|
||
assert result.method == "keyword"
|
||
# 關鍵字匹配信心度為 0 (非 AI 分析)
|
||
assert result.confidence == 0.0
|
||
assert "重啟" in result.matched_keywords
|
||
|
||
def test_scale_keywords(self, classifier):
|
||
"""測試擴縮容關鍵字"""
|
||
result = classifier.classify_sync("擴展 deployment 副本數到 5")
|
||
assert result.intent == IntentType.SCALE
|
||
assert result.method == "keyword"
|
||
|
||
def test_diagnose_keywords(self, classifier):
|
||
"""測試診斷關鍵字"""
|
||
result = classifier.classify_sync("查看 pod 日誌")
|
||
assert result.intent == IntentType.DIAGNOSE
|
||
assert result.method == "keyword"
|
||
|
||
def test_delete_keywords(self, classifier):
|
||
"""測試刪除關鍵字"""
|
||
result = classifier.classify_sync("刪除這個 pod")
|
||
assert result.intent == IntentType.DELETE
|
||
assert result.method == "keyword"
|
||
|
||
def test_unknown_text(self, classifier):
|
||
"""測試無法識別的文字"""
|
||
result = classifier.classify_sync("今天天氣真好")
|
||
assert result.intent == IntentType.UNKNOWN
|
||
|
||
|
||
class TestLlmClassify:
|
||
"""_llm_classify 測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_success(self, classifier, mock_ollama_response):
|
||
"""測試 LLM 成功分類"""
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.json.return_value = mock_ollama_response
|
||
mock_response.raise_for_status = MagicMock()
|
||
|
||
with patch("httpx.AsyncClient") as mock_client:
|
||
mock_instance = AsyncMock()
|
||
mock_instance.post.return_value = mock_response
|
||
mock_instance.__aenter__.return_value = mock_instance
|
||
mock_instance.__aexit__.return_value = None
|
||
mock_client.return_value = mock_instance
|
||
|
||
result = await classifier._llm_classify("重啟 pod")
|
||
|
||
assert result.intent == IntentType.RESTART
|
||
assert result.confidence == 0.85
|
||
assert result.method == "llm"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_timeout(self, classifier):
|
||
"""測試 LLM 超時"""
|
||
import httpx
|
||
|
||
with patch("httpx.AsyncClient") as mock_client:
|
||
mock_instance = AsyncMock()
|
||
mock_instance.post.side_effect = httpx.TimeoutException("timeout")
|
||
mock_instance.__aenter__.return_value = mock_instance
|
||
mock_instance.__aexit__.return_value = None
|
||
mock_client.return_value = mock_instance
|
||
|
||
result = await classifier._llm_classify("重啟 pod")
|
||
|
||
assert result.intent == IntentType.UNKNOWN
|
||
assert result.confidence == 0.0
|
||
assert "超時" in result.reasoning
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_invalid_json(self, classifier):
|
||
"""測試 LLM 返回無效 JSON"""
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.json.return_value = {"response": "not valid json"}
|
||
mock_response.raise_for_status = MagicMock()
|
||
|
||
with patch("httpx.AsyncClient") as mock_client:
|
||
mock_instance = AsyncMock()
|
||
mock_instance.post.return_value = mock_response
|
||
mock_instance.__aenter__.return_value = mock_instance
|
||
mock_instance.__aexit__.return_value = None
|
||
mock_client.return_value = mock_instance
|
||
|
||
result = await classifier._llm_classify("重啟 pod")
|
||
|
||
assert result.intent == IntentType.UNKNOWN
|
||
assert "解析失敗" in result.reasoning
|
||
|
||
|
||
class TestClassify:
|
||
"""完整分類流程測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_keyword_with_llm_fallback(self, classifier):
|
||
"""測試關鍵字匹配 + LLM fallback"""
|
||
# 由於關鍵字信心度為 0，會嘗試 LLM
|
||
# LLM 可能超時或失敗，最終返回關鍵字結果
|
||
result = await classifier.classify("重啟 api pod")
|
||
|
||
# 意圖應該是 RESTART (來自關鍵字或 LLM)
|
||
assert result.intent == IntentType.RESTART
|
||
# method 可能是 keyword (LLM 超時) 或 llm (LLM 成功)
|
||
assert result.method in ["keyword", "llm"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_llm_used_when_available(self, classifier, mock_ollama_response):
|
||
"""測試 LLM 可用時使用 LLM 結果"""
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.json.return_value = mock_ollama_response
|
||
mock_response.raise_for_status = MagicMock()
|
||
|
||
with patch("httpx.AsyncClient") as mock_client:
|
||
mock_instance = AsyncMock()
|
||
mock_instance.post.return_value = mock_response
|
||
mock_instance.__aenter__.return_value = mock_instance
|
||
mock_instance.__aexit__.return_value = None
|
||
mock_client.return_value = mock_instance
|
||
|
||
result = await classifier.classify("重啟 pod")
|
||
|
||
# LLM 成功時，應該使用 LLM 結果 (信心度 0.85 > 0)
|
||
assert result.intent == IntentType.RESTART
|
||
assert result.method == "llm"
|
||
assert result.confidence == 0.85
|
||
|
||
|
||
class TestGetIntentClassifier:
|
||
"""Singleton 測試"""
|
||
|
||
def test_singleton(self):
|
||
"""測試 singleton 模式"""
|
||
c1 = get_intent_classifier()
|
||
c2 = get_intent_classifier()
|
||
assert c1 is c2
|
||
|
||
|
||
class TestIntentResult:
|
||
"""IntentResult 測試"""
|
||
|
||
def test_dataclass_fields(self):
|
||
"""測試 dataclass 欄位"""
|
||
result = IntentResult(
|
||
intent=IntentType.RESTART,
|
||
confidence=0.9,
|
||
method="llm",
|
||
matched_keywords=["重啟", "pod"],
|
||
detected_resources=["api-server"],
|
||
reasoning="匹配重啟關鍵字",
|
||
)
|
||
|
||
assert result.intent == IntentType.RESTART
|
||
assert result.confidence == 0.9
|
||
assert result.method == "llm"
|
||
assert result.risk_level == RiskLevel.MEDIUM # auto-set by __post_init__
|
||
assert "重啟" in result.matched_keywords
|
||
assert "api-server" in result.detected_resources
|
||
|
||
def test_risk_level_auto_set(self):
|
||
"""測試風險等級自動設定"""
|
||
# DELETE 應該是 CRITICAL
|
||
delete_result = IntentResult(
|
||
intent=IntentType.DELETE,
|
||
confidence=0.8,
|
||
method="llm",
|
||
)
|
||
assert delete_result.risk_level == RiskLevel.CRITICAL
|
||
|
||
# DIAGNOSE 應該是 LOW
|
||
diagnose_result = IntentResult(
|
||
intent=IntentType.DIAGNOSE,
|
||
confidence=0.8,
|
||
method="llm",
|
||
)
|
||
assert diagnose_result.risk_level == RiskLevel.LOW
|