"""
P1-4: Telegram 整合驗證測試
============================
驗證 SignOz + Sentry + GitHub Webhook 的 Telegram 訊息格式
測試策略 (遵循 feedback_no_mock_testing.md):
- 直接測試訊息格式化邏輯
- 驗證 dataclass 結構正確性
- 不發送實際 Telegram 訊息
版本: v1.0
建立: 2026-03-29 (台北時區)
建立者: Claude Code (P1-4 Telegram 驗證)
"""
from src.services.telegram_gateway import (
RISK_EMOJI_MAP,
SignOzMetricsBlock,
TelegramMessage,
)
# 本地定義 (避免 import 錯誤)
RESPONSIBILITY_MAP = {
"FE": "前端",
"BE": "後端",
"INFRA": "基礎設施",
"DB": "資料庫",
"COLLAB": "協作",
}
# =============================================================================
# Test: SignOzMetricsBlock 格式化
# =============================================================================
class TestSignOzMetricsBlock:
"""測試 SignOz 指標區塊格式"""
def test_metrics_block_format_basic(self):
"""✅ 基本指標格式化"""
block = SignOzMetricsBlock(
rps=150.2,
rps_trend="up",
error_rate=0.5,
p99_latency_ms=245,
latency_trend="stable",
)
result = block.format()
assert "📊 SignOz 指標" in result
assert "RPS: 150.2 📈" in result
assert "🟢" in result # error_rate < 1%
assert "P99: 245ms" in result
def test_metrics_block_error_emoji_green(self):
"""✅ Error < 1% 顯示綠燈"""
block = SignOzMetricsBlock(error_rate=0.5)
result = block.format()
assert "🟢" in result
def test_metrics_block_error_emoji_yellow(self):
"""✅ Error 1-5% 顯示黃燈"""
block = SignOzMetricsBlock(error_rate=3.0)
result = block.format()
assert "🟡" in result
def test_metrics_block_error_emoji_red(self):
"""✅ Error >= 5% 顯示紅燈"""
block = SignOzMetricsBlock(error_rate=10.0)
result = block.format()
assert "🔴" in result
def test_metrics_trend_emojis(self):
"""✅ 趨勢 Emoji 正確對應"""
# Up trend
block_up = SignOzMetricsBlock(rps=100, rps_trend="up")
assert "📈" in block_up.format()
# Down trend
block_down = SignOzMetricsBlock(rps=100, rps_trend="down")
assert "📉" in block_down.format()
# Stable
block_stable = SignOzMetricsBlock(rps=100, rps_trend="stable")
assert "➡️" in block_stable.format()
# =============================================================================
# Test: TelegramMessage 結構
# =============================================================================
class TestTelegramMessageStructure:
"""測試 Telegram 訊息結構"""
def test_telegram_message_required_fields(self):
"""✅ 必填欄位存在"""
msg = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="api-service",
root_cause="OOM Killed",
suggested_action="Restart Pod",
estimated_downtime="~30s",
approval_id="test-123",
primary_responsibility="BE",
confidence=0.88,
namespace="awoooi-prod",
)
assert msg.status_emoji == "🚨"
assert msg.risk_level == "CRITICAL"
assert msg.resource_name == "api-service"
assert msg.root_cause == "OOM Killed"
assert msg.approval_id == "test-123"
def test_telegram_message_format_basic(self):
"""✅ 基本訊息格式化"""
msg = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="api-service-7d4b8c9f5",
root_cause="JVM Heap 配置不當",
suggested_action="刪除 Pod",
estimated_downtime="~30s",
approval_id="INC-20260321-0001",
primary_responsibility="BE",
confidence=0.88,
namespace="awoooi-prod",
)
result = msg.format()
# 驗證關鍵區塊存在
assert "CRITICAL" in result
assert "api-service" in result
assert "INC-20260321-0001" in result
assert "BE" in result or "後端" in result
assert "88%" in result
def test_telegram_message_with_signoz_metrics(self):
"""✅ 含 SignOz 指標的訊息格式"""
signoz = SignOzMetricsBlock(
rps=150.2,
error_rate=0.5,
p99_latency_ms=245,
)
msg = TelegramMessage(
status_emoji="⚠️",
risk_level="MEDIUM",
resource_name="web-service",
root_cause="High latency",
suggested_action="Scale up",
estimated_downtime="~1min",
approval_id="test-456",
primary_responsibility="INFRA",
confidence=0.75,
namespace="awoooi-prod",
signoz_metrics=signoz,
)
result = msg.format()
# 驗證 SignOz 區塊存在
assert "SignOz" in result
assert "RPS" in result
def test_telegram_message_with_anomaly_frequency(self):
"""✅ 含異常頻率統計的訊息格式 (ADR-037)"""
msg = TelegramMessage(
status_emoji="🚨",
risk_level="CRITICAL",
resource_name="api-service",
root_cause="Repeated failure",
suggested_action="Permanent fix required",
estimated_downtime="~30s",
approval_id="test-789",
primary_responsibility="BE",
confidence=0.95,
namespace="awoooi-prod",
anomaly_frequency={
"count_24h": 15,
"count_7d": 45,
"escalation_level": "ESCALATE",
},
)
result = msg.format()
# 驗證異常頻率區塊存在
assert "頻率" in result or "frequency" in result.lower() or "15" in result
# =============================================================================
# Test: 風險等級 Emoji 映射
# =============================================================================
class TestRiskEmojiMapping:
"""測試風險等級 Emoji 映射"""
def test_risk_emoji_critical(self):
"""✅ CRITICAL → 🚨"""
assert RISK_EMOJI_MAP.get("critical") == "🚨"
def test_risk_emoji_high(self):
"""✅ HIGH → 🔴"""
assert RISK_EMOJI_MAP.get("high") == "🔴"
def test_risk_emoji_medium(self):
"""✅ MEDIUM → ⚠️"""
assert RISK_EMOJI_MAP.get("medium") == "⚠️"
def test_risk_emoji_low(self):
"""✅ LOW → ℹ️"""
assert RISK_EMOJI_MAP.get("low") == "ℹ️"
# =============================================================================
# Test: 責任團隊映射
# =============================================================================
class TestResponsibilityMapping:
"""測試責任團隊映射"""
def test_responsibility_fe(self):
"""✅ FE → 前端"""
assert RESPONSIBILITY_MAP.get("FE") == "前端"
def test_responsibility_be(self):
"""✅ BE → 後端"""
assert RESPONSIBILITY_MAP.get("BE") == "後端"
def test_responsibility_infra(self):
"""✅ INFRA → 基礎設施"""
assert RESPONSIBILITY_MAP.get("INFRA") == "基礎設施"
# =============================================================================
# Test: Webhook 到 Telegram 整合驗證
# =============================================================================
class TestWebhookTelegramIntegration:
"""測試 Webhook 到 Telegram 的訊息流程"""
def test_signoz_webhook_telegram_format(self):
"""✅ SignOz Webhook 應產生正確的 Telegram 訊息參數"""
# 模擬 SignOz Webhook 的 Telegram 呼叫參數
alert_name = "HighErrorRate"
service_name = "api-service"
severity = "error"
# 驗證 risk_level 映射
risk_mapping = {
"critical": "critical",
"error": "high",
"warning": "medium",
"info": "low",
}
expected_risk = risk_mapping.get(severity, "medium")
msg = TelegramMessage(
status_emoji=RISK_EMOJI_MAP.get(expected_risk, "⚠️"),
risk_level=expected_risk.upper(),
resource_name=service_name,
root_cause=f"SignOz Alert: {alert_name}",
suggested_action="請檢查 SignOz 儀表板",
estimated_downtime="0",
approval_id="signoz-test-001",
primary_responsibility="BE",
confidence=0.7,
namespace="signoz",
)
result = msg.format()
assert "HIGH" in result
assert service_name in result
assert alert_name in result
def test_sentry_webhook_telegram_format(self):
"""✅ Sentry Webhook 應產生正確的 Telegram 訊息參數"""
level = "error"
project = "awoooi-api"
culprit = "api.views.handler"
msg = TelegramMessage(
status_emoji="🔶",
risk_level="HIGH",
resource_name=project,
root_cause=f"Sentry {level.upper()}: {culprit}",
suggested_action="檢查 Sentry Issue 詳情",
estimated_downtime="0",
approval_id="sentry-test-001",
primary_responsibility="BE",
confidence=0.8,
namespace="sentry",
)
result = msg.format()
assert "HIGH" in result
assert project in result
assert culprit in result
def test_github_webhook_telegram_format(self):
"""✅ GitHub Webhook 應產生正確的 Telegram 訊息參數"""
repo = "wooo-ai/awoooi"
msg = TelegramMessage(
status_emoji="⚠️",
risk_level="MEDIUM",
resource_name=repo,
root_cause="CI Failure: Build failed",
suggested_action="npm install missing-package",
estimated_downtime="~5min",
approval_id="github-test-001",
primary_responsibility="BE",
confidence=0.85,
namespace="github-actions",
)
result = msg.format()
assert "MEDIUM" in result
assert repo in result
assert "CI Failure" in result