Nexus Agent
85b0c3a980
Count online/offline per monitored server via Redis; filter list by UI status; expose offline_list for main table scope.
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""Tests for server_connectivity — offline/online counts and list filters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from server.application.server_connectivity import (
|
|
count_monitored_connectivity,
|
|
item_matches_online_filter,
|
|
)
|
|
|
|
|
|
def test_item_matches_online_filter_excludes_unknown():
|
|
assert item_matches_online_filter({"status": "online"}, True) is True
|
|
assert item_matches_online_filter({"status": "offline"}, True) is False
|
|
assert item_matches_online_filter({"status": "offline"}, False) is True
|
|
assert item_matches_online_filter({"status": "unknown"}, False) is False
|
|
assert item_matches_online_filter({"is_online": False, "status": "unknown"}, False) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_count_monitored_connectivity_batches_redis():
|
|
session = AsyncMock()
|
|
session.execute = AsyncMock(
|
|
return_value=MagicMock(all=lambda: [(1,), (2,), (3,)]),
|
|
)
|
|
redis = AsyncMock()
|
|
redis.pipeline = MagicMock(
|
|
return_value=MagicMock(
|
|
hgetall=MagicMock(return_value=None),
|
|
execute=AsyncMock(
|
|
side_effect=[
|
|
[{"is_online": "True"}, {}],
|
|
[{"is_online": "False"}],
|
|
],
|
|
),
|
|
),
|
|
)
|
|
|
|
out = await count_monitored_connectivity(redis, session, batch_size=2)
|
|
|
|
assert out == {"monitored": 3, "online": 1, "offline": 2}
|