test(alerts): 對齊心跳告警 SRE 群組契約

apps/api/tests/test_heartbeat_dedup_p0_4.py

@@ -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")