From 19fff8339dd0320a3408f9945e35b385f112f7c3 Mon Sep 17 00:00:00 2001 From: OG T Date: Mon, 30 Mar 2026 01:17:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(api):=20Phase=2019.4=20Terminal=20Service?= =?UTF-8?q?=20=E7=9C=9F=E5=AF=A6=20API=20=E6=95=B4=E5=90=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 整合真實後端服務，移除 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 --- apps/api/src/services/terminal_service.py | 207 +++++++++++++++--- apps/api/tests/test_terminal_service.py | 249 ++++++++++++++++++++++ 2 files changed, 421 insertions(+), 35 deletions(-) diff --git a/apps/api/src/services/terminal_service.py b/apps/api/src/services/terminal_service.py index d5f56f224..08bf83452 100644 --- a/apps/api/src/services/terminal_service.py +++ b/apps/api/src/services/terminal_service.py @@ -28,6 +28,7 @@ from src.models.terminal import ( TerminalSession, TerminalSessionStatus, ) +from src.services.approval_db import get_approval_service from src.services.openclaw import get_openclaw logger = get_logger("awoooi.terminal") @@ -489,30 +490,83 @@ class TerminalService: self, publisher, topic: str, - session_id: str, + _session_id: str, # Reserved for session-specific approval tracking _intent: str, # Reserved for intent-specific approval filtering ) -> None: - """處理簽核操作""" + """ + 處理簽核操作 + + Phase 19.4: 整合真實 Approval API + @author Claude Code + @date 2026-03-30 (台北時間) + """ await self._publish_thought(publisher, topic, "Executor", "檢查待簽核項目...") await self._publish_tool_call(publisher, topic, "Approval-Query", "executing") - await asyncio.sleep(0.5) - # TODO: Phase 19.4 - 整合真實 Approval API - await self._publish_tool_call( - publisher, topic, "Approval-Query", "completed", - result={"pending_count": 2} - ) - await asyncio.sleep(0.3) + try: + approval_service = get_approval_service() + pending_approvals = await approval_service.get_pending_approvals() - # 渲染 ApprovalCard - await self._publish_render_ui( - publisher, topic, "ApprovalCard", - { - "approvalId": f"APR-{session_id[:8].upper()}", - "riskLevel": "MEDIUM", - "kubectl": "kubectl rollout restart deployment/awoooi-api -n awoooi", - } - ) + await self._publish_tool_call( + publisher, topic, "Approval-Query", "completed", + result={"pending_count": len(pending_approvals)} + ) + await asyncio.sleep(0.3) + + if not pending_approvals: + await self._publish_thought( + publisher, topic, "Executor", + "目前沒有待簽核的項目。" + ) + return + + # 顯示待簽核清單摘要 + summary_lines = [f"發現 {len(pending_approvals)} 個待簽核項目:"] + for i, approval in enumerate(pending_approvals[:5], 1): # 最多顯示 5 個 + risk = approval.risk_level.value.upper() if approval.risk_level else "UNKNOWN" + summary_lines.append( + f" {i}. [{risk}] {approval.action[:50]}" + ) + + if len(pending_approvals) > 5: + summary_lines.append(f" ... 還有 {len(pending_approvals) - 5} 個") + + await self._publish_thought( + publisher, topic, "Executor", + "\n".join(summary_lines) + ) + + # 渲染第一個待簽核項目的 ApprovalCard + first_approval = pending_approvals[0] + risk_level = first_approval.risk_level.value.upper() if first_approval.risk_level else "MEDIUM" + + # 從 metadata 取得 kubectl 指令 (如果有) + kubectl_cmd = "kubectl get pods -n awoooi" + if first_approval.metadata: + kubectl_cmd = first_approval.metadata.get("kubectl_command", kubectl_cmd) + + await self._publish_render_ui( + publisher, topic, "ApprovalCard", + { + "approvalId": str(first_approval.id)[:8].upper(), + "riskLevel": risk_level, + "kubectl": kubectl_cmd, + "description": first_approval.description[:100] if first_approval.description else "", + "requiredSignatures": first_approval.required_signatures, + "currentSignatures": first_approval.current_signatures, + } + ) + + except Exception as e: + logger.error("approval_query_failed", error=str(e)) + await self._publish_tool_call( + publisher, topic, "Approval-Query", "failed", + result={"error": str(e)[:100]} + ) + await self._publish_thought( + publisher, topic, "System", + f"查詢簽核項目失敗: {str(e)[:50]}" + ) async def _handle_metrics_query( self, @@ -584,27 +638,110 @@ class TerminalService: _session_id: str, # Reserved for session-specific caching _intent: str, # Reserved for intent-specific status filtering ) -> None: - """處理狀態查詢""" + """ + 處理狀態查詢 + + Phase 19.4: 整合真實 K8s API + @author Claude Code + @date 2026-03-30 (台北時間) + """ await self._publish_thought(publisher, topic, "Monitor", "檢查系統狀態...") await self._publish_tool_call(publisher, topic, "K8s-Status", "executing") - await asyncio.sleep(0.5) - # TODO: Phase 19.4 - 整合真實 K8s API - await self._publish_tool_call( - publisher, topic, "K8s-Status", "completed", - result={ - "pods_running": 11, - "pods_total": 12, - "deployments_healthy": 5, - } - ) - await asyncio.sleep(0.3) + try: + # 使用 K8s Diagnostics Service 的 client + from src.services.k8s_diagnostics import _get_k8s_client - await self._publish_thought( - publisher, topic, "Monitor", - "系統狀態: 11/12 Pods 運行中，5 個 Deployment 健康。\n" - "1 個 Pod 正在重啟中 (awoooi-worker-xxx)。" - ) + client = await _get_k8s_client() + if not client: + raise RuntimeError("K8s client initialization failed") + + v1 = client.CoreV1Api() + apps_v1 = client.AppsV1Api() + + # 查詢 awoooi namespace 的 Pods + namespace = "awoooi" + pods = await v1.list_namespaced_pod(namespace=namespace) + + # 統計 Pod 狀態 + pods_total = len(pods.items) + pods_running = sum( + 1 for p in pods.items + if p.status.phase == "Running" + ) + pods_ready = sum( + 1 for p in pods.items + if p.status.phase == "Running" + and all( + c.ready for c in (p.status.container_statuses or []) + ) + ) + + # 找出有問題的 Pods + problem_pods = [] + for pod in pods.items: + if pod.status.phase != "Running": + problem_pods.append(f"{pod.metadata.name}: {pod.status.phase}") + elif pod.status.container_statuses: + for cs in pod.status.container_statuses: + if not cs.ready: + reason = "" + if cs.state and cs.state.waiting: + reason = cs.state.waiting.reason or "" + problem_pods.append(f"{pod.metadata.name}: {reason or 'NotReady'}") + break + + # 查詢 Deployments + deployments = await apps_v1.list_namespaced_deployment(namespace=namespace) + deployments_total = len(deployments.items) + deployments_healthy = sum( + 1 for d in deployments.items + if d.status.ready_replicas == d.spec.replicas + ) + + await self._publish_tool_call( + publisher, topic, "K8s-Status", "completed", + result={ + "pods_running": pods_running, + "pods_ready": pods_ready, + "pods_total": pods_total, + "deployments_healthy": deployments_healthy, + "deployments_total": deployments_total, + } + ) + await asyncio.sleep(0.3) + + # 組合狀態訊息 + status_lines = [ + f"系統狀態: {pods_ready}/{pods_total} Pods 就緒", + f"Deployments: {deployments_healthy}/{deployments_total} 健康", + ] + + if problem_pods: + status_lines.append(f"\n問題 Pods ({len(problem_pods)}):") + for pp in problem_pods[:3]: # 最多顯示 3 個 + status_lines.append(f" - {pp}") + if len(problem_pods) > 3: + status_lines.append(f" ... 還有 {len(problem_pods) - 3} 個") + else: + status_lines.append("\n所有 Pods 運行正常。") + + await self._publish_thought( + publisher, topic, "Monitor", + "\n".join(status_lines) + ) + + except Exception as e: + logger.error("k8s_status_query_failed", error=str(e)) + await self._publish_tool_call( + publisher, topic, "K8s-Status", "failed", + result={"error": str(e)[:100]} + ) + await self._publish_thought( + publisher, topic, "System", + f"K8s 狀態查詢失敗: {str(e)[:50]}\n" + "請確認 K8s 連線配置。" + ) async def _handle_general_chat( self, diff --git a/apps/api/tests/test_terminal_service.py b/apps/api/tests/test_terminal_service.py index 19995e48f..dfbc99d4c 100644 --- a/apps/api/tests/test_terminal_service.py +++ b/apps/api/tests/test_terminal_service.py @@ -15,12 +15,15 @@ Omni-Terminal SSE 架構測試 """ 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, @@ -222,3 +225,249 @@ def test_intent_type_coverage(): 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