From fea537b9d725b774a653d2d44a6d44033ec11525 Mon Sep 17 00:00:00 2001 From: v6ole Date: Tue, 4 Aug 2026 15:08:06 +0800 Subject: [PATCH] feat: deliver alert events through persistent outbox --- backend/app/services/alerter.py | 407 +++++------------ .../app/services/notification_dispatcher.py | 322 +++++++++++++ backend/app/services/pinger.py | 19 +- backend/app/services/scheduler.py | 164 ++++--- backend/tests/test_alert_workflow.py | 427 ++++++++++++++++++ backend/tests/test_notification_dispatcher.py | 327 ++++++++++++++ 6 files changed, 1297 insertions(+), 369 deletions(-) create mode 100644 backend/app/services/notification_dispatcher.py create mode 100644 backend/tests/test_alert_workflow.py create mode 100644 backend/tests/test_notification_dispatcher.py diff --git a/backend/app/services/alerter.py b/backend/app/services/alerter.py index 8cbe7d1..72b719c 100644 --- a/backend/app/services/alerter.py +++ b/backend/app/services/alerter.py @@ -1,319 +1,136 @@ -""" -告警服务 +"""Create durable alert events and their notification outbox entries.""" -职责： -1. 接收 Pinger 的状态变化事件 -2. 判断是否需要发送告警 -3. 上游心跳检测 + 全量离线抑制 -4. 企业微信消息推送（聚合告警） -5. 记录告警事件到数据库 -""" - -import asyncio -import logging from datetime import datetime -from typing import Optional -from collections import defaultdict -import httpx -from sqlalchemy import select, func +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from app.config import settings -from app.models.device import Device -from app.models.alert_event import AlertEvent, AlertTypeEnum +from app.config import Settings, settings +from app.models import ( + AlertEvent, + AlertTypeEnum, + NotificationOutbox, +) from app.services.pinger import DeviceStateChange -logger = logging.getLogger("pingwatch.alerter") - - -class PendingOfflineAlert: - """等待发送的离线告警（用于聚合）""" - def __init__(self, device: Device, alert_time: datetime): - self.device = device - self.alert_time = alert_time - class Alerter: - """ - 告警处理器。 + """Translate a persisted device transition into one atomic outbox event.""" - 核心逻辑： - - 设备 offline → 收集到待发送队列 - - 每轮结束后检查： - a) 上游心跳是否正常？ - b) 离线率是否 < 90%？ - c) 满足条件 → 聚合所有待发告警 → 一条企业微信消息 - d) 不满足 → 丢弃本轮告警，记录系统日志 - - 设备 recovered → 单独发送恢复通知 - """ + def __init__(self, runtime_settings: Settings | None = None): + self._settings = runtime_settings or settings - def __init__(self): - self._pending_alerts: list[PendingOfflineAlert] = [] - self._lock = asyncio.Lock() - # 上游心跳状态 - self._upstream_failures = 0 - self._upstream_available = True + async def record_transition( + self, + change: DeviceStateChange, + db: AsyncSession, + ) -> AlertEvent: + """Stage an event and notification without committing the caller session.""" + occurred_at = change.device.last_ping_time or datetime.now() + event_type = AlertTypeEnum(change.event_type or change.new_status) + prior_event = await self._find_prior_open_event(change, db) + duration_minutes = self._close_prior_event(prior_event, occurred_at) + event = AlertEvent( + device_id=change.device.id, + alert_type=event_type, + message=change.reason, + start_at=occurred_at, + is_resolved=event_type == AlertTypeEnum.recovered, + duration_minutes=( + duration_minutes + if event_type == AlertTypeEnum.recovered + else None + ), + related_event_id=(prior_event.id if prior_event is not None else None), + previous_status=change.old_status, + current_status=change.new_status, + ) + db.add(event) + await db.flush() - async def on_state_change(self, change: DeviceStateChange, db: AsyncSession): - """Pinger 状态变化回调""" - if change.new_status == "offline": - async with self._lock: - self._pending_alerts.append( - PendingOfflineAlert(device=change.device, alert_time=datetime.now()) - ) + if prior_event is not None: + prior_event.related_event_id = event.id - elif change.new_status == "online" and change.old_status == "offline": - # 设备恢复，立即记录并发送恢复通知 - await self._handle_recovery(change.device, db) - - async def flush_pending(self, db: AsyncSession, total_device_count: int): - """ - 每轮结束时调用：处理待发送的离线告警。 - 判断是否应该抑制告警，然后发送或丢弃。 - """ - async with self._lock: - if not self._pending_alerts: - return - pending = self._pending_alerts.copy() - self._pending_alerts.clear() - - # 1. 检查上游心跳 - upstream_ok = await self._check_upstream() - - # 2. 计算离线率 - offline_count = len(pending) - offline_ratio = offline_count / max(total_device_count, 1) - - # 3. 抑制条件 - suppressed = False - suppress_reason = "" - - if not upstream_ok: - suppressed = True - suppress_reason = "上游网络不可达（监控节点可能断网）" - elif offline_ratio >= settings.OFFLINE_SUPPRESS_RATIO: - suppressed = True - suppress_reason = f"离线率 {offline_ratio:.0%} >= {settings.OFFLINE_SUPPRESS_RATIO:.0%}，疑似监控节点断网" - - if suppressed: - logger.warning( - f"告警抑制: {suppress_reason}，" - f"本轮 {offline_count} 条告警已丢弃" + db.add( + NotificationOutbox( + alert_event_id=event.id, + message_content=self._render_notification( + change, + occurred_at, + duration_minutes, + ), + delivery_scope=( + self._settings.WECOM_TO_PARTY.strip() or "@all" + ), ) - # 记录系统告警 - db.add(AlertEvent( - device_id=0, - alert_type=AlertTypeEnum.system, - message=f"告警抑制: {suppress_reason}，丢弃 {offline_count} 条离线告警", - start_at=datetime.now(), - is_resolved=True, - notification_sent=False, - )) - await db.commit() - return + ) + await db.flush() + return event - # 4. 发送聚合告警 - if pending: - await self._send_aggregated_alert(pending, db) - - async def _check_upstream(self) -> bool: - """ - 上游心跳检测。 - 连续 UPSTREAM_PING_THRESHOLD 次失败才判定为上游断网。 - """ - try: - import subprocess - proc = await asyncio.create_subprocess_exec( - "ping", "-c", "1", "-W", "3", - settings.UPSTREAM_PING_TARGET, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - await proc.wait() - if proc.returncode == 0: - self._upstream_failures = 0 - self._upstream_available = True - return True - else: - self._upstream_failures += 1 - if self._upstream_failures >= settings.UPSTREAM_PING_THRESHOLD: - self._upstream_available = False - return False - # 未达到阈值，认为上游还可用 - return True - except Exception as e: - logger.error(f"上游心跳检测异常: {e}") - return True # 异常时保守地允许告警 - - async def _send_aggregated_alert(self, alerts: list[PendingOfflineAlert], db: AsyncSession): - """发送聚合离线告警""" - now = datetime.now() - - # 构建企业微信消息 - if len(alerts) == 1: - a = alerts[0] - msg = self._build_offline_message_single(a.device, a.alert_time) + @staticmethod + async def _find_prior_open_event( + change: DeviceStateChange, + db: AsyncSession, + ) -> AlertEvent | None: + """Find only the incident that this escalation or recovery supersedes.""" + event_type = change.event_type or change.new_status + if event_type == AlertTypeEnum.offline.value and change.old_status == "degraded": + alert_types = (AlertTypeEnum.degraded,) + elif event_type == AlertTypeEnum.recovered.value: + alert_types = (AlertTypeEnum.offline, AlertTypeEnum.degraded) else: - msg = self._build_offline_message_batch(alerts) + return None - # 发送 - success = await self._send_wecom_message(msg) - - # 记录告警事件 - for a in alerts: - db.add(AlertEvent( - device_id=a.device.id, - alert_type=AlertTypeEnum.offline, - message=a.device.name, - start_at=a.alert_time, - is_resolved=False, - notification_sent=success, - )) - - await db.commit() - - if success: - logger.info(f"已推送离线告警: {len(alerts)} 台设备") - else: - logger.error(f"企业微信推送失败: {len(alerts)} 台设备") - - async def _handle_recovery(self, device: Device, db: AsyncSession): - """处理设备恢复""" - now = datetime.now() - - # 查找未解决的离线事件 - result = await db.execute( + return await db.scalar( select(AlertEvent) - .where(AlertEvent.device_id == device.id) - .where(AlertEvent.alert_type == AlertTypeEnum.offline) - .where(AlertEvent.is_resolved == False) - .order_by(AlertEvent.created_at.desc()) + .where(AlertEvent.device_id == change.device.id) + .where(AlertEvent.alert_type.in_(alert_types)) + .where(AlertEvent.is_resolved.is_(False)) + .order_by(AlertEvent.start_at.desc(), AlertEvent.id.desc()) .limit(1) ) - event = result.scalar_one_or_none() - duration_minutes = None - if event: - delta = now - event.start_at - duration_minutes = int(delta.total_seconds() / 60) - event.end_at = now - event.duration_minutes = duration_minutes - event.is_resolved = True - - # 发送恢复通知 - msg = self._build_recovery_message(device, now, duration_minutes) - success = await self._send_wecom_message(msg) - - if event: - event.notification_sent = success - - await db.commit() - - if success: - logger.info(f"已推送恢复通知: {device.name}") - else: - logger.error(f"恢复通知推送失败: {device.name}") - - # ---------- 消息格式化 ---------- - - def _build_offline_message_single(self, device: Device, alert_time: datetime) -> str: - """单台设备离线消息""" - time_str = alert_time.strftime("%Y-%m-%d %H:%M:%S") - return ( - f"⛔ 设备离线啦！\n" - f"地址：{device.location or '未知'}\n" - f"时间：{time_str}\n" - f"项目：{device.project_name or '未分组'}\n" - f"设备类型：{self._fmt_device_type(device.device_type)}\n" - f"IP地址：{device.ip}" + @staticmethod + def _close_prior_event( + event: AlertEvent | None, + occurred_at: datetime, + ) -> int | None: + """Resolve one incident and return a non-negative minute duration.""" + if event is None: + return None + duration_minutes = max( + 0, + int((occurred_at - event.start_at).total_seconds() / 60), ) + event.end_at = occurred_at + event.duration_minutes = duration_minutes + event.is_resolved = True + return duration_minutes - def _build_offline_message_batch(self, alerts: list[PendingOfflineAlert]) -> str: - """多台设备聚合离线消息""" - now = alerts[0].alert_time - time_str = now.strftime("%Y-%m-%d %H:%M:%S") - lines = [f"⛔ 设备离线啦！（共 {len(alerts)} 台）\n"] - - for a in alerts: - dev = a.device - lines.append( - f"地址：{dev.location or '未知'}\n" - f"时间：{a.alert_time.strftime('%Y-%m-%d %H:%M:%S')}\n" - f"项目：{dev.project_name or '未分组'}\n" - f"设备类型：{self._fmt_device_type(dev.device_type)}\n" - f"IP地址：{dev.ip}\n" - f"{'---' if len(alerts) > 1 else ''}" - ) - - return "\n".join(lines).rstrip("---\n") - - def _build_recovery_message(self, device: Device, recover_time: datetime, duration: Optional[int]) -> str: - """设备恢复消息""" - time_str = recover_time.strftime("%Y-%m-%d %H:%M:%S") - duration_str = f"{duration}分钟" if duration is not None else "未知" - return ( - f"✅ 设备恢复在线！\n" - f"地址：{device.location or '未知'}\n" - f"时间：{time_str}\n" - f"项目：{device.project_name or '未分组'}\n" - f"设备类型：{self._fmt_device_type(device.device_type)}\n" - f"IP地址：{device.ip}\n" - f"离线时长：{duration_str}" - ) - - def _fmt_device_type(self, dtype) -> str: - mapping = { - "server": "服务器", - "olt": "OLT", - "switch": "交换机", - "firewall": "防火墙", - "other": "其他设备", + @staticmethod + def _render_notification( + change: DeviceStateChange, + occurred_at: datetime, + duration_minutes: int | None = None, + ) -> str: + """Render the bounded device and transition details needed by operators.""" + labels = { + "offline": "设备离线", + "degraded": "链路质量故障", + "recovered": "设备恢复", + "system": "监控系统异常", } - return mapping.get(str(dtype), str(dtype)) - - # ---------- 企业微信推送 ---------- - - async def _get_wecom_token(self) -> Optional[str]: - """获取企业微信 access_token""" - url = ( - f"https://qyapi.weixin.qq.com/cgi-bin/gettoken" - f"?corpid={settings.WECOM_CORP_ID}" - f"&corpsecret={settings.WECOM_APP_SECRET}" - ) - async with httpx.AsyncClient(timeout=10) as client: - resp = await client.get(url) - data = resp.json() - if data.get("errcode") == 0: - return data["access_token"] - else: - logger.error(f"获取企业微信 token 失败: {data}") - return None - - async def _send_wecom_message(self, content: str) -> bool: - """发送企业微信应用消息""" - if not settings.WECOM_CORP_ID or not settings.WECOM_APP_SECRET: - logger.warning("企业微信未配置，跳过推送") - logger.info(f"[模拟推送] {content}") - return True # 开发模式 - - token = await self._get_wecom_token() - if not token: - return False - - url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}" - payload = { - "touser": "@all", - "msgtype": "text", - "agentid": settings.WECOM_AGENT_ID, - "text": {"content": content}, - "safe": 0, - } - - async with httpx.AsyncClient(timeout=10) as client: - resp = await client.post(url, json=payload) - data = resp.json() - if data.get("errcode") != 0: - logger.error(f"发送企业微信消息失败: {data}") - return False - return True + event_type = change.event_type or change.new_status + device = change.device + lines = [ + f"【{labels.get(event_type, event_type)}】", + f"设备：{device.name}", + f"IP：{device.ip}", + f"位置：{device.location or '未配置'}", + f"项目：{device.project_name or '未分组'}", + f"丢包摘要：{change.reason or '无'}", + f"时间：{occurred_at:%Y-%m-%d %H:%M:%S}", + ] + if duration_minutes is not None: + lines.append(f"持续时间：{duration_minutes} 分钟") + return "\n".join(lines) diff --git a/backend/app/services/notification_dispatcher.py b/backend/app/services/notification_dispatcher.py new file mode 100644 index 0000000..3bf4366 --- /dev/null +++ b/backend/app/services/notification_dispatcher.py @@ -0,0 +1,322 @@ +"""Deliver pending notifications from the durable outbox.""" + +import asyncio +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Callable, Protocol + +import httpx +from sqlalchemy import or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import Settings, settings +from app.models import AlertEvent, NotificationOutbox, NotificationStatus + + +logger = logging.getLogger("pingwatch.notification_dispatcher") + +# A small count caps one WeCom text request while still aggregating alert bursts. +MAX_EVENTS_PER_MESSAGE = 5 +_MESSAGE_SEPARATOR = "\n\n---\n\n" +_TOKEN_EXPIRY_MARGIN_SECONDS = 60 +_HTTP_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=5.0) +_TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" +_MESSAGE_URL = "https://qyapi.weixin.qq.com/cgi-bin/message/send" + + +@dataclass(frozen=True) +class DeliveryResult: + """Sanitized delivery outcome returned by a notification transport.""" + + success: bool + error: str | None = None + retryable: bool = False + + +@dataclass(frozen=True) +class DispatchSummary: + """Counts produced by one bounded dispatcher pass.""" + + sent: int = 0 + rescheduled: int = 0 + failed: int = 0 + + +class TextNotificationClient(Protocol): + """Minimal transport boundary used by the persistent dispatcher.""" + + async def send_text(self, content: str) -> DeliveryResult: + """Deliver one text message without exposing credentials.""" + + +@dataclass(frozen=True) +class _TokenResult: + token: str | None + error: str | None = None + retryable: bool = False + + +class WeComClient: + """Minimal WeCom application client with an in-process token cache.""" + + def __init__( + self, + runtime_settings: Settings | None = None, + *, + http_client: httpx.AsyncClient | None = None, + now: Callable[[], datetime] | None = None, + ): + self._settings = runtime_settings or settings + self._http = http_client or httpx.AsyncClient(timeout=_HTTP_TIMEOUT) + self._owns_http_client = http_client is None + self._now = now or datetime.now + self._access_token: str | None = None + self._access_token_expires_at: datetime | None = None + self._token_lock = asyncio.Lock() + + async def aclose(self) -> None: + """Close only the HTTP client created by this service instance.""" + if self._owns_http_client: + await self._http.aclose() + + async def send_text(self, content: str) -> DeliveryResult: + """Deliver text while returning only sanitized failure evidence.""" + if not self._settings.wecom_notification_enabled: + return DeliveryResult(False, "delivery disabled", False) + + token_result = await self._get_access_token() + if token_result.token is None: + return DeliveryResult( + False, + token_result.error, + token_result.retryable, + ) + + payload = { + "msgtype": "text", + "agentid": self._settings.WECOM_AGENT_ID, + "text": {"content": content}, + "safe": 0, + } + to_party = self._settings.WECOM_TO_PARTY.strip() + if to_party: + payload["toparty"] = to_party + else: + payload["touser"] = "@all" + + try: + response = await self._http.post( + _MESSAGE_URL, + params={"access_token": token_result.token}, + json=payload, + ) + except (httpx.TimeoutException, httpx.TransportError): + logger.warning("WeCom message transport failure") + return DeliveryResult(False, "transport error", True) + + if response.status_code == 429 or response.status_code >= 500: + logger.warning( + "WeCom message transient HTTP failure status=%s", + response.status_code, + ) + return DeliveryResult(False, f"HTTP {response.status_code}", True) + if response.status_code >= 400: + logger.warning( + "WeCom message HTTP failure status=%s", + response.status_code, + ) + return DeliveryResult(False, f"HTTP {response.status_code}", False) + + try: + data = response.json() + if not isinstance(data, dict): + raise TypeError("response JSON must be an object") + error_code = int(data.get("errcode", -1)) + except (TypeError, ValueError): + logger.warning("WeCom message returned invalid JSON") + return DeliveryResult(False, "invalid response", True) + + if error_code == 0: + return DeliveryResult(True) + + if error_code in {40014, 42001}: + self._access_token = None + self._access_token_expires_at = None + retryable = error_code in {-1, 40014, 42001, 45009} + logger.warning("WeCom message rejected errcode=%s", error_code) + return DeliveryResult( + False, + f"WeCom errcode {error_code}", + retryable, + ) + + async def _get_access_token(self) -> _TokenResult: + """Return a cached token or refresh it once under a process-local lock.""" + if self._token_is_valid(): + return _TokenResult(self._access_token) + + async with self._token_lock: + if self._token_is_valid(): + return _TokenResult(self._access_token) + try: + response = await self._http.get( + _TOKEN_URL, + params={ + "corpid": self._settings.WECOM_CORP_ID, + "corpsecret": self._settings.WECOM_APP_SECRET, + }, + ) + except (httpx.TimeoutException, httpx.TransportError): + logger.warning("WeCom token transport failure") + return _TokenResult(None, "token transport error", True) + + if response.status_code == 429 or response.status_code >= 500: + logger.warning( + "WeCom token transient HTTP failure status=%s", + response.status_code, + ) + return _TokenResult(None, f"token HTTP {response.status_code}", True) + if response.status_code >= 400: + logger.warning( + "WeCom token HTTP failure status=%s", + response.status_code, + ) + return _TokenResult(None, f"token HTTP {response.status_code}", False) + + try: + data = response.json() + if not isinstance(data, dict): + raise TypeError("response JSON must be an object") + error_code = int(data.get("errcode", -1)) + access_token = data.get("access_token") + expires_in = int(data.get("expires_in", 0)) + except (TypeError, ValueError): + logger.warning("WeCom token endpoint returned invalid JSON") + return _TokenResult(None, "invalid token response", True) + + if error_code != 0 or not access_token or expires_in <= 0: + logger.warning("WeCom token request rejected errcode=%s", error_code) + return _TokenResult( + None, + f"WeCom token errcode {error_code}", + error_code == -1, + ) + + self._access_token = str(access_token) + cache_seconds = max(0, expires_in - _TOKEN_EXPIRY_MARGIN_SECONDS) + self._access_token_expires_at = self._now() + timedelta( + seconds=cache_seconds + ) + return _TokenResult(self._access_token) + + def _token_is_valid(self) -> bool: + return ( + self._access_token is not None + and self._access_token_expires_at is not None + and self._now() < self._access_token_expires_at + ) + + +class NotificationDispatcher: + """Claim due outbox rows and persist their delivery outcome.""" + + def __init__( + self, + client: TextNotificationClient, + runtime_settings: Settings | None = None, + ): + self._client = client + self._settings = runtime_settings or settings + + async def dispatch_due( + self, + db: AsyncSession, + now: datetime, + ) -> DispatchSummary: + """Attempt each currently due message once without committing the session.""" + rows = list( + ( + await db.execute( + select(NotificationOutbox) + .where(NotificationOutbox.status == NotificationStatus.pending) + .where( + or_( + NotificationOutbox.next_attempt_at.is_(None), + NotificationOutbox.next_attempt_at <= now, + ) + ) + .order_by(NotificationOutbox.created_at, NotificationOutbox.id) + ) + ) + .scalars() + .all() + ) + + event_ids = [row.alert_event_id for row in rows] + events = {} + if event_ids: + events = { + event.id: event + for event in ( + ( + await db.execute( + select(AlertEvent).where(AlertEvent.id.in_(event_ids)) + ) + ) + .scalars() + .all() + ) + } + + sent = rescheduled = failed = 0 + for start in range(0, len(rows), MAX_EVENTS_PER_MESSAGE): + batch = rows[start : start + MAX_EVENTS_PER_MESSAGE] + for row in batch: + row.status = NotificationStatus.sending + row.locked_at = now + try: + result = await self._client.send_text( + _MESSAGE_SEPARATOR.join(row.message_content for row in batch) + ) + except Exception as exc: # Transport implementations must not strand rows. + logger.warning( + "Notification client raised type=%s", + type(exc).__name__, + ) + result = DeliveryResult(False, "notification client error", True) + + for row in batch: + row.attempt_count += 1 + event = events.get(row.alert_event_id) + if result.success: + row.status = NotificationStatus.sent + row.sent_at = now + row.next_attempt_at = None + row.last_error = None + sent += 1 + elif ( + result.retryable + and row.attempt_count + < self._settings.wecom_notification_max_attempts + ): + delay = self._settings.wecom_retry_base_seconds * ( + 2 ** (row.attempt_count - 1) + ) + row.status = NotificationStatus.pending + row.next_attempt_at = now + timedelta(seconds=delay) + row.last_error = (result.error or "delivery failed")[:512] + rescheduled += 1 + else: + row.status = NotificationStatus.failed + row.next_attempt_at = None + row.last_error = (result.error or "delivery failed")[:512] + failed += 1 + + if event is not None: + event.notification_sent = result.success + event.notification_attempts = row.attempt_count + event.last_notification_error = row.last_error + + await db.flush() + return DispatchSummary(sent=sent, rescheduled=rescheduled, failed=failed) diff --git a/backend/app/services/pinger.py b/backend/app/services/pinger.py index 6c5a80d..19a4bb7 100644 --- a/backend/app/services/pinger.py +++ b/backend/app/services/pinger.py @@ -5,7 +5,6 @@ import time from collections import defaultdict from dataclasses import dataclass from datetime import datetime -from typing import Awaitable, Callable from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -38,16 +37,6 @@ class Pinger: def __init__(self, runtime_settings: Settings | None = None): self._settings = runtime_settings or settings self._round_num = 0 - self._on_state_change: ( - Callable[[DeviceStateChange], Awaitable[None]] | None - ) = None - - def on_state_change( - self, - callback: Callable[[DeviceStateChange], Awaitable[None]], - ) -> None: - """Register the legacy transition callback during scheduler migration.""" - self._on_state_change = callback async def run_one_round( self, @@ -128,11 +117,9 @@ class Pinger: if change is not None: changes.append(change) - await db.commit() - - if self._on_state_change is not None: - for change in changes: - await self._on_state_change(change) + # The scheduler records transition events and commits the complete + # probe/state/outbox unit. A direct caller retains the same ownership. + await db.flush() elapsed = time.monotonic() - started_at logger.info( diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index 999f01d..77293b5 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -1,94 +1,142 @@ -""" -定时任务调度器 - -使用 asyncio 循环驱动 Ping 引擎，协调 Pinger 和 Alerter。 -""" +"""Single-task scheduler for probe, event recording, and outbox delivery.""" import asyncio import logging +import time from datetime import datetime +from typing import Callable -from sqlalchemy.ext.asyncio import AsyncSession - -from app.config import settings -from app.services.pinger import Pinger -from app.services.alerter import Alerter +from app.config import Settings, settings from app.core.deps import async_session +from app.services.alerter import Alerter +from app.services.notification_dispatcher import ( + NotificationDispatcher, + WeComClient, +) +from app.services.pinger import Pinger + logger = logging.getLogger("pingwatch.scheduler") +_DEFAULT_DISPATCH_INTERVAL_SECONDS = 5.0 class PingScheduler: - """ - 调度器职责： - 1. 按间隔驱动 Ping 引擎 - 2. 每轮结束后触发 Alerter 处理待发送告警 - 3. 控制并发和清理 - """ + """Serialize monitoring cycles and retain one cancellable background task.""" - def __init__(self): - self._pinger = Pinger() - self._alerter = Alerter() + def __init__( + self, + *, + pinger=None, + alerter=None, + dispatcher=None, + session_factory: Callable = async_session, + runtime_settings: Settings | None = None, + interval_seconds: float | None = None, + dispatch_interval_seconds: float = _DEFAULT_DISPATCH_INTERVAL_SECONDS, + ): + self._settings = runtime_settings or settings + self._pinger = pinger or Pinger(self._settings) + self._alerter = alerter or Alerter(self._settings) + self._notification_client = None + self._dispatch_enabled = ( + dispatcher is not None or self._settings.wecom_notification_enabled + ) + if dispatcher is None: + self._notification_client = WeComClient(self._settings) + dispatcher = NotificationDispatcher( + self._notification_client, + self._settings, + ) + self._dispatcher = dispatcher + self._session_factory = session_factory + self._probe_interval_seconds = max( + 1.0, + float( + interval_seconds + if interval_seconds is not None + else self._settings.PING_INTERVAL_SECONDS + ), + ) + self._dispatch_interval_seconds = min( + max(1.0, float(dispatch_interval_seconds)), + 30.0, + ) self._running = False self._task: asyncio.Task | None = None + self._cycle_lock = asyncio.Lock() - # 注册状态变化回调 - self._pinger.on_state_change(self._on_state_change) + async def _run_cycle(self, *, run_probe: bool = True) -> None: + """Run one serialized transaction cycle and close its session on cancel.""" + async with self._cycle_lock: + async with self._session_factory() as db: + try: + if run_probe: + changes = await self._pinger.run_one_round(db) + for change in changes: + await self._alerter.record_transition(change, db) + await db.commit() - async def _on_state_change(self, change): - """收到设备状态变化，转给 alerter""" - async with async_session() as db: - try: - await self._alerter.on_state_change(change, db) - except Exception as e: - logger.error(f"告警处理异常: {e}", exc_info=True) - - async def _run_loop(self): - """主循环""" - logger.info("Ping 调度器已启动") - self._running = True + if self._dispatch_enabled: + await self._dispatcher.dispatch_due(db, datetime.now()) + await db.commit() + except BaseException: + await db.rollback() + raise + async def _run_loop(self) -> None: + """Probe at its configured cadence and service retries every few seconds.""" + logger.info("Ping scheduler started") + next_probe_at = 0.0 while self._running: + now = time.monotonic() + run_probe = now >= next_probe_at try: - async with async_session() as db: - # 执行一轮 ping - results = await self._pinger.run_one_round(db) - - if results: - # 直接用启用的设备数 - total = len(results) - - # 处理待发送告警 - await self._alerter.flush_pending(db, total) - + await self._run_cycle(run_probe=run_probe) + if run_probe: + next_probe_at = time.monotonic() + self._probe_interval_seconds except asyncio.CancelledError: break - except Exception as e: - logger.error(f"调度器异常: {e}", exc_info=True) + except Exception as exc: + logger.error( + "Scheduler cycle failed type=%s", + type(exc).__name__, + exc_info=True, + ) + if run_probe: + next_probe_at = time.monotonic() + self._probe_interval_seconds - # 等待下一轮 - await asyncio.sleep(settings.PING_INTERVAL_SECONDS) + seconds_until_probe = max(0.0, next_probe_at - time.monotonic()) + sleep_seconds = min( + self._dispatch_interval_seconds, + seconds_until_probe or self._dispatch_interval_seconds, + ) + try: + await asyncio.sleep(sleep_seconds) + except asyncio.CancelledError: + break + logger.info("Ping scheduler stopped") - logger.info("Ping 调度器已停止") - - def start(self): - """启动调度器（后台任务）""" - if self._running: - logger.warning("调度器已在运行") + def start(self) -> None: + """Start exactly one background scheduler task.""" + if self._task is not None and not self._task.done(): + logger.warning("Ping scheduler is already running") return + self._running = True self._task = asyncio.create_task(self._run_loop()) - async def stop(self): - """停止调度器""" + async def stop(self) -> None: + """Cancel and await the active cycle before releasing its HTTP client.""" self._running = False - if self._task: + if self._task is not None: self._task.cancel() try: await self._task except asyncio.CancelledError: pass self._task = None + if self._notification_client is not None: + await self._notification_client.aclose() + self._notification_client = None -# 全局调度器实例 scheduler = PingScheduler() diff --git a/backend/tests/test_alert_workflow.py b/backend/tests/test_alert_workflow.py new file mode 100644 index 0000000..90b02a1 --- /dev/null +++ b/backend/tests/test_alert_workflow.py @@ -0,0 +1,427 @@ +"""Transactional alert-event and notification-outbox workflow tests.""" + +import asyncio +from datetime import datetime, timedelta + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import ( + AsyncSession, + async_sessionmaker, + create_async_engine, +) + +from app.config import Settings +from app.models import ( + AlertEvent, + AlertTypeEnum, + Base, + Device, + DeviceTypeEnum, + NotificationOutbox, + NotificationStatus, + PingRecord, +) +from app.services.alerter import Alerter +from app.services.fping_runner import ProbeResult +from app.services.pinger import DeviceStateChange, Pinger +from app.services.scheduler import PingScheduler + + +@pytest.fixture +async def db_session() -> AsyncSession: + """Provide a complete isolated persistence boundary for alert tests.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + try: + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with AsyncSession(engine, expire_on_commit=False) as session: + yield session + finally: + await engine.dispose() + + +@pytest.fixture +async def offline_change(db_session: AsyncSession) -> DeviceStateChange: + """Persist a device and expose a real offline state-change value.""" + observed_at = datetime(2026, 8, 4, 9, 30) + device = Device( + name="核心交换机", + ip="10.0.0.8", + device_type=DeviceTypeEnum.switch, + location="一楼机房", + project_name="园区网", + current_status="offline", + last_ping_time=observed_at, + ) + db_session.add(device) + await db_session.commit() + return DeviceStateChange( + device=device, + old_status="online", + new_status="offline", + consecutive_failures=2, + event_type="offline", + reason="2 consecutive full-loss rounds", + ) + + +async def test_transition_creates_event_and_pending_outbox_in_one_transaction( + db_session: AsyncSession, + offline_change: DeviceStateChange, +): + """Removing either insert breaks the durable event-delivery contract.""" + event = await Alerter().record_transition(offline_change, db_session) + + outbox = await db_session.scalar( + select(NotificationOutbox).where( + NotificationOutbox.alert_event_id == event.id + ) + ) + + assert outbox is not None + assert outbox.status == NotificationStatus.pending + assert "核心交换机" in outbox.message_content + assert "10.0.0.8" in outbox.message_content + assert "一楼机房" in outbox.message_content + assert "园区网" in outbox.message_content + assert "2 consecutive full-loss rounds" in outbox.message_content + assert "2026-08-04 09:30:00" in outbox.message_content + + +async def test_event_and_outbox_rollback_together( + db_session: AsyncSession, + offline_change: DeviceStateChange, +): + """A caller rollback cannot retain an event without its notification.""" + await Alerter().record_transition(offline_change, db_session) + + await db_session.rollback() + + assert await db_session.scalar(select(AlertEvent)) is None + assert await db_session.scalar(select(NotificationOutbox)) is None + + +async def test_offline_escalation_closes_and_links_open_degraded_event( + db_session: AsyncSession, + offline_change: DeviceStateChange, +): + """Escalating degraded to offline must not leave two open incidents.""" + device = offline_change.device + device.last_ping_time = datetime(2026, 8, 4, 9, 20) + degraded = await Alerter().record_transition( + DeviceStateChange( + device=device, + old_status="online", + new_status="degraded", + consecutive_failures=0, + event_type="degraded", + reason="window packet loss 20.00%", + ), + db_session, + ) + await db_session.commit() + device.last_ping_time = datetime(2026, 8, 4, 9, 30) + + offline = await Alerter().record_transition( + DeviceStateChange( + device=device, + old_status="degraded", + new_status="offline", + consecutive_failures=2, + event_type="offline", + reason="2 consecutive full-loss rounds", + ), + db_session, + ) + + assert degraded.is_resolved is True + assert degraded.end_at == datetime(2026, 8, 4, 9, 30) + assert degraded.duration_minutes == 10 + assert degraded.related_event_id == offline.id + assert offline.related_event_id == degraded.id + + +async def test_recovery_closes_open_fault_and_notifies_with_duration( + db_session: AsyncSession, + offline_change: DeviceStateChange, +): + """Recovery closes one active incident and carries its duration to operators.""" + opened = await Alerter().record_transition(offline_change, db_session) + await db_session.commit() + device = offline_change.device + device.last_ping_time = datetime(2026, 8, 4, 10, 1) + + recovered = await Alerter().record_transition( + DeviceStateChange( + device=device, + old_status="offline", + new_status="online", + consecutive_failures=0, + event_type="recovered", + reason="3 consecutive clean rounds", + ), + db_session, + ) + outbox = await db_session.scalar( + select(NotificationOutbox).where( + NotificationOutbox.alert_event_id == recovered.id + ) + ) + + assert opened.is_resolved is True + assert opened.end_at == datetime(2026, 8, 4, 10, 1) + assert opened.duration_minutes == 31 + assert recovered.alert_type == AlertTypeEnum.recovered + assert recovered.is_resolved is True + assert recovered.duration_minutes == 31 + assert recovered.related_event_id == opened.id + assert "持续时间：31 分钟" in outbox.message_content + + +async def test_probe_state_event_and_outbox_share_the_caller_transaction( + monkeypatch, +): + """A failed caller commit cannot persist a state transition without its event.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + observed_at = datetime.now() - timedelta(seconds=30) + try: + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with AsyncSession(engine, expire_on_commit=False) as setup: + device = Device( + name="edge-atomic", + ip="10.0.0.18", + device_type=DeviceTypeEnum.switch, + current_status="online", + is_enabled=True, + offline_consecutive_rounds=2, + ) + setup.add(device) + await setup.flush() + device_id = device.id + setup.add( + PingRecord( + device_id=device_id, + is_alive=False, + response_time_ms=None, + round_num=1, + sent_count=3, + received_count=0, + packet_loss_percent=100.0, + average_rtt_ms=None, + is_valid=True, + created_at=observed_at, + ) + ) + await setup.commit() + + async def fake_run_fping_count(*args, **kwargs): + return { + "10.0.0.18": ProbeResult( + "10.0.0.18", + 3, + 0, + None, + True, + ) + } + + monkeypatch.setattr( + "app.services.pinger.run_fping_count", + fake_run_fping_count, + ) + async with AsyncSession(engine, expire_on_commit=False) as session: + changes = await Pinger().run_one_round(session) + await Alerter().record_transition(changes[0], session) + await session.rollback() + + async with AsyncSession(engine, expire_on_commit=False) as verification: + persisted_device = await verification.get(Device, device_id) + records = list( + ( + await verification.execute( + select(PingRecord).where(PingRecord.device_id == device_id) + ) + ) + .scalars() + .all() + ) + assert persisted_device.current_status == "online" + assert len(records) == 1 + assert await verification.scalar(select(AlertEvent)) is None + assert await verification.scalar(select(NotificationOutbox)) is None + finally: + await engine.dispose() + + +class RecordingSession: + """Small session boundary exposing transaction and close ordering.""" + + def __init__(self, events: list[str]): + self.events = events + self.closed = asyncio.Event() + + async def __aenter__(self): + self.events.append("session-enter") + return self + + async def __aexit__(self, exc_type, exc_value, traceback): + self.events.append("session-exit") + self.closed.set() + + async def commit(self): + self.events.append("commit") + + async def rollback(self): + self.events.append("rollback") + + +class RecordingPinger: + def __init__(self, events: list[str], changes=None): + self.events = events + self.changes = list(changes or ["transition"]) + + async def run_one_round(self, db): + self.events.append("probe") + return self.changes + + +class RecordingAlerter: + def __init__(self, events: list[str]): + self.events = events + + async def record_transition(self, change, db): + self.events.append(f"alert:{change}") + + +class RecordingDispatcher: + def __init__(self, events: list[str]): + self.events = events + + async def dispatch_due(self, db, now): + assert isinstance(now, datetime) + self.events.append("dispatch") + + +async def test_scheduler_persists_transitions_before_dispatching_due_messages(): + """Changing scheduler order cannot expose uncommitted outbox rows to dispatch.""" + events: list[str] = [] + session = RecordingSession(events) + scheduler = PingScheduler( + pinger=RecordingPinger(events), + alerter=RecordingAlerter(events), + dispatcher=RecordingDispatcher(events), + session_factory=lambda: session, + ) + + await scheduler._run_cycle() + + assert events == [ + "session-enter", + "probe", + "alert:transition", + "commit", + "dispatch", + "commit", + "session-exit", + ] + + +async def test_scheduler_lock_prevents_overlapping_cycles(): + """Even concurrent triggers cannot overlap probe or notification sessions.""" + events: list[str] = [] + + class ConcurrencyPinger: + active = 0 + maximum = 0 + + async def run_one_round(self, db): + self.active += 1 + self.maximum = max(self.maximum, self.active) + await asyncio.sleep(0.01) + self.active -= 1 + return [] + + pinger = ConcurrencyPinger() + scheduler = PingScheduler( + pinger=pinger, + alerter=RecordingAlerter(events), + dispatcher=RecordingDispatcher(events), + session_factory=lambda: RecordingSession(events), + ) + + await asyncio.gather(scheduler._run_cycle(), scheduler._run_cycle()) + + assert pinger.maximum == 1 + + +async def test_scheduler_stop_waits_for_active_session_to_close(): + """Cancellation cannot return while a probe session remains open.""" + events: list[str] = [] + session = RecordingSession(events) + probe_started = asyncio.Event() + never_complete = asyncio.Event() + + class BlockingPinger: + async def run_one_round(self, db): + probe_started.set() + await never_complete.wait() + + scheduler = PingScheduler( + pinger=BlockingPinger(), + alerter=RecordingAlerter(events), + dispatcher=RecordingDispatcher(events), + session_factory=lambda: session, + interval_seconds=1, + ) + scheduler.start() + first_task = scheduler._task + scheduler.start() + + await probe_started.wait() + await scheduler.stop() + + assert first_task is not None + assert session.closed.is_set() + assert scheduler._task is None + assert events[-2:] == ["rollback", "session-exit"] + + +async def test_scheduler_leaves_pending_outbox_untouched_when_delivery_disabled(): + """Disabling WeCom preserves queued notifications for a later enablement.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + session_factory = async_sessionmaker(engine, expire_on_commit=False) + + class EmptyPinger: + async def run_one_round(self, db): + return [] + + try: + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with session_factory() as setup: + setup.add( + NotificationOutbox( + alert_event_id=42, + message_content="保留待发送事件", + ) + ) + await setup.commit() + + runtime_settings = Settings(wecom_notification_enabled=False) + scheduler = PingScheduler( + pinger=EmptyPinger(), + alerter=Alerter(runtime_settings), + session_factory=session_factory, + runtime_settings=runtime_settings, + ) + await scheduler._run_cycle() + + async with session_factory() as verification: + outbox = await verification.scalar(select(NotificationOutbox)) + assert outbox.status == NotificationStatus.pending + assert outbox.attempt_count == 0 + await scheduler.stop() + finally: + await engine.dispose() diff --git a/backend/tests/test_notification_dispatcher.py b/backend/tests/test_notification_dispatcher.py new file mode 100644 index 0000000..f366046 --- /dev/null +++ b/backend/tests/test_notification_dispatcher.py @@ -0,0 +1,327 @@ +"""Persistent notification dispatch and retry behavior tests.""" + +import json +import logging +from datetime import datetime, timedelta + +import httpx +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine + +from app.config import Settings +from app.models import ( + AlertEvent, + AlertTypeEnum, + Base, + NotificationOutbox, + NotificationStatus, +) +from app.services.notification_dispatcher import ( + DeliveryResult, + MAX_EVENTS_PER_MESSAGE, + NotificationDispatcher, + WeComClient, +) + + +class FakeWeComClient: + """Return one controlled outcome without any external network access.""" + + def __init__(self, *results: DeliveryResult): + self._results = list(results) + self.contents: list[str] = [] + + async def send_text(self, content: str) -> DeliveryResult: + self.contents.append(content) + return self._results.pop(0) + + +@pytest.fixture +async def db_session() -> AsyncSession: + """Provide an isolated outbox database.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + try: + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with AsyncSession(engine, expire_on_commit=False) as session: + yield session + finally: + await engine.dispose() + + +async def test_retryable_failure_reschedules( + db_session: AsyncSession, +): + """A transient failure must remain pending with exponential retry evidence.""" + frozen_time = datetime(2026, 8, 4, 10, 0) + pending_message = NotificationOutbox( + alert_event_id=1, + message_content="设备离线", + ) + db_session.add(pending_message) + await db_session.commit() + client = FakeWeComClient(DeliveryResult(False, "timeout", True)) + + await NotificationDispatcher(client).dispatch_due( + db_session, + frozen_time, + ) + + assert pending_message.status == NotificationStatus.pending + assert pending_message.attempt_count == 1 + assert pending_message.next_attempt_at == datetime(2026, 8, 4, 10, 0, 30) + + +async def test_success_marks_outbox_and_event_delivered( + db_session: AsyncSession, +): + """A successful transport result updates both delivery evidence records.""" + now = datetime(2026, 8, 4, 10, 0) + event = AlertEvent( + device_id=7, + alert_type=AlertTypeEnum.offline, + message="full loss", + start_at=now, + is_resolved=False, + ) + db_session.add(event) + await db_session.flush() + message = NotificationOutbox( + alert_event_id=event.id, + message_content="设备 7 离线", + ) + db_session.add(message) + await db_session.commit() + + summary = await NotificationDispatcher( + FakeWeComClient(DeliveryResult(True)), + _enabled_settings(), + ).dispatch_due(db_session, now) + + assert summary.sent == 1 + assert message.status == NotificationStatus.sent + assert message.sent_at == now + assert message.attempt_count == 1 + assert event.notification_sent is True + assert event.notification_attempts == 1 + assert event.last_notification_error is None + + +async def test_non_retryable_or_exhausted_failure_is_terminal( + db_session: AsyncSession, +): + """The configured attempt ceiling prevents an infinite retry loop.""" + now = datetime(2026, 8, 4, 10, 0) + message = NotificationOutbox( + alert_event_id=99, + message_content="bad recipient", + attempt_count=1, + ) + db_session.add(message) + await db_session.commit() + runtime_settings = _enabled_settings(wecom_notification_max_attempts=2) + + summary = await NotificationDispatcher( + FakeWeComClient(DeliveryResult(False, "HTTP 503", True)), + runtime_settings, + ).dispatch_due(db_session, now) + + assert summary.failed == 1 + assert message.status == NotificationStatus.failed + assert message.attempt_count == 2 + assert message.next_attempt_at is None + + +async def test_only_due_messages_are_dispatched_in_fixed_safe_batches( + db_session: AsyncSession, +): + """Large bursts are split while future retries remain untouched.""" + now = datetime(2026, 8, 4, 10, 0) + due = [ + NotificationOutbox(alert_event_id=index, message_content=f"事件 {index}") + for index in range(MAX_EVENTS_PER_MESSAGE + 1) + ] + future = NotificationOutbox( + alert_event_id=100, + message_content="未来重试", + next_attempt_at=now + timedelta(minutes=1), + ) + db_session.add_all([*due, future]) + await db_session.commit() + client = FakeWeComClient(DeliveryResult(True), DeliveryResult(True)) + + summary = await NotificationDispatcher( + client, + _enabled_settings(), + ).dispatch_due(db_session, now) + + assert summary.sent == MAX_EVENTS_PER_MESSAGE + 1 + assert len(client.contents) == 2 + assert future.status == NotificationStatus.pending + assert future.attempt_count == 0 + + +def _enabled_settings(**overrides) -> Settings: + values = { + "wecom_notification_enabled": True, + "WECOM_CORP_ID": "test-corp", + "WECOM_AGENT_ID": 1000001, + "WECOM_APP_SECRET": "test-secret", + } + values.update(overrides) + return Settings(**values) + + +@pytest.mark.parametrize( + ("to_party", "expected_recipient", "unexpected_key"), + [("", ("touser", "@all"), "toparty"), ("2|3", ("toparty", "2|3"), "touser")], +) +async def test_wecom_uses_application_scope_unless_department_is_explicit( + to_party, + expected_recipient, + unexpected_key, +): + """Recipient payloads preserve the app visibility boundary by default.""" + payloads: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/gettoken"): + return httpx.Response( + 200, + json={"errcode": 0, "access_token": "token-a", "expires_in": 7200}, + ) + payloads.append(json.loads(request.content)) + return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + client = WeComClient( + _enabled_settings(WECOM_TO_PARTY=to_party), + http_client=http, + ) + result = await client.send_text("controlled test message") + + key, value = expected_recipient + assert result.success is True + assert payloads[0][key] == value + assert unexpected_key not in payloads[0] + assert payloads[0]["agentid"] == 1000001 + + +async def test_wecom_caches_token_until_expiry_margin(): + """Repeated messages do not request a new token before its safe expiry.""" + calls = {"token": 0, "message": 0} + current_time = [datetime(2026, 8, 4, 10, 0)] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/gettoken"): + calls["token"] += 1 + return httpx.Response( + 200, + json={"errcode": 0, "access_token": "token-a", "expires_in": 120}, + ) + calls["message"] += 1 + return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + client = WeComClient( + _enabled_settings(), + http_client=http, + now=lambda: current_time[0], + ) + await client.send_text("first") + current_time[0] += timedelta(seconds=30) + await client.send_text("second") + + assert calls == {"token": 1, "message": 2} + + +async def test_wecom_refreshes_token_after_safety_margin(): + """A token inside the safety margin cannot be reused for a new message.""" + token_calls = 0 + current_time = [datetime(2026, 8, 4, 10, 0)] + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal token_calls + if request.url.path.endswith("/gettoken"): + token_calls += 1 + return httpx.Response( + 200, + json={ + "errcode": 0, + "access_token": f"token-{token_calls}", + "expires_in": 120, + }, + ) + return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + client = WeComClient( + _enabled_settings(), + http_client=http, + now=lambda: current_time[0], + ) + await client.send_text("first") + current_time[0] += timedelta(seconds=61) + await client.send_text("second") + + assert token_calls == 2 + + +async def test_wecom_rejects_non_object_json_without_raising(): + """Malformed external response shapes return sanitized retry evidence.""" + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=["unexpected", "shape"]) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + result = await WeComClient( + _enabled_settings(), + http_client=http, + ).send_text("controlled") + + assert result == DeliveryResult(False, "invalid token response", True) + + +@pytest.mark.parametrize("status_code", [429, 500, 503]) +async def test_wecom_marks_rate_limit_and_server_errors_retryable(status_code): + """Transient HTTP classes reach the outbox retry path without response leakage.""" + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/gettoken"): + return httpx.Response( + 200, + json={"errcode": 0, "access_token": "token-a", "expires_in": 7200}, + ) + return httpx.Response(status_code, text="sensitive-response-body") + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + result = await WeComClient( + _enabled_settings(), + http_client=http, + ).send_text("sensitive-message-content") + + assert result.success is False + assert result.retryable is True + assert result.error == f"HTTP {status_code}" + + +async def test_wecom_transport_failure_is_sanitized_and_retryable(caplog): + """Logs and returned evidence never expose secrets, tokens, bodies, or content.""" + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/gettoken"): + return httpx.Response( + 200, + json={"errcode": 0, "access_token": "token-a", "expires_in": 7200}, + ) + raise httpx.ConnectError("response-sensitive", request=request) + + caplog.set_level(logging.WARNING) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + result = await WeComClient( + _enabled_settings(WECOM_APP_SECRET="secret-sensitive"), + http_client=http, + ).send_text("message-sensitive") + + assert result == DeliveryResult(False, "transport error", True) + assert "secret-sensitive" not in caplog.text + assert "token-a" not in caplog.text + assert "response-sensitive" not in caplog.text + assert "message-sensitive" not in caplog.text