Codex Release Bot
94 lines
2.8 KiB
Plaintext
94 lines
2.8 KiB
Plaintext
"""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,
|
|
normalize_ssh_probe_system_info,
|
|
write_ssh_health_heartbeat,
|
|
)
|
|
|
|
|
|
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, "s1"), (2, "s2"), (3, "s3")]),
|
|
)
|
|
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}
|
|
|
|
|
|
def test_normalize_ssh_probe_system_info_nested():
|
|
probe = {
|
|
"status": "online",
|
|
"channel": "ssh",
|
|
"system_info": {
|
|
"status": "healthy",
|
|
"system_info": {"cpu_usage": 12.5, "probe": "ssh"},
|
|
},
|
|
}
|
|
out = normalize_ssh_probe_system_info(probe)
|
|
assert out["cpu_usage"] == 12.5
|
|
assert out["probe_channel"] == "ssh"
|
|
|
|
|
|
def test_normalize_ssh_probe_system_info_offline_error():
|
|
probe = {"status": "offline", "channel": "ssh", "error": "timeout"}
|
|
out = normalize_ssh_probe_system_info(probe)
|
|
assert out["probe_error"] == "timeout"
|
|
assert out["probe_channel"] == "ssh"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_ssh_health_heartbeat_redis_shape():
|
|
redis = AsyncMock()
|
|
server = MagicMock()
|
|
server.id = 42
|
|
server.agent_version = "2.1.0"
|
|
|
|
ts = await write_ssh_health_heartbeat(
|
|
redis,
|
|
server,
|
|
{
|
|
"status": "online",
|
|
"channel": "ssh",
|
|
"system_info": {"status": "healthy", "system_info": {"cpu_usage": 1.0}},
|
|
},
|
|
)
|
|
|
|
assert ts
|
|
redis.hset.assert_awaited_once()
|
|
mapping = redis.hset.await_args.kwargs["mapping"]
|
|
assert mapping["is_online"] == "True"
|
|
assert mapping["agent_version"] == "2.1.0"
|
|
assert "cpu_usage" in mapping["system_info"]
|
|
redis.expire.assert_awaited_once()
|