19fff8339d
整合真實後端服務，移除 Mock 數據: _handle_approval_action: - 使用 ApprovalDBService.get_pending_approvals() - 顯示待簽核清單摘要 (最多 5 個) - 渲染第一個待簽核項目的 ApprovalCard _handle_status_query: - 使用 K8s API 查詢 Pod 狀態 - 統計 Running/Ready/Total Pods - 顯示問題 Pods (非 Running 或 NotReady) - 查詢 Deployment 健康狀態 測試覆蓋: - 6 個新增 API 整合測試 - 總計 60 個測試通過 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
474 lines
15 KiB
Python
474 lines
15 KiB
Python
"""
|
||
Phase 19.6: Terminal Service Tests
|
||
==================================
|
||
Omni-Terminal SSE 架構測試
|
||
|
||
測試內容:
|
||
1. 意圖分類 (classify_intent)
|
||
2. Service 依賴注入
|
||
3. Model 驗證
|
||
|
||
@see ADR-031 Omni-Terminal SSE Architecture
|
||
@author Claude Code (首席架構師)
|
||
@version 1.0.0
|
||
@date 2026-03-28 (台北時間)
|
||
"""
|
||
|
||
import pytest
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
from uuid import uuid4
|
||
|
||
from src.models.terminal import (
|
||
SpatialContext,
|
||
TerminalIntentRequest,
|
||
TerminalSessionStatus,
|
||
)
|
||
from src.models.approval import ApprovalRequest, ApprovalStatus, RiskLevel
|
||
from src.services.terminal_service import (
|
||
IntentType,
|
||
TerminalService,
|
||
classify_intent,
|
||
get_terminal_service,
|
||
)
|
||
|
||
# =============================================================================
|
||
# Intent Classification Tests
|
||
# =============================================================================
|
||
|
||
INTENT_CLASSIFICATION_TEST_CASES = [
|
||
# 查詢類 - 狀態
|
||
("status", IntentType.QUERY_STATUS),
|
||
("狀態", IntentType.QUERY_STATUS),
|
||
("pod status", IntentType.QUERY_STATUS),
|
||
("健康狀態", IntentType.QUERY_STATUS),
|
||
("health check", IntentType.QUERY_STATUS),
|
||
|
||
# 查詢類 - 指標
|
||
("metrics", IntentType.QUERY_METRICS),
|
||
("指標", IntentType.QUERY_METRICS),
|
||
("cpu usage", IntentType.QUERY_METRICS),
|
||
("記憶體使用量", IntentType.QUERY_METRICS),
|
||
("memory utilization", IntentType.QUERY_METRICS),
|
||
|
||
# 查詢類 - 日誌
|
||
("logs", IntentType.QUERY_LOGS),
|
||
("日誌", IntentType.QUERY_LOGS),
|
||
("error logs", IntentType.QUERY_LOGS),
|
||
("錯誤訊息", IntentType.QUERY_LOGS),
|
||
("exception trace", IntentType.QUERY_LOGS),
|
||
|
||
# 操作類 - 簽核
|
||
("approve", IntentType.ACTION_APPROVAL),
|
||
("簽核", IntentType.ACTION_APPROVAL),
|
||
("授權", IntentType.ACTION_APPROVAL),
|
||
("批准這個操作", IntentType.ACTION_APPROVAL),
|
||
("show pending approval", IntentType.ACTION_APPROVAL),
|
||
|
||
# 操作類 - 重啟
|
||
("restart", IntentType.ACTION_RESTART),
|
||
("重啟", IntentType.ACTION_RESTART),
|
||
("rollout", IntentType.ACTION_RESTART),
|
||
("重新部署", IntentType.ACTION_RESTART),
|
||
|
||
# 操作類 - 擴容
|
||
("scale", IntentType.ACTION_SCALE),
|
||
("擴容", IntentType.ACTION_SCALE),
|
||
("增加副本", IntentType.ACTION_SCALE),
|
||
("replica count", IntentType.ACTION_SCALE),
|
||
|
||
# 分析類 - RCA
|
||
("rca", IntentType.ANALYZE_RCA),
|
||
("root cause", IntentType.ANALYZE_RCA),
|
||
("根因", IntentType.ANALYZE_RCA),
|
||
("分析問題", IntentType.ANALYZE_RCA),
|
||
("為什麼 API 變慢", IntentType.ANALYZE_RCA),
|
||
("why is it slow", IntentType.ANALYZE_RCA),
|
||
|
||
# 分析類 - Incident
|
||
("incident", IntentType.ANALYZE_INCIDENT),
|
||
("事件", IntentType.ANALYZE_INCIDENT),
|
||
("alert 告警", IntentType.ANALYZE_INCIDENT),
|
||
("告警詳情", IntentType.ANALYZE_INCIDENT),
|
||
|
||
# 一般
|
||
("hello", IntentType.GENERAL),
|
||
("你好", IntentType.GENERAL),
|
||
("help", IntentType.GENERAL),
|
||
("random text", IntentType.GENERAL),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("intent,expected_type", INTENT_CLASSIFICATION_TEST_CASES)
|
||
def test_classify_intent(intent: str, expected_type: IntentType):
|
||
"""測試意圖分類準確度"""
|
||
result = classify_intent(intent)
|
||
assert result == expected_type, f"Intent '{intent}' should be classified as {expected_type}, got {result}"
|
||
|
||
|
||
def test_classify_intent_case_insensitive():
|
||
"""測試意圖分類不區分大小寫"""
|
||
assert classify_intent("STATUS") == IntentType.QUERY_STATUS
|
||
assert classify_intent("Metrics") == IntentType.QUERY_METRICS
|
||
assert classify_intent("RESTART") == IntentType.ACTION_RESTART
|
||
|
||
|
||
# =============================================================================
|
||
# Model Validation Tests
|
||
# =============================================================================
|
||
|
||
|
||
def test_spatial_context_required_fields():
|
||
"""測試 SpatialContext 必填欄位"""
|
||
context = SpatialContext(current_page="/dashboard")
|
||
assert context.current_page == "/dashboard"
|
||
assert context.focused_entity_id is None # 預設為 None
|
||
|
||
|
||
def test_spatial_context_with_entity():
|
||
"""測試 SpatialContext 帶實體 ID"""
|
||
context = SpatialContext(
|
||
current_page="/incidents",
|
||
focused_entity_id="INC-001",
|
||
)
|
||
assert context.current_page == "/incidents"
|
||
assert context.focused_entity_id == "INC-001"
|
||
|
||
|
||
def test_terminal_intent_request_validation():
|
||
"""測試 TerminalIntentRequest 驗證"""
|
||
request = TerminalIntentRequest(
|
||
intent="check system status",
|
||
context=SpatialContext(current_page="/dashboard"),
|
||
)
|
||
assert request.intent == "check system status"
|
||
assert request.context.current_page == "/dashboard"
|
||
assert request.session_id is None # 預設為 None
|
||
|
||
|
||
def test_terminal_intent_request_with_session():
|
||
"""測試帶 session_id 的 TerminalIntentRequest"""
|
||
request = TerminalIntentRequest(
|
||
intent="continue analysis",
|
||
context=SpatialContext(current_page="/"),
|
||
session_id="sess-001",
|
||
)
|
||
assert request.session_id == "sess-001"
|
||
|
||
|
||
def test_terminal_session_status_enum():
|
||
"""測試 TerminalSessionStatus 枚舉值"""
|
||
assert TerminalSessionStatus.PENDING.value == "pending"
|
||
assert TerminalSessionStatus.PROCESSING.value == "processing"
|
||
assert TerminalSessionStatus.COMPLETED.value == "completed"
|
||
assert TerminalSessionStatus.ERROR.value == "error"
|
||
assert TerminalSessionStatus.ABORTED.value == "aborted"
|
||
|
||
|
||
# =============================================================================
|
||
# Dependency Injection Tests
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_terminal_service():
|
||
"""測試 FastAPI 依賴注入函數"""
|
||
service = await get_terminal_service()
|
||
assert isinstance(service, TerminalService)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_terminal_service_creates_new_instance():
|
||
"""測試每次呼叫建立新實例 (非 Singleton)"""
|
||
service1 = await get_terminal_service()
|
||
service2 = await get_terminal_service()
|
||
assert service1 is not service2, "Should create new instance each time"
|
||
|
||
|
||
# =============================================================================
|
||
# Service Unit Tests (No External Dependencies)
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_terminal_service_instantiation():
|
||
"""測試 TerminalService 實例化"""
|
||
service = TerminalService()
|
||
assert service._sessions == {}
|
||
assert service._tasks == {}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_session_not_found():
|
||
"""測試取得不存在的 Session"""
|
||
service = TerminalService()
|
||
session = await service.get_session("nonexistent-session")
|
||
assert session is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_abort_session_not_found():
|
||
"""測試中斷不存在的 Session"""
|
||
service = TerminalService()
|
||
result = await service.abort_session("nonexistent-session")
|
||
assert result is False
|
||
|
||
|
||
# =============================================================================
|
||
# Intent Coverage Statistics
|
||
# =============================================================================
|
||
|
||
|
||
def test_intent_type_coverage():
|
||
"""確保所有 IntentType 都有測試案例"""
|
||
tested_types = {expected for _, expected in INTENT_CLASSIFICATION_TEST_CASES}
|
||
all_types = set(IntentType)
|
||
|
||
missing = all_types - tested_types
|
||
assert not missing, f"Missing test cases for IntentType: {missing}"
|
||
|
||
|
||
# =============================================================================
|
||
# Phase 19.4: API Integration Tests
|
||
# =============================================================================
|
||
# @author Claude Code
|
||
# @date 2026-03-30 (台北時間)
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_publisher():
|
||
"""模擬 SSE Publisher"""
|
||
publisher = AsyncMock()
|
||
publisher.publish = AsyncMock()
|
||
return publisher
|
||
|
||
|
||
@pytest.fixture
|
||
def sample_approval():
|
||
"""測試用 Approval"""
|
||
return ApprovalRequest(
|
||
id=uuid4(),
|
||
action="kubectl rollout restart deployment/awoooi-api -n awoooi",
|
||
description="重啟 API 服務以套用新設定",
|
||
status=ApprovalStatus.PENDING,
|
||
risk_level=RiskLevel.MEDIUM,
|
||
required_signatures=1,
|
||
current_signatures=0,
|
||
signatures=[],
|
||
requested_by="system",
|
||
)
|
||
|
||
|
||
class TestHandleApprovalAction:
|
||
"""_handle_approval_action Phase 19.4 測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approval_action_with_pending(self, mock_publisher, sample_approval):
|
||
"""測試有待簽核項目時的處理"""
|
||
service = TerminalService()
|
||
|
||
mock_approval_service = MagicMock()
|
||
mock_approval_service.get_pending_approvals = AsyncMock(
|
||
return_value=[sample_approval]
|
||
)
|
||
|
||
with patch(
|
||
"src.services.terminal_service.get_approval_service",
|
||
return_value=mock_approval_service,
|
||
):
|
||
await service._handle_approval_action(
|
||
mock_publisher,
|
||
"test_topic",
|
||
"test_session",
|
||
"approval",
|
||
)
|
||
|
||
# 驗證 publish 被呼叫多次 (thought + tool_call + render_ui)
|
||
assert mock_publisher.publish.call_count >= 3
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approval_action_no_pending(self, mock_publisher):
|
||
"""測試沒有待簽核項目時的處理"""
|
||
service = TerminalService()
|
||
|
||
mock_approval_service = MagicMock()
|
||
mock_approval_service.get_pending_approvals = AsyncMock(return_value=[])
|
||
|
||
with patch(
|
||
"src.services.terminal_service.get_approval_service",
|
||
return_value=mock_approval_service,
|
||
):
|
||
await service._handle_approval_action(
|
||
mock_publisher,
|
||
"test_topic",
|
||
"test_session",
|
||
"approval",
|
||
)
|
||
|
||
# 驗證有發送 "沒有待簽核項目" 的訊息
|
||
calls = mock_publisher.publish.call_args_list
|
||
found_no_pending = False
|
||
for call in calls:
|
||
event = call[0][0]
|
||
if hasattr(event, 'data') and isinstance(event.data, dict):
|
||
msg = event.data.get('msg', '')
|
||
if '沒有待簽核' in msg:
|
||
found_no_pending = True
|
||
break
|
||
|
||
assert found_no_pending, "Should mention no pending approvals"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approval_action_error_handling(self, mock_publisher):
|
||
"""測試查詢失敗時的錯誤處理"""
|
||
service = TerminalService()
|
||
|
||
mock_approval_service = MagicMock()
|
||
mock_approval_service.get_pending_approvals = AsyncMock(
|
||
side_effect=Exception("Database error")
|
||
)
|
||
|
||
with patch(
|
||
"src.services.terminal_service.get_approval_service",
|
||
return_value=mock_approval_service,
|
||
):
|
||
# 不應拋出例外
|
||
await service._handle_approval_action(
|
||
mock_publisher,
|
||
"test_topic",
|
||
"test_session",
|
||
"approval",
|
||
)
|
||
|
||
# 驗證有發送錯誤訊息
|
||
assert mock_publisher.publish.call_count >= 2
|
||
|
||
|
||
class TestHandleStatusQuery:
|
||
"""_handle_status_query Phase 19.4 測試"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_query_success(self, mock_publisher):
|
||
"""測試 K8s 狀態查詢成功"""
|
||
service = TerminalService()
|
||
|
||
# 模擬 K8s client
|
||
mock_pod = MagicMock()
|
||
mock_pod.metadata.name = "awoooi-api-xxx"
|
||
mock_pod.status.phase = "Running"
|
||
mock_pod.status.container_statuses = [MagicMock(ready=True)]
|
||
|
||
mock_pods_list = MagicMock()
|
||
mock_pods_list.items = [mock_pod]
|
||
|
||
mock_deployment = MagicMock()
|
||
mock_deployment.status.ready_replicas = 1
|
||
mock_deployment.spec.replicas = 1
|
||
|
||
mock_deployments_list = MagicMock()
|
||
mock_deployments_list.items = [mock_deployment]
|
||
|
||
mock_v1 = MagicMock()
|
||
mock_v1.list_namespaced_pod = AsyncMock(return_value=mock_pods_list)
|
||
|
||
mock_apps_v1 = MagicMock()
|
||
mock_apps_v1.list_namespaced_deployment = AsyncMock(
|
||
return_value=mock_deployments_list
|
||
)
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.CoreV1Api.return_value = mock_v1
|
||
mock_client.AppsV1Api.return_value = mock_apps_v1
|
||
|
||
with patch(
|
||
"src.services.k8s_diagnostics._get_k8s_client",
|
||
new=AsyncMock(return_value=mock_client),
|
||
):
|
||
await service._handle_status_query(
|
||
mock_publisher,
|
||
"test_topic",
|
||
"test_session",
|
||
"status",
|
||
)
|
||
|
||
# 驗證成功查詢
|
||
assert mock_publisher.publish.call_count >= 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_query_with_problem_pods(self, mock_publisher):
|
||
"""測試有問題 Pods 時的狀態查詢"""
|
||
service = TerminalService()
|
||
|
||
# 一個正常 Pod，一個有問題的 Pod
|
||
mock_pod_ok = MagicMock()
|
||
mock_pod_ok.metadata.name = "awoooi-api-ok"
|
||
mock_pod_ok.status.phase = "Running"
|
||
mock_pod_ok.status.container_statuses = [MagicMock(ready=True)]
|
||
|
||
mock_pod_bad = MagicMock()
|
||
mock_pod_bad.metadata.name = "awoooi-worker-bad"
|
||
mock_pod_bad.status.phase = "Pending"
|
||
mock_pod_bad.status.container_statuses = []
|
||
|
||
mock_pods_list = MagicMock()
|
||
mock_pods_list.items = [mock_pod_ok, mock_pod_bad]
|
||
|
||
mock_deployments_list = MagicMock()
|
||
mock_deployments_list.items = []
|
||
|
||
mock_v1 = MagicMock()
|
||
mock_v1.list_namespaced_pod = AsyncMock(return_value=mock_pods_list)
|
||
|
||
mock_apps_v1 = MagicMock()
|
||
mock_apps_v1.list_namespaced_deployment = AsyncMock(
|
||
return_value=mock_deployments_list
|
||
)
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.CoreV1Api.return_value = mock_v1
|
||
mock_client.AppsV1Api.return_value = mock_apps_v1
|
||
|
||
with patch(
|
||
"src.services.k8s_diagnostics._get_k8s_client",
|
||
new=AsyncMock(return_value=mock_client),
|
||
):
|
||
await service._handle_status_query(
|
||
mock_publisher,
|
||
"test_topic",
|
||
"test_session",
|
||
"status",
|
||
)
|
||
|
||
# 驗證有提到問題 Pods
|
||
calls = mock_publisher.publish.call_args_list
|
||
found_problem_pod = False
|
||
for call in calls:
|
||
event = call[0][0]
|
||
if hasattr(event, 'data') and isinstance(event.data, dict):
|
||
msg = event.data.get('msg', '')
|
||
if '問題 Pods' in msg or 'Pending' in msg:
|
||
found_problem_pod = True
|
||
break
|
||
|
||
assert found_problem_pod, "Should mention problem pods"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_query_k8s_unavailable(self, mock_publisher):
|
||
"""測試 K8s 連線失敗時的處理"""
|
||
service = TerminalService()
|
||
|
||
with patch(
|
||
"src.services.k8s_diagnostics._get_k8s_client",
|
||
new=AsyncMock(return_value=None),
|
||
):
|
||
# 不應拋出例外
|
||
await service._handle_status_query(
|
||
mock_publisher,
|
||
"test_topic",
|
||
"test_session",
|
||
"status",
|
||
)
|
||
|
||
# 驗證有發送錯誤訊息
|
||
assert mock_publisher.publish.call_count >= 2
|