""" B3: LLM 動態 Telegram 按鈕 — 單元測試 ===================================== 2026-04-27 Claude Sonnet 4.6: ADR-082 B3 測試範圍: 1. USE_LLM_DYNAMIC_BUTTONS=false → 走 YAML 路徑(現有行為) 2. USE_LLM_DYNAMIC_BUTTONS=true + recommended_actions 空 → 走 YAML 路徑 3. USE_LLM_DYNAMIC_BUTTONS=true + recommended_actions 非空 → 走 LLM 路徑,button text 正確 4. high risk action 前綴 ⚠️ 5. callback_data 是合法 JSON,t="llm_action" 6. [批准][拒絕] 第一排永遠存在 7. 超過 2 個 action → 多排 """ from __future__ import annotations import json import os from dataclasses import dataclass, field from typing import Literal from unittest.mock import MagicMock, patch import pytest # ============================================================================= # 輕量 Stub:避免載入完整 telegram_gateway(需要 Redis / httpx / OTEL) # ============================================================================= @dataclass class _RecommendedAction: name: str label: str emoji: str mcp_provider: str mcp_tool: str params: dict risk: Literal["low", "medium", "high", "critical"] reasoning: str = "" @dataclass class _ActionPlan: recommended_actions: list[_RecommendedAction] = field(default_factory=list) # ============================================================================= # 直接測試 _build_llm_action_buttons 靜態方法 # ============================================================================= def _import_builder(): """ 延遲 import,確保環境變數先設定。 返回 TelegramGateway._build_llm_action_buttons 靜態方法。 """ # 使用 importlib 動態取得,避免頂層 import 觸發 settings 初始化 import importlib import sys # 清除快取以確保 env 設定生效 for mod_name in list(sys.modules.keys()): if "telegram_gateway" in mod_name: del sys.modules[mod_name] mod = importlib.import_module("src.services.telegram_gateway") return mod.TelegramGateway._build_llm_action_buttons, mod # ============================================================================= # Fixtures # ============================================================================= @pytest.fixture def low_action(): return _RecommendedAction( name="check_pod_logs", label="查 Pod 日誌", emoji="📋", mcp_provider="k8s", mcp_tool="get_pod_logs", params={"pod": "{labels.pod}"}, risk="low", ) @pytest.fixture def high_action(): return _RecommendedAction( name="restart_pod", label="重啟 Pod", emoji="🔄", mcp_provider="k8s", mcp_tool="restart_pod", params={"pod": "{labels.pod}"}, risk="high", ) @pytest.fixture def medium_action(): return _RecommendedAction( name="scale_deployment", label="擴容", emoji="📈", mcp_provider="k8s", mcp_tool="scale_deployment", params={"replicas": "3"}, risk="medium", ) # ============================================================================= # Test 1: _build_llm_action_buttons — 基本 text 格式 # ============================================================================= class TestBuildLlmActionButtons: """直接測試靜態方法,不需要完整 gateway 初始化""" def _get_builder(self): from src.services.telegram_gateway import TelegramGateway return TelegramGateway._build_llm_action_buttons def test_low_risk_text_format(self, low_action): builder = self._get_builder() rows = builder([low_action]) assert len(rows) == 1 btn = rows[0][0] assert btn["text"] == "📋 查 Pod 日誌" def test_high_risk_prefix(self, high_action): """Test 4: high risk 前綴 ⚠️""" builder = self._get_builder() rows = builder([high_action]) btn = rows[0][0] assert btn["text"].startswith("⚠️") assert "重啟 Pod" in btn["text"] def test_medium_risk_no_prefix(self, medium_action): builder = self._get_builder() rows = builder([medium_action]) btn = rows[0][0] assert not btn["text"].startswith("⚠️") def test_callback_data_valid_json(self, low_action): """Test 5: callback_data 是合法 JSON,t="la"(llm_action 縮寫),含必要欄位""" builder = self._get_builder() rows = builder([low_action]) cb = rows[0][0]["callback_data"] payload = json.loads(cb) # t="la" 是 llm_action 的縮寫,與 YAML callback type 不衝突 assert payload["t"] == "la" # n=name, p=provider, tl=tool(縮短 key 以符合 64 bytes 限制) assert "n" in payload assert "p" in payload assert "tl" in payload def test_callback_data_within_64_bytes(self, low_action): builder = self._get_builder() rows = builder([low_action]) cb = rows[0][0]["callback_data"] assert len(cb.encode("utf-8")) <= 64 def test_long_name_truncated_within_64_bytes(self): """callback_data 超過 64 bytes 時自動截斷 name""" action = _RecommendedAction( name="a" * 50, label="長名稱動作", emoji="🔧", mcp_provider="k8s", mcp_tool="do_something", params={}, risk="low", ) from src.services.telegram_gateway import TelegramGateway rows = TelegramGateway._build_llm_action_buttons([action]) cb = rows[0][0]["callback_data"] assert len(cb.encode("utf-8")) <= 64 payload = json.loads(cb) assert payload["t"] == "la" def test_two_actions_same_row(self, low_action, medium_action): """Test 7 前置:2 個 action → 同一排""" builder = self._get_builder() rows = builder([low_action, medium_action]) assert len(rows) == 1 assert len(rows[0]) == 2 def test_three_actions_two_rows(self, low_action, medium_action, high_action): """Test 7: 超過 2 個 action → 多排(每排最多 2 個)""" builder = self._get_builder() rows = builder([low_action, medium_action, high_action]) assert len(rows) == 2 assert len(rows[0]) == 2 assert len(rows[1]) == 1 def test_empty_actions_returns_empty(self): builder = self._get_builder() rows = builder([]) assert rows == [] # ============================================================================= # Test 2-3, 6: _build_inline_keyboard 路徑切換 # ============================================================================= class TestBuildInlineKeyboardRouting: """ 測試 _build_inline_keyboard() 的路徑選擇邏輯。 用 MagicMock 模擬 security,避免初始化完整 gateway。 """ def _make_gateway_cls(self): from src.services.telegram_gateway import TelegramGateway return TelegramGateway def _make_mock_gateway(self): cls = self._make_gateway_cls() gw = object.__new__(cls) mock_security = MagicMock() mock_security.generate_callback_nonce.side_effect = ( lambda approval_id, action: f"nonce:{approval_id}:{action}" ) gw._security = mock_security return gw def _first_row_texts(self, keyboard: dict) -> list[str]: return [btn["text"] for btn in keyboard["inline_keyboard"][0]] # Test 1: flag=false → YAML 路徑(第一排還是批准/拒絕) @patch("src.services.telegram_gateway.USE_LLM_DYNAMIC_BUTTONS", False) @patch("src.services.callback_dispatcher.list_actions_for_category", return_value=[]) def test_flag_false_uses_yaml_path(self, mock_list, low_action): gw = self._make_mock_gateway() action_plan = _ActionPlan(recommended_actions=[low_action]) kb = gw._build_inline_keyboard( approval_id="APR-001", incident_id="INC-001", action_plan=action_plan, ) first_texts = self._first_row_texts(kb) assert "✅ 批准" in first_texts assert "❌ 拒絕" in first_texts # LLM 按鈕不應出現(t="la" 代表 llm_action) all_payloads_1 = [ json.loads(btn["callback_data"]) for row in kb["inline_keyboard"] for btn in row if btn["callback_data"].startswith("{") ] assert not any(p.get("t") == "la" for p in all_payloads_1) # Test 2: flag=true + actions 空 → YAML fallback @patch("src.services.telegram_gateway.USE_LLM_DYNAMIC_BUTTONS", True) @patch("src.services.callback_dispatcher.list_actions_for_category", return_value=[]) def test_flag_true_empty_actions_uses_yaml(self, mock_list): gw = self._make_mock_gateway() action_plan = _ActionPlan(recommended_actions=[]) kb = gw._build_inline_keyboard( approval_id="APR-002", incident_id="INC-002", action_plan=action_plan, ) first_texts = self._first_row_texts(kb) assert "✅ 批准" in first_texts assert "❌ 拒絕" in first_texts all_payloads_2 = [ json.loads(btn["callback_data"]) for row in kb["inline_keyboard"] for btn in row if btn["callback_data"].startswith("{") ] assert not any(p.get("t") == "la" for p in all_payloads_2) # Test 2b: flag=true + action_plan=None → YAML fallback @patch("src.services.telegram_gateway.USE_LLM_DYNAMIC_BUTTONS", True) @patch("src.services.callback_dispatcher.list_actions_for_category", return_value=[]) def test_flag_true_no_action_plan_uses_yaml(self, mock_list): gw = self._make_mock_gateway() kb = gw._build_inline_keyboard( approval_id="APR-003", incident_id="INC-003", ) first_texts = self._first_row_texts(kb) assert "✅ 批准" in first_texts assert "❌ 拒絕" in first_texts # Test 3: flag=true + actions 非空 → LLM 路徑 @patch("src.services.telegram_gateway.USE_LLM_DYNAMIC_BUTTONS", True) def test_flag_true_with_actions_uses_llm(self, low_action): gw = self._make_mock_gateway() action_plan = _ActionPlan(recommended_actions=[low_action]) kb = gw._build_inline_keyboard( approval_id="APR-004", incident_id="INC-004", action_plan=action_plan, ) # Test 6: 第一排永遠是 [批准][拒絕] first_texts = self._first_row_texts(kb) assert "✅ 批准" in first_texts assert "❌ 拒絕" in first_texts # LLM 按鈕在第二排以後(callback_data 含 "la" type) all_payloads = [ json.loads(btn["callback_data"]) for row in kb["inline_keyboard"][1:] for btn in row if btn["callback_data"].startswith("{") ] assert any(p.get("t") == "la" for p in all_payloads) # button text 正確 all_texts = [ btn["text"] for row in kb["inline_keyboard"][1:] for btn in row ] assert any("查 Pod 日誌" in t for t in all_texts) # Test 6: 第一排永遠是 [批准][拒絕](LLM 路徑) @patch("src.services.telegram_gateway.USE_LLM_DYNAMIC_BUTTONS", True) def test_approve_reject_always_first_row_in_llm_path(self, low_action, high_action, medium_action): gw = self._make_mock_gateway() action_plan = _ActionPlan(recommended_actions=[low_action, high_action, medium_action]) kb = gw._build_inline_keyboard( approval_id="APR-005", incident_id="INC-005", action_plan=action_plan, ) first_texts = self._first_row_texts(kb) assert first_texts[0] == "✅ 批准" assert first_texts[1] == "❌ 拒絕" # 總排數 = 1 (approve/reject) + 2 (3 actions, 2 per row) assert len(kb["inline_keyboard"]) == 3