Codex Release Bot
85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""Tests for server offline monitor — edge detection and heartbeat helper."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from server.application.server_connectivity import heartbeat_indicates_online
|
|
from server.background import server_offline_monitor as monitor
|
|
|
|
|
|
def test_heartbeat_indicates_online():
|
|
assert heartbeat_indicates_online({"is_online": "True"}) is True
|
|
assert heartbeat_indicates_online({"is_online": "False"}) is False
|
|
assert heartbeat_indicates_online({}) is False
|
|
assert heartbeat_indicates_online(None) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cold_start_no_alert():
|
|
monitor._prev_online.clear()
|
|
|
|
with patch.object(monitor, "_load_monitored_servers", new=AsyncMock(return_value=[(1, "srv-a")])):
|
|
mock_redis = MagicMock()
|
|
mock_pipe = MagicMock()
|
|
mock_pipe.execute = AsyncMock(return_value=[{"is_online": "False"}])
|
|
mock_redis.pipeline.return_value = mock_pipe
|
|
|
|
with patch("server.background.server_offline_monitor.get_redis", return_value=mock_redis):
|
|
with patch("server.api.websocket.broadcast_offline_alert", new=AsyncMock()) as mock_broadcast:
|
|
sent = await monitor._check_offline_transitions()
|
|
|
|
assert sent == 0
|
|
mock_broadcast.assert_not_awaited()
|
|
assert monitor._prev_online[1] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_online_to_offline_fires_once():
|
|
monitor._prev_online.clear()
|
|
monitor._prev_online[1] = True
|
|
|
|
with patch.object(monitor, "_load_monitored_servers", new=AsyncMock(return_value=[(1, "srv-a")])):
|
|
mock_redis = MagicMock()
|
|
mock_pipe = MagicMock()
|
|
mock_pipe.execute = AsyncMock(return_value=[None])
|
|
mock_redis.pipeline.return_value = mock_pipe
|
|
|
|
with patch("server.background.server_offline_monitor.get_redis", return_value=mock_redis):
|
|
with patch("server.api.websocket.broadcast_offline_alert", new=AsyncMock()) as mock_broadcast:
|
|
sent = await monitor._check_offline_transitions()
|
|
|
|
assert sent == 1
|
|
mock_broadcast.assert_awaited_once_with(1, "srv-a", "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stays_offline_no_repeat():
|
|
monitor._prev_online.clear()
|
|
monitor._prev_online[1] = False
|
|
|
|
with patch.object(monitor, "_load_monitored_servers", new=AsyncMock(return_value=[(1, "srv-a")])):
|
|
mock_redis = MagicMock()
|
|
mock_pipe = MagicMock()
|
|
mock_pipe.execute = AsyncMock(return_value=[None])
|
|
mock_redis.pipeline.return_value = mock_pipe
|
|
|
|
with patch("server.background.server_offline_monitor.get_redis", return_value=mock_redis):
|
|
with patch("server.api.websocket.broadcast_offline_alert", new=AsyncMock()) as mock_broadcast:
|
|
sent = await monitor._check_offline_transitions()
|
|
|
|
assert sent == 0
|
|
mock_broadcast.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_offline_notify_toggle_blocks_telegram():
|
|
from server.infrastructure.telegram import send_telegram_offline_alert
|
|
|
|
with patch("server.infrastructure.telegram.offline_alert_notify_enabled", return_value=False):
|
|
with patch("server.infrastructure.telegram.send_telegram", new=AsyncMock()) as mock_send:
|
|
ok = await send_telegram_offline_alert("test-srv", 42)
|
|
|
|
assert ok is False
|
|
mock_send.assert_not_awaited()
|