d89f0520f9
- 自動修復 import 排序、unused imports - 手動修復 raise from、isinstance union、unused variable - scripts/ 暫時保留 (非 CI 阻擋) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
321 lines
10 KiB
Python
321 lines
10 KiB
Python
"""
|
||
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 "📊 <b>SignOz 指標</b>" in result
|
||
assert "RPS: <code>150.2</code> 📈" in result
|
||
assert "🟢" in result # error_rate < 1%
|
||
assert "P99: <code>245ms</code>" 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
|