Codex Release Bot
37 lines
1.0 KiB
Python
37 lines
1.0 KiB
Python
"""CH-ALT-001 — recovery detection → single telegram per metric (mock)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from server.api.agent import _detect_recovery
|
|
|
|
pytestmark = pytest.mark.chain
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_flow_deduped_before_telegram(monkeypatch):
|
|
sent: list[tuple] = []
|
|
|
|
async def capture_recovery(server_id, metric, value, server_name):
|
|
sent.append((server_id, metric, value, server_name))
|
|
|
|
monkeypatch.setattr(
|
|
"server.infrastructure.telegram.send_telegram_recovery",
|
|
AsyncMock(side_effect=lambda *a, **k: sent.append(a)),
|
|
)
|
|
|
|
thresholds = {"cpu": 80, "mem": 80, "disk": 80}
|
|
prev_alerts = {"mem:88.0", "mem:91.2"}
|
|
system_info = {"mem_usage": 55.0}
|
|
|
|
recoveries = _detect_recovery(system_info, prev_alerts, thresholds)
|
|
assert recoveries == [("mem", 55.0)]
|
|
|
|
for metric, value in recoveries:
|
|
await capture_recovery(1, metric, value, "srv-1")
|
|
|
|
assert len(sent) == 1
|