test: 移除 Telegram Webhook Mock 測試
全面禁止 Mock 測試鐵律： - 移除 test_webhook_telegram_integration.py (323 lines of Mock) - 整合測試必須使用真實資料庫與服務 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
@@ -1,322 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Webhook → Telegram 全鏈路整合測試
|
||||
==================================
|
||||
Phase 5: 修復一級整合事故
|
||||
|
||||
測試涵蓋:
|
||||
1. 新告警 → 自動推送 Telegram
|
||||
2. 收斂告警 → 也必須推送 Telegram (含聚合次數)
|
||||
3. 斷言 TelegramGateway.send_approval_card 被正確參數呼叫
|
||||
4. 驗證 SOUL.md 格式資料完整性
|
||||
|
||||
使用方式:
|
||||
cd apps/api && pytest tests/test_webhook_telegram_integration.py -v
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from src.core.config import settings
|
||||
from src.main import app
|
||||
|
||||
# =============================================================================
|
||||
# Test Fixtures
|
||||
# =============================================================================
|
||||
|
||||
@pytest.fixture
|
||||
def valid_alert_payload():
|
||||
"""有效的告警 Payload"""
|
||||
return {
|
||||
"alert_type": "k8s_pod_crash",
|
||||
"severity": "critical",
|
||||
"source": "prometheus",
|
||||
"target_resource": "harbor-core-7d4b8c9f5-xk2m3",
|
||||
"namespace": "harbor",
|
||||
"message": "Pod terminated due to OOMKilled",
|
||||
"metrics": {"memory_percent": 99.8, "restart_count": 5},
|
||||
"labels": {"app": "harbor-core", "reason": "OOMKilled"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_approval_service():
|
||||
"""Mock ApprovalService"""
|
||||
mock_service = AsyncMock()
|
||||
|
||||
# Mock find_by_fingerprint 回傳 None (新告警)
|
||||
mock_service.find_by_fingerprint.return_value = None
|
||||
|
||||
# Mock create_approval_with_fingerprint 回傳模擬的 Approval
|
||||
mock_approval = MagicMock()
|
||||
mock_approval.id = UUID("12345678-1234-5678-1234-567812345678")
|
||||
mock_approval.status.value = "pending"
|
||||
mock_approval.risk_level.value = "critical"
|
||||
mock_approval.action = "kubectl delete pod harbor-core-7d4b8c9f5-xk2m3 -n harbor"
|
||||
mock_approval.hit_count = 1
|
||||
mock_service.create_approval_with_fingerprint.return_value = mock_approval
|
||||
|
||||
return mock_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_converged_approval_service():
|
||||
"""Mock ApprovalService - 收斂情境"""
|
||||
mock_service = AsyncMock()
|
||||
|
||||
# Mock find_by_fingerprint 回傳現有的 Approval (收斂)
|
||||
existing_approval = MagicMock()
|
||||
existing_approval.id = UUID("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
|
||||
existing_approval.hit_count = 5
|
||||
existing_approval.risk_level.value = "critical"
|
||||
existing_approval.action = "kubectl delete pod harbor-core -n harbor"
|
||||
mock_service.find_by_fingerprint.return_value = existing_approval
|
||||
|
||||
# Mock increment_hit_count
|
||||
updated_approval = MagicMock()
|
||||
updated_approval.id = existing_approval.id
|
||||
updated_approval.hit_count = 6 # 聚合後 +1
|
||||
updated_approval.risk_level.value = "critical"
|
||||
updated_approval.action = "kubectl delete pod harbor-core -n harbor"
|
||||
mock_service.increment_hit_count.return_value = updated_approval
|
||||
|
||||
return mock_service
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Test: 新告警 → Telegram 推送
|
||||
# =============================================================================
|
||||
|
||||
class TestNewAlertTelegramPush:
|
||||
"""新告警必須推送到 Telegram"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_new_alert_triggers_telegram_push(
|
||||
self,
|
||||
valid_alert_payload: dict,
|
||||
mock_approval_service,
|
||||
):
|
||||
"""
|
||||
[核心斷言] 新告警建立 ApprovalRecord 後，
|
||||
必須呼叫 TelegramGateway.send_approval_card()
|
||||
"""
|
||||
mock_telegram_gateway = AsyncMock()
|
||||
mock_telegram_gateway.send_approval_card = AsyncMock(return_value={"ok": True})
|
||||
|
||||
with patch("src.api.v1.webhooks.get_approval_service", return_value=mock_approval_service):
|
||||
with patch("src.api.v1.webhooks.get_openclaw") as mock_openclaw:
|
||||
# Mock OpenClaw 回傳 None (使用靜態分析)
|
||||
mock_openclaw.return_value.analyze_alert = AsyncMock(
|
||||
return_value=(None, "mock", "", None, "")
|
||||
)
|
||||
|
||||
with patch("src.api.v1.webhooks.get_telegram_gateway", return_value=mock_telegram_gateway):
|
||||
with patch.object(settings, "OPENCLAW_TG_BOT_TOKEN", "test-token"):
|
||||
with patch.object(settings, "WEBHOOK_HMAC_SECRET", ""):
|
||||
with patch.object(settings, "ENVIRONMENT", "dev"):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/alerts",
|
||||
json=valid_alert_payload,
|
||||
)
|
||||
|
||||
# 驗證 HTTP 回應
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["approval_created"] is True
|
||||
|
||||
# =====================================================================
|
||||
# [核心斷言] TelegramGateway.send_approval_card 必須被呼叫
|
||||
# =====================================================================
|
||||
# 因為使用 BackgroundTasks，需要等待一下
|
||||
import asyncio
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
mock_telegram_gateway.send_approval_card.assert_called_once()
|
||||
|
||||
# 驗證呼叫參數符合 SOUL.md 格式
|
||||
call_kwargs = mock_telegram_gateway.send_approval_card.call_args.kwargs
|
||||
assert "approval_id" in call_kwargs
|
||||
assert call_kwargs["approval_id"] == "12345678-1234-5678-1234-567812345678"
|
||||
assert "risk_level" in call_kwargs
|
||||
assert "resource_name" in call_kwargs
|
||||
assert call_kwargs["resource_name"] == "harbor-core-7d4b8c9f5-xk2m3"
|
||||
assert "root_cause" in call_kwargs
|
||||
assert "suggested_action" in call_kwargs
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Test: 收斂告警 → Telegram 推送 (含聚合次數)
|
||||
# =============================================================================
|
||||
|
||||
class TestConvergedAlertTelegramPush:
|
||||
"""收斂告警也必須推送到 Telegram"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_converged_alert_also_triggers_telegram_push(
|
||||
self,
|
||||
valid_alert_payload: dict,
|
||||
mock_converged_approval_service,
|
||||
):
|
||||
"""
|
||||
[核心斷言] 收斂告警 (相同指紋) 聚合後，
|
||||
也必須推送 Telegram，並包含聚合次數
|
||||
"""
|
||||
mock_telegram_gateway = AsyncMock()
|
||||
mock_telegram_gateway.send_approval_card = AsyncMock(return_value={"ok": True})
|
||||
|
||||
with patch("src.api.v1.webhooks.get_approval_service", return_value=mock_converged_approval_service):
|
||||
with patch("src.api.v1.webhooks.get_telegram_gateway", return_value=mock_telegram_gateway):
|
||||
with patch.object(settings, "OPENCLAW_TG_BOT_TOKEN", "test-token"):
|
||||
with patch.object(settings, "WEBHOOK_HMAC_SECRET", ""):
|
||||
with patch.object(settings, "ENVIRONMENT", "dev"):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/alerts",
|
||||
json=valid_alert_payload,
|
||||
)
|
||||
|
||||
# 驗證 HTTP 回應
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["converged"] is True
|
||||
assert data["hit_count"] == 6 # 5 + 1
|
||||
|
||||
# =====================================================================
|
||||
# [核心斷言] 收斂告警也必須呼叫 TelegramGateway
|
||||
# =====================================================================
|
||||
import asyncio
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
mock_telegram_gateway.send_approval_card.assert_called_once()
|
||||
|
||||
# 驗證聚合次數被嵌入 root_cause 字串
|
||||
call_kwargs = mock_telegram_gateway.send_approval_card.call_args.kwargs
|
||||
assert "[x6]" in call_kwargs["root_cause"], \
|
||||
f"hit_count should be embedded in root_cause, got: {call_kwargs['root_cause']}"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Test: Telegram 推送失敗不影響主流程
|
||||
# =============================================================================
|
||||
|
||||
class TestTelegramPushFailureIsolation:
|
||||
"""Telegram 推送失敗不應影響 Webhook 回應"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_telegram_failure_does_not_break_webhook(
|
||||
self,
|
||||
valid_alert_payload: dict,
|
||||
mock_approval_service,
|
||||
):
|
||||
"""
|
||||
[防禦性] Telegram API 錯誤時，Webhook 仍應回傳 200
|
||||
"""
|
||||
mock_telegram_gateway = AsyncMock()
|
||||
# 模擬 Telegram API 失敗
|
||||
mock_telegram_gateway.send_approval_card = AsyncMock(
|
||||
side_effect=Exception("Telegram API timeout")
|
||||
)
|
||||
|
||||
with patch("src.api.v1.webhooks.get_approval_service", return_value=mock_approval_service):
|
||||
with patch("src.api.v1.webhooks.get_openclaw") as mock_openclaw:
|
||||
mock_openclaw.return_value.analyze_alert = AsyncMock(
|
||||
return_value=(None, "mock", "", None, "")
|
||||
)
|
||||
with patch("src.api.v1.webhooks.get_telegram_gateway", return_value=mock_telegram_gateway):
|
||||
with patch.object(settings, "OPENCLAW_TG_BOT_TOKEN", "test-token"):
|
||||
with patch.object(settings, "WEBHOOK_HMAC_SECRET", ""):
|
||||
with patch.object(settings, "ENVIRONMENT", "dev"):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/alerts",
|
||||
json=valid_alert_payload,
|
||||
)
|
||||
|
||||
# =====================================================================
|
||||
# [核心斷言] 即使 Telegram 失敗，Webhook 仍回傳 200
|
||||
# =====================================================================
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["approval_created"] is True
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Test: SOUL.md 格式驗證
|
||||
# =============================================================================
|
||||
|
||||
class TestSOULMDFormatCompliance:
|
||||
"""驗證推送資料符合 SOUL.md 格式規範"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_telegram_payload_respects_soul_md_limits(
|
||||
self,
|
||||
mock_approval_service,
|
||||
):
|
||||
"""
|
||||
[SOUL.md] 驗證字數限制:
|
||||
- resource_name: 50 字元
|
||||
- root_cause: 100 字元
|
||||
- suggested_action: 50 字元
|
||||
"""
|
||||
# 超長資料
|
||||
long_alert_payload = {
|
||||
"alert_type": "k8s_pod_crash",
|
||||
"severity": "critical",
|
||||
"source": "prometheus",
|
||||
"target_resource": "x" * 100, # 超過 50 字元
|
||||
"namespace": "default",
|
||||
"message": "y" * 200, # 超過 100 字元
|
||||
"metrics": {},
|
||||
}
|
||||
|
||||
mock_telegram_gateway = AsyncMock()
|
||||
mock_telegram_gateway.send_approval_card = AsyncMock(return_value={"ok": True})
|
||||
|
||||
with patch("src.api.v1.webhooks.get_approval_service", return_value=mock_approval_service):
|
||||
with patch("src.api.v1.webhooks.get_openclaw") as mock_openclaw:
|
||||
mock_openclaw.return_value.analyze_alert = AsyncMock(
|
||||
return_value=(None, "mock", "", None, "")
|
||||
)
|
||||
with patch("src.api.v1.webhooks.get_telegram_gateway", return_value=mock_telegram_gateway):
|
||||
with patch.object(settings, "OPENCLAW_TG_BOT_TOKEN", "test-token"):
|
||||
with patch.object(settings, "WEBHOOK_HMAC_SECRET", ""):
|
||||
with patch.object(settings, "ENVIRONMENT", "dev"):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.post(
|
||||
"/api/v1/webhooks/alerts",
|
||||
json=long_alert_payload,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
import asyncio
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# 驗證呼叫參數已被截斷
|
||||
call_kwargs = mock_telegram_gateway.send_approval_card.call_args.kwargs
|
||||
assert len(call_kwargs["resource_name"]) <= 50
|
||||
assert len(call_kwargs["root_cause"]) <= 100
|
||||
assert len(call_kwargs["suggested_action"]) <= 50
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "--tb=short"])
|
||||
Reference in New Issue
Block a user