diff --git a/backend/app/services/webhook_adapter.py b/backend/app/services/webhook_adapter.py index 8d6fa7b..ca21139 100644 --- a/backend/app/services/webhook_adapter.py +++ b/backend/app/services/webhook_adapter.py @@ -141,10 +141,10 @@ def _post_feishu(webhook_url: str, payload: dict, secret: str, max_attempts: int except Exception as e: # noqa: BLE001 — 网络/超时, 可重试 last_err = str(e) - if attempt < _FEISHU_MAX_ATTEMPTS: + if attempt < max_attempts: time.sleep(min(2 ** (attempt - 1), 3)) # 退避: 1s, 2s - logger.warning("飞书 Webhook 推送最终失败(已重试 %d 次): %s", _FEISHU_MAX_ATTEMPTS, last_err) + logger.warning("飞书 Webhook 推送最终失败(已重试 %d 次): %s", max_attempts, last_err) return False diff --git a/backend/tests/test_notification_adapters.py b/backend/tests/test_notification_adapters.py index 7f08507..44e0ba4 100644 --- a/backend/tests/test_notification_adapters.py +++ b/backend/tests/test_notification_adapters.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import hmac import json +import logging from app.services import email_adapter, webhook_adapter @@ -98,3 +99,42 @@ def test_email_adapter_uses_starttls_login_and_multiple_recipients(monkeypatch): assert message["To"] == "one@example.com, two@example.com" assert message["Subject"] == "监控告警" assert smtp.calls[-1] == ("quit",) + + +def _unreachable(*_args, **_kwargs): + raise ConnectionError("unreachable") + + +def test_feishu_single_attempt_returns_without_backoff_and_logs_the_real_count(monkeypatch, caplog): + # 设置页「发送测试消息」传 max_attempts=1: 失败即返回, 不等退避, 日志计数如实。 + sleeps: list[float] = [] + monkeypatch.setattr(webhook_adapter.time, "sleep", lambda seconds: sleeps.append(seconds)) + monkeypatch.setattr("httpx.post", _unreachable) + with caplog.at_level(logging.WARNING, logger=webhook_adapter.__name__): + ok = webhook_adapter.send_feishu( + "https://open.feishu.cn/open-apis/bot/v2/hook/abc", "标题", "正文", max_attempts=1 + ) + assert ok is False + assert sleeps == [] + assert "已重试 1 次" in caplog.text + + +def test_feishu_production_retries_keep_backoff(monkeypatch, caplog): + # 生产路径 (默认 3 次) 的退避与日志不变: 1s、2s 两次退避, 日志写 3 次。 + sleeps: list[float] = [] + calls = {"n": 0} + + def unreachable(*_args, **_kwargs): + calls["n"] += 1 + raise ConnectionError("unreachable") + + monkeypatch.setattr(webhook_adapter.time, "sleep", lambda seconds: sleeps.append(seconds)) + monkeypatch.setattr("httpx.post", unreachable) + with caplog.at_level(logging.WARNING, logger=webhook_adapter.__name__): + ok = webhook_adapter.send_feishu( + "https://open.feishu.cn/open-apis/bot/v2/hook/abc", "标题", "正文" + ) + assert ok is False + assert calls["n"] == 3 + assert sleeps == [1, 2] + assert "已重试 3 次" in caplog.text