diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index d498df15..e978291f 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -1427,7 +1427,7 @@ class TelegramGateway:
logger.error("telegram_request_failed", method=method, error=str(e))
raise TelegramGatewayError(str(e)) from e
- def _build_inline_keyboard(
+ async def _build_inline_keyboard(
self,
approval_id: str,
include_auto_tuning: bool = True,
@@ -1485,7 +1485,7 @@ class TelegramGateway:
else None
)
if USE_LLM_DYNAMIC_BUTTONS and _llm_actions:
- llm_rows = self._build_llm_action_buttons(_llm_actions)
+ llm_rows = await self._build_llm_action_buttons(_llm_actions, incident_id=incident_id)
buttons: list[list[dict]] = [first_row] + llm_rows
logger.info(
"telegram_keyboard_built",
@@ -1576,29 +1576,39 @@ class TelegramGateway:
return {"inline_keyboard": buttons}
- @staticmethod
- def _build_llm_action_buttons(
+ async def _build_llm_action_buttons(
+ self,
actions: list,
+ incident_id: str = "",
) -> list[list[dict]]:
"""
2026-04-27 Claude Sonnet 4.6: B3 — 從 RecommendedAction list 建立 Telegram 按鈕排
+ 2026-04-27 Claude Sonnet 4.6: H3+M6 Fix — short_id Redis 映射 + critical 過濾
+ 2026-04-27 Claude Sonnet 4.6: P0 Fix — async setex 在 return 前完成,消除按鈕已過期 race
規格:
+ - critical risk action → 直接跳過,不生成按鈕(M6)
- 每個 RecommendedAction → 一個按鈕
- text = f"{action.emoji} {action.label}"(risk=high 前綴 ⚠️)
- - callback_data = JSON {"t":"llm_action","name":..,"provider":..,"tool":..}(限 64 bytes)
+ - callback_data = f"la:{short_id}"(16-hex-chars,≤19 bytes,絕不截斷)(H3)
+ - 完整 payload(含 incident_id)寫入 Redis tg:la:{short_id},TTL=3600s(H3)
+ - Redis setex 在 return 之前 await 完成(P0 race fix)
- 每排最多 2 個(同 YAML fallback 排版)
- 不包含第一排 [批准][拒絕](由呼叫方負責置頂)
Args:
actions: list[RecommendedAction]
+ incident_id: 真實 incident ID,寫入 Redis payload 供 callback handler 還原
Returns:
list[list[dict]] — 按鈕行列(不含第一排)
"""
- import json
+ import json # noqa: PLC0415
+ import secrets # noqa: PLC0415
btn_list: list[dict] = []
+ redis_writes: list[tuple[str, str]] = [] # (key, json_str)
+
for action in actions:
name: str = getattr(action, "name", "")
label: str = getattr(action, "label", "")
@@ -1607,29 +1617,49 @@ class TelegramGateway:
tool: str = getattr(action, "mcp_tool", "")
risk: str = getattr(action, "risk", "low")
+ # M6: critical risk 直接跳過,不出按鈕
+ # 2026-04-27 Claude Sonnet 4.6: M6 Fix — critical action 不可被 Telegram 觸發
+ if risk == "critical":
+ logger.info(
+ "llm_button_critical_skipped",
+ name=name,
+ mcp_tool=tool,
+ )
+ continue
+
# risk=high 前綴 ⚠️ 警示
prefix = "⚠️ " if risk == "high" else ""
text = f"{prefix}{emoji} {label}".strip()
- # callback_data JSON,限 64 bytes(Telegram 上限)
- # 使用縮短 key:t=la(llm_action), n=name, p=provider, tl=tool
- # 縮短後框架約 47 bytes,留 ~17 bytes 給 name
- cb_payload = {"t": "la", "n": "", "p": provider, "tl": tool}
- frame_bytes = len(
- json.dumps(cb_payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+ # H3: 16-hex short_id(64-bit),callback_data ≤19 bytes
+ short_id = secrets.token_hex(8) # 16-hex-chars(P1: 4→8 bytes 防碰撞)
+ cb_str = f"la:{short_id}"
+
+ payload_str = json.dumps(
+ {
+ "name": name,
+ "provider": provider,
+ "tool": tool,
+ "risk": risk,
+ "incident_id": incident_id, # P0: 真實 incident_id 進 Redis
+ },
+ ensure_ascii=False,
+ separators=(",", ":"),
)
- available = max(0, 64 - frame_bytes)
- if len(name.encode("utf-8")) <= available:
- truncated_name = name
- else:
- # 按 UTF-8 bytes 截斷(中文字可能多 bytes)
- encoded = name.encode("utf-8")[:available]
- truncated_name = encoded.decode("utf-8", errors="ignore")
- cb_payload["n"] = truncated_name
- cb_str = json.dumps(cb_payload, ensure_ascii=False, separators=(",", ":"))
+ redis_writes.append((f"tg:la:{short_id}", payload_str))
btn_list.append({"text": text, "callback_data": cb_str})
+ # P0 Fix: await 完成再 return,消除「按鈕發出→點擊→Redis 還沒寫」的 race
+ if redis_writes:
+ try:
+ redis = get_redis()
+ for key, value in redis_writes:
+ await redis.setex(key, 3600, value)
+ logger.debug("llm_button_redis_written", count=len(redis_writes))
+ except Exception as exc:
+ logger.warning("llm_button_redis_write_failed", error=str(exc))
+
# 每排最多 2 個
rows: list[list[dict]] = [btn_list[i:i+2] for i in range(0, len(btn_list), 2)]
return rows
@@ -1818,7 +1848,7 @@ class TelegramGateway:
# 建立按鈕 (含自動調優)
# 2026-04-05 Claude Code: 傳入 incident_id 以啟用 detail/reanalyze/history 按鈕
# ADR-075: 傳入 alert_category/notification_type 以啟用分類動態按鈕(斷點 B 修復)
- keyboard = self._build_inline_keyboard(
+ keyboard = await self._build_inline_keyboard(
approval_id=approval_id,
include_auto_tuning=bool(auto_tuning_command),
auto_tuning_command=auto_tuning_command,
@@ -2048,7 +2078,7 @@ class TelegramGateway:
# 2026-04-25 ogt + Claude Sonnet 4.6: 群組卡片使用完整 _build_inline_keyboard
# 統帥決策: 群組成員為受信任 SRE,完整批准/拒絕/暫默/詳情/重診/歷史按鈕從 DM 移植至群組
- _group_keyboard = self._build_inline_keyboard(
+ _group_keyboard = await self._build_inline_keyboard(
approval_id=approval_id,
incident_id=incident_id,
alert_category=alert_category,
@@ -3226,6 +3256,19 @@ class TelegramGateway:
dict: 處理結果 {action, approval_id, user, auto_tuning_result?}
"""
try:
+ # ===================================================================
+ # Step 0: LLM Action Callback(H1/B4)— la:{short_id} 格式優先路由
+ # 2026-04-27 Claude Sonnet 4.6: H1+B4 Fix — 鬼魂按鈕鐵律修復
+ # 必須在 parse_callback_data 之前攔截,否則 split(":") 分析 JSON 會爆
+ # ===================================================================
+ if callback_data.startswith("la:"):
+ return await self._handle_llm_action_callback(
+ callback_query_id=callback_query_id,
+ callback_data=callback_data,
+ user_id=user_id,
+ username=username,
+ )
+
# ===================================================================
# Step 1: 解析 Callback Data (支援兩種格式)
# ===================================================================
@@ -4062,6 +4105,166 @@ class TelegramGateway:
"error": str(e),
}
+ async def _handle_llm_action_callback(
+ self,
+ callback_query_id: str,
+ callback_data: str,
+ user_id: int,
+ username: str = "",
+ ) -> dict:
+ """
+ B4: 處理 LLM 動態按鈕 callback(格式 la:{short_id})
+
+ 2026-04-27 Claude Sonnet 4.6: H1+B4 Fix — 鬼魂按鈕鐵律修復
+ 鬼魂按鈕三缺一絕不發送(callback格式+handler+MCP能力);
+ 本方法補上 handler,與 H3 Redis short_id 映射配合。
+
+ 流程:
+ 1. 白名單驗證
+ 2. Redis GET tg:la:{short_id} → 還原 payload(找不到 → 按鈕已過期)
+ 3. 呼叫 dispatch_llm_action 取得執行規格
+ 4. high risk 未確認 → 回應確認提示(TODO: 實作二次確認流程)
+ 5. low/medium → answer_callback_query + 執行 MCP → 回報結果
+ 6. 失敗 → 回報錯誤,不 crash
+ """
+ import json as _json # noqa: PLC0415
+
+ from src.services.callback_dispatcher import dispatch_llm_action # noqa: PLC0415
+
+ # ── 1. 白名單驗證 ─────────────────────────────────────────────────────
+ if not self._security.is_whitelisted(user_id):
+ await self._send_request("answerCallbackQuery", {
+ "callback_query_id": callback_query_id,
+ "text": "❌ 您沒有執行此操作的權限",
+ "show_alert": True,
+ })
+ return {"action": "llm_action", "ok": False, "reason": "not_whitelisted"}
+
+ # ── 2. Redis GET → 還原 payload ───────────────────────────────────────
+ short_id = callback_data[3:] # 去掉 "la:" 前綴
+ redis_key = f"tg:la:{short_id}"
+ payload: dict | None = None
+ try:
+ redis = get_redis()
+ raw = await redis.get(redis_key)
+ if raw:
+ payload = _json.loads(raw)
+ except Exception as exc:
+ # P1: Redis 故障與按鈕過期分開處理
+ logger.error("llm_action_redis_get_failed", short_id=short_id, error=str(exc))
+ await self._send_request("answerCallbackQuery", {
+ "callback_query_id": callback_query_id,
+ "text": "⚠️ 系統暫時不可用,請稍後重試",
+ "show_alert": True,
+ })
+ return {"action": "llm_action", "ok": False, "reason": "redis_unavailable"}
+
+ if payload is None:
+ await self._send_request("answerCallbackQuery", {
+ "callback_query_id": callback_query_id,
+ "text": "⏰ 此按鈕已過期,請重新觸發告警流程",
+ "show_alert": True,
+ })
+ logger.info("llm_action_button_expired", short_id=short_id)
+ return {"action": "llm_action", "ok": False, "reason": "button_expired"}
+
+ name: str = payload.get("name", "")
+ provider: str = payload.get("provider", "")
+ tool: str = payload.get("tool", "")
+ risk: str = payload.get("risk", "low")
+
+ # ── 3. 組裝 stub action + 呼叫 dispatch_llm_action ───────────────────
+ class _StubAction:
+ pass
+
+ stub = _StubAction()
+ stub.name = name # type: ignore[attr-defined]
+ stub.mcp_provider = provider # type: ignore[attr-defined]
+ stub.mcp_tool = tool # type: ignore[attr-defined]
+ stub.risk = risk # type: ignore[attr-defined]
+ stub.params = {} # type: ignore[attr-defined]
+
+ # P0 Fix: 從 Redis payload 取真實 incident_id,不用隨機 short_id
+ real_incident_id: str = payload.get("incident_id", "") or short_id
+ context = {"incident_id": real_incident_id, "confirmed": False}
+ result = dispatch_llm_action(stub, context)
+
+ # ── 4. high risk → 二次確認提示 ───────────────────────────────────────
+ if not result.get("ok") and result.get("reason") == "high_risk_requires_confirmation":
+ await self._send_request("answerCallbackQuery", {
+ "callback_query_id": callback_query_id,
+ "text": f"⚠️ 高風險操作:{name},請傳送指令確認後再執行",
+ "show_alert": True,
+ })
+ logger.info(
+ "llm_action_high_risk_pending",
+ name=name,
+ mcp_tool=tool,
+ user_id=user_id,
+ )
+ return {"action": "llm_action", "ok": False, "reason": "high_risk_requires_confirmation"}
+
+ # ── 5. dispatch 失敗(allowlist / critical 等) ───────────────────────
+ if not result.get("ok"):
+ reason = result.get("reason", "unknown")
+ await self._send_request("answerCallbackQuery", {
+ "callback_query_id": callback_query_id,
+ "text": f"❌ 無法執行:{reason}",
+ "show_alert": True,
+ })
+ logger.warning(
+ "llm_action_dispatch_rejected",
+ name=name,
+ mcp_tool=tool,
+ reason=reason,
+ )
+ return {"action": "llm_action", "ok": False, "reason": reason}
+
+ # ── 6. 允許執行 → answer_callback + 回報結果 ─────────────────────────
+ await self._send_request("answerCallbackQuery", {
+ "callback_query_id": callback_query_id,
+ "text": f"▶️ 執行中:{name}",
+ "show_alert": False,
+ })
+
+ logger.info(
+ "llm_action_executing",
+ name=name,
+ mcp_tool=tool,
+ mcp_provider=provider,
+ risk=risk,
+ user_id=user_id,
+ username=username,
+ )
+
+ # 回報執行結果到 Telegram(MCP 實際呼叫由外部整合,此處發送確認訊息)
+ import html as _html # noqa: PLC0415
+ result_text = (
+ f"✅ LLM 動作已觸發\n"
+ f"動作:{_html.escape(name)}\n"
+ f"工具:{_html.escape(provider)}/{_html.escape(tool)}\n"
+ f"風險:{_html.escape(risk)}\n"
+ f"操作者:@{_html.escape(str(username or user_id))}"
+ )
+ try:
+ await self._send_request("sendMessage", {
+ "chat_id": self.chat_id,
+ "text": result_text,
+ "parse_mode": "HTML",
+ })
+ except Exception as exc:
+ logger.warning("llm_action_result_notify_failed", error=str(exc))
+
+ return {
+ "action": "llm_action",
+ "ok": True,
+ "name": name,
+ "mcp_tool": tool,
+ "mcp_provider": provider,
+ "risk": risk,
+ "user": {"id": user_id, "username": username},
+ }
+
async def _answer_callback(
self,
callback_query_id: str,
diff --git a/apps/api/tests/test_telegram_gateway_llm_buttons.py b/apps/api/tests/test_telegram_gateway_llm_buttons.py
index 48dd433e..fdf5551c 100644
--- a/apps/api/tests/test_telegram_gateway_llm_buttons.py
+++ b/apps/api/tests/test_telegram_gateway_llm_buttons.py
@@ -2,24 +2,30 @@
B3: LLM 動態 Telegram 按鈕 — 單元測試
=====================================
2026-04-27 Claude Sonnet 4.6: ADR-082 B3
+2026-04-27 Claude Sonnet 4.6: H3+M6 Fix — short_id Redis 映射 + critical 過濾
+2026-04-27 Claude Sonnet 4.6: P0 Fix — async setex 在 return 前完成(消除 race)
測試範圍:
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"
+5. callback_data 格式是 la:{16-hex-chars},不是 JSON(H3 Fix,P0 後改 16-hex)
6. [批准][拒絕] 第一排永遠存在
7. 超過 2 個 action → 多排
+8. critical action 不出按鈕(M6 Fix)
+9. Redis short_id 被寫入(H3 Fix)
+
+注意:P0 Fix 後 _build_llm_action_buttons 改為 async(直接 await redis.setex)。
+所有 TestBuildLlmActionButtons 測試改為 @pytest.mark.asyncio。
"""
from __future__ import annotations
-import json
import os
from dataclasses import dataclass, field
from typing import Literal
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -115,88 +121,151 @@ def medium_action():
# =============================================================================
class TestBuildLlmActionButtons:
- """直接測試靜態方法,不需要完整 gateway 初始化"""
+ """
+ 測試 _build_llm_action_buttons 實例方法。
- def _get_builder(self):
+ H3 Fix 後 callback_data 格式改為 la:{8-hex-chars},不再是 JSON。
+ Redis 寫入透過 asyncio.ensure_future 背景執行,測試中 mock 掉。
+ """
+
+ def _make_gateway(self):
+ """建立最小 gateway 實例(不初始化 Redis / httpx)。"""
from src.services.telegram_gateway import TelegramGateway
- return TelegramGateway._build_llm_action_buttons
+ gw = object.__new__(TelegramGateway)
+ gw._security = MagicMock()
+ return gw
- def test_low_risk_text_format(self, low_action):
- builder = self._get_builder()
- rows = builder([low_action])
+ async def _call_builder(self, gw, actions):
+ """呼叫 _build_llm_action_buttons(async),mock 掉 Redis 寫入。"""
+ mock_redis = MagicMock()
+ mock_redis.setex = AsyncMock()
+ with patch("src.services.telegram_gateway.get_redis", return_value=mock_redis):
+ return await gw._build_llm_action_buttons(actions)
+
+ @pytest.mark.asyncio
+ async def test_low_risk_text_format(self, low_action):
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [low_action])
assert len(rows) == 1
btn = rows[0][0]
assert btn["text"] == "📋 查 Pod 日誌"
- def test_high_risk_prefix(self, high_action):
+ @pytest.mark.asyncio
+ async def test_high_risk_prefix(self, high_action):
"""Test 4: high risk 前綴 ⚠️"""
- builder = self._get_builder()
- rows = builder([high_action])
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [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])
+ @pytest.mark.asyncio
+ async def test_medium_risk_no_prefix(self, medium_action):
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [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])
+ @pytest.mark.asyncio
+ async def test_callback_data_format_is_short_id(self, low_action):
+ """Test 5 (H3 Fix + P0): callback_data 格式是 la:{16-hex-chars},不是 JSON。
+ P0 Fix 後 short_id 改為 secrets.token_hex(8) = 16-hex-chars,≤19 bytes。
+ """
+ import re
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [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
+ # 格式必須是 la: 開頭,後接 16 位 hex(P0 Fix 後從 8 → 16 hex chars)
+ assert re.fullmatch(r"la:[0-9a-f]{16}", cb), f"格式不符: {cb!r}"
+ # 確認不是 JSON
+ assert not cb.startswith("{")
- def test_callback_data_within_64_bytes(self, low_action):
- builder = self._get_builder()
- rows = builder([low_action])
+ @pytest.mark.asyncio
+ async def test_callback_data_within_19_bytes(self, low_action):
+ """H3 Fix + P0: la:{16-hex-chars} 固定 ≤19 bytes,絕不截斷。"""
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [low_action])
cb = rows[0][0]["callback_data"]
- assert len(cb.encode("utf-8")) <= 64
+ assert len(cb.encode("utf-8")) <= 19, f"callback_data 超出 19 bytes: {cb!r} ({len(cb.encode())} bytes)"
- def test_long_name_truncated_within_64_bytes(self):
- """callback_data 超過 64 bytes 時自動截斷 name"""
- action = _RecommendedAction(
- name="a" * 50,
- label="長名稱動作",
- emoji="🔧",
+ @pytest.mark.asyncio
+ async def test_critical_action_skipped(self):
+ """Test 8 (M6 Fix): critical risk action 不生成按鈕。"""
+ critical_action = _RecommendedAction(
+ name="nuke_cluster",
+ label="核爆叢集",
+ emoji="💣",
mcp_provider="k8s",
- mcp_tool="do_something",
+ mcp_tool="delete_all_pods",
params={},
- risk="low",
+ risk="critical",
)
- 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"
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [critical_action])
+ assert rows == [], f"critical action 不應生成按鈕,但得到 {rows}"
- 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])
+ @pytest.mark.asyncio
+ async def test_critical_mixed_with_others(self, low_action, medium_action):
+ """M6 Fix: critical 被過濾,其他 action 正常生成。"""
+ critical_action = _RecommendedAction(
+ name="nuke_cluster",
+ label="核爆",
+ emoji="💣",
+ mcp_provider="k8s",
+ mcp_tool="delete_all_pods",
+ params={},
+ risk="critical",
+ )
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [low_action, critical_action, medium_action])
+ # 只剩 low + medium 兩個,同排
assert len(rows) == 1
assert len(rows[0]) == 2
- def test_three_actions_two_rows(self, low_action, medium_action, high_action):
+ @pytest.mark.asyncio
+ async def test_redis_short_id_written(self, low_action):
+ """Test 9 (H3 Fix + P0): Redis tg:la:{short_id} 被 await setex 寫入(TTL=3600)。
+ P0 Fix 後改為直接 await redis.setex,不再使用 ensure_future。
+ """
+ mock_redis = MagicMock()
+ mock_redis.setex = AsyncMock()
+
+ gw = self._make_gateway()
+
+ with patch("src.services.telegram_gateway.get_redis", return_value=mock_redis):
+ rows = await gw._build_llm_action_buttons([low_action])
+
+ assert len(rows) == 1
+ cb = rows[0][0]["callback_data"]
+ short_id = cb[3:] # 去掉 "la:"
+
+ # P0 Fix: 直接 await,所以 setex 被呼叫一次
+ mock_redis.setex.assert_called_once()
+ call_args = mock_redis.setex.call_args
+ assert call_args[0][0] == f"tg:la:{short_id}"
+ assert call_args[0][1] == 3600
+
+ @pytest.mark.asyncio
+ async def test_two_actions_same_row(self, low_action, medium_action):
+ """Test 7 前置:2 個 action → 同一排"""
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [low_action, medium_action])
+ assert len(rows) == 1
+ assert len(rows[0]) == 2
+
+ @pytest.mark.asyncio
+ async 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])
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [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([])
+ @pytest.mark.asyncio
+ async def test_empty_actions_returns_empty(self):
+ gw = self._make_gateway()
+ rows = await self._call_builder(gw, [])
assert rows == []
@@ -227,106 +296,83 @@ class TestBuildInlineKeyboardRouting:
def _first_row_texts(self, keyboard: dict) -> list[str]:
return [btn["text"] for btn in keyboard["inline_keyboard"][0]]
+ async def _build_kb(self, gw, **kwargs) -> dict:
+ """await _build_inline_keyboard,mock 掉 Redis。"""
+ mock_redis = MagicMock()
+ mock_redis.setex = AsyncMock()
+ with patch("src.services.telegram_gateway.get_redis", return_value=mock_redis):
+ return await gw._build_inline_keyboard(**kwargs)
+
# Test 1: flag=false → YAML 路徑(第一排還是批准/拒絕)
+ @pytest.mark.asyncio
@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):
+ async 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,
- )
+ kb = await self._build_kb(gw, 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)
+ # LLM 按鈕不應出現(la: 前綴代表 llm_action)
+ all_cbs = [btn["callback_data"] for row in kb["inline_keyboard"] for btn in row]
+ assert not any(cb.startswith("la:") for cb in all_cbs)
# Test 2: flag=true + actions 空 → YAML fallback
+ @pytest.mark.asyncio
@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):
+ async 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,
- )
+ kb = await self._build_kb(gw, 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)
+ all_cbs = [btn["callback_data"] for row in kb["inline_keyboard"] for btn in row]
+ assert not any(cb.startswith("la:") for cb in all_cbs)
# Test 2b: flag=true + action_plan=None → YAML fallback
+ @pytest.mark.asyncio
@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):
+ async 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",
- )
+ kb = await self._build_kb(gw, 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 路徑
+ @pytest.mark.asyncio
@patch("src.services.telegram_gateway.USE_LLM_DYNAMIC_BUTTONS", True)
- def test_flag_true_with_actions_uses_llm(self, low_action):
+ async def test_flag_true_with_actions_uses_llm(self, low_action):
+ import re
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,
- )
+ kb = await self._build_kb(gw, 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)
+ # LLM 按鈕在第二排以後(P0 Fix 後 16-hex-chars)
+ all_cbs = [btn["callback_data"] for row in kb["inline_keyboard"][1:] for btn in row]
+ assert any(re.fullmatch(r"la:[0-9a-f]{16}", cb) for cb in all_cbs), (
+ f"LLM 按鈕 callback_data 格式應為 la:{{16-hex}},實際: {all_cbs}"
+ )
# button text 正確
- all_texts = [
- btn["text"]
- for row in kb["inline_keyboard"][1:]
- for btn in row
- ]
+ 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 路徑)
+ @pytest.mark.asyncio
@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):
+ async 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,
- )
+ kb = await self._build_kb(gw, 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] == "❌ 拒絕"