ea23972f7a
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 9m10s
ADR-082 §B2：dispatch_llm_action() 風險閘控 + allowlist + 模板渲染 23 tests pass Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
335 lines
12 KiB
Python
335 lines
12 KiB
Python
"""
|
||
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
|