andreas
c7326da7d9
Extracts host info assembly (owner, managers, hbc version/type, last packet timestamp, threshold configs) into a testable module-level helper, with 10 covering tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""Tests for _build_host_info helper in http.py."""
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
from hbd.server.http import _build_host_info
|
|
|
|
|
|
class _FakeConn:
|
|
def __init__(self, lastbeat):
|
|
self.lastbeat = lastbeat
|
|
|
|
|
|
class _FakeHost:
|
|
def __init__(self, name="myhost", owner=None, managers=None,
|
|
connections=None, os_data=None):
|
|
self.name = name
|
|
self.owner = owner
|
|
self.managers = managers or []
|
|
self.connections = connections or {}
|
|
self._os_data = os_data
|
|
|
|
def get_latest_plugin_data(self, plugin_name):
|
|
if plugin_name == "os_info" and self._os_data is not None:
|
|
return (1234567890.0, self._os_data)
|
|
return None
|
|
|
|
|
|
def test_build_host_info_basic_fields():
|
|
host = _FakeHost(owner="alice", managers=["bob", "carol"])
|
|
result = _build_host_info(host)
|
|
assert result["owner"] == "alice"
|
|
assert result["managers"] == ["bob", "carol"]
|
|
assert result["hbc_version"] is None
|
|
assert result["hbc_type"] is None
|
|
assert result["last_packet"] is None
|
|
assert result["thresholds"] is None
|
|
|
|
|
|
def test_build_host_info_no_owner():
|
|
host = _FakeHost()
|
|
result = _build_host_info(host)
|
|
assert result["owner"] is None
|
|
assert result["managers"] == []
|
|
|
|
|
|
def test_build_host_info_reads_hbc_from_os_info():
|
|
host = _FakeHost(os_data={"hbc_version": "5.3.0", "hbc_type": "full"})
|
|
result = _build_host_info(host)
|
|
assert result["hbc_version"] == "5.3.0"
|
|
assert result["hbc_type"] == "full"
|
|
|
|
|
|
def test_build_host_info_hbc_none_when_no_os_info():
|
|
host = _FakeHost(os_data=None)
|
|
result = _build_host_info(host)
|
|
assert result["hbc_version"] is None
|
|
assert result["hbc_type"] is None
|
|
|
|
|
|
def test_build_host_info_last_packet_is_max_lastbeat():
|
|
host = _FakeHost(connections={
|
|
"IPv4": _FakeConn(1000.0),
|
|
"IPv6": _FakeConn(2000.0),
|
|
})
|
|
result = _build_host_info(host)
|
|
assert result["last_packet"] == 2000.0
|
|
|
|
|
|
def test_build_host_info_last_packet_none_when_no_connections():
|
|
host = _FakeHost(connections={})
|
|
result = _build_host_info(host)
|
|
assert result["last_packet"] is None
|
|
|
|
|
|
def test_build_host_info_thresholds_none_without_checker():
|
|
host = _FakeHost()
|
|
result = _build_host_info(host, threshold_checker=None)
|
|
assert result["thresholds"] is None
|
|
|
|
|
|
def test_build_host_info_thresholds_sorted_by_metric():
|
|
from hbd.server.threshold import ThresholdConfig
|
|
tc_cpu = ThresholdConfig("cpu_monitor.cpu_percent", warning=80.0, critical=95.0)
|
|
tc_mem = ThresholdConfig("memory_monitor.memory_percent", warning=85.0, critical=98.0)
|
|
|
|
checker = MagicMock()
|
|
checker.get_thresholds_for_host.return_value = {
|
|
"memory_monitor.memory_percent": tc_mem,
|
|
"cpu_monitor.cpu_percent": tc_cpu,
|
|
}
|
|
|
|
host = _FakeHost()
|
|
result = _build_host_info(host, threshold_checker=checker)
|
|
|
|
assert result["thresholds"] is not None
|
|
assert len(result["thresholds"]) == 2
|
|
assert result["thresholds"][0]["metric"] == "cpu_monitor.cpu_percent"
|
|
assert result["thresholds"][0]["warning"] == 80.0
|
|
assert result["thresholds"][0]["critical"] == 95.0
|
|
assert result["thresholds"][0]["operator"] == ">"
|
|
assert result["thresholds"][1]["metric"] == "memory_monitor.memory_percent"
|
|
|
|
|
|
def test_build_host_info_thresholds_empty_list_when_no_thresholds():
|
|
checker = MagicMock()
|
|
checker.get_thresholds_for_host.return_value = {}
|
|
host = _FakeHost()
|
|
result = _build_host_info(host, threshold_checker=checker)
|
|
assert result["thresholds"] == []
|
|
|
|
|
|
def test_build_host_info_threshold_null_warning_critical():
|
|
from hbd.server.threshold import ThresholdConfig
|
|
tc = ThresholdConfig("rtt.myhost", warning=None, critical=500.0)
|
|
checker = MagicMock()
|
|
checker.get_thresholds_for_host.return_value = {"rtt.myhost": tc}
|
|
host = _FakeHost()
|
|
result = _build_host_info(host, threshold_checker=checker)
|
|
assert result["thresholds"][0]["warning"] is None
|
|
assert result["thresholds"][0]["critical"] == 500.0
|
|
|
|
|
|
def test_build_host_info_nagios_operator_serialized():
|
|
from hbd.server.threshold import ThresholdConfig
|
|
tc = ThresholdConfig("nagios_runner.check_http", operator="nagios")
|
|
checker = MagicMock()
|
|
checker.get_thresholds_for_host.return_value = {"nagios_runner.check_http": tc}
|
|
host = _FakeHost()
|
|
result = _build_host_info(host, threshold_checker=checker)
|
|
assert result["thresholds"][0]["operator"] == "nagios"
|