test(alerts): 對齊心跳告警 SRE 群組契約
All checks were successful
CD Pipeline / tests (push) Successful in 1m37s
Code Review / ai-code-review (push) Successful in 14s
CD Pipeline / build-and-deploy (push) Successful in 4m0s
CD Pipeline / post-deploy-checks (push) Successful in 1m39s

This commit is contained in:
Your Name
2026-06-12 11:11:00 +08:00
parent ee2cc2bfc3
commit 1276149114
2 changed files with 43 additions and 11 deletions

View File

@@ -61,6 +61,14 @@ def _make_report(warnings: list[str] | None = None):
)
@pytest.fixture
def sre_group_configured(monkeypatch):
"""Heartbeat 正式推送只能在 AwoooI SRE 戰情室設定存在時成立。"""
from src.services.telegram_gateway import settings
monkeypatch.setattr(settings, "SRE_GROUP_CHAT_ID", "-1003711974679")
@pytest.fixture
def gateway_with_fake_redis():
"""構造 telegram gateway 實例 + 注入 fake redis"""
@@ -79,7 +87,11 @@ class TestHeartbeatDedup:
"""P0 #4 heartbeat 降頻邏輯"""
@pytest.mark.asyncio
async def test_healthy_first_send_goes_through(self, gateway_with_fake_redis):
async def test_healthy_first_send_goes_through(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""健康狀態第一次推送(無 silent marker→ 推送"""
gw, fake_redis = gateway_with_fake_redis
@@ -93,11 +105,15 @@ class TestHeartbeatDedup:
assert result is True
assert "heartbeat:silent_last_sent" in fake_redis._store
# 應該有呼叫 send_to_group 或 send_notification其一
assert gw.send_to_group.called or gw.send_notification.called
gw.send_to_group.assert_called_once()
gw.send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_healthy_second_send_within_6h_skipped(self, gateway_with_fake_redis):
async def test_healthy_second_send_within_6h_skipped(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""健康狀態 6h 內第二次推送 → 跳過"""
gw, fake_redis = gateway_with_fake_redis
fake_redis.preset("heartbeat:silent_last_sent") # 模擬已有 silent marker
@@ -116,7 +132,11 @@ class TestHeartbeatDedup:
gw.send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_warnings_unchanged_skipped(self, gateway_with_fake_redis):
async def test_warnings_unchanged_skipped(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""有 warnings 跟上次同 hash → 跳過"""
gw, fake_redis = gateway_with_fake_redis
warnings = ["Pod api-x Failed", "Redis: down"]
@@ -139,7 +159,11 @@ class TestHeartbeatDedup:
gw.send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_warnings_changed_pushes(self, gateway_with_fake_redis):
async def test_warnings_changed_pushes(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""有 warnings 但跟上次不同 → 立即推送"""
gw, fake_redis = gateway_with_fake_redis
# 預設舊的 hash跟新 warnings 不同)
@@ -156,11 +180,15 @@ class TestHeartbeatDedup:
result = await gw.send_heartbeat()
assert result is True
# 應該推送
assert gw.send_to_group.called or gw.send_notification.called
gw.send_to_group.assert_called_once()
gw.send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_warnings_to_healthy_clears_warnings_hash(self, gateway_with_fake_redis):
async def test_warnings_to_healthy_clears_warnings_hash(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""從有事 → 健康:清掉 warnings_hash marker下次有事可立即推"""
gw, fake_redis = gateway_with_fake_redis
fake_redis.preset("heartbeat:warnings_hash", "old1234567890")
@@ -177,7 +205,11 @@ class TestHeartbeatDedup:
assert "heartbeat:silent_last_sent" in fake_redis._store
@pytest.mark.asyncio
async def test_healthy_to_warnings_clears_silent_marker(self, gateway_with_fake_redis):
async def test_healthy_to_warnings_clears_silent_marker(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""從健康 → 有事:清掉 silent marker下次靜默過 6h 才再推"""
gw, fake_redis = gateway_with_fake_redis
fake_redis.preset("heartbeat:silent_last_sent")

View File

@@ -17,7 +17,7 @@
- 路由殘留掃描:`.gitea` / `apps/api/src` / `apps/api/tests` / `scripts/ops` / `k8s/awoooi-prod` 未命中舊 `TELEGRAM_ALERT_CHAT_ID`、舊 `TELEGRAM_CHAT_ID`、SRE/OpenClaw chat fallback 混用、個人 fallback 或 direct OpenClaw bot sendMessage。
- `python3.11 -m py_compile`Telegram gateway、notification matrix、Telegram provider、recurrence notifier、failover alerter、post verifier、rate limiter、approval execution、Telegram API 與相關 jobs 通過。
- `bash -n`:相關 ops scripts 與 CI notify script 通過。
- `DATABASE_URL='postgresql+asyncpg://test:test@localhost/test' PYTHONPATH=. python3.11 -m pytest -q tests/test_notification_matrix_group_cutover.py tests/test_alert_converged_recurrence.py tests/test_failover_alerter.py tests/test_telegram_button_consistency.py``39 passed`
- `DATABASE_URL='postgresql+asyncpg://test:test@localhost/test' PYTHONPATH=. python3.11 -m pytest -q tests/test_heartbeat_dedup_p0_4.py tests/test_notification_matrix_group_cutover.py tests/test_alert_converged_recurrence.py tests/test_failover_alerter.py tests/test_telegram_button_consistency.py``45 passed`heartbeat 測試同步改成必須設定 `SRE_GROUP_CHAT_ID` 並只允許 `send_to_group`,不得用個人 / 舊群組 fallback
**完成度同步**