diff --git a/apps/api/tests/test_heartbeat_dedup_p0_4.py b/apps/api/tests/test_heartbeat_dedup_p0_4.py index ceee594a..e2b223c9 100644 --- a/apps/api/tests/test_heartbeat_dedup_p0_4.py +++ b/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") diff --git a/docs/LOGBOOK.md b/docs/LOGBOOK.md index 1a39ae18..8638bd2e 100644 --- a/docs/LOGBOOK.md +++ b/docs/LOGBOOK.md @@ -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。 **完成度同步**: